Files
ruvnet--RuView/python/wifi_densepose/client/primitives.py
T
ruv f21daf9aa8 feat(adr-117/p4): pure-Python WS/MQTT client layer
New sub-package `wifi_densepose.client` (no PyO3, no Rust deps):

- ws.SensingClient — asyncio websockets>=12 wrapper for the Rust
  sensing-server /ws/sensing endpoint. Yields typed dataclasses
  (ConnectionEstablishedMessage, EdgeVitalsMessage, PoseDataMessage)
  with raw-payload fallback for forward-compat with unknown types.
  Malformed frames log+drop without breaking the stream.

- mqtt.RuViewMqttClient — paho-mqtt v2 wrapper using the explicit
  CallbackAPIVersion.VERSION2 API. Per-instance unique client_id by
  default (rumqttc memory lesson). MQTT v5-spec-correct topic
  wildcard matcher: + as whole-level wildcard, # matches the prefix
  itself plus all sub-levels. Auto-resubscribes on reconnect.
  Handler exceptions are caught and logged so a misbehaving callback
  can't crash the network loop.

- primitives.SemanticPrimitiveListener — typed router for the 10
  HA-MIND fused inference outputs from ADR-115 §3.12
  (SomeoneSleeping, PossibleDistress, RoomActive, ElderlyInactivity-
  Anomaly, MeetingInProgress, BathroomOccupied, FallRiskElevated,
  BedExit, NoMovementSafety, MultiRoomTransition). Decodes both
  JSON payloads with confidence+explanation AND plain HA state
  strings ("ON"/"OFF"/numeric). Pluggable into RuViewMqttClient.

- ha.HABlueprintHelper — read-only parser for the
  homeassistant/<kind>/wifi_densepose_<node>/<id>/config payload
  family. Aggregator queries: entities_for_node, by_device_class,
  nodes. Useful for blueprint authors + dashboard introspection.

Test coverage (63 new tests, 156 total in Python suite):
- test_client_ha — 18 tests (topic+payload parsing, aggregator)
- test_client_primitives — 13 tests (enum coverage, listener routing)
- test_client_mqtt — 17 tests (matcher parametrize, dispatch path,
  on_connect, exception isolation) — no broker needed
- test_client_ws — 6 tests including end-to-end against an in-process
  websockets.serve() fixture exercising all 4 message types plus a
  malformed-frame survival check

Post-bridge wheel size: 238 KB (well under ADR §5.4 5 MB budget).

Refs: docs/adr/ADR-117-pip-wifi-densepose-modernization.md §5.6
Refs: docs/adr/ADR-115-home-assistant-integration.md §3.12
Refs: #785

Co-Authored-By: claude-flow <ruv@ruv.net>
2026-05-24 11:31:29 -04:00

223 lines
7.8 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""ADR-117 P4 — Typed listener for HA-MIND semantic primitives.
ADR-115 §3.12 defines 10 fused inference outputs that the sensing-server
publishes under the HA-DISCO MQTT namespace. This module gives clients
a typed handle on them so they can write `if event.kind ==
SemanticPrimitive.SomeoneSleeping: ...` instead of pattern-matching
strings.
The 10 v1 primitives (ADR-115 §3.12.1):
| Enum value | Topic suffix | Output kind |
|---|---|---|
| `SomeoneSleeping` | `someone_sleeping` | binary_sensor |
| `PossibleDistress` | `possible_distress` | binary_sensor + event |
| `RoomActive` | `room_active` | binary_sensor |
| `ElderlyInactivityAnomaly` | `elderly_inactivity` | binary_sensor + event |
| `MeetingInProgress` | `meeting_in_progress` | binary_sensor |
| `BathroomOccupied` | `bathroom_occupied` | binary_sensor |
| `FallRiskElevated` | `fall_risk_elevated` | sensor (0100) + event |
| `BedExit` | `bed_exit` | event |
| `NoMovementSafety` | `no_movement_safety` | binary_sensor + event |
| `MultiRoomTransition` | `multi_room_transition` | event |
"""
from __future__ import annotations
import enum
import json
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
# ─── Enum ────────────────────────────────────────────────────────────
class SemanticPrimitive(enum.Enum):
"""One of the 10 HA-MIND fused inference outputs."""
SomeoneSleeping = "someone_sleeping"
PossibleDistress = "possible_distress"
RoomActive = "room_active"
ElderlyInactivityAnomaly = "elderly_inactivity"
MeetingInProgress = "meeting_in_progress"
BathroomOccupied = "bathroom_occupied"
FallRiskElevated = "fall_risk_elevated"
BedExit = "bed_exit"
NoMovementSafety = "no_movement_safety"
MultiRoomTransition = "multi_room_transition"
@classmethod
def from_object_id(cls, object_id: str) -> Optional["SemanticPrimitive"]:
for v in cls:
if v.value == object_id:
return v
return None
# ─── Event payload ───────────────────────────────────────────────────
@dataclass(frozen=True)
class SemanticPrimitiveEvent:
"""A single fired event for one semantic primitive.
`state` semantics depend on the primitive kind:
- binary_sensor: "ON" / "OFF"
- sensor: numeric string (e.g. "73" for fall_risk_elevated 0100)
- event: "fired" or an event-class string like "bed_exit_detected"
"""
kind: SemanticPrimitive
node_id: str
state: str
confidence: float = 0.0
explanation: tuple[str, ...] = ()
timestamp: float = 0.0
raw: dict[str, Any] = field(default_factory=dict, hash=False, compare=False)
# ─── Listener ────────────────────────────────────────────────────────
Callback = Callable[[SemanticPrimitiveEvent], None]
class SemanticPrimitiveListener:
"""Routes raw MQTT state messages to per-primitive callbacks.
Designed to plug into RuViewMqttClient:
```python
from wifi_densepose.client import (
RuViewMqttClient, SemanticPrimitive, SemanticPrimitiveListener
)
listener = SemanticPrimitiveListener()
listener.on(SemanticPrimitive.SomeoneSleeping, lambda e: print(e))
client = RuViewMqttClient()
client.on_message(
"homeassistant/+/wifi_densepose_+/+/state",
listener.handle_mqtt_message,
)
client.start()
```
The listener itself never touches MQTT — it's a pure router. You
feed it `(topic, payload)` pairs and it figures out which primitive
the topic refers to and decodes the payload.
"""
# Matches state topics for any of the 10 primitives.
# homeassistant/<kind>/wifi_densepose_<node>/<primitive_slug>/state
_SLUGS = {p.value for p in SemanticPrimitive}
def __init__(self) -> None:
self._handlers: dict[Optional[SemanticPrimitive], list[Callback]] = {}
def on(self, primitive: SemanticPrimitive, cb: Callback) -> None:
"""Register a callback for a specific primitive."""
self._handlers.setdefault(primitive, []).append(cb)
def on_any(self, cb: Callback) -> None:
"""Register a callback that fires for ALL primitives. Useful
for logging or dashboards."""
self._handlers.setdefault(None, []).append(cb)
def handle_mqtt_message(self, topic: str, payload: Any) -> Optional[SemanticPrimitiveEvent]:
"""Decode one MQTT message into a SemanticPrimitiveEvent and
fire the matching callbacks. Returns the event (or None if the
topic was not a semantic-primitive state topic)."""
parts = topic.split("/")
# Shape: homeassistant / <kind> / wifi_densepose_<node> / <slug> / state
if len(parts) != 5:
return None
if parts[0] != "homeassistant" or parts[4] != "state":
return None
node_prefix = parts[2]
if not node_prefix.startswith("wifi_densepose_"):
return None
slug = parts[3]
if slug not in self._SLUGS:
return None
primitive = SemanticPrimitive.from_object_id(slug)
if primitive is None: # pragma: no cover — guarded above
return None
node_id = node_prefix[len("wifi_densepose_"):]
event = _decode_event(primitive, node_id, payload)
# Dispatch — primitive-specific first, then "any" handlers.
for cb in self._handlers.get(primitive, ()):
cb(event)
for cb in self._handlers.get(None, ()):
cb(event)
return event
def _decode_event(
primitive: SemanticPrimitive,
node_id: str,
payload: Any,
) -> SemanticPrimitiveEvent:
"""Decode a raw state payload into a typed event.
HA state payloads come in two shapes:
1. Plain string ("ON", "OFF", "73") — used by binary_sensor/sensor
with no json_attributes_topic.
2. JSON object with `state` + `confidence` + `explanation` fields —
used by HA-MIND semantic primitives per ADR-115 §3.12.4.
Both are supported transparently.
"""
if isinstance(payload, bytes):
try:
payload = payload.decode("utf-8")
except UnicodeDecodeError:
return SemanticPrimitiveEvent(
kind=primitive, node_id=node_id, state="", raw={}
)
if isinstance(payload, dict):
body = payload
elif isinstance(payload, str):
# Try to JSON-decode; if it's not JSON, treat as a plain state string.
try:
decoded = json.loads(payload)
except json.JSONDecodeError:
return SemanticPrimitiveEvent(
kind=primitive,
node_id=node_id,
state=payload,
raw={"state": payload},
)
if isinstance(decoded, dict):
body = decoded
else:
return SemanticPrimitiveEvent(
kind=primitive,
node_id=node_id,
state=str(decoded),
raw={"state": decoded},
)
else:
return SemanticPrimitiveEvent(
kind=primitive, node_id=node_id, state=str(payload), raw={}
)
expl = body.get("explanation") or body.get("reason") or ()
if isinstance(expl, str):
expl_tuple: tuple[str, ...] = (expl,)
else:
expl_tuple = tuple(str(x) for x in expl)
return SemanticPrimitiveEvent(
kind=primitive,
node_id=node_id,
state=str(body.get("state", "")),
confidence=float(body.get("confidence", 0.0)),
explanation=expl_tuple,
timestamp=float(body.get("timestamp", 0.0)),
raw=body,
)