Merge: bazénové čerpadlo přes Shelly (telemetrie + signal ovládání)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/services/shelly_client.py

@@ -0,0 +1,112 @@
+"""
+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

backend/services/telemetry_collector.py

@@ -397,6 +397,7 @@ async def run_telemetry_loop(conn: asyncpg.Connection) -> float:
             await poll_inverter(sid, conn)
             await poll_ev_chargers(sid, conn)
             await poll_heat_pump(sid, conn)
+            await poll_pool_pumps(sid, conn)
         except Exception as e:
             logger.error("Telemetry loop error site %s: %s", sid, e)
     return loop.time() - start
@@ -421,3 +422,42 @@ async def run_telemetry_loop_wrapper(pool: asyncpg.Pool) -> None:
         if elapsed > 50:
             logger.warning("Telemetry loop took %.1fs (>50s)", elapsed)
         await asyncio.sleep(max(0.0, 60.0 - elapsed))
+
+
+async def poll_pool_pumps(site_id: int, db: asyncpg.Connection) -> None:
+    """Poll bazénových čerpadel přes Shelly relé (Gen2 RPC Switch.GetStatus), 60 s.
+
+    Shelly nedrží historii — stavíme ji 1min vzorky jako u ostatních zařízení.
+    """
+    # Lokální import: minimální dotyk hlavičky souboru (souběžné změny na main).
+    from services.shelly_client import get_switch_status, shelly_base_url
+
+    rows = await db.fetch(
+        """
+        select pump_id as id, code, shelly_switch_id, protocol, host, port
+        from ems.vw_asset_pool_pump_http_poll
+        where site_id = $1
+        """,
+        site_id,
+    )
+    measured_at = datetime.now(timezone.utc)
+    for row in rows:
+        code = row["code"]
+        base = shelly_base_url(row["protocol"], row["host"], row["port"])
+        try:
+            status = await get_switch_status(base, int(row["shelly_switch_id"] or 0))
+        except Exception as e:
+            # Při výpadku čtení NIC nezapisovat — fabrikovaná nula by špinila
+            # historii spotřeby (stejný princip jako u EV nabíječek výše).
+            logger.warning("pool pump %s (%s) read failed: %s", code, base, e)
+            continue
+
+        await db.execute(
+            "select ems.fn_telemetry_pool_pump_sample($1::int, $2::int, $3::timestamptz, $4::boolean, $5::int, $6::bigint)",
+            site_id,
+            row["id"],
+            measured_at,
+            status.output,
+            int(round(status.apower_w)) if status.apower_w is not None else None,
+            int(round(status.aenergy_total_wh)) if status.aenergy_total_wh is not None else None,
+        )