""" Hue Controller module for MIDI-to-Hue application. Manages connections to the Hue Bridge and light controls. """ from phue import Bridge import time from threading import Timer from typing import Dict, Any, List, Optional class ThrottledUpdater: """ Updates Hue lights with rate limiting to prevent overwhelming the bridge. Collects and consolidates multiple rapid updates into fewer bridge calls. """ def __init__(self, bridge, update_interval: float = 0.05): """ Initialize the throttled updater. Args: bridge: The Hue Bridge object update_interval: Time between updates in seconds """ self.bridge = bridge self.update_interval = update_interval self.last_update_time = 0 self.pending_updates = {} # {light_name: {param: value, ...}, ...} self.update_timer = None print(f"ThrottledUpdater initialized with interval: {update_interval * 1000:.0f}ms") def schedule_update(self, light_name: str, param: str, value: Any) -> None: """ Schedule an update for a specific light and parameter. Args: light_name: Name of the light to update param: Parameter to update (e.g., 'bri', 'hue', 'on') value: New value for the parameter """ # Store the most recent value for this light and parameter if light_name not in self.pending_updates: self.pending_updates[light_name] = {} self.pending_updates[light_name][param] = value current_time = time.time() time_since_last_update = current_time - self.last_update_time # If no timer is active and it's been longer than our interval, update immediately if self.update_timer is None and time_since_last_update >= self.update_interval: self._perform_updates() # Otherwise, ensure we have a timer scheduled elif self.update_timer is None: # Calculate remaining time until next update remaining_time = max(0, self.update_interval - time_since_last_update) self.update_timer = Timer(remaining_time, self._perform_updates) self.update_timer.start() def _perform_updates(self) -> None: """Apply all pending updates to the Hue bridge.""" if self.update_timer: self.update_timer.cancel() self.update_timer = None # Apply all pending updates for light_name, params in self.pending_updates.items(): for param, value in params.items(): try: self.bridge.set_light(light_name, param, value) print(f"Updated {light_name} {param} = {value}") except Exception as e: print(f"Error updating light {light_name}: {e}") # Clear pending updates and update timestamp self.pending_updates = {} self.last_update_time = time.time() class HueController: """Manages interactions with the Philips Hue system.""" def __init__(self, bridge_ip: str, update_interval: float = 0.05): """ Initialize the Hue controller. Args: bridge_ip: IP address of the Hue Bridge update_interval: Time between updates in seconds """ self.bridge_ip = bridge_ip self.bridge = None self.updater = None self.update_interval = update_interval self.connect() def connect(self) -> None: """Connect to the Hue Bridge and initialize the updater.""" try: self.bridge = Bridge(self.bridge_ip) self.bridge.connect() self.updater = ThrottledUpdater(self.bridge, self.update_interval) print(f"Connected to Hue Bridge at {self.bridge_ip}") except Exception as e: print(f"Failed to connect to Hue Bridge: {e}") raise def list_lights(self) -> None: """Print available lights for reference.""" print("Available Hue lights:") for light in self.bridge.lights: print(f" {light.name} - Current brightness: {light.brightness}") def get_light_by_name(self, light_name: str): """Get a light object by its name.""" try: return self.bridge.get_light_objects('name')[light_name] except KeyError: print(f"Light '{light_name}' not found.") return None def update_light(self, light_name: str, parameter: str, value: Any) -> None: """ Schedule an update for a light parameter. Args: light_name: Name of the light to update parameter: Parameter to update (e.g., 'bri', 'hue', 'on') value: New value for the parameter """ if value is None: print("Skipping update due to None value") return self.updater.schedule_update(light_name, parameter, value)