171 lines
4.6 KiB
Python
171 lines
4.6 KiB
Python
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"]
|