Philips Hue MCP Server

""" Philips Hue Bridge client wrapper. This module provides an async wrapper around the aiohue library for interacting with Philips Hue Bridge API. """ from typing import Dict, List, Optional, Any import logging try: from aiohue.v2 import HueBridgeV2 except ImportError: HueBridgeV2 = None logger = logging.getLogger(__name__) # Brightness value constants BRIGHTNESS_MIN_HUE_API = 0 BRIGHTNESS_MAX_HUE_API = 254 BRIGHTNESS_MIN_PERCENT = 0.0 BRIGHTNESS_MAX_PERCENT = 100.0 # Color temperature constants (in mireds) COLOR_TEMP_MIN_MIREDS = 153 COLOR_TEMP_MAX_MIREDS = 500 # CIE xy color space constants CIE_XY_MIN = 0.0 CIE_XY_MAX = 1.0 # Resource type identifier DEVICE_RESOURCE_TYPE = "device" def _convert_brightness_to_percentage(brightness: int) -> float: """ Convert Hue API brightness value (0-254) to percentage (0-100). Args: brightness: Brightness value in Hue API format (0-254) Returns: Brightness as percentage (0.0-100.0) """ return (brightness / BRIGHTNESS_MAX_HUE_API) * BRIGHTNESS_MAX_PERCENT def _is_device_resource(resource_type: str) -> bool: """ Check if a resource type identifier represents a device. Args: resource_type: The resource type string to check Returns: True if the resource is a device type """ return resource_type == DEVICE_RESOURCE_TYPE class HueClient: """Async client for interacting with Philips Hue Bridge.""" def __init__(self, bridge_ip: str, api_key: str): """ Initialize Hue client. Args: bridge_ip: IP address of the Hue Bridge api_key: API key/username for authentication """ if HueBridgeV2 is None: raise ImportError( "aiohue library not installed. " "Install it with: pip install aiohue" ) self.bridge_ip = bridge_ip self.api_key = api_key self.bridge: Optional[HueBridgeV2] = None self._connected = False async def connect(self) -> None: """Connect to the Hue Bridge.""" try: self.bridge = HueBridgeV2(self.bridge_ip, self.api_key) await self.bridge.initialize() self._connected = True logger.info(f"Connected to Hue Bridge at {self.bridge_ip}") except Exception as e: logger.error(f"Failed to connect to Hue Bridge: {e}") raise async def disconnect(self) -> None: """Disconnect from the Hue Bridge.""" if self.bridge: await self.bridge.close() self._connected = False logger.info("Disconnected from Hue Bridge") def is_connected(self) -> bool: """Check if connected to the bridge.""" return self._connected async def get_lights(self) -> Dict[str, Any]: """ Get all lights from the bridge. Returns: Dictionary of light resources """ if not self.bridge: raise RuntimeError("Not connected to bridge") lights = {} for light in self.bridge.lights: # Get device name device = self.bridge.devices.get(light.owner.rid) name = device.metadata.name if device and device.metadata else f"Light {light.id[:8]}" lights[light.id] = { "id": light.id, "name": name, "on": light.on.on if light.on else False, "brightness": light.dimming.brightness if light.dimming else None, "color_temp": light.color_temperature.mirek if (light.color_temperature and light.color_temperature.mirek_valid) else None, "color": { "xy": [light.color.xy.x, light.color.xy.y] if light.color and light.color.xy else None, }, "reachable": True, # v2 API doesn't have reachable status } return lights async def get_light(self, light_id: str) -> Optional[Dict[str, Any]]: """ Get a specific light by ID. Args: light_id: The light's unique identifier Returns: Light data dictionary or None if not found """ lights = await self.get_lights() return lights.get(light_id) async def set_light_state( self, light_id: str, on: Optional[bool] = None, brightness: Optional[int] = None, color_temp: Optional[int] = None, xy: Optional[List[float]] = None, transition_time: Optional[float] = None, ) -> bool: """ Set the state of a light. Args: light_id: The light's unique identifier on: Turn light on (True) or off (False) brightness: Brightness level (0-254) color_temp: Color temperature in mireds (153-500) xy: Color in CIE xy format [x, y] transition_time: Transition time in seconds Returns: True if successful """ if not self.bridge: raise RuntimeError("Not connected to bridge") # Verify light exists light = self.bridge.lights.get(light_id) if not light: raise ValueError(f"Light {light_id} not found") # Use the specific methods provided by aiohue LightsController # These are the ONLY correct ways to control lights in aiohue v2 API # DO NOT use update() or pass kwargs - it will fail! if on is not None: # Use dedicated turn_on/turn_off methods if on: await self.bridge.lights.turn_on(light_id) else: await self.bridge.lights.turn_off(light_id) if brightness is not None: # aiohue v2 uses 0-100 percentage for brightness, not 0-254 brightness_percentage = _convert_brightness_to_percentage(brightness) await self.bridge.lights.set_brightness(light_id, brightness_percentage) if color_temp is not None: # Set color temperature in mireds (153-500) await self.bridge.lights.set_color_temperature(light_id, color_temp) if xy is not None: # aiohue expects x and y as separate positional parameters await self.bridge.lights.set_color(light_id, xy[0], xy[1]) # Note: transition_time is not directly supported by all methods in aiohue # It would need to be implemented differently if required return True async def get_groups(self) -> Dict[str, Any]: """ Get all groups/rooms from the bridge. Returns: Dictionary of group resources """ if not self.bridge: raise RuntimeError("Not connected to bridge") groups = {} for group in self.bridge.groups: # Only include Room objects (not GroupedLight) if hasattr(group, 'metadata') and group.metadata: groups[group.id] = { "id": group.id, "name": group.metadata.name, "type": group.type.value if hasattr(group, 'type') else "room", "lights": [svc.rid for svc in group.children] if hasattr(group, 'children') else [], } return groups async def get_scenes(self) -> Dict[str, Any]: """ Get all scenes from the bridge. Returns: Dictionary of scene resources """ if not self.bridge: raise RuntimeError("Not connected to bridge") scenes = {} for scene in self.bridge.scenes: scenes[scene.id] = { "id": scene.id, "name": scene.metadata.name if scene.metadata else f"Scene {scene.id[:8]}", "group": scene.group.rid if hasattr(scene, 'group') else None, } return scenes async def activate_scene(self, scene_id: str) -> bool: """ Activate a scene. Args: scene_id: The scene's unique identifier Returns: True if successful """ if not self.bridge: raise RuntimeError("Not connected to bridge") scene = self.bridge.scenes.get(scene_id) if not scene: raise ValueError(f"Scene {scene_id} not found") await self.bridge.scenes.recall(scene_id) return True def _collect_light_ids_from_group(self, group) -> List[str]: """ Collect all light IDs from a group's devices. In Hue API v2, groups contain device references in their 'children' attribute. This method finds all lights owned by those devices. Args: group: The group resource object Returns: List of light IDs belonging to the group Raises: ValueError: If group has no children or no lights found """ if not hasattr(group, 'children'): raise ValueError(f"Group {group.id} has no children") light_ids = [] for child in group.children: # Each child is a ResourceIdentifier - check if it points to a device if _is_device_resource(child.rtype.value): # Find all lights owned by this device for light in self.bridge.lights: if hasattr(light, 'owner') and light.owner.rid == child.rid: light_ids.append(light.id) if not light_ids: raise ValueError(f"No lights found in group {group.id}") return light_ids async def set_group_state( self, group_id: str, on: Optional[bool] = None, brightness: Optional[int] = None, color_temp: Optional[int] = None, xy: Optional[List[float]] = None, ) -> bool: """ Set the state of all lights in a group. Args: group_id: The group's unique identifier on: Turn lights on (True) or off (False) brightness: Brightness level (0-254) color_temp: Color temperature in mireds (153-500) xy: Color in CIE xy format [x, y] Returns: True if successful """ if not self.bridge: raise RuntimeError("Not connected to bridge") group = self.bridge.groups.get(group_id) if not group: raise ValueError(f"Group {group_id} not found") # Collect all light IDs from the group's devices light_ids = self._collect_light_ids_from_group(group) # Apply state to each light in the group # Note: This could be optimized by using asyncio.gather for parallel execution for light_id in light_ids: await self.set_light_state( light_id=light_id, on=on, brightness=brightness, color_temp=color_temp, xy=xy, ) return True