7f22311172
- Switch.GetStatus (output, apower W, aenergy.total Wh), Switch.Set - jen Gen2 RPC, Gen1 odpověď parser odmítá; timeout, bez retry smyček - testy: čistý parser + RPC přes httpx.MockTransport (bez sítě) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""
|
|
Shelly Gen2+ RPC klient (HTTP, httpx) — Switch.GetStatus / Switch.Set.
|
|
|
|
Záměrně POUZE Gen2 RPC (`/rpc/<Method>?...`). Gen1 REST (`/relay/0?turn=on`)
|
|
nepodporujeme — všechna nasazovaná relé (Plus/Pro řada) mluví Gen2 a fallback
|
|
by jen maskoval chybnou konfiguraci. Viz docs/04-modules/pool-shelly.md.
|
|
|
|
Žádné retry smyčky: telemetrii volá poll cyklus každých 60 s a další pokus
|
|
zajistí sám; ovládání jde přes signal_service (vlastní retry + verify).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
DEFAULT_TIMEOUT_S = 5.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ShellySwitchStatus:
|
|
"""Stav Switch komponenty ze Switch.GetStatus."""
|
|
|
|
output: bool
|
|
apower_w: float | None
|
|
aenergy_total_wh: float | None
|
|
|
|
|
|
def shelly_base_url(protocol: str | None, host: str, port: int | None) -> str:
|
|
"""Base URL Shelly z řádku ems.site_endpoint (protocol/host/port)."""
|
|
p = (protocol or "http").lower()
|
|
if p not in ("http", "https"):
|
|
p = "http"
|
|
prt = int(port or (443 if p == "https" else 80))
|
|
return f"{p}://{host}:{prt}"
|
|
|
|
|
|
def parse_switch_status(data: dict[str, Any]) -> ShellySwitchStatus:
|
|
"""Čistý parser odpovědi Switch.GetStatus (testovatelné bez HTTP).
|
|
|
|
Gen2: {"id":0,"output":true,"apower":745.3,"aenergy":{"total":12345.678,...},...}
|
|
`aenergy.total` je ve Wh; `apower` ve W. Obojí volitelné (ne každý model měří).
|
|
"""
|
|
if "output" not in data:
|
|
raise ValueError("Shelly Switch.GetStatus: missing 'output' (not a Gen2 RPC response?)")
|
|
output = bool(data["output"])
|
|
|
|
apower_w: float | None = None
|
|
if data.get("apower") is not None:
|
|
apower_w = float(data["apower"])
|
|
|
|
aenergy_total_wh: float | None = None
|
|
aenergy = data.get("aenergy")
|
|
if isinstance(aenergy, dict) and aenergy.get("total") is not None:
|
|
aenergy_total_wh = float(aenergy["total"])
|
|
|
|
return ShellySwitchStatus(
|
|
output=output,
|
|
apower_w=apower_w,
|
|
aenergy_total_wh=aenergy_total_wh,
|
|
)
|
|
|
|
|
|
async def get_switch_status(
|
|
base_url: str,
|
|
switch_id: int = 0,
|
|
*,
|
|
timeout: float = DEFAULT_TIMEOUT_S,
|
|
client: httpx.AsyncClient | None = None,
|
|
) -> ShellySwitchStatus:
|
|
"""GET {base}/rpc/Switch.GetStatus?id=N → ShellySwitchStatus.
|
|
|
|
`client` lze injektovat (testy, sdílený klient); jinak se vytvoří jednorázový.
|
|
"""
|
|
url = f"{base_url.rstrip('/')}/rpc/Switch.GetStatus"
|
|
params = {"id": int(switch_id)}
|
|
if client is not None:
|
|
resp = await client.get(url, params=params)
|
|
else:
|
|
async with httpx.AsyncClient(timeout=timeout) as c:
|
|
resp = await c.get(url, params=params)
|
|
resp.raise_for_status()
|
|
return parse_switch_status(resp.json())
|
|
|
|
|
|
async def set_switch(
|
|
base_url: str,
|
|
on: bool,
|
|
switch_id: int = 0,
|
|
*,
|
|
timeout: float = DEFAULT_TIMEOUT_S,
|
|
client: httpx.AsyncClient | None = None,
|
|
) -> bool | None:
|
|
"""GET {base}/rpc/Switch.Set?id=N&on=true|false. Vrátí was_on (předchozí stav), pokud ho Shelly poslalo.
|
|
|
|
Pozn.: produkční ovládání bazénu jde přes signal_service (journal + verify);
|
|
tato funkce je pro ruční zásahy / budoucí přímé použití.
|
|
"""
|
|
url = f"{base_url.rstrip('/')}/rpc/Switch.Set"
|
|
# Gen2 RPC parsuje query parametry jako JSON — bool musí být 'true'/'false'.
|
|
params = {"id": int(switch_id), "on": "true" if on else "false"}
|
|
if client is not None:
|
|
resp = await client.get(url, params=params)
|
|
else:
|
|
async with httpx.AsyncClient(timeout=timeout) as c:
|
|
resp = await c.get(url, params=params)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
was_on = data.get("was_on") if isinstance(data, dict) else None
|
|
return bool(was_on) if was_on is not None else None
|