diff --git a/infra/hermes/shelly_mcp.py b/infra/hermes/shelly_mcp.py index b43818695..9836c4f34 100755 --- a/infra/hermes/shelly_mcp.py +++ b/infra/hermes/shelly_mcp.py @@ -2,6 +2,7 @@ """Shelly MCP — Status und Steuerung der Muldenstein-Shellys per LAN-HTTP. Inventar fest verdrahtet (CT999 projekte/shelly-inventar.md). +Kinds: cover | pulse | sensor | switch (Gen2/3 Plug). Sicherheit: - Rollo Bad unten (.96): Bewegung blockiert (Motor defekt) - Garagentor / Haustuer-Impuls: nur mit confirm=true, nie in Cron @@ -81,6 +82,13 @@ DEVICES: dict[str, dict] = { "kind": "sensor", "aliases": ["garage_decke_b", "decke_b", "garage-b"], }, + "plug_50": { + "ip": "192.168.178.50", + "name": "Shelly Plug MG3", + "gen": 3, + "kind": "switch", + "aliases": ["plug_50", "shelly_50", "plug50", "steckdose_50", "mg3"], + }, } # flatten alias lookup @@ -168,7 +176,7 @@ def tool_list(_args: dict) -> str: ) lines.append(f" Aliase: {', '.join(d.get('aliases', [key]))}") lines.append("") - lines.append("Tools: shelly_list, shelly_status, shelly_cover, shelly_pulse") + lines.append("Tools: shelly_list, shelly_status, shelly_cover, shelly_pulse, shelly_switch") lines.append( "Cron: Rollos OK; Tor/Tuersummer verboten; Bad unten nie bewegen." ) @@ -230,14 +238,26 @@ def _status_gen2(d: dict) -> dict: if not isinstance(st, dict): out["error"] = str(st) return out - cover = st.get("cover:0") or {} - out["cover"] = { - "state": cover.get("state"), - "current_pos": cover.get("current_pos"), - "apower": cover.get("apower"), - "voltage": cover.get("voltage"), - "source": cover.get("source"), - } + if d["kind"] == "cover": + cover = st.get("cover:0") or {} + out["cover"] = { + "state": cover.get("state"), + "current_pos": cover.get("current_pos"), + "apower": cover.get("apower"), + "voltage": cover.get("voltage"), + "source": cover.get("source"), + } + if d["kind"] == "switch": + sw = st.get("switch:0") or {} + out["switch"] = { + "output": sw.get("output"), + "apower": sw.get("apower"), + "voltage": sw.get("voltage"), + "current": sw.get("current"), + "aenergy_total": (sw.get("aenergy") or {}).get("total"), + "temperature_c": (sw.get("temperature") or {}).get("tC"), + "source": sw.get("source"), + } wifi = st.get("wifi") or {} out["wifi_rssi"] = wifi.get("rssi") if d.get("motor_broken"): @@ -251,7 +271,7 @@ def tool_status(args: dict) -> str: return resolved key, d = resolved try: - data = _status_gen2(d) if d["gen"] == 2 else _status_gen1(d) + data = _status_gen2(d) if int(d.get("gen", 1)) >= 2 else _status_gen1(d) data["key"] = key return json.dumps(data, ensure_ascii=False, indent=2) except Exception as e: # noqa: BLE001 @@ -327,7 +347,7 @@ def tool_cover(args: dict) -> str: if not 0 <= pos <= 100: return "position muss 0-100 sein." try: - if d["gen"] == 2: + if int(d.get("gen", 1)) >= 2: return _cover_gen2(d["ip"], action, pos) return _cover_gen1_roller(d["ip"], action, pos) except Exception as e: # noqa: BLE001 @@ -363,6 +383,44 @@ def tool_pulse(args: dict) -> str: return f"Fehler shelly_pulse {d['name']}: {e}" + +def tool_switch(args: dict) -> str: + """Gen2/3 Switch/Plug: on|off|toggle.""" + resolved = _resolve(args.get("target") or args.get("device") or args.get("name")) + if isinstance(resolved, str): + return resolved + key, d = resolved + if d["kind"] != "switch": + return ( + f"{d['name']} ist kein Schalter/Plug (kind={d['kind']}). " + "Fuer Rollos: shelly_cover. Fuer Impuls: shelly_pulse." + ) + if int(d.get("gen", 1)) < 2: + return f"{d['name']}: Gen1-Switch noch nicht implementiert." + action = str(args.get("action") or "").strip().lower() + if action in ("an", "ein", "on"): + action = "on" + elif action in ("aus", "off"): + action = "off" + elif action in ("toggle", "umschalten", "schalten"): + action = "toggle" + if action not in ("on", "off", "toggle"): + return "action muss on|off|toggle sein." + ip = d["ip"] + try: + method = {"on": "Switch.Set", "off": "Switch.Set", "toggle": "Switch.Toggle"}[action] + params = {"id": 0} + if action in ("on", "off"): + params["on"] = action == "on" + res = _http_post_json(f"http://{ip}/rpc", {"id": 1, "method": method, "params": params}) + return json.dumps( + {"ok": True, "device": d["name"], "ip": ip, "action": action, "result": res}, + ensure_ascii=False, + ) + except Exception as e: # noqa: BLE001 + return f"Fehler shelly_switch {d['name']}: {e}" + + TOOLS = [ { "name": "shelly_list", @@ -377,7 +435,7 @@ TOOLS = [ "description": ( "Live-Status eines Shellys (Rollo-Position, Relay, Sensor-Temps). " "Ziel per Alias/IP/Name: kueche, computerraum, gaestezimmer, bad_unten, " - "garagentor, haustuer, garage_decke_a/b." + "garagentor, haustuer, garage_decke_a/b, plug_50." ), "inputSchema": { "type": "object", @@ -435,6 +493,27 @@ TOOLS = [ "required": ["target", "confirm"], }, }, + { + "name": "shelly_switch", + "description": ( + "Schaltet einen Shelly-Plug/Switch (on|off|toggle). " + "Ziel z.B. plug_50 / 192.168.178.50 / Shelly Plug MG3." + ), + "inputSchema": { + "type": "object", + "properties": { + "target": { + "type": "string", + "description": "Alias, IP oder Name (z.B. plug_50)", + }, + "action": { + "type": "string", + "description": "on | off | toggle", + }, + }, + "required": ["target", "action"], + }, + }, ] HANDLERS = { @@ -442,6 +521,7 @@ HANDLERS = { "shelly_status": tool_status, "shelly_cover": tool_cover, "shelly_pulse": tool_pulse, + "shelly_switch": tool_switch, }