b934600380
- Sync and async HTTP clients for Ignis Core WiZ server - 23 endpoints: auth, devices, groups, control, schedules, stats, API keys - Pydantic models with client-side validation - 108 unit tests - README with role table and usage examples
279 lines
7.0 KiB
Python
279 lines
7.0 KiB
Python
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]
|