from .accessories.light import ( OnOffLightAccessory, DimmableLightAccessory, ColorLightAccessory, ) # 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: bridge.add_accessory(accessory) accessory_map[device.device_id] = accessory logger.info(f"Added accessory: {device.friendly_name} ({device.type})") 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 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 # Light accessories if device_type == "light": if features.get("color_hsb"): return ColorLightAccessory(driver, device, api_client) elif features.get("brightness"): return DimmableLightAccessory(driver, device, api_client) else: return OnOffLightAccessory(driver, device, api_client) # Thermostat elif device_type == "thermostat": return ThermostatAccessory(driver, device, api_client) # Contact sensor elif device_type == "contact": return ContactAccessory(driver, device, api_client) # Temperature/Humidity sensor elif device_type == "temp_humidity": return TempHumidityAccessory(driver, device, api_client) # Outlet/Switch elif device_type == "outlet": return OutletAccessory(driver, device, api_client) # Cover/Blinds (optional) elif device_type == "cover": # TODO: Implement CoverAccessory based on homekit_mapping.md return CoverAccessory(driver, device, api_client) # 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()