Michael Sauer
2c77b2d325
Aufgebaut wie ha-ninlight: die Abbilder liegen unter ninstrip_firmware im Konfigurationsverzeichnis, Home Assistant liefert sie selbst unter einer einmaligen Adresse aus, und zwar genau die Bytes, die vorher geprueft wurden. Dazu firmware.py (Anhaenger, Versionsreihenfolge, WLAN-Pruefung, Katalog), http.py, urls.py und shared.py. Drei Unterschiede zu ninlight, alle aus der Firmware heraus: * Verglichen wird device_type - die Platine -, nicht product. Letzteres ist bei allen Platinen 'ninstrip'; die Abbilder unterscheiden sich aber hart, die K1Y invertiert das Datensignal und die KL-RGBC3 nicht. Firmware ohne dieses Feld bekommt gar kein Angebot. * Die WLAN-Pruefung sucht die SSID unverpackt: ein ESP32-Abbild ist nicht gzip-verpackt wie das des W600. An einem mit sdkconfig.local gebauten Abbild gegengeprueft - die SSID steht genau einmal NUL-gerahmt drin. * Der Fehlerschluessel heisst device_url_too_long statt lamp_url_too_long. Dazu das Testgeruest, das es hier bisher gar nicht gab: 204 Tests in tests_unit/ (ohne HA-Instanz) und tests/ (mit), Abdeckung 97 %, dazu scripts/check.sh als Ersatz fuer die fehlende CI. Die Nutzdaten in tests_unit/samples.py sind an der Stehlampe abgefragt. Beim Gegenpruefen per Mutation fiel auf, dass die Platinenpruefung unmittelbar vor dem Senden ungeprueft war - der Test dazu ist jetzt da und faellt, wenn man die Pruefung entfernt.
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""Was sich alle ninstrip-Eintraege teilen: der Abbild-Katalog und die einmaligen Schluessel."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|
from homeassistant.helpers.event import async_track_time_interval
|
|
from homeassistant.util.hass_dict import HassKey
|
|
|
|
from .const import CATALOG_SCAN_INTERVAL, DOMAIN, FIRMWARE_DIR_NAME, SIGNAL_CATALOG_UPDATED
|
|
from .firmware import FirmwareCatalog
|
|
from .http import NinstripFirmwareView, OtaRegistry
|
|
|
|
|
|
def firmware_directory(hass: HomeAssistant) -> Path:
|
|
return Path(hass.config.path(FIRMWARE_DIR_NAME))
|
|
|
|
|
|
@dataclass
|
|
class NinstripShared:
|
|
"""Einmal je Home Assistant."""
|
|
|
|
catalog: FirmwareCatalog
|
|
registry: OtaRegistry
|
|
scan_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
|
unsub_scan: CALLBACK_TYPE | None = None
|
|
|
|
|
|
DATA_SHARED: HassKey[NinstripShared] = HassKey(DOMAIN)
|
|
|
|
|
|
@callback
|
|
def async_get_shared(hass: HomeAssistant) -> NinstripShared:
|
|
"""Beim ersten Zugriff anlegen und die Ausgabe genau einmal anmelden."""
|
|
shared = hass.data.get(DATA_SHARED)
|
|
if shared is None:
|
|
shared = NinstripShared(FirmwareCatalog(firmware_directory(hass)), OtaRegistry())
|
|
hass.data[DATA_SHARED] = shared
|
|
hass.http.register_view(NinstripFirmwareView(hass, shared.registry))
|
|
return shared
|
|
|
|
|
|
async def async_scan_catalog(hass: HomeAssistant, shared: NinstripShared) -> None:
|
|
async with shared.scan_lock:
|
|
await hass.async_add_executor_job(shared.catalog.scan)
|
|
async_dispatcher_send(hass, SIGNAL_CATALOG_UPDATED)
|
|
|
|
|
|
async def async_start_catalog(hass: HomeAssistant, shared: NinstripShared) -> None:
|
|
if shared.unsub_scan is not None:
|
|
return
|
|
|
|
async def _turnus(_now: datetime) -> None:
|
|
await async_scan_catalog(hass, shared)
|
|
|
|
shared.unsub_scan = async_track_time_interval(
|
|
hass,
|
|
_turnus,
|
|
CATALOG_SCAN_INTERVAL,
|
|
name="ninstrip firmware catalog",
|
|
cancel_on_shutdown=True,
|
|
)
|
|
await async_scan_catalog(hass, shared)
|
|
|
|
|
|
@callback
|
|
def async_stop_catalog(shared: NinstripShared) -> None:
|
|
if shared.unsub_scan is not None:
|
|
shared.unsub_scan()
|
|
shared.unsub_scan = None
|