90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""The HA Probe integration.
|
|
|
|
Receives pushes from a gerviba/systemmonitor probe instance over a webhook and
|
|
turns them into sensors. See hu.gerviba.probe.api.ApiService on the probe side
|
|
for the wire format: a POST of {"devices": [...], "entities": [...]} to
|
|
/api/webhook/<webhook_id>.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from aiohttp.hdrs import METH_POST
|
|
from aiohttp.web import Request, Response
|
|
|
|
from homeassistant.components import webhook
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
|
|
|
from .device import parse_device_meta, sync_device_registry
|
|
from .const import CONF_WEBHOOK_ID, DOMAIN, PLATFORMS, HaProbeRuntimeData, signal_entities
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
type HaProbeConfigEntry = ConfigEntry[HaProbeRuntimeData]
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: HaProbeConfigEntry) -> bool:
|
|
"""Set up a HA Probe config entry."""
|
|
entry.runtime_data = HaProbeRuntimeData()
|
|
webhook_id = entry.data[CONF_WEBHOOK_ID]
|
|
|
|
async def handle_webhook(
|
|
hass: HomeAssistant, webhook_id: str, request: Request
|
|
) -> Response:
|
|
"""Handle an incoming push from the probe."""
|
|
try:
|
|
payload = await request.json()
|
|
except ValueError:
|
|
_LOGGER.warning("Received invalid JSON on the HA Probe webhook")
|
|
return Response(status=400)
|
|
|
|
if not isinstance(payload, dict):
|
|
_LOGGER.warning("Received an unexpected HA Probe webhook payload")
|
|
return Response(status=400)
|
|
|
|
for device in payload.get("devices") or ():
|
|
ha_id = device.get("haId")
|
|
display_name = device.get("haDisplayName")
|
|
if ha_id and display_name:
|
|
entry.runtime_data.devices[ha_id] = display_name
|
|
meta = parse_device_meta(device)
|
|
entry.runtime_data.device_meta[ha_id] = meta
|
|
sync_device_registry(hass, entry.entry_id, ha_id, meta)
|
|
|
|
entities = payload.get("entities") or ()
|
|
if entities:
|
|
async_dispatcher_send(hass, signal_entities(entry.entry_id), entities)
|
|
|
|
return Response(status=200)
|
|
|
|
webhook.async_register(
|
|
hass,
|
|
DOMAIN,
|
|
entry.title,
|
|
webhook_id,
|
|
handle_webhook,
|
|
local_only=True,
|
|
allowed_methods=[METH_POST],
|
|
)
|
|
|
|
@callback
|
|
def _unregister_webhook() -> None:
|
|
webhook.async_unregister(hass, webhook_id)
|
|
|
|
entry.async_on_unload(_unregister_webhook)
|
|
|
|
entry.runtime_data.webhook_url = webhook.async_generate_url(hass, webhook_id)
|
|
_LOGGER.info("HA Probe webhook URL for %s: %s", entry.title, entry.runtime_data.webhook_url)
|
|
|
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
|
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: HaProbeConfigEntry) -> bool:
|
|
"""Unload a HA Probe config entry."""
|
|
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|