Add missing files

custom_components/ha_probe/device.py

@@ -0,0 +1,92 @@
+"""Device metadata helpers for the HA Probe integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.device_registry import DeviceInfo
+
+from .const import DOMAIN
+
+DEFAULT_MANUFACTURER = "gerviba/systemmonitor"
+DEFAULT_MODEL = "Probe"
+
+# Probe payload key -> normalized metadata key.
+_PAYLOAD_KEYS = {
+    "manufacturer": "manufacturer",
+    "model": "model",
+    "configurationUrl": "configuration_url",
+    "swVersion": "sw_version",
+    "viaDevice": "via_device",
+}
+
+
+def parse_device_meta(device: dict[str, Any]) -> dict[str, str]:
+    """Pick the optional metadata fields a probe device record carries."""
+    return {
+        key: value
+        for payload_key, key in _PAYLOAD_KEYS.items()
+        if isinstance(value := device.get(payload_key), str) and value
+    }
+
+
+def device_identifier(entry_id: str, device_ha_id: str) -> tuple[str, str]:
+    """Registry identifier of a probe device."""
+    return (DOMAIN, f"{entry_id}_{device_ha_id}")
+
+
+def build_device_info(
+    entry_id: str, device_ha_id: str, name: str, meta: dict[str, str]
+) -> DeviceInfo:
+    """DeviceInfo for a probe device, using whatever metadata is known."""
+    info = DeviceInfo(
+        identifiers={device_identifier(entry_id, device_ha_id)},
+        name=name,
+        manufacturer=meta.get("manufacturer") or DEFAULT_MANUFACTURER,
+        model=meta.get("model") or DEFAULT_MODEL,
+    )
+    if sw_version := meta.get("sw_version"):
+        info["sw_version"] = sw_version
+    if configuration_url := meta.get("configuration_url"):
+        info["configuration_url"] = configuration_url
+    if via_device := meta.get("via_device"):
+        info["via_device"] = device_identifier(entry_id, via_device)
+    return info
+
+
+def sync_device_registry(
+    hass: HomeAssistant,
+    entry_id: str,
+    device_ha_id: str,
+    meta: dict[str, str],
+) -> None:
+    """Push changed metadata (e.g. a new docker image) to an existing device.
+
+    Entities only hand their DeviceInfo to HA when first created, so later
+    changes from the probe have to be applied to the registry directly.
+    """
+    registry = dr.async_get(hass)
+    device = registry.async_get_device(
+        identifiers={device_identifier(entry_id, device_ha_id)}
+    )
+    if device is None:
+        return
+
+    changes: dict[str, Any] = {
+        "manufacturer": meta.get("manufacturer") or DEFAULT_MANUFACTURER,
+        "model": meta.get("model") or DEFAULT_MODEL,
+    }
+    if sw_version := meta.get("sw_version"):
+        changes["sw_version"] = sw_version
+    if configuration_url := meta.get("configuration_url"):
+        changes["configuration_url"] = configuration_url
+    if via_device := meta.get("via_device"):
+        parent = registry.async_get_device(
+            identifiers={device_identifier(entry_id, via_device)}
+        )
+        if parent is not None:
+            changes["via_device_id"] = parent.id
+
+    registry.async_update_device(device.id, **changes)