Petr Polezhaev
b21817ab9f
- PLAN_CORE: монорепозиторий (CMakeLists в корне; components/fglair,
custom_components/fglair, include/fgl-aircon, src/{ayla,aircon},
src/ayla/platform); логическое разделение ayla (протокол+цикл) /
aircon (конверсии+шаблоны+API); тесты зеркалят слои (tests/{ayla,aircon});
конверсии: шаблон / линейные коэффициенты / функция-указатель
(лямбды ESPHome); оценка httpd/json (jsmn вендор, свой мини-httpd на
BSD-сокетах, conan не нужен); README библиотеки в M4.
- PLAN_ESPHOME: host вместо ip_address (DNS + Ayla-mDNS :10276),
секреты в примерах, кастомные конверсии через !lambda, advanced-пример
(триггеры режимов + LVGL с пропусками), README-план, скрипт приёмки
(aioesphomeapi + HA REST).
- PLAN_HOME_ASSISTANT: шаг config flow с превью рассчитанных значений
шаблона (через cffi в C-ядро, без дублей), README с HACS-инструкцией
и заглушками под скриншоты с описаниями, приёмка (long-lived token).
- Убраны реальные dsn/ip/lanip_key/key_id из примеров; probe_mdns.py
принимает DSN аргументом.
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Пассивная mDNS-проба модуля кондиционера: A-запрос <DSN>.local
|
|
на 224.0.0.251:5353 и :10276 (нестандартный порт Ayla из APK).
|
|
Модуль отвечает ТОЛЬКО на :10276 (проверено на приборе).
|
|
|
|
Использование: python probe_mdns.py <DSN> (например AC000W00ABCD1234)"""
|
|
import socket, struct, sys, time
|
|
|
|
def qname(name):
|
|
out = b""
|
|
for lbl in name.split("."):
|
|
out += bytes([len(lbl)]) + lbl.encode()
|
|
return out + b"\x00"
|
|
|
|
def query_packet(txid):
|
|
header = struct.pack(">HHHHHH", txid, 0x0100, 1, 0, 0, 0)
|
|
return header + qname(HOST) + struct.pack(">HH", 1, 1)
|
|
|
|
def parse_name(buf, off):
|
|
parts = []
|
|
while True:
|
|
l = buf[off]
|
|
if l == 0:
|
|
off += 1
|
|
break
|
|
if l & 0xC0 == 0xC0:
|
|
ptr = struct.unpack(">H", buf[off:off+2])[0] & 0x3FFF
|
|
parts.append(parse_name(buf, ptr)[0])
|
|
off += 2
|
|
break
|
|
parts.append(buf[off+1:off+1+l].decode("latin1"))
|
|
off += 1 + l
|
|
return ".".join(parts), off
|
|
|
|
def parse_answers(buf):
|
|
try:
|
|
qd = struct.unpack(">H", buf[4:6])[0]
|
|
an = struct.unpack(">H", buf[6:8])[0]
|
|
off = 12
|
|
for _ in range(qd):
|
|
_, off = parse_name(buf, off)
|
|
off += 4
|
|
found = []
|
|
for _ in range(an):
|
|
name, off = parse_name(buf, off)
|
|
rtype, rclass, ttl, rdlen = struct.unpack(">HHIH", buf[off:off+10])
|
|
off += 10
|
|
rdata = buf[off:off+rdlen]
|
|
if rtype == 1 and rdlen == 4:
|
|
found.append((name, ".".join(str(b) for b in rdata), rclass, ttl))
|
|
off += rdlen
|
|
return found
|
|
except Exception as e:
|
|
return [("parse-error", str(e), 0, 0)]
|
|
|
|
def probe(port, txid, wait=4.0):
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
s.bind(("", 0))
|
|
pkt = query_packet(txid)
|
|
for _ in range(3):
|
|
s.sendto(pkt, ("224.0.0.251", port))
|
|
time.sleep(0.5)
|
|
s.settimeout(wait)
|
|
try:
|
|
while True:
|
|
data, addr = s.recvfrom(4096)
|
|
print(f" [:{port}] from {addr}: {len(data)} bytes")
|
|
for rec in parse_answers(data):
|
|
print(f" {rec}")
|
|
except socket.timeout:
|
|
pass
|
|
finally:
|
|
s.close()
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print(__doc__); sys.exit(1)
|
|
HOST = sys.argv[1] + ".local"
|
|
print("mDNS A?", HOST)
|
|
print("-- querying :5353")
|
|
probe(5353, 0x1234)
|
|
print("-- querying :10276")
|
|
probe(10276, 0x1235)
|