415 lines
8.3 KiB
Python
415 lines
8.3 KiB
Python
from datetime import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from app.my_dataclasses import (
|
|
MIN_FAN_SPEED,
|
|
MAX_FAN_SPEED,
|
|
MIN_TARGET_TEMP,
|
|
MAX_TARGET_TEMP,
|
|
SUPPORTED_AIR_MODES,
|
|
)
|
|
|
|
from .models import (
|
|
ScheduleAction,
|
|
ScheduleActionType,
|
|
ScheduleConfig,
|
|
SchedulePoint,
|
|
ScheduledSettings,
|
|
)
|
|
|
|
|
|
WEEKDAYS = (
|
|
"mon",
|
|
"tue",
|
|
"wed",
|
|
"thu",
|
|
"fri",
|
|
"sat",
|
|
"sun",
|
|
)
|
|
|
|
|
|
SET_FIELDS = {
|
|
"power",
|
|
"speed",
|
|
"heater",
|
|
"target_temp",
|
|
"mode",
|
|
"sound",
|
|
"light",
|
|
}
|
|
|
|
|
|
def _parse_time(value: Any) -> time:
|
|
if not isinstance(value, str):
|
|
raise ValueError(
|
|
f"Schedule time must be a string: {value!r}"
|
|
)
|
|
|
|
try:
|
|
parsed = time.fromisoformat(value)
|
|
|
|
except ValueError as exc:
|
|
raise ValueError(
|
|
f"Invalid schedule time: {value!r}"
|
|
) from exc
|
|
|
|
# Не разрешаем секунды.
|
|
if parsed.second != 0 or parsed.microsecond != 0:
|
|
raise ValueError(
|
|
f"Schedule time must use HH:MM: {value!r}"
|
|
)
|
|
|
|
return parsed
|
|
|
|
|
|
def _parse_on_off(
|
|
name: str,
|
|
value: Any,
|
|
) -> bool:
|
|
# PyYAML может автоматически превратить
|
|
# on/off в True/False.
|
|
if type(value) is bool:
|
|
return value
|
|
|
|
if isinstance(value, str):
|
|
normalized = value.strip().lower()
|
|
|
|
if normalized == "on":
|
|
return True
|
|
|
|
if normalized == "off":
|
|
return False
|
|
|
|
raise ValueError(
|
|
f"{name} must be 'on' or 'off'"
|
|
)
|
|
|
|
|
|
def _parse_settings(
|
|
data: dict[str, Any],
|
|
) -> ScheduledSettings:
|
|
|
|
unknown = set(data) - SET_FIELDS
|
|
|
|
if unknown:
|
|
raise ValueError(
|
|
f"Unknown SET fields: {sorted(unknown)}"
|
|
)
|
|
|
|
values: dict[str, Any] = {}
|
|
|
|
if "power" in data:
|
|
values["power"] = _parse_on_off(
|
|
"power",
|
|
data["power"],
|
|
)
|
|
|
|
if "speed" in data:
|
|
speed = data["speed"]
|
|
|
|
if type(speed) is not int:
|
|
raise ValueError(
|
|
"speed must be an integer"
|
|
)
|
|
|
|
if not MIN_FAN_SPEED <= speed <= MAX_FAN_SPEED:
|
|
raise ValueError(
|
|
f"speed must be between "
|
|
f"{MIN_FAN_SPEED} and {MAX_FAN_SPEED}"
|
|
)
|
|
|
|
values["speed"] = speed
|
|
|
|
if "heater" in data:
|
|
values["heater"] = _parse_on_off(
|
|
"heater",
|
|
data["heater"],
|
|
)
|
|
|
|
if "target_temp" in data:
|
|
temperature = data["target_temp"]
|
|
|
|
if type(temperature) is not int:
|
|
raise ValueError(
|
|
"target_temp must be an integer"
|
|
)
|
|
|
|
if not (
|
|
MIN_TARGET_TEMP
|
|
<= temperature
|
|
<= MAX_TARGET_TEMP
|
|
):
|
|
raise ValueError(
|
|
f"target_temp must be between "
|
|
f"{MIN_TARGET_TEMP} and "
|
|
f"{MAX_TARGET_TEMP}"
|
|
)
|
|
|
|
values["target_temp"] = temperature
|
|
|
|
if "mode" in data:
|
|
mode = data["mode"]
|
|
|
|
if mode not in SUPPORTED_AIR_MODES:
|
|
raise ValueError(
|
|
f"Unsupported air mode: {mode!r}"
|
|
)
|
|
|
|
values["mode"] = mode
|
|
|
|
if "sound" in data:
|
|
values["sound"] = _parse_on_off(
|
|
"sound",
|
|
data["sound"],
|
|
)
|
|
|
|
if "light" in data:
|
|
values["light"] = _parse_on_off(
|
|
"light",
|
|
data["light"],
|
|
)
|
|
|
|
settings = ScheduledSettings(**values)
|
|
|
|
if settings.empty:
|
|
raise ValueError(
|
|
"SET action must contain "
|
|
"at least one setting"
|
|
)
|
|
|
|
return settings
|
|
|
|
|
|
def _parse_action(
|
|
data: Any,
|
|
) -> ScheduleAction:
|
|
|
|
if not isinstance(data, dict):
|
|
raise ValueError(
|
|
"Schedule action must be an object"
|
|
)
|
|
|
|
action_type_raw = data.get("type")
|
|
|
|
try:
|
|
action_type = ScheduleActionType(
|
|
action_type_raw
|
|
)
|
|
|
|
except ValueError as exc:
|
|
raise ValueError(
|
|
f"Unsupported action type: "
|
|
f"{action_type_raw!r}"
|
|
) from exc
|
|
|
|
if action_type == ScheduleActionType.AUTO:
|
|
|
|
extra = set(data) - {
|
|
"type",
|
|
"speed",
|
|
}
|
|
|
|
if extra:
|
|
raise ValueError(
|
|
"AUTO action supports only "
|
|
f"'speed': {sorted(extra)}"
|
|
)
|
|
|
|
if "speed" not in data:
|
|
raise ValueError(
|
|
"AUTO action must contain "
|
|
"fallback speed"
|
|
)
|
|
|
|
settings = _parse_settings(
|
|
{
|
|
"speed": data["speed"],
|
|
}
|
|
)
|
|
|
|
return ScheduleAction(
|
|
type=ScheduleActionType.AUTO,
|
|
settings=settings,
|
|
)
|
|
|
|
settings_data = {
|
|
key: value
|
|
for key, value in data.items()
|
|
if key != "type"
|
|
}
|
|
|
|
return ScheduleAction(
|
|
type=ScheduleActionType.SET,
|
|
settings=_parse_settings(
|
|
settings_data
|
|
),
|
|
)
|
|
|
|
def _parse_template(
|
|
name: str,
|
|
data: Any,
|
|
) -> tuple[SchedulePoint, ...]:
|
|
|
|
if not isinstance(data, list):
|
|
raise ValueError(
|
|
f"Template {name!r} must be a list"
|
|
)
|
|
|
|
if not data:
|
|
raise ValueError(
|
|
f"Template {name!r} cannot be empty"
|
|
)
|
|
|
|
points: list[SchedulePoint] = []
|
|
|
|
for item in data:
|
|
|
|
if not isinstance(item, dict):
|
|
raise ValueError(
|
|
f"Invalid point in template {name!r}"
|
|
)
|
|
|
|
if "time" not in item:
|
|
raise ValueError(
|
|
f"Point in {name!r} has no time"
|
|
)
|
|
|
|
if "action" not in item:
|
|
raise ValueError(
|
|
f"Point in {name!r} has no action"
|
|
)
|
|
|
|
points.append(
|
|
SchedulePoint(
|
|
at=_parse_time(item["time"]),
|
|
action=_parse_action(
|
|
item["action"]
|
|
),
|
|
)
|
|
)
|
|
|
|
points.sort(
|
|
key=lambda point: point.at
|
|
)
|
|
|
|
seen_times = set()
|
|
|
|
for point in points:
|
|
if point.at in seen_times:
|
|
raise ValueError(
|
|
f"Duplicate time {point.at} "
|
|
f"in template {name!r}"
|
|
)
|
|
|
|
seen_times.add(point.at)
|
|
|
|
return tuple(points)
|
|
|
|
|
|
def load_schedule(
|
|
path: str | Path,
|
|
) -> ScheduleConfig:
|
|
|
|
path = Path(path)
|
|
|
|
with path.open(
|
|
"r",
|
|
encoding="utf-8",
|
|
) as file:
|
|
raw = yaml.safe_load(file)
|
|
|
|
if not isinstance(raw, dict):
|
|
raise ValueError(
|
|
"Schedule root must be an object"
|
|
)
|
|
|
|
version = raw.get("version", 1)
|
|
|
|
if version != 1:
|
|
raise ValueError(
|
|
f"Unsupported schedule version: {version}"
|
|
)
|
|
|
|
enabled = raw.get("enabled", True)
|
|
|
|
if type(enabled) is not bool:
|
|
raise ValueError(
|
|
"enabled must be true or false"
|
|
)
|
|
|
|
timezone = raw.get(
|
|
"timezone",
|
|
"local",
|
|
)
|
|
|
|
if timezone != "local":
|
|
raise ValueError(
|
|
"Only timezone: local "
|
|
"is currently supported"
|
|
)
|
|
|
|
raw_templates = raw.get("templates")
|
|
|
|
if not isinstance(raw_templates, dict):
|
|
raise ValueError(
|
|
"templates must be an object"
|
|
)
|
|
|
|
templates = {
|
|
name: _parse_template(
|
|
name,
|
|
template,
|
|
)
|
|
for name, template
|
|
in raw_templates.items()
|
|
}
|
|
|
|
raw_days = raw.get("days")
|
|
|
|
if not isinstance(raw_days, dict):
|
|
raise ValueError(
|
|
"days must be an object"
|
|
)
|
|
|
|
days: dict[str, str] = {}
|
|
|
|
for weekday in WEEKDAYS:
|
|
|
|
if weekday not in raw_days:
|
|
raise ValueError(
|
|
f"Missing schedule day: {weekday}"
|
|
)
|
|
|
|
template_name = raw_days[weekday]
|
|
|
|
if template_name not in templates:
|
|
raise ValueError(
|
|
f"Unknown template "
|
|
f"{template_name!r} "
|
|
f"for {weekday}"
|
|
)
|
|
|
|
days[weekday] = template_name
|
|
|
|
unknown_days = (
|
|
set(raw_days)
|
|
- set(WEEKDAYS)
|
|
)
|
|
|
|
if unknown_days:
|
|
raise ValueError(
|
|
f"Unknown weekdays: "
|
|
f"{sorted(unknown_days)}"
|
|
)
|
|
|
|
return ScheduleConfig(
|
|
version=version,
|
|
enabled=enabled,
|
|
timezone=timezone,
|
|
templates=templates,
|
|
days=days,
|
|
) |