132 lines
5 KiB
Python
132 lines
5 KiB
Python
"""
|
|
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)
|