41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
import logging
|
|
from typing import Dict, List, Optional
|
|
from app.models.device import DeviceSchema, GroupModel
|
|
from app.core.discovery import DiscoveryService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StateManager:
|
|
def __init__(self):
|
|
# Храним устройства как Pydantic объекты
|
|
self.devices: Dict[str, DeviceSchema] = {}
|
|
# Группы как модели SQLAlchemy
|
|
self.groups: Dict[str, GroupModel] = {}
|
|
|
|
def update_device(self, device_data: dict):
|
|
"""Обновляет или добавляет устройство в состояние."""
|
|
|
|
mac = device_data["mac"]
|
|
|
|
# Используем DeviceSchema вместо Device
|
|
device = DeviceSchema(
|
|
id=mac, ip=device_data["ip"], name=f"WiZ {mac[-4:]}", room="Default"
|
|
)
|
|
self.devices[mac] = device
|
|
|
|
def get_group_ips(self, group_id: str) -> List[str]:
|
|
"""Возвращает список IP всех ламп, входящих в группу."""
|
|
group = self.groups.get(group_id)
|
|
if not group:
|
|
return []
|
|
|
|
# Извлекаем IP по MAC-адресам, которые хранятся в группе
|
|
return [
|
|
self.devices[d_id].ip for d_id in group.device_ids if d_id in self.devices
|
|
]
|
|
|
|
|
|
state_manager = StateManager()
|
|
discovery_service = DiscoveryService()
|