""" HomeKit Bridge Main Module Implementiert eine HAP-Python Bridge, die Geräte über die REST-API lädt und über HomeKit verfügbar macht. Für detaillierte Implementierungsanweisungen, Tests und Deployment-Informationen siehe README.md in diesem Verzeichnis. """ import os import logging import signal import sys import threading from typing import Optional from pyhap.accessory_driver import AccessoryDriver from pyhap.accessory import Bridge from .accessories.light import ( OnOffLightAccessory, DimmableLightAccessory, ColorLightAccessory, ) from .accessories.thermostat import ThermostatAccessory from .accessories.contact import ContactAccessory from .accessories.sensor import TempHumidityAccessory from .accessories.outlet import OutletAccessory from .api_client import ApiClient from .device_registry import DeviceRegistry # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Environment configuration HOMEKIT_NAME = os.getenv("HOMEKIT_NAME", "Home Automation Bridge") HOMEKIT_PIN = os.getenv("HOMEKIT_PIN", "031-45-154") HOMEKIT_PORT = int(os.getenv("HOMEKIT_PORT", "51826")) API_BASE = os.getenv("API_BASE", "http://api:8001") HOMEKIT_API_TOKEN = os.getenv("HOMEKIT_API_TOKEN") PERSIST_FILE = os.getenv("HOMEKIT_PERSIST_FILE", "homekit.state") def build_bridge(driver: AccessoryDriver, api_client: ApiClient) -> Bridge: """ Build the HomeKit Bridge with all device accessories. Args: driver: HAP-Python AccessoryDriver api_client: API client for communication with backend Returns: Bridge accessory with all device accessories attached """ logger.info("Loading devices from API...") registry = DeviceRegistry.load_from_api(api_client) devices = registry.get_all() logger.info(f"Loaded {len(devices)} devices from API") # Create bridge bridge = Bridge(driver, HOMEKIT_NAME) accessory_map = {} # device_id -> Accessory instance for device in devices: try: accessory = create_accessory_for_device(device, api_client, driver) if accessory: # Set room information in the accessory (HomeKit will use this for suggestions) if device.room: # Store room info for potential future use accessory._room_name = device.room bridge.add_accessory(accessory) accessory_map[device.device_id] = accessory logger.info(f"Added accessory: {device.friendly_name} ({device.type}) in room: {device.room or 'Unknown'}") else: logger.warning(f"No accessory mapping for device: {device.name} ({device.type})") except Exception as e: logger.error(f"Failed to create accessory for {device.name}: {e}", exc_info=True) # Store accessory_map on bridge for realtime updates bridge._accessory_map = accessory_map logger.info(f"Bridge built with {len(accessory_map)} accessories") return bridge def get_accessory_name(device) -> str: """ Build accessory name including room information. Args: device: Device object from DeviceRegistry Returns: Name string like "Device Name (Room)" or just "Device Name" if no room """ base_name = device.friendly_name or device.name if device.room: return f"{base_name} ({device.room})" return base_name def create_accessory_for_device(device, api_client: ApiClient, driver: AccessoryDriver): """ Create appropriate HomeKit accessory based on device type and features. Maps device types to HomeKit accessories according to homekit_mapping.md. """ device_type = device.type features = device.features display_name = get_accessory_name(device) # Light accessories if device_type == "light": if features.get("color_hsb"): return ColorLightAccessory(driver, device, api_client, display_name=display_name) elif features.get("brightness"): return DimmableLightAccessory(driver, device, api_client, display_name=display_name) else: return OnOffLightAccessory(driver, device, api_client, display_name=display_name) # Thermostat elif device_type == "thermostat": return ThermostatAccessory(driver, device, api_client, display_name=display_name) # Contact sensor elif device_type == "contact": return ContactAccessory(driver, device, api_client, display_name=display_name) # Temperature/Humidity sensor elif device_type == "temp_humidity_sensor": return TempHumidityAccessory(driver, device, api_client, display_name=display_name) # Relay/Outlet elif device_type == "relay": return OutletAccessory(driver, device, api_client, display_name=display_name) # Cover/Blinds (optional) elif device_type == "cover": # TODO: Implement CoverAccessory based on homekit_mapping.md logger.warning(f"Cover accessory not yet implemented for {device.name}") return None # TODO: Add more device types as needed (lock, motion, etc.) return None def realtime_event_loop(api_client: ApiClient, bridge: Bridge, stop_event: threading.Event): """ Background thread that listens to realtime events and updates accessories. Args: api_client: API client bridge: HomeKit bridge with accessories stop_event: Threading event to signal shutdown """ logger.info("Starting realtime event loop...") while not stop_event.is_set(): try: for event in api_client.stream_realtime(): if stop_event.is_set(): break # Handle state update events if event.get("type") == "state": device_id = event.get("device_id") payload = event.get("payload", {}) # Find corresponding accessory accessory = bridge._accessory_map.get(device_id) if accessory and hasattr(accessory, 'update_state'): try: accessory.update_state(payload) logger.debug(f"Updated state for {device_id}: {payload}") except Exception as e: logger.error(f"Error updating accessory {device_id}: {e}") except Exception as e: if not stop_event.is_set(): logger.error(f"Realtime stream error: {e}. Reconnecting in 5s...") stop_event.wait(5) # Backoff before reconnect logger.info("Realtime event loop stopped") def main(): logger.info("=" * 60) logger.info(f"Starting HomeKit Bridge: {HOMEKIT_NAME}") logger.info(f"API Base: {API_BASE}") logger.info(f"HomeKit Port: {HOMEKIT_PORT}") logger.info(f"PIN: {HOMEKIT_PIN}") logger.info("=" * 60) # Create API client api_client = ApiClient( base_url=API_BASE, token=HOMEKIT_API_TOKEN, timeout=10 ) # Test API connectivity try: devices = api_client.get_devices() logger.info(f"API connectivity OK - {len(devices)} devices available") except Exception as e: logger.error(f"Failed to connect to API at {API_BASE}: {e}") logger.error("Please check API_BASE and network connectivity") sys.exit(1) # Create AccessoryDriver driver = AccessoryDriver( port=HOMEKIT_PORT, persist_file=PERSIST_FILE ) # Build bridge with all accessories try: bridge = build_bridge(driver, api_client) except Exception as e: logger.error(f"Failed to build bridge: {e}", exc_info=True) sys.exit(1) # Add bridge to driver driver.add_accessory(accessory=bridge) # Setup realtime event thread stop_event = threading.Event() realtime_thread = threading.Thread( target=realtime_event_loop, args=(api_client, bridge, stop_event), daemon=True ) # Signal handlers for graceful shutdown def signal_handler(sig, frame): logger.info("Received shutdown signal, stopping...") stop_event.set() driver.stop() signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # Start realtime thread realtime_thread.start() # Start the bridge logger.info(f"HomeKit Bridge started on port {HOMEKIT_PORT}") logger.info(f"Pair with PIN: {HOMEKIT_PIN}") logger.info("Press Ctrl+C to stop") try: driver.start() except KeyboardInterrupt: logger.info("KeyboardInterrupt received") finally: logger.info("Stopping bridge...") stop_event.set() realtime_thread.join(timeout=5) logger.info("Bridge stopped") if __name__ == "__main__": main()