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
parent 071684da15
commit f6e670b408
9 changed files with 941 additions and 9 deletions
@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""Sonda Aira Home: verifica che il canale cloud funzioni e legge lo stato dell'impianto.
+
+Credenziali (in ordine di priorita'):
+  1. variabili d'ambiente AIRA_EMAIL / AIRA_PASSWORD
+  2. file JSON $AIRA_CONFIG_DIR/aira.json  {"email": "...", "password": "..."}
+  3. token salvati in $AIRA_CONFIG_DIR/aira_tokens.json (refresh automatico)
+
+I token vengono risalvati (permessi 600) a ogni login riuscito, cosi' le volte
+successive non serve piu' la password.
+
+Uso:
+    .venv/bin/python aira_probe.py            # stato impianto
+    .venv/bin/python aira_probe.py --json     # dump completo
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import stat
+import sys
+from pathlib import Path
+
+BASE = Path(os.environ.get("AIRA_CONFIG_DIR") or (Path.home() / ".hermes" / "profiles" / "carlo"))
+CREDS_FILE = BASE / "aira.json"
+TOKENS_FILE = BASE / "aira_tokens.json"
+
+
+def _write_private(path: Path, data: dict) -> None:
+    """Scrive JSON con permessi 600 (mai credenziali leggibili da altri)."""
+    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 load_credentials() -> tuple[str, str] | None:
+    email = os.environ.get("AIRA_EMAIL")
+    password = os.environ.get("AIRA_PASSWORD")
+    if email and password:
+        return email, password
+    if CREDS_FILE.exists():
+        data = json.loads(CREDS_FILE.read_text())
+        if data.get("email") and data.get("password"):
+            return data["email"], data["password"]
+    return None
+
+
+def load_tokens() -> dict | None:
+    if TOKENS_FILE.exists():
+        data = json.loads(TOKENS_FILE.read_text())
+        if all(data.get(k) for k in ("id_token", "access_token", "refresh_token")):
+            return data
+    return None
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--json", action="store_true", help="dump completo dello stato")
+    ap.add_argument("--pool", type=int, default=None,
+                    help="indice user pool (default: da aira_config.json, altrimenti 0)")
+    args = ap.parse_args()
+
+    from pyairahome import AiraHome
+    from pyairahome.config import Settings
+
+    # Il pool giusto va letto dalla config, mai indovinato: un pool sbagliato
+    # risponde "credenziali non valide" anche con la password corretta.
+    cfg = BASE / "aira_config.json"
+    pool_index = args.pool
+    if pool_index is None and cfg.exists():
+        pool_index = json.loads(cfg.read_text()).get("user_pool_index")
+    if pool_index is None:
+        pool_index = 0
+    print(f"[auth] user pool [{pool_index}] {Settings.USER_POOL_IDS[pool_index]}")
+
+    aira = AiraHome(user_pool_id=Settings.USER_POOL_IDS[pool_index])
+    logged = False
+
+    tokens = load_tokens()
+    if tokens:
+        try:
+            aira.cloud.login_with_tokens(
+                tokens["id_token"], tokens["access_token"], tokens["refresh_token"]
+            )
+            logged = True
+            print("[auth] sessione ripristinata dai token salvati")
+        except Exception as exc:  # token scaduti/revocati
+            print(f"[auth] token non piu' validi ({type(exc).__name__}), serve il login")
+
+    if not logged:
+        creds = load_credentials()
+        if not creds:
+            print(
+                "[auth] credenziali assenti.\n"
+                f"       Esegui prima:  .venv/bin/python setup_creds.py\n"
+                f"       (scrive {CREDS_FILE})",
+                file=sys.stderr,
+            )
+            return 2
+        email, password = creds
+        aira.cloud.login_with_credentials(email, password)
+        logged = True
+        print(f"[auth] login cloud riuscito come {email}")
+
+    # Salva i token per i prossimi avvii.
+    try:
+        _write_private(TOKENS_FILE, aira.cloud.get_tokens().dict())
+        print(f"[auth] token salvati in {TOKENS_FILE} (600)")
+    except Exception as exc:
+        print(f"[auth] impossibile salvare i token: {exc}")
+
+    devices = aira.cloud.get_devices()
+    dev_list = devices.get("devices", [])
+    print(f"\n[impianti] trovati: {len(dev_list)}")
+
+    for i, dev in enumerate(dev_list):
+        device_id = dev["id"]["value"]
+        household_id = dev.get("device_id", {}).get("household_id", {}).get("value")
+        name = dev.get("name", {}).get("value", f"impianto {i}")
+        print(f"  - [{i}] {name}  id={device_id}  household={household_id}")
+
+    if not dev_list:
+        print("[!] nessun impianto associato all'account")
+        aira.close()
+        return 1
+
+    device_id = dev_list[0]["id"]["value"]
+    states = aira.cloud.get_states(device_id)
+
+    if args.json:
+        print("\n=== STATO (dump completo) ===")
+        print(json.dumps(states, indent=2, default=str))
+    else:
+        print("\n=== STATO ===")
+        print(json.dumps(states, indent=2, default=str)[:6000])
+
+    aira.close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())