Spaces:
Running
Running
zzstoatzz commited on
Commit ·
0b63457
1
Parent(s): 220b6cc
improvements
Browse files
examples/smart_home/src/smart_home/lights/server.py
CHANGED
|
@@ -101,36 +101,95 @@ def list_groups() -> list[str]:
|
|
| 101 |
|
| 102 |
|
| 103 |
@lights_mcp.tool()
|
| 104 |
-
def list_scenes() -> list[str]:
|
| 105 |
-
"""Lists
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 106 |
if not (bridge := _get_bridge()):
|
| 107 |
return ["Error: Bridge not connected"]
|
| 108 |
try:
|
| 109 |
-
|
| 110 |
-
|
| 111 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 112 |
except (PhueException, Exception) as e:
|
| 113 |
-
|
|
|
|
| 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 |
-
#
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 123 |
result = bridge.run_scene(group_name=group_name, scene_name=scene_name)
|
| 124 |
-
|
| 125 |
if result:
|
| 126 |
return {
|
| 127 |
"group": group_name,
|
| 128 |
"activated_scene": scene_name,
|
| 129 |
"success": True,
|
| 130 |
-
"phue2_result": result,
|
| 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,
|
|
@@ -139,7 +198,7 @@ def activate_scene(group_name: str, scene_name: str) -> dict[str, Any]:
|
|
| 139 |
}
|
| 140 |
|
| 141 |
except (KeyError, PhueException, Exception) as e:
|
| 142 |
-
#
|
| 143 |
return handle_phue_error(f"{group_name}/{scene_name}", "activate_scene", e)
|
| 144 |
|
| 145 |
|
|
@@ -193,3 +252,41 @@ def set_group_attributes(group_name: str, attributes: HueAttributes) -> dict[str
|
|
| 193 |
}
|
| 194 |
except (KeyError, PhueException, ValueError, Exception) as e:
|
| 195 |
return handle_phue_error(group_name, "set_group_attributes", e)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 101 |
|
| 102 |
|
| 103 |
@lights_mcp.tool()
|
| 104 |
+
def list_scenes() -> dict[str, list[str]] | list[str]:
|
| 105 |
+
"""Lists Hue scenes, grouped by the light group they belong to.
|
| 106 |
+
|
| 107 |
+
Returns:
|
| 108 |
+
dict[str, list[str]]: A dictionary mapping group names to a list of scene names within that group.
|
| 109 |
+
list[str]: An error message list if the bridge connection fails or an error occurs.
|
| 110 |
+
"""
|
| 111 |
if not (bridge := _get_bridge()):
|
| 112 |
return ["Error: Bridge not connected"]
|
| 113 |
try:
|
| 114 |
+
scenes_data = bridge.get_scene() # Returns dict {scene_id: {details...}}
|
| 115 |
+
groups_data = bridge.get_group() # Returns dict {group_id: {details...}}
|
| 116 |
+
|
| 117 |
+
# Create a lookup for group name by group ID
|
| 118 |
+
group_id_to_name = {gid: ginfo["name"] for gid, ginfo in groups_data.items()}
|
| 119 |
+
|
| 120 |
+
scenes_by_group: dict[str, list[str]] = {}
|
| 121 |
+
for scene_id, scene_details in scenes_data.items():
|
| 122 |
+
scene_name = scene_details.get("name")
|
| 123 |
+
# Scenes might be associated with a group via 'group' key or lights
|
| 124 |
+
# Using 'group' key if available is more direct for group scenes
|
| 125 |
+
group_id = scene_details.get("group")
|
| 126 |
+
if scene_name and group_id and group_id in group_id_to_name:
|
| 127 |
+
group_name = group_id_to_name[group_id]
|
| 128 |
+
if group_name not in scenes_by_group:
|
| 129 |
+
scenes_by_group[group_name] = []
|
| 130 |
+
# Avoid duplicate scene names within a group listing (though unlikely)
|
| 131 |
+
if scene_name not in scenes_by_group[group_name]:
|
| 132 |
+
scenes_by_group[group_name].append(scene_name)
|
| 133 |
+
|
| 134 |
+
# Sort scenes within each group for consistent output
|
| 135 |
+
for group_name in scenes_by_group:
|
| 136 |
+
scenes_by_group[group_name].sort()
|
| 137 |
+
|
| 138 |
+
return scenes_by_group
|
| 139 |
except (PhueException, Exception) as e:
|
| 140 |
+
# Return error as list to match other list-returning tools on error
|
| 141 |
+
return [f"Error listing scenes by group: {e}"]
|
| 142 |
|
| 143 |
|
| 144 |
@lights_mcp.tool()
|
| 145 |
def activate_scene(group_name: str, scene_name: str) -> dict[str, Any]:
|
| 146 |
+
"""Activates a specific scene within a specified light group, verifying the scene belongs to the group."""
|
| 147 |
if not (bridge := _get_bridge()):
|
| 148 |
return {"error": "Bridge not connected", "success": False}
|
| 149 |
try:
|
| 150 |
+
# 1. Find the target group ID
|
| 151 |
+
groups_data = bridge.get_group()
|
| 152 |
+
target_group_id = None
|
| 153 |
+
for gid, ginfo in groups_data.items():
|
| 154 |
+
if ginfo.get("name") == group_name:
|
| 155 |
+
target_group_id = gid
|
| 156 |
+
break
|
| 157 |
+
if not target_group_id:
|
| 158 |
+
return {"error": f"Group '{group_name}' not found", "success": False}
|
| 159 |
+
|
| 160 |
+
# 2. Find the target scene and check its group association
|
| 161 |
+
scenes_data = bridge.get_scene()
|
| 162 |
+
scene_found = False
|
| 163 |
+
scene_in_correct_group = False
|
| 164 |
+
for sid, sinfo in scenes_data.items():
|
| 165 |
+
if sinfo.get("name") == scene_name:
|
| 166 |
+
scene_found = True
|
| 167 |
+
# Check if this scene is associated with the target group ID
|
| 168 |
+
if sinfo.get("group") == target_group_id:
|
| 169 |
+
scene_in_correct_group = True
|
| 170 |
+
break # Found the scene in the correct group
|
| 171 |
+
|
| 172 |
+
if not scene_found:
|
| 173 |
+
return {"error": f"Scene '{scene_name}' not found", "success": False}
|
| 174 |
+
|
| 175 |
+
if not scene_in_correct_group:
|
| 176 |
+
return {
|
| 177 |
+
"error": f"Scene '{scene_name}' does not belong to group '{group_name}'",
|
| 178 |
+
"success": False,
|
| 179 |
+
}
|
| 180 |
+
|
| 181 |
+
# 3. Activate the scene (now that we've verified it)
|
| 182 |
result = bridge.run_scene(group_name=group_name, scene_name=scene_name)
|
| 183 |
+
|
| 184 |
if result:
|
| 185 |
return {
|
| 186 |
"group": group_name,
|
| 187 |
"activated_scene": scene_name,
|
| 188 |
"success": True,
|
| 189 |
+
"phue2_result": result,
|
| 190 |
}
|
| 191 |
else:
|
| 192 |
+
# This case might indicate the scene/group exists but activation failed internally
|
| 193 |
return {
|
| 194 |
"group": group_name,
|
| 195 |
"scene": scene_name,
|
|
|
|
| 198 |
}
|
| 199 |
|
| 200 |
except (KeyError, PhueException, Exception) as e:
|
| 201 |
+
# Handle potential errors during bridge communication or data parsing
|
| 202 |
return handle_phue_error(f"{group_name}/{scene_name}", "activate_scene", e)
|
| 203 |
|
| 204 |
|
|
|
|
| 252 |
}
|
| 253 |
except (KeyError, PhueException, ValueError, Exception) as e:
|
| 254 |
return handle_phue_error(group_name, "set_group_attributes", e)
|
| 255 |
+
|
| 256 |
+
|
| 257 |
+
@lights_mcp.tool()
|
| 258 |
+
def list_lights_by_group() -> dict[str, list[str]] | list[str]:
|
| 259 |
+
"""Lists Hue lights, grouped by the room/group they belong to.
|
| 260 |
+
|
| 261 |
+
Returns:
|
| 262 |
+
dict[str, list[str]]: A dictionary mapping group names to a list of light names within that group.
|
| 263 |
+
list[str]: An error message list if the bridge connection fails or an error occurs.
|
| 264 |
+
"""
|
| 265 |
+
if not (bridge := _get_bridge()):
|
| 266 |
+
return ["Error: Bridge not connected"]
|
| 267 |
+
try:
|
| 268 |
+
groups_data = bridge.get_group() # dict {group_id: {details}}
|
| 269 |
+
lights_data = bridge.get_light_objects("id") # dict {light_id: {details}}
|
| 270 |
+
|
| 271 |
+
lights_by_group: dict[str, list[str]] = {}
|
| 272 |
+
for group_id, group_details in groups_data.items():
|
| 273 |
+
group_name = group_details.get("name")
|
| 274 |
+
light_ids = group_details.get("lights", [])
|
| 275 |
+
if group_name and light_ids:
|
| 276 |
+
light_names = []
|
| 277 |
+
for light_id in light_ids:
|
| 278 |
+
# phue uses string IDs for lights in group, but int IDs in get_light_objects
|
| 279 |
+
light_id_int = int(light_id)
|
| 280 |
+
if light_id_int in lights_data:
|
| 281 |
+
light_name = lights_data[light_id_int].name
|
| 282 |
+
if light_name:
|
| 283 |
+
light_names.append(light_name)
|
| 284 |
+
if light_names:
|
| 285 |
+
light_names.sort() # Keep light list sorted
|
| 286 |
+
lights_by_group[group_name] = light_names
|
| 287 |
+
|
| 288 |
+
return lights_by_group
|
| 289 |
+
|
| 290 |
+
except (PhueException, Exception) as e:
|
| 291 |
+
# Return error as list
|
| 292 |
+
return [f"Error listing lights by group: {e}"]
|