129 lines
3.3 KiB
Python
129 lines
3.3 KiB
Python
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 |