kaneda/harbour-aira
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
fddd21e16a200ec91ccb7fd5c54695168e7b0deb
harbour-aira/bridge/aira_bridge.py
T
Carlo f6e670b408 bridge nel repo: sorgenti, unit systemd e istruzioni per i tester 
- bridge/ con aira_bridge.py, setup_creds.py, aira_probe.py, test offline
- percorsi di configurazione via AIRA_CONFIG_DIR (default invariato)
- README del bridge in inglese: installazione, servizio, firewall, sicurezza
- README principale aggiornato (il bridge non è più 'fuori dal repo')
2026-10-10 07:44:41 +02:00

433 lines
18 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.3.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
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
# --------------------------------------------------------------------------- #
# 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 []
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"),
}
# --------------------------------------------------------------------------- #
# 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 == "/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