from datetime import datetime from typing import Any, List, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator class IgnisError(Exception): def __init__(self, status_code: int, detail: str): self.status_code = status_code self.detail = detail super().__init__(f"[{status_code}] {detail}") SCENES: dict[str, int] = { "ocean": 1, "romance": 2, "party": 3, "fireplace": 5, "cozy": 6, "forest": 10, "pastel_colors": 11, "wake_up": 12, "bedtime": 13, "warm_white": 14, "daylight": 15, "cool_white": 16, "night_light": 17, "focus": 18, "relax": 19, "true_colors": 20, "tv_time": 21, "plant_growth": 22, "spring": 23, "summer": 24, "fall": 25, "deep_dive": 26, "jungle": 27, "mojito": 28, "club": 29, "christmas": 30, "halloween": 31, "candlelight": 32, "golden_white": 33, "pulse": 34, "steampunk": 35, } class CommandRequest(BaseModel): model_config = ConfigDict(extra="forbid") state: bool | None = None brightness: int | None = Field(default=None, ge=10, le=100) scene: str | None = None temp: int | None = Field(default=None, ge=2200, le=6500) r: int | None = Field(default=None, ge=0, le=255) g: int | None = Field(default=None, ge=0, le=255) b: int | None = Field(default=None, ge=0, le=255) @property def has_rgb(self) -> bool: return all(channel is not None for channel in (self.r, self.g, self.b)) @model_validator(mode="after") def validate_payload(self): if all( value is None for value in ( self.state, self.brightness, self.scene, self.temp, self.r, self.g, self.b, ) ): raise ValueError("Никаких команд не передано") rgb_values = (self.r, self.g, self.b) if any(value is not None for value in rgb_values) and not self.has_rgb: raise ValueError("Поля r, g и b нужно передавать вместе") exclusive_modes = ( self.scene is not None, self.temp is not None, self.has_rgb, ) if sum(1 for enabled in exclusive_modes if enabled) > 1: raise ValueError("Можно передать только один режим из scene, temp или rgb") return self def to_wiz_params(self) -> dict[str, Any]: params: dict[str, Any] = {} if self.state is not None: params["state"] = self.state if self.brightness is not None: params["dimming"] = self.brightness if self.scene is not None: if self.scene not in SCENES: raise ValueError("Неизвестная сцена") params["sceneId"] = SCENES[self.scene] elif self.temp is not None: params["temp"] = self.temp elif self.has_rgb: params["r"] = self.r params["g"] = self.g params["b"] = self.b return params class ScheduleOnceRequest(CommandRequest): target_id: str = Field(min_length=1) run_at: datetime | None = None hours_from_now: int | None = Field(default=None, ge=0) is_group: bool = True @model_validator(mode="after") def validate_schedule_target(self): has_run_at = self.run_at is not None has_hours_from_now = self.hours_from_now is not None if has_run_at == has_hours_from_now: raise ValueError("Передайте ровно одно из полей run_at или hours_from_now") return self class ScheduleCronRequest(CommandRequest): target_id: str = Field(min_length=1) hour: str = Field(min_length=1) minute: str = Field(min_length=1) day_of_week: str = "*" is_group: bool = True class DeviceControlResponse(BaseModel): device_id: str applied: dict[str, Any] result: dict[str, Any] | None = None status: Literal["ok"] class GroupCommandResult(BaseModel): ip: str ok: bool kind: str result: dict[str, Any] | None = None error: str | None = None class GroupControlResponse(BaseModel): status: Literal["ok", "partial"] applied: dict[str, Any] sent_to: list[str] success_count: int failure_count: int results: list[GroupCommandResult] class BlinkResponse(BaseModel): status: Literal["blink_done"] original: bool class DeviceStatusResponse(BaseModel): device_id: str status: dict[str, Any] class GroupStatusItem(BaseModel): ip: str status: dict[str, Any] | None = None error: str | None = None kind: str | None = None class GroupStatusResponse(BaseModel): group_id: str results: list[GroupStatusItem] class ScheduleCreateResponse(BaseModel): status: str job_id: str run_at: str | None = None class ScheduleTaskItem(BaseModel): id: str target_id: str is_group: bool state: bool | None = None action_params: dict[str, Any] trigger_type: str next_run: str | None = None hour: str | None = None minute: str | None = None day_of_week: str | None = None job_present: bool class ScheduleTasksResponse(BaseModel): tasks: list[ScheduleTaskItem] class DeleteStatusResponse(BaseModel): status: Literal["deleted"] class RescanResponse(BaseModel): status: Literal["ok"] found: int added: int updated: int removed_offline: int pending_removal: int online: int class ServerBuildInfoResponse(BaseModel): version: str | None = None git_sha: str | None = None build_date: str | None = None class ServerUrlInfoResponse(BaseModel): observed_base_url: str | None = None configured_public_base_url: str | None = None effective_public_base_url: str | None = None class ServerConfigurationStatusResponse(BaseModel): configured: bool master_key_configured: bool scan_network_configured: bool public_base_url_configured: bool build_metadata_complete: bool class ServerDiscoveryInfoResponse(BaseModel): last_scan_at: str | None = None last_scan_mode: str | None = None online: int | None = None found: int | None = None added: int | None = None updated: int | None = None removed_offline: int | None = None pending_removal: int | None = None class ServerInfoResponse(BaseModel): app_name: str instance_name: str | None = None timezone: str | None = None uptime_seconds: int diagnostics_visible: bool started_at: str | None = None build: ServerBuildInfoResponse | None = None urls: ServerUrlInfoResponse | None = None configuration: ServerConfigurationStatusResponse | None = None discovery: ServerDiscoveryInfoResponse | None = None class KeyActionRequest(BaseModel): key: str class DeviceSchema(BaseModel): id: str ip: str name: str room: str class GroupCreateSchema(BaseModel): id: str name: str macs: List[str]