259 lines
9.3 KiB
Python
259 lines
9.3 KiB
Python
"""Пробное подключение при настройке: старт сессии, ONLINE ≤ timeout, чтение
|
||
базовых свойств и зондирование необязательных (presence).
|
||
|
||
Выполняется в executor (блокирующие вызовы ядра). Модуль отвечает только на
|
||
свойства из своей таблицы: отсутствие push = свойства нет (проверено на
|
||
AP-WC1E). Поэтому необязательные свойства не валят trial, а фиксируются в
|
||
``answered`` и используются как дефолты фич и для определения шаблона.
|
||
"""
|
||
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
|
||
from pyfglair.templates import template_info
|
||
|
||
# Обязательные свойства: есть во всех шаблонах, должны ответить.
|
||
REQUIRED_PROPS: tuple[Prop, ...] = (
|
||
Prop.OPERATION_MODE,
|
||
Prop.FAN_SPEED,
|
||
)
|
||
|
||
# Зондируемые свойства (в пределах выбранного шаблона): наличие в ответе —
|
||
# информация для дефолтов фич и распознавания шаблона.
|
||
OPTIONAL_PROPS: tuple[Prop, ...] = (
|
||
Prop.DEVICE_CAPABILITIES,
|
||
Prop.DISPLAY_TEMPERATURE,
|
||
Prop.AF_VERTICAL_DIRECTION,
|
||
Prop.AF_HORIZONTAL_DIRECTION,
|
||
Prop.AF_VERTICAL_NUM_DIR,
|
||
Prop.AF_HORIZONTAL_NUM_DIR,
|
||
Prop.AF_VERTICAL_SWING,
|
||
Prop.AF_HORIZONTAL_SWING,
|
||
Prop.HUMAN_DET_AUTO_SAVE,
|
||
Prop.WIFI_LED_ENABLE,
|
||
Prop.FILTER_SIGN_RESET_DISPLAY,
|
||
Prop.MONITOR1,
|
||
)
|
||
|
||
# Подписи, по которым распознаётся шаблон при неизвестной oem_model:
|
||
# B отвечает move_step1, F — monitor1/filter_sign_reset (у A их нет).
|
||
PROBE_PROPS: dict[Template, tuple[Prop, ...]] = {
|
||
Template.A: (
|
||
Prop.AF_VERTICAL_DIRECTION,
|
||
Prop.DISPLAY_TEMPERATURE,
|
||
Prop.FILTER_SIGN_RESET_DISPLAY,
|
||
),
|
||
Template.B: (
|
||
Prop.AF_VERTICAL_MOVE_STEP1,
|
||
Prop.AF_HORIZONTAL_MOVE_STEP1,
|
||
),
|
||
Template.F: (
|
||
Prop.MONITOR1,
|
||
Prop.FILTER_SIGN_RESET,
|
||
Prop.AF_VERTICAL_DIRECTION,
|
||
),
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class TrialResult:
|
||
ok: bool
|
||
reason: str = ""
|
||
state: State = State.IDLE
|
||
error: Error = Error.NONE
|
||
properties: dict[Prop, Value] = field(default_factory=dict)
|
||
answered: set[Prop] = field(default_factory=set)
|
||
|
||
|
||
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 classify_template(answered: set[Prop]) -> Template | None:
|
||
"""Шаблон по ответам зондов; None — недостаточно данных."""
|
||
if Prop.AF_VERTICAL_MOVE_STEP1 in answered and (
|
||
Prop.AF_VERTICAL_DIRECTION not in answered
|
||
):
|
||
return Template.B
|
||
if Prop.MONITOR1 in answered or Prop.FILTER_SIGN_RESET in answered:
|
||
if Prop.AF_VERTICAL_DIRECTION in answered:
|
||
return Template.F
|
||
if (
|
||
Prop.AF_VERTICAL_DIRECTION in answered
|
||
or Prop.DISPLAY_TEMPERATURE in answered
|
||
or Prop.FILTER_SIGN_RESET_DISPLAY in answered
|
||
):
|
||
return Template.A
|
||
return None
|
||
|
||
|
||
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] = REQUIRED_PROPS,
|
||
optional_props: Iterable[Prop] = OPTIONAL_PROPS,
|
||
props_timeout: float = 3.0,
|
||
) -> TrialResult:
|
||
"""Поднимает сессию, ждёт ONLINE, читает обязательные и зондирует
|
||
необязательные свойства.
|
||
|
||
Никогда не бросает: любая неудача — ``TrialResult(ok=False, reason=...)``.
|
||
"""
|
||
lock = threading.Lock()
|
||
box = {"state": State.IDLE, "error": Error.NONE}
|
||
values: dict[Prop, Value] = {}
|
||
valid = {info.prop for info in template_info(template)}
|
||
wanted = tuple(prop for prop in props if prop in valid)
|
||
optional = tuple(
|
||
prop for prop in optional_props if prop in valid and prop not in wanted
|
||
)
|
||
|
||
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
|
||
|
||
try:
|
||
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,
|
||
)
|
||
except Exception: # pragma: no cover - защитный путь
|
||
return TrialResult(ok=False, reason="start_failed")
|
||
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,
|
||
)
|
||
|
||
requested = wanted + optional
|
||
if requested:
|
||
session.batch_begin()
|
||
for prop in requested:
|
||
session.get_prop(prop)
|
||
session.batch_commit()
|
||
prop_deadline = min(
|
||
deadline, time.monotonic() + props_timeout
|
||
)
|
||
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)
|
||
missing = [prop for prop in wanted if prop not in snapshot]
|
||
if missing:
|
||
return TrialResult(
|
||
ok=False,
|
||
reason="properties_missing",
|
||
state=State.ONLINE,
|
||
properties=snapshot,
|
||
answered=set(snapshot),
|
||
)
|
||
return TrialResult(
|
||
ok=True,
|
||
state=State.ONLINE,
|
||
properties=snapshot,
|
||
answered=set(snapshot),
|
||
)
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
def probe_template(
|
||
host: str,
|
||
dsn: str,
|
||
lanip_key: str,
|
||
lanip_key_id: int,
|
||
*,
|
||
device_port: int = 80,
|
||
listen_port: int = 10275,
|
||
timeout: float = 10.0,
|
||
) -> tuple[Template | None, dict[Prop, Value], str | None]:
|
||
"""Определяет шаблон по ответам, перебирая A/B/F короткими сессиями.
|
||
|
||
Возвращает (template, properties, fatal_reason): если модуль недоступен
|
||
(или нет слотов/ключ не тот), перебор бессмыслен и причина возвращается
|
||
третьим элементом, чтобы вызывающий не открывал ещё одну сессию.
|
||
"""
|
||
answered: set[Prop] = set()
|
||
properties: dict[Prop, Value] = {}
|
||
for template in (Template.A, Template.B, Template.F):
|
||
result = trial_connect(
|
||
host, dsn, lanip_key, lanip_key_id,
|
||
device_port=device_port, template=template,
|
||
listen_port=listen_port, timeout=timeout,
|
||
props=PROBE_PROPS[template] + (Prop.OPERATION_MODE,),
|
||
optional_props=(),
|
||
props_timeout=2.0,
|
||
)
|
||
if not result.ok:
|
||
if result.reason == "properties_missing":
|
||
# Модуль доступен, но свойств шаблона нет — пробуем дальше.
|
||
continue
|
||
# Недоступен/нет слотов/ключ — перебор шаблонов бессмыслен.
|
||
return None, properties, result.reason
|
||
answered |= result.answered
|
||
properties.update(result.properties)
|
||
detected = classify_template(answered)
|
||
if detected is not None:
|
||
return detected, properties, None
|
||
return classify_template(answered), properties, None
|