diff --git a/examples/smart_home/src/smart_home/lights/hue_utils.py b/examples/smart_home/src/smart_home/lights/hue_utils.py new file mode 100644 index 000000000..478897203 --- /dev/null +++ b/examples/smart_home/src/smart_home/lights/hue_utils.py @@ -0,0 +1,34 @@ +from typing import Any + +from phue2 import Bridge +from phue2.exceptions import PhueException + +from smart_home.settings import settings + + +def _get_bridge() -> Bridge | None: + """Attempts to connect to the Hue bridge using settings.""" + try: + return Bridge( + ip=str(settings.hue_bridge_ip), + username=settings.hue_bridge_username, + save_config=False, + ) + except Exception: + # Broad exception to catch potential connection issues + # TODO: Add more specific logging or error handling + return None + + +def handle_phue_error( + light_or_group: str, operation: str, error: Exception +) -> dict[str, Any]: + """Creates a standardized error response for phue2 operations.""" + base_info = {"target": light_or_group, "operation": operation, "success": False} + if isinstance(error, KeyError): + base_info["error"] = f"Target '{light_or_group}' not found" + elif isinstance(error, PhueException): + base_info["error"] = f"phue2 error during {operation}: {error}" + else: + base_info["error"] = f"Unexpected error during {operation}: {error}" + return base_info diff --git a/examples/smart_home/src/smart_home/lights/server.py b/examples/smart_home/src/smart_home/lights/server.py index 34a4a937d..ce0eabcc8 100644 --- a/examples/smart_home/src/smart_home/lights/server.py +++ b/examples/smart_home/src/smart_home/lights/server.py @@ -1,23 +1,25 @@ -from typing import Any +from typing import Annotated, Any, Literal, TypedDict -from phue2 import Bridge -from phue2.exceptions import ( - PhueException, -) +from phue2.exceptions import PhueException +from pydantic import Field +from typing_extensions import NotRequired from fastmcp import FastMCP -from smart_home.settings import settings +from smart_home.lights.hue_utils import _get_bridge, handle_phue_error -def _get_bridge() -> Bridge | None: - try: - return Bridge( - ip=str(settings.hue_bridge_ip), - username=settings.hue_bridge_username, - save_config=False, - ) - except Exception: - return None +class HueAttributes(TypedDict, total=False): + """TypedDict for optional light attributes.""" + + on: NotRequired[bool] + bri: NotRequired[Annotated[int, Field(ge=0, le=254)]] + hue: NotRequired[Annotated[int, Field(ge=0, le=65535)]] + sat: NotRequired[Annotated[int, Field(ge=0, le=254)]] + xy: NotRequired[list[float]] + ct: NotRequired[Annotated[int, Field(ge=153, le=500)]] + alert: NotRequired[Literal["none", "select", "lselect"]] + effect: NotRequired[Literal["none", "colorloop"]] + transitiontime: NotRequired[int] # deciseconds lights_mcp = FastMCP( @@ -27,8 +29,6 @@ lights_mcp = FastMCP( ], ) -# --- Resources --- - @lights_mcp.tool() def read_all_lights() -> list[str]: @@ -38,10 +38,9 @@ def read_all_lights() -> list[str]: try: light_dict = bridge.get_light_objects("list") return [light.name for light in light_dict] - except PhueException as e: + except (PhueException, Exception) as e: + # Simplified error handling for list return type return [f"Error listing lights: {e}"] - except Exception as e: - return [f"Unexpected error listing lights: {e}"] # --- Tools --- @@ -60,24 +59,8 @@ def toggle_light(light_name: str, state: bool) -> dict[str, Any]: "success": True, "phue2_result": result, } - except KeyError: - return { - "light": light_name, - "error": f"Light '{light_name}' not found", - "success": False, - } - except PhueException as e: - return { - "light": light_name, - "error": f"phue2 error toggling light: {e}", - "success": False, - } - except Exception as e: - return { - "light": light_name, - "error": f"Unexpected error toggling light: {e}", - "success": False, - } + except (KeyError, PhueException, Exception) as e: + return handle_phue_error(light_name, "toggle_light", e) @lights_mcp.tool() @@ -86,6 +69,7 @@ def set_brightness(light_name: str, brightness: int) -> dict[str, Any]: if not (bridge := _get_bridge()): return {"error": "Bridge not connected", "success": False} if not 0 <= brightness <= 254: + # Keep specific input validation error here return { "light": light_name, "error": "Brightness must be between 0 and 254", @@ -99,21 +83,113 @@ def set_brightness(light_name: str, brightness: int) -> dict[str, Any]: "success": True, "phue2_result": result, } - except KeyError: + except (KeyError, PhueException, Exception) as e: + return handle_phue_error(light_name, "set_brightness", e) + + +@lights_mcp.tool() +def list_groups() -> list[str]: + """Lists the names of all available Hue light groups.""" + if not (bridge := _get_bridge()): + return ["Error: Bridge not connected"] + try: + # phue2 get_group() returns a dict {id: {details}} including name + groups = bridge.get_group() + return [group_details["name"] for group_details in groups.values()] + except (PhueException, Exception) as e: + return [f"Error listing groups: {e}"] + + +@lights_mcp.tool() +def list_scenes() -> list[str]: + """Lists the names of all available Hue scenes.""" + if not (bridge := _get_bridge()): + return ["Error: Bridge not connected"] + try: + # phue2 get_scene() returns a dict {id: {details}} including name + scenes = bridge.get_scene() + return [scene_details["name"] for scene_details in scenes.values()] + except (PhueException, Exception) as e: + return [f"Error listing scenes: {e}"] + + +@lights_mcp.tool() +def activate_scene(group_name: str, scene_name: str) -> dict[str, Any]: + """Activates a specific scene within a specified light group.""" + if not (bridge := _get_bridge()): + return {"error": "Bridge not connected", "success": False} + try: + # Note: phue2 run_scene uses group_name and scene_name directly + result = bridge.run_scene(group_name=group_name, scene_name=scene_name) + # run_scene returns True on success, we'll make the response richer + if result: + return { + "group": group_name, + "activated_scene": scene_name, + "success": True, + "phue2_result": result, # Include the raw True/False + } + else: + # This case might indicate the scene/group exists but activation failed + return { + "group": group_name, + "scene": scene_name, + "error": "Scene activation failed (phue2 returned False)", + "success": False, + } + + except (KeyError, PhueException, Exception) as e: + # KeyError likely means group or scene name is wrong + return handle_phue_error(f"{group_name}/{scene_name}", "activate_scene", e) + + +@lights_mcp.tool() +def set_light_attributes(light_name: str, attributes: HueAttributes) -> dict[str, Any]: + """Sets multiple attributes (e.g., hue, sat, bri, ct, xy, transitiontime) for a specific light.""" + if not (bridge := _get_bridge()): + return {"error": "Bridge not connected", "success": False} + + # Basic validation (more specific validation could be added) + if not isinstance(attributes, dict) or not attributes: + return { + "error": "Attributes must be a non-empty dictionary", + "success": False, + "light": light_name, + } + + try: + result = bridge.set_light(light_name, attributes) return { "light": light_name, - "error": f"Light '{light_name}' not found", - "success": False, + "set_attributes": attributes, + "success": True, + "phue2_result": result, } - except PhueException as e: + except (KeyError, PhueException, ValueError, Exception) as e: + # ValueError might occur for invalid attribute values + return handle_phue_error(light_name, "set_light_attributes", e) + + +@lights_mcp.tool() +def set_group_attributes(group_name: str, attributes: HueAttributes) -> dict[str, Any]: + """Sets multiple attributes for all lights within a specific group.""" + if not (bridge := _get_bridge()): + return {"error": "Bridge not connected", "success": False} + + if not isinstance(attributes, dict) or not attributes: return { - "light": light_name, - "error": f"phue2 error setting brightness: {e}", + "error": "Attributes must be a non-empty dictionary", "success": False, + "group": group_name, } - except Exception as e: + + try: + result = bridge.set_group(group_name, attributes) return { - "light": light_name, - "error": f"Unexpected error setting brightness: {e}", - "success": False, + "group": group_name, + "set_attributes": attributes, + "success": True, + "phue2_result": result, } + except (KeyError, PhueException, ValueError, Exception) as e: + return handle_phue_error(group_name, "set_group_attributes", e)