"""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}")