truetype
b078b850f8
Flask app con gauge Chart.js custom, polling Shelly Cloud API v2, storico SQLite, due widget gauge (produzione/consumo) con scala graduata.
105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
"""Shelly Cloud API client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
class ShellyError(Exception):
|
|
"""Shelly API error."""
|
|
|
|
|
|
class ShellyClient:
|
|
"""Client for Shelly Cloud Control API v2."""
|
|
|
|
def __init__(
|
|
self,
|
|
server: str,
|
|
auth_key: str,
|
|
device_id: str,
|
|
timeout: float = 10,
|
|
):
|
|
self.server = server.rstrip("/")
|
|
self.auth_key = auth_key
|
|
self.device_id = device_id
|
|
self.timeout = timeout
|
|
self._session = requests.Session()
|
|
|
|
def _post(self, path: str, body: dict) -> Any:
|
|
"""POST JSON to API endpoint with auth_key as query param."""
|
|
url = f"{self.server}/v2/devices/api/{path}"
|
|
params = {"auth_key": self.auth_key}
|
|
try:
|
|
resp = self._session.post(
|
|
url, json=body, params=params, timeout=self.timeout
|
|
)
|
|
except requests.exceptions.ConnectionError as exc:
|
|
raise ShellyError(f"Connection error: {exc}") from exc
|
|
except requests.exceptions.Timeout as exc:
|
|
raise ShellyError(f"Timeout: {exc}") from exc
|
|
|
|
if resp.status_code == 429:
|
|
time.sleep(2)
|
|
return self._post(path, body)
|
|
|
|
if resp.status_code == 401:
|
|
raise ShellyError("Authentication failed — check auth_key")
|
|
|
|
if not resp.ok:
|
|
raise ShellyError(f"HTTP {resp.status_code}: {resp.text[:200]}")
|
|
|
|
if not resp.content:
|
|
return {}
|
|
|
|
return resp.json()
|
|
|
|
def get_device_status(self) -> dict:
|
|
"""Fetch full status of the device."""
|
|
data = self._post("get", {
|
|
"ids": [self.device_id],
|
|
"select": ["status"],
|
|
})
|
|
if isinstance(data, list) and len(data) > 0:
|
|
return data[0]
|
|
return {}
|
|
|
|
def parse_readings(self, status: dict) -> dict:
|
|
"""Extract useful readings from device status."""
|
|
s = status.get("status", status)
|
|
|
|
em1_0 = s.get("em1:0", {})
|
|
em1_1 = s.get("em1:1", {})
|
|
em1data_0 = s.get("em1data:0", {})
|
|
em1data_1 = s.get("em1data:1", {})
|
|
switch = s.get("switch:0", {})
|
|
wifi = s.get("wifi", {})
|
|
|
|
return {
|
|
# Canale 0 — produzione pannelli → inverter
|
|
"ch0_act_power": em1_0.get("act_power"),
|
|
"ch0_aprt_power": em1_0.get("aprt_power"),
|
|
"ch0_current": em1_0.get("current"),
|
|
"ch0_voltage": em1_0.get("voltage"),
|
|
"ch0_pf": em1_0.get("pf"),
|
|
"ch0_total_act_energy": em1data_0.get("total_act_energy"),
|
|
"ch0_total_ret_energy": em1data_0.get("total_act_ret_energy"),
|
|
# Canale 1 — inverter → casa/rete
|
|
"ch1_act_power": em1_1.get("act_power"),
|
|
"ch1_aprt_power": em1_1.get("aprt_power"),
|
|
"ch1_current": em1_1.get("current"),
|
|
"ch1_voltage": em1_1.get("voltage"),
|
|
"ch1_pf": em1_1.get("pf"),
|
|
"ch1_total_act_energy": em1data_1.get("total_act_energy"),
|
|
"ch1_total_ret_energy": em1data_1.get("total_act_ret_energy"),
|
|
# Switch / stato dispositivo
|
|
"switch_output": switch.get("output"),
|
|
"temperature": switch.get("temperature", {}).get("tC"),
|
|
"wifi_rssi": wifi.get("rssi"),
|
|
"online": s.get("online", status.get("online")),
|
|
"unixtime": s.get("sys", {}).get("unixtime"),
|
|
"uptime": s.get("sys", {}).get("uptime"),
|
|
}
|