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: allowed_fields = { "type", "speed", "target_temp", } extra = set(data) - allowed_fields if extra: raise ValueError( "AUTO action supports only " "'speed' and 'target_temp': " f"{sorted(extra)}" ) if "speed" not in data: raise ValueError( "AUTO action requires 'speed'" ) settings_data = { "speed": data["speed"], } if "target_temp" in data: settings_data["target_temp"] = ( data["target_temp"] ) return ScheduleAction( type=ScheduleActionType.AUTO, settings=_parse_settings(settings_data), ) 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, )