resources/libraries/python/MLRsearch/PerDurationDatabase.py

   1 # Copyright (c) 2021 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Module defining PerDurationDatabase class."""
  15
  16 import copy
  17
  18
  19 class PerDurationDatabase:
  20     """List-like structure holding measurement results for one duration.
  21
  22     This is a building block for MeasurementDatabase.
  23
  24     This class hold measurements for single target duration value only,
  25     so the logic is quite simple.
  26
  27     Several utility methods are added, accomplishing tasks useful for MLRsearch
  28     (to be called by MeasurementDatabase).
  29     """
  30
  31     def __init__(self, duration, measurements):
  32         """Store (deep copy of) measurement results and normalize them.
  33
  34         The results have to have the corresponding target duration,
  35         and there should be no duplicate target_tr values.
  36         Empty iterable (zero measurements) is an acceptable input.
  37
  38         :param duration: All measurements have to have this target duration [s].
  39         :param measurements: The measurement results to store.
  40         :type duration: float
  41         :type measurements: Iterable[ReceiveRateMeasurement]
  42         :raises ValueError: If duration does not match or if TR duplicity.
  43         """
  44         self.duration = duration
  45         self.measurements = [copy.deepcopy(meas) for meas in measurements]
  46         self._normalize()
  47
  48     def __repr__(self):
  49         """Return string executable to get equivalent instance.
  50
  51         :returns: Code to construct equivalent instance.
  52         :rtype: str
  53         """
  54         return (
  55             u"PerDurationDatabase("
  56             f"duration={self.duration!r},"
  57             f"measurements={self.measurements!r})"
  58         )
  59
  60     def _normalize(self):
  61         """Sort by target_tr, fail on detecting duplicate target_tr.
  62
  63         Also set effective loss ratios.
  64
  65         :raises ValueError: If duration does not match or if TR duplicity.
  66         """
  67         measurements = self.measurements
  68         measurements.sort(key=lambda measurement: measurement.target_tr)
  69         # Detect duplicated TRs.
  70         previous_tr = None
  71         for measurement in measurements:
  72             current_tr = measurement.target_tr
  73             if current_tr == previous_tr:
  74                 raise ValueError(
  75                     u"Transmit rate conflict:"
  76                     f" {measurement!r} {previous_tr!r}"
  77                 )
  78             previous_tr = current_tr
  79         # Update effective ratios.
  80         ratio_previous = None
  81         for measurement in measurements:
  82             if ratio_previous is None:
  83                 ratio_previous = measurement.loss_ratio
  84                 measurement.effective_loss_ratio = ratio_previous
  85                 continue
  86             ratio_previous = max(ratio_previous, measurement.loss_ratio)
  87             measurement.effective_loss_ratio = ratio_previous
  88
  89     def add(self, measurement):
  90         """Add measurement and normalize.
  91
  92         :param measurement: Measurement result to add to the database.
  93         :type measurement: ReceiveRateMeasurement
  94         """
  95         # TODO: We should deepcopy either everywhere or nowhere.
  96         self.measurements.append(measurement)
  97         self._normalize()
  98
  99     def get_valid_bounds(self, ratio):
 100         """Return None or a valid measurement for two tightest bounds.
 101
 102         The validity of a measurement to act as a bound is determined
 103         by comparing the argument ratio with measurement's effective loss ratio.
 104
 105         Both lower and upper bounds are returned, both tightest and second
 106         tightest. If some value is not available, None is returned instead.
 107
 108         :param ratio: Target ratio, valid has to be lower or equal.
 109         :type ratio: float
 110         :returns: Tightest lower bound, tightest upper bound,
 111             second tightest lower bound, second tightest upper bound.
 112         :rtype: 4-tuple of Optional[ReceiveRateMeasurement]
 113         """
 114         lower_1, upper_1, lower_2, upper_2 = None, None, None, None
 115         for measurement in self.measurements:
 116             if measurement.effective_loss_ratio > ratio:
 117                 if upper_1 is None:
 118                     upper_1 = measurement
 119                     continue
 120                 upper_2 = measurement
 121                 break
 122             lower_1, lower_2 = measurement, lower_1
 123         return lower_1, upper_1, lower_2, upper_2