from __future__ import annotations
import json
import os
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List
STATE_DIR = os.path.join(os.path.dirname(__file__), "devices")
def _path(alias: str) -> str:
return os.path.join(STATE_DIR, f"{alias}.json")
@dataclass
class ChannelState:
state: str = "off" # "on" | "off" for relay; "idle" | "clicked" for button
last_event: str = ""
last_event_time: str = ""
@dataclass
class DeviceState:
alias: str
device_type: str
device_id: str
device_name: str
port: int
server_url: str
mac_address: str
firmware_version: str = "virtual-1.0.0"
status: str = "setup" # "setup" | "normal"
token: str | None = None
channels: List[Dict[str, Any]] = field(default_factory=list)
channels_schema: List[int] = field(default_factory=list)
uptime_start: float = field(default_factory=lambda: __import__("time").time())
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "DeviceState":
return cls(**data)
def ensure_dir() -> None:
os.makedirs(STATE_DIR, exist_ok=True)
def load(alias: str) -> DeviceState | None:
ensure_dir()
path = _path(alias)
if not os.path.exists(path):
return None
with open(path, "r", encoding="utf-8") as f:
return DeviceState.from_dict(json.load(f))
def save(state: DeviceState) -> None:
ensure_dir()
path = _path(state.alias)
with open(path, "w", encoding="utf-8") as f:
json.dump(state.to_dict(), f, indent=2, ensure_ascii=False)
def remove(alias: str) -> bool:
path = _path(alias)
if os.path.exists(path):
os.remove(path)
return True
return False
def list_all() -> List[str]:
ensure_dir()
names = []
for fname in os.listdir(STATE_DIR):
if fname.endswith(".json"):
names.append(fname[:-5])
return sorted(names)