petr.polezhaev
1fc845d7f5
- ESPHome connect/command/импорт оборачиваются в AcceptanceError; при сбое connect канал закрывается (loop закрыт), CLI даёт rc 2 без traceback - self-test: connect error, command error, отсутствие aioesphomeapi через main (rc 2) — всего 15 acceptance-тестов - RELEASE_HA: repaired-wheel в dist/manylinux + удаление linux-тега перед upload; README компонента — корректная ссылка на docs/RELEASE_HA.md - translations/ru: options data «LAN-ключ»; план §2/§6 синхронизирован
554 lines
20 KiB
Python
554 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""Приёмка HA (и опционально ESPHome) на живом кондиционере.
|
||
|
||
Полуавтоматический скрипт: третью LAN-сессию НЕ открывает (оба слота заняты
|
||
HA + ESPHome); работает через REST API Home Assistant по long-lived token.
|
||
|
||
Профиль → Security → Long-lived access tokens → создать токен.
|
||
|
||
Запуск:
|
||
python tests/acceptance/test_esphome_ha.py quick \
|
||
--ha-url http://homeassistant.local:8123 --ha-token TOKEN \
|
||
--entity climate.ac
|
||
python tests/acceptance/test_esphome_ha.py long \
|
||
--ha-url ... --ha-token ... --entity climate.ac \
|
||
--hours 24 --interval 3600 --report acceptance.csv
|
||
|
||
Режимы:
|
||
quick — матрица изменений (hvac/fan/уставка/swing), последовательно и
|
||
«burst» (без пауз), каждое действие проверяется по состоянию в
|
||
HA и завершается возвратом к исходному состоянию;
|
||
long — одно изменение в --interval секунд (по умолчанию 3600) на
|
||
протяжении --hours часов, проверка и возврат, CSV-отчёт
|
||
дописывается по ходу.
|
||
|
||
ESPHome-сторона (опционально): --esphome-host/--esphome-key/--esphome-entity
|
||
и установленный `aioesphomeapi` — шаги «режим» и «уставка» отправляются
|
||
через ESPHome native API, результат сверяется по состоянию в HA (то есть
|
||
проверяется сквозная цепочка ESPHome → модуль → HA). Fan/swing через
|
||
ESPHome в этой версии не отправляются.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import csv
|
||
import datetime as dt
|
||
import json
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from dataclasses import dataclass
|
||
from typing import Any, Callable, Optional
|
||
|
||
DEFAULT_TIMEOUT = 15.0
|
||
SETTLE_TIMEOUT = 30.0
|
||
|
||
|
||
class AcceptanceError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class HaRest:
|
||
"""Минимальный клиент REST API Home Assistant (stdlib)."""
|
||
|
||
def __init__(self, base_url: str, token: str, timeout: float = DEFAULT_TIMEOUT):
|
||
self.base_url = base_url.rstrip("/")
|
||
self.timeout = timeout
|
||
self.headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
def _request(self, method: str, path: str, payload: Any = None) -> Any:
|
||
data = json.dumps(payload).encode() if payload is not None else None
|
||
request = urllib.request.Request(
|
||
self.base_url + path, data=data, headers=self.headers, method=method
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=self.timeout) as resp:
|
||
body = resp.read().decode() or "null"
|
||
try:
|
||
return resp.status, json.loads(body)
|
||
except ValueError as err:
|
||
raise AcceptanceError(
|
||
f"{method} {path}: ответ не JSON: {body[:120]!r}"
|
||
) from err
|
||
except urllib.error.HTTPError as err:
|
||
try:
|
||
body = err.read().decode(errors="replace")
|
||
finally:
|
||
err.close()
|
||
raise AcceptanceError(
|
||
f"{method} {path}: HTTP {err.code}: {body[:200]}"
|
||
) from err
|
||
except OSError as err:
|
||
raise AcceptanceError(f"{method} {path}: {err}") from err
|
||
|
||
def state(self, entity_id: str) -> dict:
|
||
_, data = self._request("GET", f"/api/states/{entity_id}")
|
||
if not isinstance(data, dict):
|
||
raise AcceptanceError(f"{entity_id}: неожиданный ответ /api/states")
|
||
return data
|
||
|
||
def call(self, domain: str, service: str, data: dict) -> None:
|
||
self._request("POST", f"/api/services/{domain}/{service}", data)
|
||
|
||
def wait_state(
|
||
self,
|
||
entity_id: str,
|
||
predicate: Callable[[dict], bool],
|
||
timeout: float = SETTLE_TIMEOUT,
|
||
) -> dict:
|
||
deadline = time.monotonic() + timeout
|
||
last: Optional[dict] = None
|
||
while time.monotonic() < deadline:
|
||
last = self.state(entity_id)
|
||
if predicate(last):
|
||
return last
|
||
time.sleep(0.5)
|
||
raise AcceptanceError(
|
||
f"{entity_id}: состояние не достигнуто за {timeout} с: "
|
||
f"{last and last.get('state')}"
|
||
)
|
||
|
||
|
||
class EsphomeChannel:
|
||
"""Канал ESPHome native API: режим и уставка (best-effort)."""
|
||
|
||
def __init__(self, host: str, key: str, timeout: float = DEFAULT_TIMEOUT):
|
||
self.host = host
|
||
self.key = key
|
||
self.timeout = timeout
|
||
self._loop = asyncio.new_event_loop()
|
||
self._aio = None
|
||
self._client = None
|
||
self._entity_key: Optional[int] = None
|
||
|
||
def connect(self) -> None:
|
||
self._loop.run_until_complete(self._connect())
|
||
|
||
async def _connect(self) -> None:
|
||
try:
|
||
import aioesphomeapi # noqa: PLC0415
|
||
except ImportError as err:
|
||
raise AcceptanceError(
|
||
"ESPHome-канал: пакет aioesphomeapi не установлен"
|
||
) from err
|
||
|
||
self._aio = aioesphomeapi
|
||
try:
|
||
await self._connect_inner()
|
||
except AcceptanceError:
|
||
raise
|
||
except Exception as err: # noqa: BLE001
|
||
raise AcceptanceError(
|
||
f"ESPHome {self.host}: подключение/список сущностей: {err}"
|
||
) from err
|
||
|
||
async def _connect_inner(self) -> None:
|
||
self._client = self._aio.APIClient(
|
||
self.host, 6053, noise_psk=self.key or None
|
||
)
|
||
await asyncio.wait_for(
|
||
self._client.connect(login=True), timeout=self.timeout
|
||
)
|
||
entities, _services = await self._client.list_entities_services()
|
||
climate_info = getattr(self._aio, "ClimateInfo", None)
|
||
for entity in entities:
|
||
if climate_info is not None and isinstance(entity, climate_info):
|
||
self._entity_key = entity.key
|
||
break
|
||
if entity.__class__.__name__ == "ClimateInfo":
|
||
self._entity_key = entity.key
|
||
break
|
||
if self._entity_key is None:
|
||
raise AcceptanceError("ESPHome: climate-сущность не найдена")
|
||
|
||
def apply(self, step: "Step") -> bool:
|
||
"""Отправляет шаг через ESPHome; False — шаг не поддержан каналом."""
|
||
if step.name not in ("hvac_mode", "temperature"):
|
||
return False
|
||
self._loop.run_until_complete(self._apply(step))
|
||
return True
|
||
|
||
async def _apply(self, step: "Step") -> None:
|
||
mode = None
|
||
target = None
|
||
if step.name == "hvac_mode":
|
||
mode = getattr(self._aio.ClimateMode, str(step.expected).upper())
|
||
else:
|
||
target = float(step.expected)
|
||
try:
|
||
await asyncio.wait_for(
|
||
self._client.climate_command(
|
||
key=self._entity_key, mode=mode, target_temperature=target
|
||
),
|
||
timeout=self.timeout,
|
||
)
|
||
except Exception as err: # noqa: BLE001
|
||
raise AcceptanceError(f"ESPHome: команда не отправлена: {err}") from err
|
||
|
||
def close(self) -> None:
|
||
try:
|
||
if self._client is not None:
|
||
try:
|
||
self._loop.run_until_complete(
|
||
asyncio.wait_for(
|
||
self._client.disconnect(), timeout=self.timeout
|
||
)
|
||
)
|
||
except Exception: # noqa: BLE001 - best-effort
|
||
pass
|
||
finally:
|
||
self._loop.close()
|
||
|
||
|
||
@dataclass
|
||
class Step:
|
||
name: str
|
||
domain: str
|
||
service: str
|
||
service_data: dict
|
||
attribute: str
|
||
expected: Any
|
||
|
||
def predicate(self) -> Callable[[dict], bool]:
|
||
def check(state: dict) -> bool:
|
||
if self.attribute == "state":
|
||
return state.get("state") == self.expected
|
||
value = state.get("attributes", {}).get(self.attribute)
|
||
return _matches(value, self.expected)
|
||
|
||
return check
|
||
|
||
|
||
def _pick(options: list, preferred: str) -> Optional[str]:
|
||
if not options:
|
||
return None
|
||
if preferred in options:
|
||
return preferred
|
||
return options[0]
|
||
|
||
|
||
def _matches(observed: Any, expected: Any) -> bool:
|
||
if isinstance(expected, (int, float)) and isinstance(observed, (int, float)):
|
||
return abs(float(observed) - float(expected)) < 0.11
|
||
return observed == expected
|
||
|
||
|
||
def _observed(state: Optional[dict], step: Step) -> Any:
|
||
if state is None:
|
||
return None
|
||
if step.attribute == "state":
|
||
return state.get("state")
|
||
return state.get("attributes", {}).get(step.attribute)
|
||
|
||
|
||
def build_steps(state: dict) -> list[Step]:
|
||
"""Адаптивная матрица: только поддерживаемые режимы/диапазоны."""
|
||
attrs = state.get("attributes", {})
|
||
steps: list[Step] = []
|
||
|
||
hvac_modes = [mode for mode in attrs.get("hvac_modes", []) if mode != "off"]
|
||
mode = _pick(hvac_modes, "cool")
|
||
if mode:
|
||
steps.append(
|
||
Step("hvac_mode", "climate", "set_hvac_mode",
|
||
{"hvac_mode": mode}, "state", mode)
|
||
)
|
||
|
||
fan_mode = _pick(attrs.get("fan_modes", []), "low")
|
||
if fan_mode:
|
||
steps.append(
|
||
Step("fan_mode", "climate", "set_fan_mode",
|
||
{"fan_mode": fan_mode}, "fan_mode", fan_mode)
|
||
)
|
||
|
||
swing = _pick(attrs.get("swing_modes", []), "on")
|
||
if swing:
|
||
steps.append(
|
||
Step("swing_mode", "climate", "set_swing_mode",
|
||
{"swing_mode": swing}, "swing_mode", swing)
|
||
)
|
||
|
||
temperature = attrs.get("temperature")
|
||
low = attrs.get("min_temp", 16.0)
|
||
high = attrs.get("max_temp", 30.0)
|
||
if temperature is not None and low is not None and high is not None:
|
||
target = temperature + 1.0
|
||
if target > high:
|
||
target = max(low, high - 1.0)
|
||
steps.append(
|
||
Step("temperature", "climate", "set_temperature",
|
||
{"temperature": target}, "temperature", target)
|
||
)
|
||
return steps
|
||
|
||
|
||
def _original_predicate(original: dict) -> Callable[[dict], bool]:
|
||
state = original.get("state")
|
||
attrs = original.get("attributes", {})
|
||
|
||
def check(current: dict) -> bool:
|
||
if current.get("state") != state:
|
||
return False
|
||
current_attrs = current.get("attributes", {})
|
||
for key in ("fan_mode", "swing_mode", "temperature"):
|
||
if attrs.get(key) is not None and not _matches(
|
||
current_attrs.get(key), attrs.get(key)
|
||
):
|
||
return False
|
||
return True
|
||
|
||
return check
|
||
|
||
|
||
def _restore(
|
||
ha: HaRest, entity_id: str, original: dict, *, wait: bool = True
|
||
) -> None:
|
||
attrs = original.get("attributes", {})
|
||
state = original.get("state")
|
||
if state == "off" and "off" in attrs.get("hvac_modes", []):
|
||
ha.call("climate", "set_hvac_mode",
|
||
{"entity_id": entity_id, "hvac_mode": "off"})
|
||
elif state and state != "off":
|
||
ha.call("climate", "set_hvac_mode",
|
||
{"entity_id": entity_id, "hvac_mode": state})
|
||
if attrs.get("fan_mode") is not None:
|
||
ha.call("climate", "set_fan_mode",
|
||
{"entity_id": entity_id, "fan_mode": attrs["fan_mode"]})
|
||
if attrs.get("temperature") is not None:
|
||
ha.call("climate", "set_temperature",
|
||
{"entity_id": entity_id, "temperature": attrs["temperature"]})
|
||
if attrs.get("swing_mode") is not None:
|
||
ha.call("climate", "set_swing_mode",
|
||
{"entity_id": entity_id, "swing_mode": attrs["swing_mode"]})
|
||
if wait:
|
||
ha.wait_state(entity_id, _original_predicate(original))
|
||
|
||
|
||
def _apply_step(
|
||
ha: HaRest, esphome: Optional[EsphomeChannel], entity_id: str, step: Step
|
||
) -> None:
|
||
if esphome is not None and esphome.apply(step):
|
||
return
|
||
ha.call(step.domain, step.service,
|
||
{"entity_id": entity_id, **step.service_data})
|
||
|
||
|
||
def _write_report(
|
||
path: str, rows: list[list], write_header: bool = True
|
||
) -> None:
|
||
mode = "w" if write_header else "a"
|
||
with open(path, mode, newline="", encoding="utf-8") as fh:
|
||
writer = csv.writer(fh)
|
||
if write_header:
|
||
writer.writerow(
|
||
["timestamp", "step", "expected", "observed", "result"]
|
||
)
|
||
writer.writerows(rows)
|
||
|
||
|
||
def run_quick(
|
||
ha: HaRest,
|
||
entity_id: str,
|
||
*,
|
||
burst: bool = False,
|
||
esphome: Optional[EsphomeChannel] = None,
|
||
settle_timeout: float = SETTLE_TIMEOUT,
|
||
) -> list[dict]:
|
||
original = ha.state(entity_id)
|
||
steps = build_steps(original)
|
||
if not steps:
|
||
raise AcceptanceError("climate-сущность не поддерживает ни одного шага")
|
||
results: list[dict] = []
|
||
try:
|
||
if burst and esphome is None:
|
||
for step in steps:
|
||
_apply_step(ha, esphome, entity_id, step)
|
||
for step in steps:
|
||
try:
|
||
state = ha.wait_state(
|
||
entity_id, step.predicate(), settle_timeout
|
||
)
|
||
observed = _observed(state, step)
|
||
except AcceptanceError as err:
|
||
observed = str(err)
|
||
results.append({
|
||
"step": step.name,
|
||
"expected": step.expected,
|
||
"observed": observed,
|
||
})
|
||
else:
|
||
for step in steps:
|
||
_apply_step(ha, esphome, entity_id, step)
|
||
try:
|
||
state = ha.wait_state(
|
||
entity_id, step.predicate(), settle_timeout
|
||
)
|
||
observed = _observed(state, step)
|
||
except AcceptanceError as err:
|
||
observed = str(err)
|
||
results.append({
|
||
"step": step.name,
|
||
"expected": step.expected,
|
||
"observed": observed,
|
||
})
|
||
finally:
|
||
_restore(ha, entity_id, original, wait=False)
|
||
try:
|
||
restored = ha.wait_state(
|
||
entity_id, _original_predicate(original), settle_timeout
|
||
)
|
||
except AcceptanceError:
|
||
restored = ha.state(entity_id)
|
||
results.append({
|
||
"step": "restore",
|
||
"expected": original.get("state"),
|
||
"observed": restored.get("state"),
|
||
})
|
||
return results
|
||
|
||
|
||
def run_long(
|
||
ha: HaRest,
|
||
entity_id: str,
|
||
*,
|
||
hours: float,
|
||
interval: float,
|
||
report_path: Optional[str],
|
||
esphome: Optional[EsphomeChannel] = None,
|
||
settle_timeout: float = SETTLE_TIMEOUT,
|
||
) -> tuple[int, int]:
|
||
original = ha.state(entity_id)
|
||
steps = build_steps(original)
|
||
if not steps:
|
||
raise AcceptanceError("climate-сущность не поддерживает ни одного шага")
|
||
iterations = max(1, int(hours * 3600 / interval))
|
||
ok = failed = 0
|
||
rows: list[list] = []
|
||
for index in range(iterations):
|
||
step = steps[index % len(steps)]
|
||
timestamp = dt.datetime.now(dt.timezone.utc).isoformat()
|
||
try:
|
||
_apply_step(ha, esphome, entity_id, step)
|
||
state = ha.wait_state(
|
||
entity_id, step.predicate(), settle_timeout
|
||
)
|
||
observed = _observed(state, step)
|
||
passed = _matches(observed, step.expected)
|
||
except AcceptanceError as err:
|
||
observed = str(err)
|
||
passed = False
|
||
finally:
|
||
try:
|
||
_restore(ha, entity_id, original)
|
||
except AcceptanceError as err:
|
||
observed = f"restore failed: {err}"
|
||
passed = False
|
||
ok += 1 if passed else 0
|
||
failed += 0 if passed else 1
|
||
row = [timestamp, step.name, step.expected, observed,
|
||
"ok" if passed else "fail"]
|
||
rows.append(row)
|
||
if report_path:
|
||
_write_report(report_path, [row], write_header=index == 0)
|
||
if index + 1 < iterations:
|
||
time.sleep(interval)
|
||
return ok, failed
|
||
|
||
|
||
def _build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
parser.add_argument("mode", nargs="?", choices=["quick", "long"],
|
||
default="quick")
|
||
parser.add_argument("--long", dest="long_mode", action="store_true",
|
||
help="алиас режима long")
|
||
parser.add_argument("--ha-url", required=True)
|
||
parser.add_argument("--ha-token", required=True)
|
||
parser.add_argument("--entity", required=True,
|
||
help="entity_id climate-сущности FGLair")
|
||
parser.add_argument("--burst", action="store_true",
|
||
help="quick: без пауз между изменениями (без ESPHome)")
|
||
parser.add_argument("--hours", type=float, default=24.0)
|
||
parser.add_argument("--interval", type=float, default=3600.0)
|
||
parser.add_argument("--report", help="CSV-отчёт (quick и long)")
|
||
parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
|
||
parser.add_argument("--settle-timeout", type=float,
|
||
default=SETTLE_TIMEOUT)
|
||
parser.add_argument("--esphome-host")
|
||
parser.add_argument("--esphome-key", default="")
|
||
return parser
|
||
|
||
|
||
def _make_esphome(args: argparse.Namespace) -> Optional[EsphomeChannel]:
|
||
if not args.esphome_host:
|
||
return None
|
||
channel = EsphomeChannel(args.esphome_host, args.esphome_key, args.timeout)
|
||
try:
|
||
channel.connect()
|
||
except AcceptanceError:
|
||
channel.close()
|
||
raise
|
||
except Exception as err: # noqa: BLE001
|
||
channel.close()
|
||
raise AcceptanceError(
|
||
f"ESPHome {args.esphome_host}: подключение не удалось: {err}"
|
||
) from err
|
||
return channel
|
||
|
||
|
||
def main(argv: Optional[list[str]] = None) -> int:
|
||
args = _build_parser().parse_args(argv)
|
||
if args.long_mode:
|
||
args.mode = "long"
|
||
esphome: Optional[EsphomeChannel] = None
|
||
try:
|
||
esphome = _make_esphome(args)
|
||
ha = HaRest(args.ha_url, args.ha_token, args.timeout)
|
||
if args.mode == "quick":
|
||
results = run_quick(
|
||
ha, args.entity, burst=args.burst, esphome=esphome,
|
||
settle_timeout=args.settle_timeout,
|
||
)
|
||
failed = [
|
||
row for row in results
|
||
if not _matches(row.get("observed"), row.get("expected"))
|
||
]
|
||
for row in results:
|
||
print(f"{row['step']}: expected={row['expected']} "
|
||
f"observed={row.get('observed', '-')}")
|
||
if args.report:
|
||
_write_report(args.report, [
|
||
[dt.datetime.now(dt.timezone.utc).isoformat(),
|
||
row["step"], row["expected"], row.get("observed"),
|
||
"ok" if _matches(row.get("observed"),
|
||
row.get("expected")) else "fail"]
|
||
for row in results
|
||
])
|
||
if failed:
|
||
print(f"ПРОВАЛЕНО: {len(failed)}", file=sys.stderr)
|
||
return 1
|
||
return 0
|
||
ok, failed = run_long(
|
||
ha, args.entity,
|
||
hours=args.hours, interval=args.interval,
|
||
report_path=args.report, esphome=esphome,
|
||
settle_timeout=args.settle_timeout,
|
||
)
|
||
print(f"long: ok={ok} fail={failed}"
|
||
+ (f", отчёт: {args.report}" if args.report else ""))
|
||
return 1 if failed else 0
|
||
except AcceptanceError as err:
|
||
print(f"приёмка: {err}", file=sys.stderr)
|
||
return 2
|
||
finally:
|
||
if esphome is not None:
|
||
esphome.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|