ignis-core/app/api/schemas.py

from datetime import datetime
from typing import Any, Literal

from fastapi import HTTPException
from pydantic import BaseModel, ConfigDict, Field, model_validator

from app.drivers.wiz import WizDriver


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 WizDriver.SCENES:
                raise HTTPException(status_code=400, detail="Неизвестная сцена")
            params["sceneId"] = WizDriver.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