0.5.0: riordino come l'app ufficiale (home, Dati, Impianto)
- home: temperatura esterna e in casa/assente in alto, messaggio di stato, scheda impianto con temperatura stanza e acqua calda; comandi sotto - pagina Dati: calore prodotto, elettricita' usata, COP acqua calda e riscaldamento, risparmio, CO2 evitata, Smart Tariff Control (mese per mese) - pagina Impianto: stato, termostato a parete, versioni software, errori - bridge 0.4.0: rotta /stats (servizio statistiche del cloud) e blocco dispositivo in /summary (ID breve, versioni, termostato) - C++: parseStats, formatNumber, formatMoney, monthLabel (separatore e valuta secondo la lingua); 61/61 verifiche del nucleo - anteprime aggiornate (dati, impianto) in inglese e italiano
This commit is contained in:
+220
-1
@@ -53,7 +53,7 @@ CREDS = BASE / "aira.json"
|
||||
TOKENS = BASE / "aira_tokens.json"
|
||||
POOL_CFG = BASE / "aira_config.json"
|
||||
|
||||
VERSION = "0.3.0"
|
||||
VERSION = "0.4.0"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -97,6 +97,10 @@ class AiraSession:
|
||||
# Ultimo stato letto: serve a rimandare alla pompa il valore attuale
|
||||
# della zona che NON si sta modificando (mandarla a zero la azzererebbe).
|
||||
self._states = None
|
||||
# Statistiche: sono chiamate piu' lente e cambiano lentamente, quindi si
|
||||
# tengono in cache (il telefono puo' chiederle a ogni apertura di pagina).
|
||||
self._stats = None
|
||||
self._stats_at = 0.0
|
||||
|
||||
def _login(self):
|
||||
from pyairahome import AiraHome
|
||||
@@ -156,6 +160,26 @@ class AiraSession:
|
||||
def last_states(self):
|
||||
return self._states
|
||||
|
||||
def household_id(self) -> str:
|
||||
devs = self.call(lambda a: a.cloud.get_devices())["devices"]
|
||||
if not devs:
|
||||
raise RuntimeError("nessun impianto associato all'account")
|
||||
return (devs[0].get("device_id") or {}).get("household_id", {}).get("value", "")
|
||||
|
||||
def stats(self, ttl: int = 600) -> dict:
|
||||
"""Statistiche dal servizio di sola lettura, con cache (default 10 minuti)."""
|
||||
import time
|
||||
|
||||
now = time.time()
|
||||
if self._stats is not None and (now - self._stats_at) < ttl:
|
||||
return self._stats
|
||||
|
||||
did = self.device_id()
|
||||
hid = self.household_id()
|
||||
out = self.call(lambda a: build_stats(a, did, hid))
|
||||
self._stats, self._stats_at = out, now
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# riassunto leggibile
|
||||
@@ -170,6 +194,27 @@ def summarize(st: dict) -> dict:
|
||||
return d
|
||||
|
||||
errs = st.get("errors") or []
|
||||
|
||||
# ---- dati del dispositivo (per la pagina "Impianto": accessori + riepilogo,
|
||||
# come la scheda "Accessori" e il "Riepilogo impostazioni" dell'app ufficiale)
|
||||
hp_id = g("heat_pump_id", "value") or ""
|
||||
versions = st.get("versions") or {}
|
||||
thermos = st.get("thermostats") or []
|
||||
t0 = thermos[0] if thermos and isinstance(thermos[0], dict) else {}
|
||||
t_upd = t0.get("last_update") or {}
|
||||
t_zone = str(t0.get("zone") or "")
|
||||
|
||||
def zona_numero(nome: str) -> int:
|
||||
"""'ZONE_1' -> 1 (0 se non riconosciuta)."""
|
||||
coda = nome.rsplit("_", 1)[-1]
|
||||
return int(coda) if coda.isdigit() else 0
|
||||
|
||||
def decimo(valore):
|
||||
"""Il termostato manda i decimali come interi: 228 -> 22.8, 734 -> 73.4."""
|
||||
if not isinstance(valore, (int, float)):
|
||||
return None
|
||||
return round(float(valore) / 10.0, 1)
|
||||
|
||||
return {
|
||||
"outdoor_temperature": g("current_outdoor_temperature"),
|
||||
"indoor_temperature": g("deprecated_current_indoor_temperature"),
|
||||
@@ -205,9 +250,180 @@ def summarize(st: dict) -> dict:
|
||||
"zone2_cooling_setpoint": g("zone_setpoints_cooling", "zone_2"),
|
||||
"errors_count": len(errs) if isinstance(errs, list) else 0,
|
||||
"pump_updated": g("aws_iot_received_time") or st.get("time"),
|
||||
# Identificativo e versioni: l'app ufficiale mostra le prime 6 cifre
|
||||
# dell'UUID come "Dispositivo ID".
|
||||
"device_id": hp_id[:6] if hp_id else None,
|
||||
"device_uuid": hp_id or None,
|
||||
"version_platform": versions.get("linux_build_id"),
|
||||
"version_connectivity": versions.get("connectivity_manager"),
|
||||
"version_climate_control": versions.get("climate_control_villa"),
|
||||
"version_outdoor_unit": versions.get("outdoor_unit_application"),
|
||||
"version_outdoor_eeprom": versions.get("outdoor_unit_eeprom"),
|
||||
"connected_via": "Cloud",
|
||||
"timezone": g("time_zone", "time_zone_identifier"),
|
||||
# Termostato a parete (se presente).
|
||||
"thermostats_count": len(thermos) if isinstance(thermos, list) else 0,
|
||||
"thermostat_zone": zona_numero(t_zone) if t0 else None,
|
||||
"thermostat_article": t0.get("article_number"),
|
||||
"thermostat_temperature": decimo(t_upd.get("actual_temperature")),
|
||||
"thermostat_humidity": decimo(t_upd.get("humidity")),
|
||||
"thermostat_rssi": t0.get("rssi"),
|
||||
"thermostat_battery_low": bool(t_upd.get("warning_low_battery_level")),
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# statistiche (HeatPumpStatisticsService: sola lettura, nessun comando)
|
||||
# --------------------------------------------------------------------------- #
|
||||
def money_value(units, scale) -> float:
|
||||
"""Money del protocollo: units / 10^scale. (2710000000, 9) -> 2.71."""
|
||||
try:
|
||||
return round(float(units) / (10 ** int(scale)), 2)
|
||||
except (TypeError, ValueError, ZeroDivisionError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def wh_to_kwh(wh):
|
||||
"""Wh -> kWh con un decimale; None se il dato manca o e' zero."""
|
||||
if not isinstance(wh, (int, float)) or not wh:
|
||||
return None
|
||||
return round(float(wh) / 1000.0, 1)
|
||||
|
||||
|
||||
def _ym(chiave) -> str:
|
||||
"""YearMonth -> '2026-09'."""
|
||||
return f"{getattr(chiave, 'year', 0) or 0:04d}-{getattr(chiave, 'month', 0) or 0:02d}"
|
||||
|
||||
|
||||
def _data(chiave) -> str:
|
||||
"""LocalDateTime -> '2026-10-01'."""
|
||||
return (f"{getattr(chiave, 'year', 0) or 0:04d}-"
|
||||
f"{getattr(chiave, 'month', 0) or 0:02d}-"
|
||||
f"{getattr(chiave, 'day', 0) or 0:02d}")
|
||||
|
||||
|
||||
def build_stats(aira, heat_pump_id: str, household_id: str,
|
||||
months: int = 6, days: int = 30) -> dict:
|
||||
"""Statistiche delle stesse voci che l'app ufficiale mostra in "Dati":
|
||||
calore prodotto, elettricita' usata, efficienza (COP) di acqua calda e
|
||||
riscaldamento, risparmi e risparmio Smart Tariff Control.
|
||||
|
||||
Tutto in sola lettura: non invia nessun comando all'impianto.
|
||||
"""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from google.protobuf.timestamp_pb2 import Timestamp
|
||||
from pyairahome.device.heat_pump.statistics.v1 import service_pb2 as S
|
||||
|
||||
# _stats_service_stub non e' esposto dalla libreria: serve per GetCop,
|
||||
# GetSavings e GetOptimisationSavings (get_insights copre solo gli insight).
|
||||
stub = aira.cloud._stats_service_stub # noqa: SLF001
|
||||
call = aira.cloud.call_service
|
||||
now = datetime.now()
|
||||
|
||||
out = {
|
||||
"period_days": days,
|
||||
"currency": "EUR",
|
||||
"month": {"label": None, "heat_kwh": None, "energy_kwh": None,
|
||||
"cop_heating": None, "cop_dhw": None},
|
||||
"cop_monthly": [],
|
||||
"insights": [],
|
||||
"savings": None,
|
||||
"smart_tariff": None,
|
||||
"monthly_savings": [],
|
||||
}
|
||||
|
||||
# ---- COP mensile (l'"Efficienza energetica" dell'app ufficiale) ---------- #
|
||||
req = S.GetCopRequest()
|
||||
req.heat_pump_id.value = heat_pump_id
|
||||
req.granularity = S.COP_GRANULARITY_MONTHLY
|
||||
inizio = now - timedelta(days=31 * months)
|
||||
req.start_date.year, req.start_date.month = inizio.year, inizio.month
|
||||
if now.month == 12:
|
||||
req.end_date_exclusive.year, req.end_date_exclusive.month = now.year + 1, 1
|
||||
else:
|
||||
req.end_date_exclusive.year, req.end_date_exclusive.month = now.year, now.month + 1
|
||||
for e in call(stub, "GetCop", req, timeout=25, raw=True).entries:
|
||||
out["cop_monthly"].append({
|
||||
"month": _ym(e.start_date),
|
||||
"heating_cop": round(e.room_heating_cop_value, 2) if e.room_heating_cop_value else None,
|
||||
"dhw_cop": round(e.dhw_cop_value, 2) if e.dhw_cop_value else None,
|
||||
"avg_outdoor": round(e.average_outdoor_temperature, 1) if e.average_outdoor_temperature else None,
|
||||
})
|
||||
# L'efficienza da mostrare e' quella dell'ultimo mese con un dato: se il
|
||||
# riscaldamento non ha ancora dati (come nell'app ufficiale: "Nessun dato")
|
||||
# resta il COP dell'acqua calda, che invece c'e'.
|
||||
for voce in reversed(out["cop_monthly"]):
|
||||
if voce["dhw_cop"] or voce["heating_cop"]:
|
||||
out["month"]["cop_heating"] = voce["heating_cop"]
|
||||
out["month"]["cop_dhw"] = voce["dhw_cop"]
|
||||
break
|
||||
|
||||
# ---- Calore prodotto ed elettricita' usata ------------------------------ #
|
||||
req = S.GetHeatPumpInsightsRequest()
|
||||
req.heat_pump_id.value = heat_pump_id
|
||||
req.granularity = S.GRANULARITY_DAILY
|
||||
da = now - timedelta(days=days)
|
||||
req.start_time.year, req.start_time.month, req.start_time.day = da.year, da.month, da.day
|
||||
req.end_time_exclusive.year = now.year
|
||||
req.end_time_exclusive.month = now.month
|
||||
req.end_time_exclusive.day = now.day
|
||||
per_mese: dict = {}
|
||||
for e in call(stub, "GetHeatPumpInsights", req, timeout=25, raw=True).insights:
|
||||
giorno = _data(e.start_time)
|
||||
out["insights"].append({
|
||||
"date": giorno,
|
||||
"energy_kwh": wh_to_kwh(e.energy_consumption_wh),
|
||||
"heat_kwh": wh_to_kwh(e.delivered_heat_wh),
|
||||
})
|
||||
voce = per_mese.setdefault(giorno[:7], {"heat": 0.0, "energy": 0.0})
|
||||
voce["heat"] += float(e.delivered_heat_wh or 0)
|
||||
voce["energy"] += float(e.energy_consumption_wh or 0)
|
||||
if per_mese:
|
||||
ultimo = max(per_mese)
|
||||
out["month"]["label"] = ultimo
|
||||
out["month"]["heat_kwh"] = wh_to_kwh(per_mese[ultimo]["heat"])
|
||||
out["month"]["energy_kwh"] = wh_to_kwh(per_mese[ultimo]["energy"])
|
||||
|
||||
# ---- Risparmi del periodo ----------------------------------------------- #
|
||||
req = S.GetSavingsRequest()
|
||||
req.heat_pump_id.value = heat_pump_id
|
||||
ts = Timestamp()
|
||||
ts.FromDatetime(da)
|
||||
req.start_time.CopyFrom(ts)
|
||||
ts = Timestamp()
|
||||
ts.FromDatetime(now)
|
||||
req.end_time.CopyFrom(ts)
|
||||
sav = call(stub, "GetSavings", req, timeout=25, raw=True)
|
||||
out["currency"] = sav.currency_code or "EUR"
|
||||
out["savings"] = {
|
||||
"amount": round(sav.cost_savings, 2),
|
||||
# Attenzione al nome: il campo e' co_2_savings_kg (con l'underscore),
|
||||
# non "co2_savings_kg".
|
||||
"co2_kg": round(sav.co_2_savings_kg, 1),
|
||||
"show": bool(sav.show),
|
||||
"points": int(sav.number_of_data_points or 0),
|
||||
}
|
||||
|
||||
# ---- Risparmio Smart Tariff Control (mese mobile) ----------------------- #
|
||||
if household_id:
|
||||
req = S.GetOptimisationSavingsRequest()
|
||||
req.household_id.value = household_id
|
||||
opt = call(stub, "GetOptimisationSavings", req, timeout=25, raw=True)
|
||||
rm = opt.rolling_month_optimisation
|
||||
out["smart_tariff"] = {
|
||||
"amount": money_value(rm.savings.units, rm.savings.scale),
|
||||
"hours": int(rm.weather_optimisation_hours or 0),
|
||||
}
|
||||
for m in opt.monthly_savings:
|
||||
out["monthly_savings"].append({
|
||||
"month": _ym(m.month),
|
||||
"amount": money_value(m.savings.units, m.savings.scale),
|
||||
})
|
||||
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# comandi ammessi (whitelist: mai esporre l'intera libreria su HTTP)
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -367,6 +583,9 @@ class Handler(BaseHTTPRequestHandler):
|
||||
return self._send(200, {"ok": True, "summary": summarize(self.session.states())})
|
||||
if path == "/state":
|
||||
return self._send(200, {"ok": True, "state": self.session.states()})
|
||||
if path == "/stats":
|
||||
# Statistiche (sola lettura, con cache breve lato bridge).
|
||||
return self._send(200, {"ok": True, "stats": self.session.stats()})
|
||||
if path == "/commands":
|
||||
return self._send(200, {"ok": True, "commands": COMMANDS})
|
||||
return self._send(404, {"ok": False, "error": "rotta sconosciuta"})
|
||||
|
||||
Reference in New Issue
Block a user