class Co2SpeedPolicy: def __init__( self, *, base_speed: int, thresholds: tuple[ tuple[int, int], ... ], hysteresis: int, ): if hysteresis < 0: raise ValueError( "hysteresis must be >= 0" ) if not thresholds: raise ValueError( "thresholds cannot be empty" ) previous_ppm = None speeds = [base_speed] for ppm, speed in thresholds: if previous_ppm is not None: if ppm <= previous_ppm: raise ValueError( "CO2 thresholds must " "be strictly increasing" ) if speed in speeds: raise ValueError( "AUTO speeds must be unique" ) speeds.append(speed) previous_ppm = ppm self._base_speed = base_speed self._thresholds = thresholds self._hysteresis = hysteresis self._speeds = tuple(speeds) def select_speed(self, co2: int, current_speed: int | None) -> int: if co2 < 0: raise ValueError( "CO2 cannot be negative" ) # -------------------------------------------------- # Первое решение AUTO. # # Гистерезис пока применять не к чему: # предыдущей AUTO-скорости ещё нет. # -------------------------------------------------- if ( current_speed is None or current_speed not in self._speeds ): return self._select_initial_speed(co2) index = self._speeds.index(current_speed) # -------------------------------------------------- # CO2 растёт. # # Проверяем пороги перехода вверх. # За один вызов можем перепрыгнуть # сразу несколько скоростей. # -------------------------------------------------- while index < len( self._thresholds ): ppm, _ = self._thresholds[index] if co2 < ppm: break index += 1 # -------------------------------------------------- # CO2 падает. # # Для перехода вниз используем: # # threshold - hysteresis # # Поэтому скорость не будет прыгать # туда-сюда около одного порога. # -------------------------------------------------- while index > 0: ppm, _ = self._thresholds[index - 1] down_threshold = ppm - self._hysteresis if co2 > down_threshold: break index -= 1 return self._speeds[index] def _select_initial_speed(self, co2: int) -> int: speed = self._base_speed for ppm, candidate_speed in ( self._thresholds ): if co2 < ppm: break speed = candidate_speed return speed