seems to work
This commit is contained in:
138
apps/homekit/device_registry.py
Normal file
138
apps/homekit/device_registry.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""
|
||||
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
|
||||
friendly_name: str # Display title from /layout (fallback to name)
|
||||
room: Optional[str] # Room name from layout
|
||||
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()
|
||||
layout_data = api_client.get_layout()
|
||||
|
||||
# Build lookup: device_id -> (room_name, title)
|
||||
layout_map = {}
|
||||
if isinstance(layout_data, dict) and 'rooms' in layout_data:
|
||||
rooms_list = layout_data['rooms']
|
||||
if isinstance(rooms_list, list):
|
||||
for room in rooms_list:
|
||||
if isinstance(room, dict):
|
||||
room_name = room.get('name', 'Unknown')
|
||||
devices_in_room = room.get('devices', [])
|
||||
for device_info in devices_in_room:
|
||||
if isinstance(device_info, dict):
|
||||
device_id = device_info.get('device_id')
|
||||
title = device_info.get('title', '')
|
||||
if device_id:
|
||||
layout_map[device_id] = (room_name, title)
|
||||
|
||||
# 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
|
||||
|
||||
# Get layout info
|
||||
room_name, title = layout_map.get(device_id, (None, ''))
|
||||
|
||||
# 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=dev_data.get('name', device_id),
|
||||
friendly_name=title or dev_data.get('name', device_id),
|
||||
room=room_name,
|
||||
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]
|
||||
Reference in New Issue
Block a user