zzstoatzz commited on
Commit
220b6cc
·
1 Parent(s): dc7ce68

more tools

Browse files
examples/smart_home/src/smart_home/lights/hue_utils.py ADDED
@@ -0,0 +1,34 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from typing import Any
2
+
3
+ from phue2 import Bridge
4
+ from phue2.exceptions import PhueException
5
+
6
+ from smart_home.settings import settings
7
+
8
+
9
+ def _get_bridge() -> Bridge | None:
10
+ """Attempts to connect to the Hue bridge using settings."""
11
+ try:
12
+ return Bridge(
13
+ ip=str(settings.hue_bridge_ip),
14
+ username=settings.hue_bridge_username,
15
+ save_config=False,
16
+ )
17
+ except Exception:
18
+ # Broad exception to catch potential connection issues
19
+ # TODO: Add more specific logging or error handling
20
+ return None
21
+
22
+
23
+ def handle_phue_error(
24
+ light_or_group: str, operation: str, error: Exception
25
+ ) -> dict[str, Any]:
26
+ """Creates a standardized error response for phue2 operations."""
27
+ base_info = {"target": light_or_group, "operation": operation, "success": False}
28
+ if isinstance(error, KeyError):
29
+ base_info["error"] = f"Target '{light_or_group}' not found"
30
+ elif isinstance(error, PhueException):
31
+ base_info["error"] = f"phue2 error during {operation}: {error}"
32
+ else:
33
+ base_info["error"] = f"Unexpected error during {operation}: {error}"
34
+ return base_info
examples/smart_home/src/smart_home/lights/server.py CHANGED
@@ -1,23 +1,25 @@
1
- from typing import Any
2
 
3
- from phue2 import Bridge
4
- from phue2.exceptions import (
5
- PhueException,
6
- )
7
 
8
  from fastmcp import FastMCP
9
- from smart_home.settings import settings
10
 
11
 
12
- def _get_bridge() -> Bridge | None:
13
- try:
14
- return Bridge(
15
- ip=str(settings.hue_bridge_ip),
16
- username=settings.hue_bridge_username,
17
- save_config=False,
18
- )
19
- except Exception:
20
- return None
 
 
 
21
 
22
 
23
  lights_mcp = FastMCP(
@@ -27,8 +29,6 @@ lights_mcp = FastMCP(
27
  ],
28
  )
29
 
30
- # --- Resources ---
31
-
32
 
33
  @lights_mcp.tool()
34
  def read_all_lights() -> list[str]:
@@ -38,10 +38,9 @@ def read_all_lights() -> list[str]:
38
  try:
39
  light_dict = bridge.get_light_objects("list")
40
  return [light.name for light in light_dict]
41
- except PhueException as e:
 
42
  return [f"Error listing lights: {e}"]
43
- except Exception as e:
44
- return [f"Unexpected error listing lights: {e}"]
45
 
46
 
47
  # --- Tools ---
@@ -60,24 +59,8 @@ def toggle_light(light_name: str, state: bool) -> dict[str, Any]:
60
  "success": True,
61
  "phue2_result": result,
62
  }
63
- except KeyError:
64
- return {
65
- "light": light_name,
66
- "error": f"Light '{light_name}' not found",
67
- "success": False,
68
- }
69
- except PhueException as e:
70
- return {
71
- "light": light_name,
72
- "error": f"phue2 error toggling light: {e}",
73
- "success": False,
74
- }
75
- except Exception as e:
76
- return {
77
- "light": light_name,
78
- "error": f"Unexpected error toggling light: {e}",
79
- "success": False,
80
- }
81
 
82
 
83
  @lights_mcp.tool()
@@ -86,6 +69,7 @@ def set_brightness(light_name: str, brightness: int) -> dict[str, Any]:
86
  if not (bridge := _get_bridge()):
87
  return {"error": "Bridge not connected", "success": False}
88
  if not 0 <= brightness <= 254:
 
89
  return {
90
  "light": light_name,
91
  "error": "Brightness must be between 0 and 254",
@@ -99,21 +83,113 @@ def set_brightness(light_name: str, brightness: int) -> dict[str, Any]:
99
  "success": True,
100
  "phue2_result": result,
101
  }
102
- except KeyError:
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
103
  return {
104
- "light": light_name,
105
- "error": f"Light '{light_name}' not found",
106
  "success": False,
 
107
  }
108
- except PhueException as e:
 
 
109
  return {
110
  "light": light_name,
111
- "error": f"phue2 error setting brightness: {e}",
112
- "success": False,
 
113
  }
114
- except Exception as e:
 
 
 
 
 
 
 
 
 
 
 
115
  return {
116
- "light": light_name,
117
- "error": f"Unexpected error setting brightness: {e}",
118
  "success": False,
 
 
 
 
 
 
 
 
 
 
119
  }
 
 
 
1
+ from typing import Annotated, Any, Literal, TypedDict
2
 
3
+ from phue2.exceptions import PhueException
4
+ from pydantic import Field
5
+ from typing_extensions import NotRequired
 
6
 
7
  from fastmcp import FastMCP
8
+ from smart_home.lights.hue_utils import _get_bridge, handle_phue_error
9
 
10
 
11
+ class HueAttributes(TypedDict, total=False):
12
+ """TypedDict for optional light attributes."""
13
+
14
+ on: NotRequired[bool]
15
+ bri: NotRequired[Annotated[int, Field(ge=0, le=254)]]
16
+ hue: NotRequired[Annotated[int, Field(ge=0, le=65535)]]
17
+ sat: NotRequired[Annotated[int, Field(ge=0, le=254)]]
18
+ xy: NotRequired[list[float]]
19
+ ct: NotRequired[Annotated[int, Field(ge=153, le=500)]]
20
+ alert: NotRequired[Literal["none", "select", "lselect"]]
21
+ effect: NotRequired[Literal["none", "colorloop"]]
22
+ transitiontime: NotRequired[int] # deciseconds
23
 
24
 
25
  lights_mcp = FastMCP(
 
29
  ],
30
  )
31
 
 
 
32
 
33
  @lights_mcp.tool()
34
  def read_all_lights() -> list[str]:
 
38
  try:
39
  light_dict = bridge.get_light_objects("list")
40
  return [light.name for light in light_dict]
41
+ except (PhueException, Exception) as e:
42
+ # Simplified error handling for list return type
43
  return [f"Error listing lights: {e}"]
 
 
44
 
45
 
46
  # --- Tools ---
 
59
  "success": True,
60
  "phue2_result": result,
61
  }
62
+ except (KeyError, PhueException, Exception) as e:
63
+ return handle_phue_error(light_name, "toggle_light", e)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
64
 
65
 
66
  @lights_mcp.tool()
 
69
  if not (bridge := _get_bridge()):
70
  return {"error": "Bridge not connected", "success": False}
71
  if not 0 <= brightness <= 254:
72
+ # Keep specific input validation error here
73
  return {
74
  "light": light_name,
75
  "error": "Brightness must be between 0 and 254",
 
83
  "success": True,
84
  "phue2_result": result,
85
  }
86
+ except (KeyError, PhueException, Exception) as e:
87
+ return handle_phue_error(light_name, "set_brightness", e)
88
+
89
+
90
+ @lights_mcp.tool()
91
+ def list_groups() -> list[str]:
92
+ """Lists the names of all available Hue light groups."""
93
+ if not (bridge := _get_bridge()):
94
+ return ["Error: Bridge not connected"]
95
+ try:
96
+ # phue2 get_group() returns a dict {id: {details}} including name
97
+ groups = bridge.get_group()
98
+ return [group_details["name"] for group_details in groups.values()]
99
+ except (PhueException, Exception) as e:
100
+ return [f"Error listing groups: {e}"]
101
+
102
+
103
+ @lights_mcp.tool()
104
+ def list_scenes() -> list[str]:
105
+ """Lists the names of all available Hue scenes."""
106
+ if not (bridge := _get_bridge()):
107
+ return ["Error: Bridge not connected"]
108
+ try:
109
+ # phue2 get_scene() returns a dict {id: {details}} including name
110
+ scenes = bridge.get_scene()
111
+ return [scene_details["name"] for scene_details in scenes.values()]
112
+ except (PhueException, Exception) as e:
113
+ return [f"Error listing scenes: {e}"]
114
+
115
+
116
+ @lights_mcp.tool()
117
+ def activate_scene(group_name: str, scene_name: str) -> dict[str, Any]:
118
+ """Activates a specific scene within a specified light group."""
119
+ if not (bridge := _get_bridge()):
120
+ return {"error": "Bridge not connected", "success": False}
121
+ try:
122
+ # Note: phue2 run_scene uses group_name and scene_name directly
123
+ result = bridge.run_scene(group_name=group_name, scene_name=scene_name)
124
+ # run_scene returns True on success, we'll make the response richer
125
+ if result:
126
+ return {
127
+ "group": group_name,
128
+ "activated_scene": scene_name,
129
+ "success": True,
130
+ "phue2_result": result, # Include the raw True/False
131
+ }
132
+ else:
133
+ # This case might indicate the scene/group exists but activation failed
134
+ return {
135
+ "group": group_name,
136
+ "scene": scene_name,
137
+ "error": "Scene activation failed (phue2 returned False)",
138
+ "success": False,
139
+ }
140
+
141
+ except (KeyError, PhueException, Exception) as e:
142
+ # KeyError likely means group or scene name is wrong
143
+ return handle_phue_error(f"{group_name}/{scene_name}", "activate_scene", e)
144
+
145
+
146
+ @lights_mcp.tool()
147
+ def set_light_attributes(light_name: str, attributes: HueAttributes) -> dict[str, Any]:
148
+ """Sets multiple attributes (e.g., hue, sat, bri, ct, xy, transitiontime) for a specific light."""
149
+ if not (bridge := _get_bridge()):
150
+ return {"error": "Bridge not connected", "success": False}
151
+
152
+ # Basic validation (more specific validation could be added)
153
+ if not isinstance(attributes, dict) or not attributes:
154
  return {
155
+ "error": "Attributes must be a non-empty dictionary",
 
156
  "success": False,
157
+ "light": light_name,
158
  }
159
+
160
+ try:
161
+ result = bridge.set_light(light_name, attributes)
162
  return {
163
  "light": light_name,
164
+ "set_attributes": attributes,
165
+ "success": True,
166
+ "phue2_result": result,
167
  }
168
+ except (KeyError, PhueException, ValueError, Exception) as e:
169
+ # ValueError might occur for invalid attribute values
170
+ return handle_phue_error(light_name, "set_light_attributes", e)
171
+
172
+
173
+ @lights_mcp.tool()
174
+ def set_group_attributes(group_name: str, attributes: HueAttributes) -> dict[str, Any]:
175
+ """Sets multiple attributes for all lights within a specific group."""
176
+ if not (bridge := _get_bridge()):
177
+ return {"error": "Bridge not connected", "success": False}
178
+
179
+ if not isinstance(attributes, dict) or not attributes:
180
  return {
181
+ "error": "Attributes must be a non-empty dictionary",
 
182
  "success": False,
183
+ "group": group_name,
184
+ }
185
+
186
+ try:
187
+ result = bridge.set_group(group_name, attributes)
188
+ return {
189
+ "group": group_name,
190
+ "set_attributes": attributes,
191
+ "success": True,
192
+ "phue2_result": result,
193
  }
194
+ except (KeyError, PhueException, ValueError, Exception) as e:
195
+ return handle_phue_error(group_name, "set_group_attributes", e)