harbour-aira/bridge/aira_probe.py

#!/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())