"""Authoritative presence-radar coordinator (#485). The coordinator is deliberately the only component which interprets radar entities. Browsers, recordings and room outputs consume the same normalized frame so freshness and coordinate semantics cannot drift between features. """ from __future__ import annotations import asyncio import hashlib import json import math import secrets from collections.abc import Callable from datetime import UTC, datetime from typing import Any import voluptuous as vol from homeassistant.core import Event, HomeAssistant, callback from homeassistant.helpers.event import ( async_call_later, async_track_state_change_event, async_track_state_report_event, ) from .radar_geometry import ( clipped_arc_segments, length_cm, point_in_polygon, polar_to_cartesian, polygon_is_convex, project_local, ) from .radar_validation import ( radar_registry_evidence, radar_source_entity_ids, validate_radar_draft, ) from .store import HouseplanData MAX_FRAME_HZ = 4 PAIR_MAX_AGE_S = 3.0 PAIR_MAX_SKEW_S = 1.5 _EXPLICIT_ABSENT = object() def _fingerprint(value: Any) -> str: payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(payload.encode()).hexdigest()[:24] def _reported_timestamp(state: Any) -> float: value = getattr(state, "last_reported", None) or getattr(state, "last_updated", None) if value is None: return 0.0 if value.tzinfo is None: value = value.replace(tzinfo=UTC) return value.timestamp() def _numeric_state(state: Any) -> float | None: if state is None or str(state.state).strip().lower() in {"", "unknown", "unavailable"}: return None try: value = float(state.state) except (TypeError, ValueError): return None return value if math.isfinite(value) else None def _binary_state(state: Any) -> bool | None: if state is None: return None value = str(state.state).lower() if value == "on": return True if value == "off": return False return None class RadarCoordinator: """One bounded state-report listener graph for the whole integration.""" def __init__(self, hass: HomeAssistant, runtime: HouseplanData) -> None: self.hass = hass self.runtime = runtime self.server_session_id = secrets.token_hex(12) self.closed = False self.config_rev = 0 self.config: dict[str, Any] = {} self.radars: dict[str, dict[str, Any]] = {} self._seq: dict[str, int] = {} self._frames: dict[str, dict[str, Any]] = {} self._last_signature: dict[str, str] = {} self._listeners: set[Callable[[str, dict[str, Any]], None]] = set() self._internal_listeners: set[Callable[[str, dict[str, Any]], None]] = set() self._external_cleanups: set[Callable[[], None]] = set() self._unsub_sources: list[Callable[[], None]] = [] self._unsub_config: Callable[[], None] | None = None self._unsub_tick: Callable[[], None] | None = None self._pending: asyncio.TimerHandle | None = None self._last_publish_monotonic = 0.0 self._immediate_source_ids: set[str] = set() self._last_acl_epoch = -1 self._lock = asyncio.Lock() async def async_setup(self) -> None: await self.async_refresh() if self.closed: return @callback def config_updated(_event: Event) -> None: self.hass.async_create_task(self.async_refresh()) self._unsub_config = self.hass.bus.async_listen( "houseplan_config_updated", config_updated ) async def async_refresh(self) -> None: async with self._lock: if self.closed: return stored = await self.runtime.config_store.async_load() or {} if self.closed: return config = stored.get("config") or {} registry = radar_registry_evidence(self.hass) self.config = config self.config_rev = int(stored.get("rev", 0)) self.radars = {} for marker in config.get("markers") or []: if not isinstance(marker, dict) or not isinstance(marker.get("radar"), dict) \ or marker["radar"].get("version") != 1 \ or marker["radar"].get("enabled") is not True \ or marker.get("removed") is True: continue marker_id = str(marker.get("id")) try: validate_radar_draft(config, marker_id, marker["radar"], registry) except vol.Invalid: # Change-aware config compatibility may retain an old or # future-invalid block. It stays lossless but never runs. continue self.radars[marker_id] = marker self._resubscribe_sources() self._publish_all(force=True) def teardown(self) -> None: self.closed = True if self._pending: self._pending.cancel() self._pending = None if self._unsub_tick: self._unsub_tick() self._unsub_tick = None if self._unsub_config: self._unsub_config() self._unsub_config = None for unsub in self._unsub_sources: unsub() self._unsub_sources.clear() # Setup subscriptions own draft-specific HA listeners which are not # part of the saved source graph. Closing the config entry must tear # those down too, even if the browser WebSocket stays connected. for cleanup in tuple(self._external_cleanups): cleanup() self._external_cleanups.clear() self._listeners.clear() self._internal_listeners.clear() self._frames.clear() @callback def add_external_cleanup(self, cleanup: Callable[[], None]) -> Callable[[], None]: """Own a setup listener for the lifetime of this coordinator.""" self._external_cleanups.add(cleanup) @callback def unregister() -> None: self._external_cleanups.discard(cleanup) return unregister def _resubscribe_sources(self) -> None: for unsub in self._unsub_sources: unsub() self._unsub_sources.clear() entity_ids = sorted({ entity_id for marker in self.radars.values() for entity_id in radar_source_entity_ids(marker.get("radar")) }) if not entity_ids: self._immediate_source_ids.clear() return immediate: set[str] = set() for marker in self.radars.values(): sources = marker.get("radar", {}).get("sources") or {} immediate.update( value for key in ("occupancy_entity", "count_entity", "availability_entity") if isinstance((value := sources.get(key)), str) ) for group in ("slots", "ranges"): immediate.update( item["presence_entity"] for item in sources.get(group) or [] if isinstance(item, dict) and isinstance(item.get("presence_entity"), str) ) immediate.update( item["entity_id"] for item in sources.get("zones") or [] if isinstance(item, dict) and isinstance(item.get("entity_id"), str) ) self._immediate_source_ids = immediate @callback def source_reported(event: Event) -> None: entity_id = str(event.data.get("entity_id") or "") state = event.data.get("new_state") immediate_transition = entity_id in self._immediate_source_ids \ or state is None or str(getattr(state, "state", "")).lower() \ in {"", "unknown", "unavailable"} self._schedule_publish(immediate=immediate_transition) # state_reported catches repeated coordinate values. state_changed is # retained for HA versions/integrations which do not emit a report event # for every change; one 250 ms coalescer deduplicates the overlap and # enforces the public four-frames-per-second ceiling. self._unsub_sources.append(async_track_state_report_event( self.hass, entity_ids, source_reported )) self._unsub_sources.append(async_track_state_change_event( self.hass, entity_ids, source_reported )) @callback def _schedule_publish(self, *, immediate: bool = False) -> None: if self.closed: return if self._pending is not None: if not immediate: return # Availability and occupancy edges must not wait behind an already # queued coordinate batch. Replacing that handle remains bounded: # the two HA event streams still collapse into this one callback. self._pending.cancel() self._pending = None elapsed = self.hass.loop.time() - self._last_publish_monotonic delay = 0 if immediate else max(0.1, 1 / MAX_FRAME_HZ - elapsed) self._pending = self.hass.loop.call_later(delay, self._flush_pending) @callback def _flush_pending(self) -> None: self._pending = None self._last_publish_monotonic = self.hass.loop.time() self._publish_all() def _ensure_tick(self) -> None: if self._unsub_tick or not (self._listeners or self._internal_listeners): return @callback def tick(_now: datetime) -> None: self._unsub_tick = None self._publish_all() self._ensure_tick() self._unsub_tick = async_call_later(self.hass, 1.0, tick) def add_listener( self, listener: Callable[[str, dict[str, Any]], None], *, publish_initial: bool = True, ) -> Callable[[], None]: self._listeners.add(listener) self._ensure_tick() if publish_initial: for marker_id, frame in self._frames.items(): listener(marker_id, self.public_frame(frame)) @callback def remove() -> None: self._listeners.discard(listener) self._stop_tick_if_idle() return remove def add_internal_listener(self, listener: Callable[[str, dict[str, Any]], None]) -> Callable[[], None]: self._internal_listeners.add(listener) self._ensure_tick() @callback def remove() -> None: self._internal_listeners.discard(listener) self._stop_tick_if_idle() return remove def _stop_tick_if_idle(self) -> None: if not (self._listeners or self._internal_listeners) and self._unsub_tick: self._unsub_tick() self._unsub_tick = None def space_for_marker(self, marker_id: str) -> str | None: marker = self.radars.get(marker_id) return str(marker.get("space")) if marker and marker.get("space") else None def has_space(self, space_id: str) -> bool: return any(str(space.get("id")) == space_id for space in self.config.get("spaces") or []) def marker_config(self, marker_id: str) -> dict[str, Any] | None: return next( (marker for marker in self.config.get("markers") or [] if isinstance(marker, dict) and str(marker.get("id")) == marker_id), None, ) def source_ids(self, marker_id: str) -> set[str]: marker = self.radars.get(marker_id) return radar_source_entity_ids(marker.get("radar")) if marker else set() def frame(self, marker_id: str) -> dict[str, Any] | None: return self._frames.get(marker_id) def frames_for_space(self, space_id: str) -> list[dict[str, Any]]: return [frame for marker_id, frame in self._frames.items() if self.space_for_marker(marker_id) == space_id][:32] def _publish_all(self, force: bool = False) -> None: now = datetime.now(UTC).timestamp() acl_epoch = int(now // 60) force = force or acl_epoch != self._last_acl_epoch self._last_acl_epoch = acl_epoch live_ids = set(self.radars) for stale_id in set(self._frames) - live_ids: self._frames.pop(stale_id, None) self._last_signature.pop(stale_id, None) clear = self._base_frame(stale_id, now, "disabled") self._deliver(stale_id, clear) for marker_id, marker in self.radars.items(): frame = self._build_frame(marker_id, marker, now) self._annotate_smoothing(marker_id, marker, frame) signature = _fingerprint({key: value for key, value in frame.items() if key not in {"seq", "reported_at", "expires_at"}}) if not force and signature == self._last_signature.get(marker_id): continue self._last_signature[marker_id] = signature self._frames[marker_id] = frame self._deliver(marker_id, frame) def _annotate_smoothing( self, marker_id: str, marker: dict[str, Any], frame: dict[str, Any], ) -> None: """Permit a cosmetic transition only where it cannot invent travel.""" previous = self._frames.get(marker_id) if not previous or previous.get("health") not in {"ok", "stale"} \ or frame.get("health") not in {"ok", "stale"} \ or previous.get("source_generation") != frame.get("source_generation") \ or previous.get("calibration_revision") != frame.get("calibration_revision") \ or previous.get("_filter_revision") != frame.get("_filter_revision"): return room_exists, room_poly = self._room_polygon(marker, marker["radar"]) if not room_exists or room_poly is not None and not polygon_is_convex(room_poly): return cell_cm = self._space_cell_cm(marker) if cell_cm is None: return old_targets = { target.get("slot"): target for target in previous.get("targets") or [] if target.get("included") is True } for target in frame.get("targets") or []: old = old_targets.get(target.get("slot")) if not old or target.get("included") is not True \ or float(target.get("reported_at", 0)) <= float(old.get("reported_at", 0)) \ or float(target.get("reported_at", 0)) > float(old.get("expires_at", 0)): continue distance_cm = math.hypot( float(target["x"]) - float(old["x"]), float(target["y"]) - float(old["y"]), ) * 240 * cell_cm if distance_cm <= 100: target["smooth"] = True def _deliver(self, marker_id: str, frame: dict[str, Any]) -> None: for listener in tuple(self._internal_listeners): listener(marker_id, frame) public = self.public_frame(frame) for listener in tuple(self._listeners): listener(marker_id, public) def _base_frame(self, marker_id: str, now: float, health: str) -> dict[str, Any]: self._seq[marker_id] = self._seq.get(marker_id, 0) + 1 return { "server_session_id": self.server_session_id, "marker_id": marker_id, "source_generation": "", "calibration_revision": "", "seq": self._seq[marker_id], "reported_at": now, "expires_at": now, "health": health, "reported_presence": None, "complete": False, "targets": [], "ranges": [], "zones": [], } def _build_frame(self, marker_id: str, marker: dict[str, Any], now: float) -> dict[str, Any]: radar = marker["radar"] frame = self._base_frame(marker_id, now, "ok") source_material = { "installation_id": radar.get("mount", {}).get("installation_id"), "mount": {key: radar.get("mount", {}).get(key) for key in ("x", "y")}, "profile": radar.get("profile"), "sources": radar.get("sources"), "space": marker.get("space"), } calibration_material = { "heading_deg": radar.get("mount", {}).get("heading_deg"), "calibration": radar.get("calibration"), } frame["source_generation"] = _fingerprint(source_material) frame["calibration_revision"] = _fingerprint(calibration_material) profile = radar.get("profile") room_exists, room_poly = self._room_polygon(marker, radar) frame["_filter_revision"] = _fingerprint({ "room_id": radar.get("room_id"), "polygon": room_poly, }) if not room_exists: frame["health"] = "needs_setup" return frame current_cell_cm = self._space_cell_cm(marker) projects_geometry = profile in { "esphome_ld2450_v1", "cartesian_v1", "polar_v1", "range_v1", } if projects_geometry and (current_cell_cm is None or not math.isclose( float(radar.get("calibration", {}).get("cell_cm", -1)), current_cell_cm, rel_tol=0, abs_tol=1e-9, )): frame["health"] = "needs_setup" return frame sources = radar.get("sources") or {} availability = _binary_state(self.hass.states.get(sources.get("availability_entity"))) \ if sources.get("availability_entity") else True occupancy = _binary_state(self.hass.states.get(sources.get("occupancy_entity"))) \ if sources.get("occupancy_entity") else None frame["reported_presence"] = occupancy if availability is not True: frame["health"] = "unavailable" if availability is None else "off" return frame if occupancy is False: frame["complete"] = True frame["expires_at"] = now + 60 frame["health"] = "off" return frame any_stale = False explicit_slots = 0 for slot in sources.get("slots") or []: try: local = self._slot_local(profile, slot, now) except (TypeError, ValueError): local = None if local is _EXPLICIT_ABSENT: explicit_slots += 1 continue if local is None: gate = _binary_state(self.hass.states.get(slot.get("presence_entity"))) \ if slot.get("presence_entity") else None if gate is False: explicit_slots += 1 else: any_stale = True continue x_cm, y_cm, t1, t2, quality = local gate = _binary_state(self.hass.states.get(slot.get("presence_entity"))) \ if slot.get("presence_entity") else True if gate is not True: if gate is False: explicit_slots += 1 else: any_stale = True continue try: point = project_local(radar["mount"], radar["calibration"], x_cm, y_cm) except (TypeError, ValueError): any_stale = True continue tolerance = 0.1 / (240 * current_cell_cm) if current_cell_cm else 1e-6 included = room_poly is None or point_in_polygon(point, room_poly, tolerance) frame["targets"].append({ "slot": str(slot.get("id")), "x": point[0], "y": point[1], "reported_at": max(t1, t2), "expires_at": min(t1, t2) + PAIR_MAX_AGE_S, "pair_quality": quality, "included": included, "reason": None if included else "outside_room", # Private values are removed by public_frame and consumed only # by the bounded local recording/analysis services. "_local_x_cm": x_cm, "_local_y_cm": y_cm, "_component_times": [t1, t2], }) explicit_slots += 1 for source in sources.get("ranges") or []: state = self.hass.states.get(source.get("entity_id")) value = _numeric_state(state) reported = _reported_timestamp(state) gate = _binary_state(self.hass.states.get(source.get("presence_entity"))) \ if source.get("presence_entity") else True if value is None or now - reported > PAIR_MAX_AGE_S or gate is not True: any_stale = any_stale or gate is not False continue try: radius_cm = length_cm(value, source.get("unit")) except (TypeError, ValueError): any_stale = True continue radius = radius_cm / (240 * radar["calibration"]["cell_cm"]) segments = clipped_arc_segments( (radar["mount"]["x"], radar["mount"]["y"]), radius, radar["mount"].get("heading_deg", 0), radar["mount"].get("fov_deg"), room_poly, ) frame["ranges"].append({ "id": str(source.get("id")), "x": radar["mount"]["x"], "y": radar["mount"]["y"], "radius": radius, "heading_deg": radar["mount"].get("heading_deg", 0), "fov_deg": radar["mount"].get("fov_deg"), "segments": segments, "reported_at": reported, "expires_at": reported + PAIR_MAX_AGE_S, }) for source in sources.get("zones") or []: state = self.hass.states.get(source.get("entity_id")) value = _binary_state(state) if source.get("kind") == "occupancy" else _numeric_state(state) frame["zones"].append({"id": str(source.get("id")), "state": value}) if profile == "presence_v1": frame["complete"] = occupancy is not None frame["health"] = "ok" if occupancy is not None else "unknown" frame["expires_at"] = None return frame if profile == "zones_v1": frame["complete"] = all(zone["state"] is not None for zone in frame["zones"]) frame["health"] = "ok" if frame["complete"] else "unknown" frame["expires_at"] = None return frame if profile == "range_v1": frame["complete"] = bool(frame["ranges"]) and not any_stale else: frame["complete"] = explicit_slots == len(sources.get("slots") or []) and not any_stale if occupancy is True and not frame["targets"] and not frame["ranges"]: frame["health"] = "position_unavailable" elif any_stale: has_current_evidence = bool(frame["ranges"]) if profile == "range_v1" \ else explicit_slots > 0 frame["health"] = "partial" if has_current_evidence else "stale" expires = [item["expires_at"] for item in [*frame["targets"], *frame["ranges"]]] frame["expires_at"] = min(expires) if expires else now count = _numeric_state(self.hass.states.get(sources.get("count_entity"))) \ if sources.get("count_entity") else None count_state = self.hass.states.get(sources.get("count_entity")) \ if sources.get("count_entity") else None count_reported = _reported_timestamp(count_state) count_fresh = count is not None and now - count_reported <= PAIR_MAX_AGE_S if count_fresh and count == 0: if frame["targets"]: frame["health"] = "inconsistent" frame["complete"] = False elif not frame["ranges"]: frame["health"] = "off" frame["complete"] = True frame["expires_at"] = count_reported + PAIR_MAX_AGE_S return frame def _slot_local(self, profile: str, slot: dict[str, Any], now: float): if profile == "polar_v1": first = self.hass.states.get(slot.get("distance_entity")) second = self.hass.states.get(slot.get("angle_entity")) a, b = _numeric_state(first), _numeric_state(second) t1, t2 = _reported_timestamp(first), _reported_timestamp(second) if a is None or b is None or now - min(t1, t2) > PAIR_MAX_AGE_S or abs(t1 - t2) > PAIR_MAX_SKEW_S: return None x, y = polar_to_cartesian(a, b, unit=slot["unit"], angle_unit=slot["angle_unit"], zero=slot["angle_zero"], clockwise=slot["angle_clockwise"]) return x, y, t1, t2, "bounded_latest" first = self.hass.states.get(slot.get("x_entity")) second = self.hass.states.get(slot.get("y_entity")) a, b = _numeric_state(first), _numeric_state(second) t1, t2 = _reported_timestamp(first), _reported_timestamp(second) if a is None or b is None or now - min(t1, t2) > PAIR_MAX_AGE_S or abs(t1 - t2) > PAIR_MAX_SKEW_S: return None unit = "mm" if profile == "esphome_ld2450_v1" else slot["unit"] # The reference ESPHome LD2450 connection reports both axes as zero # for an unused slot. A single zero remains a legal coordinate; only # the verified pair is negative evidence. Generic profiles keep (0,0) # legal because they have no adapter-specific absence contract. if profile == "esphome_ld2450_v1" and a == 0 and b == 0: return _EXPLICIT_ABSENT x, y = length_cm(a, unit), length_cm(b, unit) if slot.get("swap_xy"): x, y = y, x return x * slot.get("x_sign", 1), y * slot.get("y_sign", 1), t1, t2, "bounded_latest" def _room_polygon( self, marker: dict[str, Any], radar: dict[str, Any], ) -> tuple[bool, list[Any] | None]: space_id = str(marker.get("space") or "") room_id = str(radar.get("room_id") or "") for space in self.config.get("spaces") or []: if str(space.get("id")) != space_id: continue for room in space.get("rooms") or []: if str(room.get("id")) == room_id: poly = room.get("poly") return True, poly if isinstance(poly, list) and len(poly) >= 3 else None return False, None def _space_cell_cm(self, marker: dict[str, Any]) -> float | None: space_id = str(marker.get("space") or "") for space in self.config.get("spaces") or []: if str(space.get("id")) != space_id: continue value = space.get("cell_cm", 5) try: result = float(value) except (TypeError, ValueError): return None return result if math.isfinite(result) and result > 0 else None return None @staticmethod def public_frame(frame: dict[str, Any]) -> dict[str, Any]: """Projection safe for View: no raw coordinates, ids or private times.""" public: dict[str, Any] = {} for key, value in frame.items(): if key.startswith("_"): continue if key not in {"targets", "ranges", "zones"}: public[key] = value continue items = value if key == "targets": items = [entry for entry in value if entry.get("included")] public[key] = [ {item_key: item_value for item_key, item_value in item.items() if not item_key.startswith("_")} for item in items ] return public def inspect(self, marker_id: str) -> dict[str, Any]: marker = self.radars.get(marker_id) if marker is None: return {"marker_id": marker_id, "health": "not_configured", "config_rev": self.config_rev} frame = self._frames.get(marker_id) or self._build_frame( marker_id, marker, datetime.now(UTC).timestamp() ) radar = marker["radar"] return { "marker_id": marker_id, "config_rev": self.config_rev, "profile": radar.get("profile"), "capabilities": self._capabilities(radar), "frame": self.public_frame(frame), "sources": [ {"entity_id": entity_id, "state": str(self.hass.states.get(entity_id).state) if self.hass.states.get(entity_id) else None, "reported_at": _reported_timestamp(self.hass.states.get(entity_id))} for entity_id in sorted(radar_source_entity_ids(radar)) ], } def inspect_draft(self, marker: dict[str, Any], radar: dict[str, Any]) -> dict[str, Any]: """Return one permission-gated setup snapshot for an unsaved draft.""" marker_id = str(marker.get("id")) candidate = {**marker, "radar": radar} frame = self._build_frame(marker_id, candidate, datetime.now(UTC).timestamp()) result = { "marker_id": marker_id, "config_rev": self.config_rev, "profile": radar.get("profile"), "capabilities": self._capabilities(radar), "frame": self.public_frame(frame), "local_targets": [ { "slot": target.get("slot"), "x_cm": target.get("_local_x_cm"), "y_cm": target.get("_local_y_cm"), "reported_at": target.get("reported_at"), "pair_quality": target.get("pair_quality"), } for target in frame.get("targets") or [] ], "sources": [ {"entity_id": entity_id, "state": str(self.hass.states.get(entity_id).state) if self.hass.states.get(entity_id) else None, "reported_at": _reported_timestamp(self.hass.states.get(entity_id))} for entity_id in sorted(radar_source_entity_ids(radar)) ], } return result @staticmethod def _capabilities(radar: dict[str, Any]) -> list[str]: profile = radar.get("profile") out = [] if profile in {"esphome_ld2450_v1", "cartesian_v1", "polar_v1"}: out.append("coordinates") if profile == "range_v1": out.append("range") if profile == "zones_v1": out.append("zone_state") if (radar.get("sources") or {}).get("occupancy_entity"): out.append("reported_presence") if ((radar.get("zones") or {}).get("hardware") or {}).get("adapter") == "esphome_ld2450_numbers_v1": out.append("hardware_zones") return out