93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
"""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)
|