""" Device Registry for HomeKit Bridge Loads devices from API and joins with layout information. """ import logging from typing import Dict, List, Optional from dataclasses import dataclass logger = logging.getLogger(__name__) @dataclass class Device: """Represents a device with combined info from /devices and /layout.""" device_id: str type: str # "light", "thermostat", "relay", "contact", "temp_humidity", "cover" name: str # Short name from /devices homekit_aid: int # HomeKit Accessory ID features: Dict[str, bool] # Feature flags (e.g., {"power": true, "brightness": true}) read_only: bool # True for sensors that don't accept commands class DeviceRegistry: """Registry of all devices loaded from the API.""" def __init__(self, devices: List[Device]): """ Initialize registry with devices. Args: devices: List of Device objects """ self._devices = devices self._by_id = {d.device_id: d for d in devices} @classmethod def load_from_api(cls, api_client) -> 'DeviceRegistry': """ Load devices from API and join with layout information. Args: api_client: ApiClient instance Returns: DeviceRegistry with all devices """ # Get devices and layout devices_data = api_client.get_devices() # Create Device objects devices = [] for dev_data in devices_data: device_id = dev_data.get('device_id') if not device_id: logger.warning(f"Device without device_id: {dev_data}") continue # Check for required homekit_aid field homekit_aid = dev_data.get('homekit_aid') if homekit_aid is None: logger.error(f"Device {device_id} is missing required homekit_aid field - skipping") continue # Determine if read-only (sensors don't accept set commands) device_type = dev_data.get('type', '') read_only = device_type in ['contact', 'temp_humidity', 'motion', 'smoke'] device = Device( device_id=device_id, type=device_type, name=device_id, homekit_aid=homekit_aid, features=dev_data.get('features', {}), read_only=read_only ) devices.append(device) logger.info(f"Loaded {len(devices)} devices from API") return cls(devices) def get_all(self) -> List[Device]: """Get all devices.""" return self._devices.copy() def get_by_id(self, device_id: str) -> Optional[Device]: """ Get device by ID. Args: device_id: Device identifier Returns: Device or None if not found """ return self._by_id.get(device_id) def get_by_type(self, device_type: str) -> List[Device]: """ Get all devices of a specific type. Args: device_type: Device type (e.g., "light", "thermostat") Returns: List of matching devices """ return [d for d in self._devices if d.type == device_type] def get_by_room(self, room: str) -> List[Device]: """ Get all devices in a specific room. Args: room: Room name Returns: List of devices in the room """ return [d for d in self._devices if d.room == room]