fgl-aircon/custom_components/fglair/trial.py

"""Пробное подключение при настройке: старт сессии, ONLINE ≤ timeout, чтение
базовых свойств. Выполняется в executor (блокирующие вызовы ядра).
"""
from __future__ import annotations

import threading
import time
from dataclasses import dataclass, field
from typing import Iterable

from pyfglair import Config, Error, Prop, Session, State, Template, Value

BASE_PROPS: tuple[Prop, ...] = (
    Prop.OPERATION_MODE,
    Prop.FAN_SPEED,
    Prop.DISPLAY_TEMPERATURE,
)


@dataclass
class TrialResult:
    ok: bool
    reason: str = ""
    state: State = State.IDLE
    error: Error = Error.NONE
    properties: dict[Prop, Value] = field(default_factory=dict)


def _failure_reason(state: State, error: Error) -> str:
    if state == State.KEY_ERROR or error == Error.KEY_MISMATCH:
        return "key_mismatch"
    if error == Error.NO_SLOT:
        return "no_slots"
    if error in (Error.UNREACHABLE, Error.ACTIVATION_TIMEOUT):
        return "unreachable"
    return "timeout"


def trial_connect(
    host: str,
    dsn: str,
    lanip_key: str,
    lanip_key_id: int,
    *,
    device_port: int = 80,
    template: Template = Template.A,
    listen_port: int = 10275,
    keepalive_ms: int = 15000,
    timeout: float = 10.0,
    props: Iterable[Prop] = BASE_PROPS,
) -> TrialResult:
    """Поднимает сессию, ждёт ONLINE и читает базовые свойства.

    Никогда не бросает: любая неудача — ``TrialResult(ok=False, reason=...)``.
    """
    lock = threading.Lock()
    box = {"state": State.IDLE, "error": Error.NONE}
    values: dict[Prop, Value] = {}
    wanted = tuple(props)

    def on_state(state: State, error: Error) -> None:
        with lock:
            box["state"] = state
            box["error"] = error

    def on_property(event) -> None:
        with lock:
            values[event.prop] = event.value

    session = Session(
        Config(
            host=host,
            device_port=device_port,
            dsn=dsn,
            lanip_key=lanip_key,
            lanip_key_id=lanip_key_id,
            template=template,
            listen_port=listen_port,
            keepalive_ms=keepalive_ms,
        ),
        loop=None,
        on_state=on_state,
        on_property=on_property,
    )
    try:
        try:
            started = session.start()
        except Exception:  # pragma: no cover - защитный путь
            return TrialResult(ok=False, reason="start_failed")
        if not started:
            return TrialResult(ok=False, reason="start_failed")

        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            with lock:
                state, error = box["state"], box["error"]
            if state == State.ONLINE:
                break
            if state == State.KEY_ERROR:
                return TrialResult(
                    ok=False, reason=_failure_reason(state, error),
                    state=state, error=error,
                )
            time.sleep(0.05)
        else:
            with lock:
                state, error = box["state"], box["error"]
            return TrialResult(
                ok=False, reason=_failure_reason(state, error),
                state=state, error=error,
            )

        session.batch_begin()
        for prop in wanted:
            session.get_prop(prop)
        session.batch_commit()

        prop_deadline = min(deadline, time.monotonic() + 2.0)
        while time.monotonic() < prop_deadline:
            with lock:
                if all(prop in values for prop in wanted):
                    break
            time.sleep(0.05)

        with lock:
            snapshot = dict(values)
        return TrialResult(ok=True, state=State.ONLINE, properties=snapshot)
    finally:
        session.close()