Carlo
6ef6f5bfd8
La copia del bridge dentro l'app va tenuta allineata: contiene la correzione del boost, dell'assenza, di ACS off e del riscaldamento forzato, che la libreria accetta solo con un argomento. Il pacchetto Sailfish non cambia (il bridge non entra nell'RPM), quindi non serve ricostruire l'RPM.
206 lines
10 KiB
Python
206 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Verifica offline dei comandi del bridge. NON invia nulla all'impianto:
|
|
costruisce i messaggi protobuf e ne controlla il contenuto.
|
|
|
|
.venv/bin/python test_bridge_commands.py
|
|
"""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
import aira_bridge as b # noqa: E402
|
|
|
|
esiti = []
|
|
|
|
|
|
def check(nome, cond, extra=""):
|
|
esiti.append((nome, bool(cond), extra))
|
|
print(f"{'OK ' if cond else 'FALLITO'} {nome}{(' -> ' + extra) if extra else ''}")
|
|
|
|
|
|
# --- contesto: come lo stato reale dell'impianto --------------------------- #
|
|
ctx = {"zone_setpoints_heating": {"zone_1": 19.0, "zone_2": 0.0},
|
|
"zone_setpoints_cooling": {"zone_1": 22.0, "zone_2": 0.0}}
|
|
|
|
# --- 1) riscaldamento, zona 1 --------------------------------------------- #
|
|
cmd = b.build_command("set_heating_setpoint", 21.5, 1, ctx)
|
|
msg = cmd._message
|
|
check("zona 1: obiettivo applicato", abs(msg.zone_setpoints.zone_1 - 21.5) < 1e-6,
|
|
f"zone_1={msg.zone_setpoints.zone_1}")
|
|
check("zona 1: zona 2 preservata", msg.zone_setpoints.zone_2 == 0.0,
|
|
f"zone_2={msg.zone_setpoints.zone_2}")
|
|
check("zona 1: kind = riscaldamento (1)", msg.kind == 1, f"kind={msg.kind}")
|
|
check("messaggio serializzabile", len(msg.SerializeToString()) > 0,
|
|
f"{len(msg.SerializeToString())} byte")
|
|
|
|
# --- 2) raffrescamento, zona 2 (con zona 1 da preservare) ------------------ #
|
|
ctx2 = {"zone_setpoints_cooling": {"zone_1": 22.0, "zone_2": 24.0},
|
|
"zone_setpoints_heating": {"zone_1": 19.0, "zone_2": 0.0}}
|
|
cmd = b.build_command("set_cooling_setpoint", 23.0, 2, ctx2)
|
|
msg = cmd._message
|
|
check("zona 2: obiettivo applicato", abs(msg.zone_setpoints.zone_2 - 23.0) < 1e-6,
|
|
f"zone_2={msg.zone_setpoints.zone_2}")
|
|
check("zona 2: zona 1 preservata (22.0)", msg.zone_setpoints.zone_1 == 22.0,
|
|
f"zone_1={msg.zone_setpoints.zone_1}")
|
|
check("raffrescamento: kind = 2", msg.kind == 2, f"kind={msg.kind}")
|
|
|
|
# --- 3) senza contesto: non deve rompersi -------------------------------- #
|
|
cmd = b.build_command("set_heating_setpoint", 20.0)
|
|
check("senza contesto: nessuna eccezione", cmd._message.zone_setpoints.zone_1 == 20.0)
|
|
|
|
# --- 4) valori rifiutati -------------------------------------------------- #
|
|
def rifiuta(nome, *args, **kw):
|
|
try:
|
|
b.build_command(*args, **kw)
|
|
except ValueError as exc:
|
|
check(nome, True, str(exc))
|
|
except Exception as exc: # inatteso
|
|
check(nome, False, f"eccezione sbagliata: {type(exc).__name__}: {exc}")
|
|
else:
|
|
check(nome, False, "accettato, doveva essere rifiutato")
|
|
|
|
|
|
rifiuta("rifiuta 40 °C (troppo alta)", "set_heating_setpoint", 40.0, 1, ctx)
|
|
rifiuta("rifiuta 3 °C (troppo bassa)", "set_heating_setpoint", 3.0, 1, ctx)
|
|
rifiuta("rifiuta zona 3", "set_heating_setpoint", 20.0, 3, ctx)
|
|
rifiuta("rifiuta valore non numerico", "set_heating_setpoint", "caldo", 1, ctx)
|
|
rifiuta("rifiuta comando sconosciuto", "factory_reset", None, 1, ctx)
|
|
|
|
# --- 5) il riassunto espone i nuovi campi --------------------------------- #
|
|
stato_file = Path("/tmp/state.json")
|
|
if stato_file.exists():
|
|
st = json.loads(stato_file.read_text())["state"]
|
|
s = b.summarize(st)
|
|
for k in ("zone1_setpoint", "zone1_temperature", "zone1_mode",
|
|
"zone2_setpoint", "zones"):
|
|
check(f"riassunto contiene {k}", k in s, repr(s.get(k)))
|
|
check("zona 1: obiettivo presente", s.get("zone1_setpoint") is not None,
|
|
f"{s.get('zone1_setpoint')} °C")
|
|
|
|
# force_heating e' un oggetto {enabled, remaining_time}: leggerlo con bool()
|
|
# darebbe sempre vero (era il bug). Qui si verifica la lettura del campo.
|
|
spento = b.summarize(dict(st, force_heating={"enabled": False, "remaining_time": "0s"}))
|
|
acceso = b.summarize(dict(st, force_heating={"enabled": True, "remaining_time": "1h"}))
|
|
check("riassunto: forzato spento -> falso", spento.get("force_heating") is False,
|
|
repr(spento.get("force_heating")))
|
|
check("riassunto: forzato acceso -> vero", acceso.get("force_heating") is True,
|
|
repr(acceso.get("force_heating")))
|
|
check("riassunto: durata forzato riportata", acceso.get("force_heating_remaining") == "1h",
|
|
repr(acceso.get("force_heating_remaining")))
|
|
for k in ("hot_water_heating", "configured_modes", "pump_active_state",
|
|
"outdoor_defrost", "signature_lights"):
|
|
check(f"riassunto contiene {k}", k in s, repr(s.get(k)))
|
|
else:
|
|
print("(salto il controllo del riassunto: /tmp/state.json assente)")
|
|
|
|
# --- 6) statistiche: conversioni pure ------------------------------------- #
|
|
check("money_value: 2710000000 / 10^9 = 2.71", b.money_value(2710000000, 9) == 2.71,
|
|
repr(b.money_value(2710000000, 9)))
|
|
check("money_value: valori assenti -> 0.0", b.money_value(None, None) == 0.0)
|
|
check("wh_to_kwh: 29300 Wh = 29.3 kWh", b.wh_to_kwh(29300) == 29.3, repr(b.wh_to_kwh(29300)))
|
|
check("wh_to_kwh: 70844.5 Wh = 70.8 kWh", b.wh_to_kwh(70844.53878839477) == 70.8,
|
|
repr(b.wh_to_kwh(70844.53878839477)))
|
|
check("wh_to_kwh: zero -> None (nessun dato)", b.wh_to_kwh(0) is None)
|
|
check("wh_to_kwh: valore assente -> None", b.wh_to_kwh(None) is None)
|
|
|
|
class _YM: # finto YearMonth / LocalDateTime
|
|
def __init__(self, y, m, d=0):
|
|
self.year, self.month, self.day = y, m, d
|
|
|
|
check("_ym: 2026-09", b._ym(_YM(2026, 9)) == "2026-09", b._ym(_YM(2026, 9)))
|
|
check("_data: 2026-10-01", b._data(_YM(2026, 10, 1)) == "2026-10-01", b._data(_YM(2026, 10, 1)))
|
|
|
|
# --- 7) riassunto: blocco dispositivo (stato sintetico, nessun file) ------- #
|
|
# I valori del termostato arrivano con i decimali come interi (228 -> 22.8).
|
|
finto = {
|
|
"heat_pump_id": {"value": "1f4b26c4-1800-4b21-8ea2-236925a92aeb"},
|
|
"versions": {"linux_build_id": "6.10.0", "connectivity_manager": "3.30.1",
|
|
"climate_control_villa": "3.9.12", "outdoor_unit_application": "1.30.0",
|
|
"outdoor_unit_eeprom": "0.114.0"},
|
|
"thermostats": [{"article_number": "201884", "zone": "ZONE_1", "rssi": -79,
|
|
"last_update": {"actual_temperature": 228, "humidity": 734,
|
|
"warning_low_battery_level": False}}],
|
|
"time_zone": {"time_zone_identifier": "Europe/Rome"},
|
|
"zone_setpoints_heating": {"zone_1": 19.0, "zone_2": 0.0},
|
|
"zone_temperatures": {"zone_1": 22.8, "zone_2": 0.0},
|
|
}
|
|
d = b.summarize(finto)
|
|
check("dispositivo: id breve di 6 cifre", d.get("device_id") == "1f4b26", repr(d.get("device_id")))
|
|
check("dispositivo: uuid completo", d.get("device_uuid") == "1f4b26c4-1800-4b21-8ea2-236925a92aeb")
|
|
check("dispositivo: versione piattaforma", d.get("version_platform") == "6.10.0")
|
|
check("dispositivo: versione connettività", d.get("version_connectivity") == "3.30.1")
|
|
check("dispositivo: versione unità esterna", d.get("version_outdoor_unit") == "1.30.0")
|
|
check("dispositivo: connessione", d.get("connected_via") == "Cloud")
|
|
check("dispositivo: fuso orario", d.get("timezone") == "Europe/Rome")
|
|
check("termostato: zona 1 da 'ZONE_1'", d.get("thermostat_zone") == 1, repr(d.get("thermostat_zone")))
|
|
check("termostato: 228 -> 22.8 °C", d.get("thermostat_temperature") == 22.8,
|
|
repr(d.get("thermostat_temperature")))
|
|
check("termostato: 734 -> 73.4 %", d.get("thermostat_humidity") == 73.4,
|
|
repr(d.get("thermostat_humidity")))
|
|
check("termostato: segnale riportato", d.get("thermostat_rssi") == -79)
|
|
check("termostato: batteria non scarica", d.get("thermostat_battery_low") is False)
|
|
check("senza termostato: nessun errore",
|
|
b.summarize({k: v for k, v in finto.items() if k != "thermostats"}).get("thermostat_zone") is None)
|
|
|
|
# --- 5) comandi che la libreria accetta solo con i loro argomenti ---------- #
|
|
# Qui sta la ragione del blocco: il boost senza durata solleva TypeError e il
|
|
# bridge risponde "richiesta non valida" all'app.
|
|
from pyairahome.commands import CommandBase # noqa: E402
|
|
|
|
boost = b.build_command("hot_water_boost_on")
|
|
check("boost: un'ora di default",
|
|
boost._message.hot_water_boost_duration.seconds == 3600,
|
|
f"{boost._message.hot_water_boost_duration.seconds}s")
|
|
check("boost: durata scelta in minuti (90 -> 5400s)",
|
|
b.build_command("hot_water_boost_on", 90)
|
|
._message.hot_water_boost_duration.seconds == 5400)
|
|
rifiuta("boost: durata fuori intervallo", "hot_water_boost_on", 5)
|
|
rifiuta("boost: durata non numerica", "hot_water_boost_on", "presto")
|
|
|
|
acs = b.build_command("hot_water_heating_off")
|
|
check("ACS spento: un'ora di default", acs._message.time.seconds == 3600,
|
|
f"{acs._message.time.seconds}s")
|
|
|
|
forzato = b.build_command("force_heating_on")
|
|
check("forzato: un'ora di default", forzato._message.duration.seconds == 3600,
|
|
f"{forzato._message.duration.seconds}s")
|
|
|
|
assenza = b.build_command("away_mode_on")
|
|
m = assenza._message
|
|
check("assenza: da adesso a +7 giorni",
|
|
abs((m.end_time.seconds - m.current_time.seconds) - 7 * 86400) <= 1,
|
|
f"{m.end_time.seconds - m.current_time.seconds}s")
|
|
check("assenza: fine dopo l'inizio", m.end_time.seconds > m.current_time.seconds)
|
|
check("assenza: casa tenuta a 15 °C", abs(m.target_room_temperature - 15.0) < 1e-6,
|
|
f"{m.target_room_temperature}")
|
|
rifiuta("assenza: zero giorni", "away_mode_on", 0)
|
|
|
|
# Nessun comando della whitelist deve restare senza un argomento obbligatorio:
|
|
# e' il difetto che ha fermato il boost, e la whitelist e' esattamente la lista
|
|
# che l'app puo' chiamare, quindi va provata tutta.
|
|
campione = {"set_hot_water_target": 55, "set_heating_setpoint": 20,
|
|
"set_cooling_setpoint": 20, "hot_water_boost_on": 60,
|
|
"hot_water_heating_off": 60, "away_mode_on": 7,
|
|
"force_heating_on": 60}
|
|
for cid in b.COMMANDS:
|
|
try:
|
|
cmd = b.build_command(cid, campione.get(cid), 1, ctx)
|
|
# La busta e' quello che la libreria manda davvero: se il comando non
|
|
# avesse un nome di campo valido, si romperebbe qui. (I comandi senza
|
|
# parametri, come ping, hanno il messaggio interno vuoto: controllare
|
|
# quello sarebbe sbagliato.)
|
|
busta = CommandBase.build_command(cmd)
|
|
byte = len(busta.SerializeToString())
|
|
check(f"whitelist: {cid} si costruisce e si incarta",
|
|
bool(cmd.get_field()) and byte > 0,
|
|
f"campo {cmd.get_field()}, {byte} byte")
|
|
except Exception as exc:
|
|
check(f"whitelist: {cid} si costruisce e si incarta", False,
|
|
f"{type(exc).__name__}: {exc}")
|
|
|
|
# --- esito ---------------------------------------------------------------- #
|
|
falliti = [n for n, ok, _ in esiti if not ok]
|
|
print(f"\n{len(esiti) - len(falliti)}/{len(esiti)} controlli superati")
|
|
sys.exit(1 if falliti else 0)
|