kaneda/harbour-aira
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
36033daf3f5c5f1b92b46c446739fed3ebd8a2ae
harbour-aira/bridge/aira_bridge.py
T
kaneda 4f279a18b0 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
2026-10-10 08:59:03 +02:00

652 lines
27 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Bridge HTTP locale verso l'impianto Aira (per l'app Sailfish harbour-aira).
Perche' esiste: il cloud Aira parla **gRPC** (HTTP/2) e Qt 5.6, il toolkit di
Sailfish, non ha HTTP/2. Un client nativo non puo' quindi parlare direttamente
con Aira: questo servizio traduce HTTP/1.1 + JSON -> gRPC, e tiene le
credenziali/token SOLO sul server.
Avvio:
python aira_bridge.py # ascolta su 0.0.0.0:8790
python aira_bridge.py --port 8791 # porta diversa
Config: $AIRA_CONFIG_DIR/aira_bridge.json {"host","port","token"}
(il default di AIRA_CONFIG_DIR e' la cartella usata dall'autore; su un'altra
macchina conviene impostarla, es. AIRA_CONFIG_DIR=$HOME/.config/aira)
Il token viene generato al primo avvio se assente.
Endpoint (tutti richiedono `Authorization: Bearer <token>`, tranne /health):
GET /health -> {ok, version}
GET /summary -> valori leggibili per la dashboard
GET /state -> stato completo (raw)
GET /commands -> comandi disponibili con i loro parametri
POST /command -> {"id": "...", "value": ...}
Log su stderr. stdout NON e' usato per il protocollo (solo log di servizio).
"""
from __future__ import annotations
import argparse
import json
import os
import secrets
import stat
import sys
import threading
import traceback
import warnings
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
warnings.filterwarnings("ignore")
# Tutti i file di configurazione stanno in una cartella sola (permessi 600):
# aira.json credenziali Aira (email + password)
# aira_tokens.json token di sessione Cognito (si rinnovano da soli)
# aira_config.json indice del pool di utenti Cognito
# aira_bridge.json host/porta/token del bridge (token generato al primo avvio)
# Il default e' la cartella usata dall'autore: su un'altra macchina si imposta
# AIRA_CONFIG_DIR (es. AIRA_CONFIG_DIR=$HOME/.config/aira).
BASE = Path(os.environ.get("AIRA_CONFIG_DIR") or (Path.home() / ".hermes" / "profiles" / "carlo"))
BRIDGE_CFG = BASE / "aira_bridge.json"
CREDS = BASE / "aira.json"
TOKENS = BASE / "aira_tokens.json"
POOL_CFG = BASE / "aira_config.json"
VERSION = "0.4.0"
# --------------------------------------------------------------------------- #
# configurazione e credenziali
# --------------------------------------------------------------------------- #
def load_bridge_config() -> dict:
cfg = {"host": "0.0.0.0", "port": 8790, "token": ""}
if BRIDGE_CFG.exists():
cfg.update(json.loads(BRIDGE_CFG.read_text()))
if not cfg.get("token"):
cfg["token"] = secrets.token_urlsafe(32)
write_private(BRIDGE_CFG, cfg)
print(f"[bridge] token generato -> {BRIDGE_CFG} (600)", file=sys.stderr)
return cfg
def write_private(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(data, indent=2))
os.chmod(tmp, stat.S_IRUSR | stat.S_IWUSR)
tmp.replace(path)
def pool_id() -> str:
from pyairahome.config import Settings
idx = 0
if POOL_CFG.exists():
idx = json.loads(POOL_CFG.read_text()).get("user_pool_index", 0)
return Settings.USER_POOL_IDS[idx]
# --------------------------------------------------------------------------- #
# sessione Aira (una sola, riusata; il login si fa solo se i token non bastano)
# --------------------------------------------------------------------------- #
class AiraSession:
def __init__(self) -> None:
self._lock = threading.Lock()
self._client = None
# 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
aira = AiraHome(user_pool_id=pool_id())
if TOKENS.exists():
t = json.loads(TOKENS.read_text())
try:
aira.cloud.login_with_tokens(t["id_token"], t["access_token"], t["refresh_token"])
return aira
except Exception as exc:
print(f"[bridge] token non validi ({type(exc).__name__}), provo le credenziali", file=sys.stderr)
if not CREDS.exists():
raise RuntimeError(f"credenziali assenti: {CREDS}")
c = json.loads(CREDS.read_text())
aira.cloud.login_with_credentials(c["email"], c["password"])
try:
write_private(TOKENS, aira.cloud.get_tokens().dict())
except Exception as exc:
print(f"[bridge] token non salvati: {exc}", file=sys.stderr)
return aira
def _ensure(self):
if self._client is None:
self._client = self._login()
return self._client
def call(self, fn):
"""Esegue `fn(client)` sotto lock; se la sessione e' morta, rifa' il login una volta."""
from pyairahome.utils import TokenError
from pyairahome.utils.exceptions import AuthenticationError
with self._lock:
try:
return fn(self._ensure())
except (TokenError, AuthenticationError):
print("[bridge] sessione scaduta: nuovo login", file=sys.stderr)
self._client = None
return fn(self._ensure())
def device_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]["id"]["value"]
def states(self) -> dict:
did = self.device_id()
st = self.call(lambda a: a.cloud.get_states(did))
out = st["heat_pump_states"][0]
self._states = out
return out
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
# --------------------------------------------------------------------------- #
def summarize(st: dict) -> dict:
def g(*path, default=None):
d = st
for p in path:
if not isinstance(d, dict) or p not in d:
return default
d = d[p]
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"),
"dhw_current": g("current_hot_water_temperature"),
"dhw_target": g("target_hot_water_temperature"),
"operating_status": g("operating_status"),
"away_mode": bool(st.get("away_mode_enabled")),
"night_mode": bool(st.get("night_mode_enabled")),
"manual_mode": bool(st.get("manual_mode_enabled")),
# force_heating E' un oggetto {enabled, remaining_time}: leggerlo con
# bool() darebbe sempre vero, perche' un dict non vuoto e' truthy.
"force_heating": bool(g("force_heating", "enabled")),
"force_heating_remaining": g("force_heating", "remaining_time"),
"inline_heater": bool(st.get("inline_heater_active")),
# Stato delle funzioni, per la pagina dei comandi: senza questi l'utente
# preme alla cieca senza sapere cosa e' gia' attivo.
"hot_water_heating": bool(g("hot_water", "heating_enabled")),
"configured_modes": g("configured_pump_modes"),
"pump_active_state": g("pump_active_state"),
"outdoor_defrost": bool(st.get("outdoor_unit_defrost_enabled")),
"signature_lights": bool(g("signature_element", "enabled")),
"led_pattern": g("led_pattern"),
"zones": st.get("num_zones"),
# Riscaldamento/raffrescamento per zona: obiettivo, temperatura della
# stanza e stato della pompa (zona 2 a zero se non configurata).
"zone1_setpoint": g("zone_setpoints_heating", "zone_1"),
"zone1_temperature": g("zone_temperatures", "zone_1"),
"zone1_mode": g("current_pump_mode_state", "zone_1"),
"zone1_cooling_setpoint": g("zone_setpoints_cooling", "zone_1"),
"zone2_setpoint": g("zone_setpoints_heating", "zone_2"),
"zone2_temperature": g("zone_temperatures", "zone_2"),
"zone2_mode": g("current_pump_mode_state", "zone_2"),
"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)
# --------------------------------------------------------------------------- #
COMMANDS = {
"ping": {"help": "verifica che l'impianto risponda", "value": None},
"set_hot_water_target": {"help": "temperatura ACS obiettivo in °C", "value": "45-60"},
"set_heating_setpoint": {"help": "temperatura obiettivo della casa in °C", "value": "5-30", "zone": "1 o 2 (opzionale, default 1)"},
"set_cooling_setpoint": {"help": "temperatura obiettivo del raffrescamento in °C", "value": "5-30", "zone": "1 o 2 (opzionale, default 1)"},
"hot_water_boost_on": {"help": "avvia il boost ACS (una tantum)", "value": None},
"hot_water_boost_off": {"help": "ferma il boost ACS", "value": None},
"hot_water_heating_on": {"help": "abilita il riscaldamento ACS", "value": None},
"hot_water_heating_off": {"help": "disabilita il riscaldamento ACS", "value": None},
"away_mode_on": {"help": "modalita' assenza", "value": None},
"away_mode_off": {"help": "disattiva l'assenza", "value": None},
"night_mode_1h": {"help": "night mode per un'ora", "value": None},
"force_heating_on": {"help": "forza il riscaldamento", "value": None},
"force_heating_off": {"help": "disattiva il riscaldamento forzato", "value": None},
"heating_on": {"help": "abilita la funzione riscaldamento", "value": None},
"heating_off": {"help": "disabilita la funzione riscaldamento", "value": None},
"legionella_cycle": {"help": "avvia un ciclo antilegionella", "value": None},
}
def build_command(cmd_id: str, value=None, zone=1, context=None):
"""Costruisce l'oggetto comando della libreria. Importata solo qui per non
pagarne il costo all'avvio del bridge.
`context` e' l'ultimo stato letto: serve ai comandi per zona, che vanno
inviati completi (zona 1 E zona 2) per non azzerare quella non toccata.
"""
from pyairahome.commands import (
ActivateHotWaterBoosting,
ActivateNightModeForOneHour,
ClearAwayMode,
DeactivateHotWaterBoosting,
DisableForceHeating,
DisableHeatingFunction,
DisableHotWaterHeating,
EnableForceHeating,
EnableHeatingFunction,
EnableHotWaterHeating,
Ping,
RunLegionellaCycle,
SetAwayMode,
SetTargetHotWaterTemperature,
SetZoneSetpoints,
)
simple = {
"ping": Ping,
"hot_water_boost_on": ActivateHotWaterBoosting,
"hot_water_boost_off": DeactivateHotWaterBoosting,
"hot_water_heating_on": EnableHotWaterHeating,
"hot_water_heating_off": DisableHotWaterHeating,
"away_mode_on": SetAwayMode,
"away_mode_off": ClearAwayMode,
"night_mode_1h": ActivateNightModeForOneHour,
"force_heating_on": EnableForceHeating,
"force_heating_off": DisableForceHeating,
"heating_on": EnableHeatingFunction,
"heating_off": DisableHeatingFunction,
"legionella_cycle": RunLegionellaCycle,
}
if cmd_id == "set_hot_water_target":
try:
t = float(value)
except (TypeError, ValueError):
raise ValueError("set_hot_water_target richiede un numero in °C")
if not 45.0 <= t <= 60.0:
raise ValueError("temperatura ACS fuori intervallo ammesso (45-60 °C)")
return SetTargetHotWaterTemperature(t)
if cmd_id in ("set_heating_setpoint", "set_cooling_setpoint"):
heating = cmd_id == "set_heating_setpoint"
nome = "heating" if heating else "cooling"
try:
t = float(value)
except (TypeError, ValueError):
raise ValueError(f"{cmd_id} richiede un numero in °C")
if not 5.0 <= t <= 30.0:
raise ValueError("temperatura della casa fuori intervallo ammesso (5-30 °C)")
try:
z = int(zone)
except (TypeError, ValueError):
raise ValueError("zona non valida: usare 1 o 2")
if z not in (1, 2):
raise ValueError("zona ammessa: 1 o 2")
# Il messaggio va inviato completo: la zona non toccata viaggia col suo
# valore attuale, altrimenti a zero verrebbe spenta.
def attuale(zona: int) -> float:
if not isinstance(context, dict):
return 0.0
valori = context.get(f"zone_setpoints_{nome}") or {}
v = valori.get(f"zone_{zona}")
return float(v) if isinstance(v, (int, float)) else 0.0
from pyairahome.device.heat_pump.command.v1.set_zone_setpoints_pb2 import (
ZoneTemperatures,
)
zs = ZoneTemperatures(
zone_1=t if z == 1 else attuale(1),
zone_2=t if z == 2 else attuale(2),
)
# Kind: 1 = riscaldamento, 2 = raffrescamento (enum del protocollo).
return SetZoneSetpoints(zone_setpoints=zs, kind=1 if heating else 2)
cls = simple.get(cmd_id)
if cls is None:
raise ValueError(f"comando non ammesso: {cmd_id}")
return cls()
# --------------------------------------------------------------------------- #
# server HTTP
# --------------------------------------------------------------------------- #
class Handler(BaseHTTPRequestHandler):
server_version = "aira-bridge/" + VERSION
session: AiraSession = None # impostato in main()
token: str = ""
# ---- utilita' ---------------------------------------------------------- #
def log_message(self, fmt, *args): # log su stderr, non su stdout
print(f"[bridge] {self.address_string()} {fmt % args}", file=sys.stderr)
def _send(self, code: int, payload: dict) -> None:
body = json.dumps(payload, default=str).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _auth_ok(self) -> bool:
if self.path.split("?")[0] == "/health":
return True
hdr = self.headers.get("Authorization", "")
return hdr == f"Bearer {self.token}"
def _body(self) -> dict:
n = int(self.headers.get("Content-Length") or 0)
if not n:
return {}
return json.loads(self.rfile.read(n).decode() or "{}")
# ---- rotte ------------------------------------------------------------- #
def do_GET(self):
if not self._auth_ok():
return self._send(401, {"ok": False, "error": "token mancante o errato"})
path = self.path.split("?")[0]
try:
if path == "/health":
return self._send(200, {"ok": True, "version": VERSION})
if path == "/summary":
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"})
except Exception as exc:
traceback.print_exc(file=sys.stderr)
return self._send(502, {"ok": False, "error": f"{type(exc).__name__}: {exc}"})
def do_POST(self):
if not self._auth_ok():
return self._send(401, {"ok": False, "error": "token mancante o errato"})
if self.path.split("?")[0] != "/command":
return self._send(404, {"ok": False, "error": "rotta sconosciuta"})
try:
body = self._body()
cmd_id = body.get("id", "")
cmd = build_command(cmd_id, body.get("value"), body.get("zone") or 1,
self.session.last_states())
except ValueError as exc:
return self._send(400, {"ok": False, "error": str(exc)})
except Exception as exc:
return self._send(400, {"ok": False, "error": f"richiesta non valida: {exc}"})
try:
did = self.session.device_id()
def run(a):
out = []
for update in a.cloud.run_command(did, cmd):
out.append(update)
return out
result = self.session.call(run)
print(f"[bridge] comando {cmd_id} eseguito", file=sys.stderr)
return self._send(200, {"ok": True, "id": cmd_id, "result": result[-1] if result else None})
except Exception as exc:
traceback.print_exc(file=sys.stderr)
return self._send(502, {"ok": False, "error": f"{type(exc).__name__}: {exc}"})
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--host", default=None)
ap.add_argument("--port", type=int, default=None)
args = ap.parse_args()
cfg = load_bridge_config()
host = args.host or cfg["host"]
port = args.port or cfg["port"]
Handler.session = AiraSession()
Handler.token = cfg["token"]
srv = ThreadingHTTPServer((host, port), Handler)
print(f"[bridge] in ascolto su http://{host}:{port} (v{VERSION})", file=sys.stderr)
try:
srv.serve_forever()
except KeyboardInterrupt:
print("[bridge] arresto", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink