doc: поправил зависимости и расписание
This commit is contained in:
Fedorov Dmitriy
@@ -0,0 +1,390 @@
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from app.my_dataclasses import (
|
||||
MIN_FAN_SPEED,
|
||||
MAX_FAN_SPEED,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(
|
||||
frozen=True,
|
||||
slots=True,
|
||||
)
|
||||
class Co2Config:
|
||||
base_speed: int
|
||||
hysteresis: int
|
||||
|
||||
thresholds: tuple[
|
||||
tuple[int, int],
|
||||
...
|
||||
]
|
||||
|
||||
|
||||
@dataclass(
|
||||
frozen=True,
|
||||
slots=True,
|
||||
)
|
||||
class AutoConfig:
|
||||
version: int
|
||||
check_interval: float
|
||||
co2: Co2Config
|
||||
temperature: TemperatureConfig
|
||||
|
||||
@dataclass(
|
||||
frozen=True,
|
||||
slots=True,
|
||||
)
|
||||
class TemperatureConfig:
|
||||
hysteresis: float
|
||||
|
||||
|
||||
def _require_dict(name: str, value: Any) -> dict:
|
||||
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(
|
||||
f"{name} must be an object"
|
||||
)
|
||||
|
||||
return value
|
||||
|
||||
|
||||
def _parse_speed(name: str, value: Any) -> int:
|
||||
|
||||
if type(value) is not int:
|
||||
raise ValueError(
|
||||
f"{name} must be an integer"
|
||||
)
|
||||
|
||||
if not (
|
||||
MIN_FAN_SPEED
|
||||
<= value
|
||||
<= MAX_FAN_SPEED
|
||||
):
|
||||
raise ValueError(
|
||||
f"{name} must be between "
|
||||
f"{MIN_FAN_SPEED} and "
|
||||
f"{MAX_FAN_SPEED}"
|
||||
)
|
||||
|
||||
return value
|
||||
|
||||
|
||||
def _parse_thresholds(value: Any) -> tuple[
|
||||
tuple[int, int],
|
||||
...
|
||||
]:
|
||||
|
||||
if not isinstance(value, list):
|
||||
raise ValueError(
|
||||
"co2.thresholds must be a list"
|
||||
)
|
||||
|
||||
if not value:
|
||||
raise ValueError(
|
||||
"co2.thresholds cannot be empty"
|
||||
)
|
||||
|
||||
result = []
|
||||
|
||||
previous_ppm = None
|
||||
previous_speed = None
|
||||
|
||||
for index, item in enumerate(value):
|
||||
|
||||
item = _require_dict(
|
||||
f"co2.thresholds[{index}]",
|
||||
item,
|
||||
)
|
||||
|
||||
unknown = set(item) - {
|
||||
"ppm",
|
||||
"speed",
|
||||
}
|
||||
|
||||
if unknown:
|
||||
raise ValueError(
|
||||
f"Unknown fields in "
|
||||
f"co2.thresholds[{index}]: "
|
||||
f"{sorted(unknown)}"
|
||||
)
|
||||
|
||||
if "ppm" not in item:
|
||||
raise ValueError(
|
||||
f"co2.thresholds[{index}].ppm "
|
||||
f"is required"
|
||||
)
|
||||
|
||||
if "speed" not in item:
|
||||
raise ValueError(
|
||||
f"co2.thresholds[{index}].speed "
|
||||
f"is required"
|
||||
)
|
||||
|
||||
ppm = item["ppm"]
|
||||
|
||||
if type(ppm) is not int:
|
||||
raise ValueError(
|
||||
f"co2.thresholds[{index}].ppm "
|
||||
f"must be an integer"
|
||||
)
|
||||
|
||||
if ppm <= 0:
|
||||
raise ValueError(
|
||||
f"co2.thresholds[{index}].ppm "
|
||||
f"must be > 0"
|
||||
)
|
||||
|
||||
speed = _parse_speed(
|
||||
(
|
||||
f"co2.thresholds"
|
||||
f"[{index}].speed"
|
||||
),
|
||||
item["speed"],
|
||||
)
|
||||
|
||||
if (
|
||||
previous_ppm is not None
|
||||
and ppm <= previous_ppm
|
||||
):
|
||||
raise ValueError(
|
||||
"CO2 thresholds must be "
|
||||
"strictly increasing"
|
||||
)
|
||||
|
||||
if (
|
||||
previous_speed is not None
|
||||
and speed <= previous_speed
|
||||
):
|
||||
raise ValueError(
|
||||
"AUTO speeds must be "
|
||||
"strictly increasing"
|
||||
)
|
||||
|
||||
result.append(
|
||||
(
|
||||
ppm,
|
||||
speed,
|
||||
)
|
||||
)
|
||||
|
||||
previous_ppm = ppm
|
||||
previous_speed = speed
|
||||
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def _parse_co2(value: Any) -> Co2Config:
|
||||
|
||||
data = _require_dict(
|
||||
"co2",
|
||||
value,
|
||||
)
|
||||
|
||||
unknown = set(data) - {
|
||||
"base_speed",
|
||||
"hysteresis",
|
||||
"thresholds",
|
||||
}
|
||||
|
||||
if unknown:
|
||||
raise ValueError(
|
||||
f"Unknown CO2 config fields: "
|
||||
f"{sorted(unknown)}"
|
||||
)
|
||||
|
||||
if "base_speed" not in data:
|
||||
raise ValueError(
|
||||
"co2.base_speed is required"
|
||||
)
|
||||
|
||||
if "hysteresis" not in data:
|
||||
raise ValueError(
|
||||
"co2.hysteresis is required"
|
||||
)
|
||||
|
||||
if "thresholds" not in data:
|
||||
raise ValueError(
|
||||
"co2.thresholds is required"
|
||||
)
|
||||
|
||||
base_speed = _parse_speed(
|
||||
"co2.base_speed",
|
||||
data["base_speed"],
|
||||
)
|
||||
|
||||
hysteresis = data["hysteresis"]
|
||||
|
||||
if type(hysteresis) is not int:
|
||||
raise ValueError(
|
||||
"co2.hysteresis must be "
|
||||
"an integer"
|
||||
)
|
||||
|
||||
if hysteresis < 0:
|
||||
raise ValueError(
|
||||
"co2.hysteresis must be >= 0"
|
||||
)
|
||||
|
||||
thresholds = _parse_thresholds(
|
||||
data["thresholds"]
|
||||
)
|
||||
|
||||
first_speed = thresholds[0][1]
|
||||
|
||||
if first_speed <= base_speed:
|
||||
raise ValueError(
|
||||
"First threshold speed must be "
|
||||
"greater than base_speed"
|
||||
)
|
||||
|
||||
return Co2Config(
|
||||
base_speed=base_speed,
|
||||
hysteresis=hysteresis,
|
||||
thresholds=thresholds,
|
||||
)
|
||||
|
||||
|
||||
def load_auto_config(path: str | Path) -> AutoConfig:
|
||||
|
||||
path = Path(path)
|
||||
|
||||
with path.open(
|
||||
"r",
|
||||
encoding="utf-8",
|
||||
) as file:
|
||||
raw = yaml.safe_load(file)
|
||||
|
||||
data = _require_dict(
|
||||
"AUTO config",
|
||||
raw,
|
||||
)
|
||||
|
||||
unknown = set(data) - {
|
||||
"version",
|
||||
"check_interval",
|
||||
"co2",
|
||||
"temperature",
|
||||
}
|
||||
|
||||
if unknown:
|
||||
raise ValueError(
|
||||
f"Unknown AUTO config fields: "
|
||||
f"{sorted(unknown)}"
|
||||
)
|
||||
|
||||
version = data.get("version")
|
||||
|
||||
if version != 1:
|
||||
raise ValueError(
|
||||
f"Unsupported AUTO config "
|
||||
f"version: {version!r}"
|
||||
)
|
||||
|
||||
check_interval = data.get(
|
||||
"check_interval"
|
||||
)
|
||||
|
||||
if (
|
||||
type(check_interval) not in {
|
||||
int,
|
||||
float,
|
||||
}
|
||||
):
|
||||
raise ValueError(
|
||||
"check_interval must be a number"
|
||||
)
|
||||
|
||||
if check_interval <= 0:
|
||||
raise ValueError(
|
||||
"check_interval must be > 0"
|
||||
)
|
||||
|
||||
# --------------------------------------------------
|
||||
# CO2
|
||||
# --------------------------------------------------
|
||||
|
||||
if "co2" not in data:
|
||||
raise ValueError(
|
||||
"co2 config is required"
|
||||
)
|
||||
|
||||
co2 = _parse_co2(
|
||||
data["co2"]
|
||||
)
|
||||
|
||||
# --------------------------------------------------
|
||||
# Temperature
|
||||
# --------------------------------------------------
|
||||
|
||||
if "temperature" not in data:
|
||||
raise ValueError(
|
||||
"temperature config is required"
|
||||
)
|
||||
|
||||
temperature_data = _require_dict(
|
||||
"temperature",
|
||||
data["temperature"],
|
||||
)
|
||||
|
||||
unknown_temperature = (
|
||||
set(temperature_data)
|
||||
- {
|
||||
"hysteresis",
|
||||
}
|
||||
)
|
||||
|
||||
if unknown_temperature:
|
||||
raise ValueError(
|
||||
f"Unknown temperature config fields: "
|
||||
f"{sorted(unknown_temperature)}"
|
||||
)
|
||||
|
||||
temperature_hysteresis = (
|
||||
temperature_data.get(
|
||||
"hysteresis"
|
||||
)
|
||||
)
|
||||
|
||||
if (
|
||||
type(temperature_hysteresis)
|
||||
not in {
|
||||
int,
|
||||
float,
|
||||
}
|
||||
):
|
||||
raise ValueError(
|
||||
"temperature.hysteresis "
|
||||
"must be a number"
|
||||
)
|
||||
|
||||
temperature_hysteresis = float(
|
||||
temperature_hysteresis
|
||||
)
|
||||
|
||||
if temperature_hysteresis < 0:
|
||||
raise ValueError(
|
||||
"temperature.hysteresis "
|
||||
"must be >= 0"
|
||||
)
|
||||
|
||||
temperature = TemperatureConfig(
|
||||
hysteresis=temperature_hysteresis,
|
||||
)
|
||||
|
||||
# --------------------------------------------------
|
||||
# Result
|
||||
# --------------------------------------------------
|
||||
|
||||
return AutoConfig(
|
||||
version=version,
|
||||
check_interval=float(
|
||||
check_interval
|
||||
),
|
||||
co2=co2,
|
||||
temperature=temperature,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user