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.
279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""Firmware-Aktualisierung fuer ninstrip, samt WLAN-Sicherung.
|
|
|
|
Aufgebaut wie ha-ninlight. Zwei Unterschiede, beide aus der Firmware heraus:
|
|
|
|
* Die Platine steht in `device_type` (system/info), nicht in `product` - das
|
|
ist bei allen Platinen gleich. Firmware ohne dieses Feld bekommt gar kein
|
|
Angebot: ohne die Platine zu kennen, koennte ein K1Y-Abbild auf einer
|
|
KL-RGBC3 landen, und die invertiert das Datensignal nicht.
|
|
* Die WLAN-Pruefung sucht die SSID unverpackt im Abbild (siehe `firmware`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from typing import Any
|
|
|
|
from homeassistant.components.update import (
|
|
UpdateDeviceClass,
|
|
UpdateEntity,
|
|
UpdateEntityFeature,
|
|
)
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import issue_registry as ir
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
|
|
from .api import NinstripError
|
|
from .const import (
|
|
DOMAIN,
|
|
MAX_DEVICE_URL_LENGTH,
|
|
OTA_POLL_INTERVAL_S,
|
|
OTA_TIMEOUT_S,
|
|
SIGNAL_CATALOG_UPDATED,
|
|
)
|
|
from .coordinator import NinstripConfigEntry, NinstripCoordinator
|
|
from .entity import NinstripEntity
|
|
from .firmware import FirmwareImage
|
|
from .firmware import contains_ssid as _enthaelt_ssid
|
|
from .firmware import is_newer as _ist_neuer
|
|
from .http import device_base_url
|
|
from .shared import DATA_SHARED
|
|
from .urls import firmware_url
|
|
|
|
PARALLEL_UPDATES = 0 # mehrere Geraete duerfen gleichzeitig; jedes fuer sich ist gesperrt
|
|
|
|
|
|
def haengt_an_kompiliertem_wlan(settings: dict[str, Any]) -> bool:
|
|
"""Nur ein ausdrueckliches 'stored' macht ein Geraet vom Abbild unabhaengig."""
|
|
wifi = settings.get("wifi")
|
|
return not (isinstance(wifi, dict) and wifi.get("source") == "stored")
|
|
|
|
|
|
def _ssid(info: dict[str, Any]) -> str:
|
|
wifi = info.get("wifi")
|
|
return str(wifi.get("ssid") or "") if isinstance(wifi, dict) else ""
|
|
|
|
|
|
def _fehler(key: str, **placeholders: str) -> HomeAssistantError:
|
|
return HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key=key,
|
|
translation_placeholders=placeholders or None,
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: NinstripConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
async_add_entities([NinstripUpdateEntity(entry.runtime_data)])
|
|
|
|
|
|
class NinstripUpdateEntity(NinstripEntity, UpdateEntity):
|
|
"""Bietet das neueste passende Abbild aus dem Firmware-Ordner an."""
|
|
|
|
_attr_device_class = UpdateDeviceClass.FIRMWARE
|
|
_attr_supported_features = UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS
|
|
|
|
def __init__(self, coordinator: NinstripCoordinator) -> None:
|
|
super().__init__(coordinator, "firmware")
|
|
self._offer: FirmwareImage | None = None
|
|
self._evaluated_for: tuple[str, str, str] | None = None
|
|
self._eval_lock = asyncio.Lock()
|
|
self._install_lock = asyncio.Lock()
|
|
self._attr_in_progress = False
|
|
|
|
@property
|
|
def installed_version(self) -> str | None:
|
|
return self.coordinator.data.info.get("fw_version")
|
|
|
|
@property
|
|
def latest_version(self) -> str | None:
|
|
return self._offer.version if self._offer is not None else self.installed_version
|
|
|
|
def version_is_newer(self, latest_version: str, installed_version: str) -> bool:
|
|
"""Der eigene `git describe`-Vergleich statt HAs AwesomeVersion.
|
|
|
|
HA haelt "0.1.0-47-g71a57" und "0.1.0-36-gca9bb" fuer denselben
|
|
Versionskern und vergleicht dann die Zeichenketten - das faellt genau
|
|
dort falsch aus, wo diese Firmware lebt. `firmware.is_newer` hat die
|
|
Reihenfolge, nach der `self._offer` ueberhaupt erst entstanden ist;
|
|
so widersprechen sich Zustand und Angebot nie.
|
|
"""
|
|
return _ist_neuer(latest_version, installed_version)
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
await super().async_added_to_hass()
|
|
self.async_on_remove(
|
|
async_dispatcher_connect(self.hass, SIGNAL_CATALOG_UPDATED, self._pruefung_anstossen)
|
|
)
|
|
self._pruefung_anstossen()
|
|
|
|
@callback
|
|
def _pruefung_anstossen(self) -> None:
|
|
# Nie pruefen, waehrend eine Aktualisierung laeuft: das fragte ein
|
|
# gerade flashendes Geraet nach seinen Einstellungen und koennte
|
|
# Angebot und Hinweis unter der laufenden Aktualisierung wegziehen.
|
|
# `_async_install_locked` stoesst selbst genau eine Pruefung an, sobald
|
|
# die Merker wieder zurueckgesetzt sind.
|
|
if self._attr_in_progress:
|
|
return
|
|
self.coordinator.config_entry.async_create_task(
|
|
self.hass, self._async_pruefen(), name="ninstrip firmware evaluation"
|
|
)
|
|
|
|
@callback
|
|
def _handle_coordinator_update(self) -> None:
|
|
if not self._attr_in_progress and self._pruefschluessel() != self._evaluated_for:
|
|
self._pruefung_anstossen()
|
|
super()._handle_coordinator_update()
|
|
|
|
def _pruefschluessel(self) -> tuple[str, str, str]:
|
|
info = self.coordinator.data.info
|
|
return (str(info.get("device_type")), str(info.get("fw_version")), _ssid(info))
|
|
|
|
async def _async_pruefen(self) -> None:
|
|
async with self._eval_lock:
|
|
catalog = self.hass.data[DATA_SHARED].catalog
|
|
key = self._pruefschluessel()
|
|
device_type, installed, ssid = key
|
|
offer: FirmwareImage | None = None
|
|
blocked: FirmwareImage | None = None
|
|
# Firmware vor v0.1.0 meldet keinen device_type. Dann bleibt offen,
|
|
# fuer welche Platine sie gebaut wurde - also kein Angebot.
|
|
newer = (
|
|
catalog.newer_images(device_type, installed)
|
|
if self.coordinator.data.info.get("device_type")
|
|
else []
|
|
)
|
|
if newer:
|
|
try:
|
|
settings = await self.coordinator.async_call(lambda c: c.get_settings())
|
|
except HomeAssistantError:
|
|
return # das bisherige Angebot bleibt; der naechste Durchlauf versucht es
|
|
abhaengig = haengt_an_kompiliertem_wlan(settings)
|
|
for image in newer: # neuestes zuerst
|
|
if not abhaengig or await self.hass.async_add_executor_job(
|
|
catalog.contains_ssid, image, ssid
|
|
):
|
|
offer = image
|
|
break
|
|
if blocked is None:
|
|
blocked = image
|
|
self._offer = offer
|
|
self._evaluated_for = key
|
|
self._wlan_hinweis(blocked, ssid)
|
|
self.async_write_ha_state()
|
|
|
|
@callback
|
|
def _wlan_hinweis(self, blocked: FirmwareImage | None, ssid: str) -> None:
|
|
entry = self.coordinator.config_entry
|
|
issue_id = f"wifi_guard_{entry.entry_id}"
|
|
if blocked is None:
|
|
ir.async_delete_issue(self.hass, DOMAIN, issue_id)
|
|
return
|
|
ir.async_create_issue(
|
|
self.hass,
|
|
DOMAIN,
|
|
issue_id,
|
|
is_fixable=False,
|
|
severity=ir.IssueSeverity.WARNING,
|
|
translation_key="wifi_guard",
|
|
translation_placeholders={
|
|
"name": entry.title,
|
|
"version": blocked.version,
|
|
"device_type": blocked.device_type,
|
|
"ssid": ssid or "?",
|
|
},
|
|
)
|
|
|
|
async def async_install(self, version: str | None, backup: bool, **kwargs: Any) -> None:
|
|
# Nur der Aufruf, der diese Sperre haelt, darf unten in_progress und
|
|
# ota_in_progress anfassen: ohne diese Pruefung koennte ein zweiter
|
|
# Aufruf bis zu den Gegenproben durchlaufen - HAs eigene Sperre greift
|
|
# erst, wenn _attr_in_progress schon gesetzt ist - und sein `finally`
|
|
# wuerde die Merker zuruecksetzen, waehrend der erste noch flasht.
|
|
if self._install_lock.locked():
|
|
raise _fehler("update_busy")
|
|
async with self._install_lock:
|
|
await self._async_install_locked()
|
|
|
|
async def _async_install_locked(self) -> None:
|
|
shared = self.hass.data[DATA_SHARED]
|
|
image = self._offer
|
|
if image is None:
|
|
raise _fehler("no_update")
|
|
base = device_base_url(self.hass, self.coordinator.config_entry.options)
|
|
# Die Bytes hier EINMAL lesen und genau diese pruefen: die Ausgabe
|
|
# liefert dieselbe Kopie aus dem Speicher aus. Was danach mit der Datei
|
|
# passiert - etwa ein Bau, der sie mittendrin ueberschreibt - aendert
|
|
# nichts mehr an dem, was beim Geraet ankommt.
|
|
data = await self.hass.async_add_executor_job(shared.catalog.verify_and_read, image)
|
|
if data is None:
|
|
raise _fehler("image_invalid")
|
|
|
|
# Unmittelbar vor dem Senden noch einmal nachsehen: die Firmware des
|
|
# Geraets kann sich seit der Pruefung geaendert haben (etwa ein altes
|
|
# Angebot, das eine fehlgeschlagene Pruefung ueberlebt hat) und darf
|
|
# dann nicht mehr eingespielt werden.
|
|
info = await self.coordinator.async_call(lambda c: c.get_info())
|
|
if not _ist_neuer(image.version, str(info.get("fw_version"))):
|
|
raise _fehler("no_update")
|
|
settings = await self.coordinator.async_call(lambda c: c.get_settings())
|
|
if info.get("device_type") != image.device_type:
|
|
raise _fehler("image_invalid")
|
|
ssid = _ssid(info)
|
|
if haengt_an_kompiliertem_wlan(settings) and not await self.hass.async_add_executor_job(
|
|
_enthaelt_ssid, data, ssid
|
|
):
|
|
raise _fehler("wifi_guard", ssid=ssid or "?")
|
|
|
|
key = shared.registry.create(data)
|
|
url = firmware_url(base, key)
|
|
if len(url) > MAX_DEVICE_URL_LENGTH:
|
|
shared.registry.revoke(key)
|
|
raise _fehler("device_url_too_long")
|
|
|
|
self._attr_in_progress = True
|
|
self.coordinator.ota_in_progress = True
|
|
self.async_write_ha_state()
|
|
try:
|
|
await self.coordinator.async_call(lambda c: c.fw_update(url))
|
|
await self._async_warten_auf(image.version, info.get("uptime"))
|
|
finally:
|
|
shared.registry.revoke(key)
|
|
self._attr_in_progress = False
|
|
self.coordinator.ota_in_progress = False
|
|
await self.coordinator.async_refresh()
|
|
self.async_write_ha_state()
|
|
self._pruefung_anstossen()
|
|
|
|
async def _async_warten_auf(self, target: str, start_uptime: Any) -> None:
|
|
"""Wartet auf die Zielversion, einen Fehler oder einen Neustart mit der alten.
|
|
|
|
Die Firmware loescht last_ota_error, wenn eine Aktualisierung beginnt
|
|
(nl_ota.c) - was hier auftaucht, gehoert also zu diesem Versuch.
|
|
"""
|
|
deadline = time.monotonic() + OTA_TIMEOUT_S
|
|
while True:
|
|
await asyncio.sleep(OTA_POLL_INTERVAL_S)
|
|
if time.monotonic() >= deadline:
|
|
raise _fehler("update_timeout")
|
|
try:
|
|
async with self.coordinator.lock:
|
|
info = await self.coordinator.client.get_info()
|
|
except NinstripError:
|
|
continue # flasht oder startet gerade neu
|
|
version = str(info.get("fw_version"))
|
|
if version == target:
|
|
return
|
|
if error := info.get("last_ota_error"):
|
|
raise _fehler("update_failed", reason=str(error))
|
|
uptime = info.get("uptime")
|
|
if isinstance(uptime, int) and isinstance(start_uptime, int) and uptime < start_uptime:
|
|
raise _fehler("update_failed", reason=f"neu gestartet mit {version}")
|