ClimatController/app/auto/config.py

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,
    )