"""
Event Bus Client
"""
from __future__ import annotations
import json
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Iterator
import grpc # noqa: F401 — tests patch events.grpc.insecure_channel
from ._transport import GrpcClient
from .proto import event_pb2, event_pb2_grpc
def _json_default(o: Any) -> Any:
"""Coerce numpy scalars/arrays to native Python for JSON serialization.
ML pipeline payloads routinely carry numpy.float32 (confidences, bbox
coords); json.dumps cannot serialize them and raises TypeError, which
crashes publish()/publish_batch() callers. Duck-typed so the SDK does
not hard-depend on numpy.
"""
if hasattr(o, "item"):
try:
return o.item() # np.float32/np.int64 -> python scalar
except Exception:
pass
if hasattr(o, "tolist"):
return o.tolist() # np.ndarray -> list
raise TypeError(
f"Object of type {o.__class__.__name__} is not JSON serializable",
)
[docs]
@dataclass
class Event:
topic: str
payload: dict[str, Any]
source: str = field(default="")
event_id: str = ""
timestamp_ns: int = 0
metadata: dict[str, str] = field(default_factory=dict)
def __post_init__(self):
if not self.source:
self.source = self._get_app_id()
if not self.timestamp_ns:
self.timestamp_ns = self._get_timestamp()
@staticmethod
def _get_app_id() -> str:
import os
return os.getenv("APP_ID", "unknown")
@staticmethod
def _get_timestamp() -> int:
return int(time.time() * 1e9)
[docs]
def to_json(self) -> str:
return json.dumps(self.payload, default=_json_default)
[docs]
@classmethod
def from_proto(cls, msg: event_pb2.Event) -> Event:
payload = {}
if msg.payload:
try:
payload = json.loads(msg.payload.decode("utf-8"))
except json.JSONDecodeError:
payload = {"raw": msg.payload.decode("utf-8", errors="replace")}
return cls(
topic=msg.topic,
payload=payload,
source=msg.source,
event_id=msg.event_id,
timestamp_ns=msg.timestamp_ns,
metadata=dict(msg.metadata),
)
[docs]
@dataclass
class TopicInfo:
topic: str
subscriber_count: int
total_messages: int
last_message_ts: int
[docs]
class EventClient(GrpcClient):
"""
Event Bus Client
Usage::
events = EventClient()
events.publish("app/alert", {"type": "person_detected"})
for event in events.subscribe("model/*/detections"):
print(f"Received: {event.topic}")
"""
_stub_factory = event_pb2_grpc.EventBusStub
_endpoint_env = "EVENT_BUS_ENDPOINT"
_endpoint_default = "unix:///run/aipc/event-bus.sock"
# Channel lifecycle and stub caching live in GrpcClient; close() below
# additionally joins the subscription threads.
[docs]
def __init__(self, endpoint: str | None = None):
super().__init__(endpoint)
self.app_id = self._get_app_id()
self._subscriptions: list[threading.Thread] = []
self._running = True
def _get_app_id(self) -> str:
import os
return os.getenv("APP_ID", "unknown")
[docs]
def close(self) -> None:
self._running = False
for t in self._subscriptions:
if t.is_alive():
t.join(timeout=1.0)
super().close()
[docs]
def publish(
self,
topic: str,
payload: dict[str, Any],
persistent: bool = False,
ttl_ms: int | None = None,
metadata: dict[str, str] | None = None,
compact: bool = False,
) -> str:
if self.stub is None:
self.connect()
event = event_pb2.Event(
topic=topic,
timestamp_ns=int(time.time() * 1e9),
source=self.app_id,
# compact drops the ", "/": " separators — smaller on the wire,
# and required by consumers that string-scan the payload (the
# camera-daemon overlay parser looks for "bbox":[ literally)
payload=json.dumps(
payload,
default=_json_default,
separators=(",", ":") if compact else None,
).encode("utf-8"),
payload_type="json",
)
if metadata:
event.metadata.update(metadata)
request = event_pb2.PublishRequest(event=event, persistent=persistent, ttl_ms=ttl_ms or 0)
response = self.stub.Publish(request)
if not response.status.success:
raise RuntimeError(f"Publish failed: {response.status.message}")
return response.event_id
[docs]
def publish_batch(self, events: list[dict[str, Any]], persistent: bool = False) -> None:
if self.stub is None:
self.connect()
def generate_requests():
for e in events:
event = event_pb2.Event(
topic=e["topic"],
timestamp_ns=int(time.time() * 1e9),
source=self.app_id,
payload=json.dumps(e["payload"], default=_json_default).encode("utf-8"),
payload_type="json",
)
yield event_pb2.PublishRequest(event=event, persistent=persistent)
response = self.stub.PublishBatch(generate_requests())
if not response.success:
raise RuntimeError(f"Batch publish failed: {response.message}")
[docs]
def subscribe(
self,
topic: str,
filters: dict[str, str] | None = None,
queue_size: int = 100,
drop_old: bool = True,
) -> Iterator[Event]:
if self.stub is None:
self.connect()
request = event_pb2.SubscribeRequest(
topic=topic, subscriber_id=self.app_id, queue_size=queue_size, drop_old=drop_old
)
if filters:
request.filters.update(filters)
for event_msg in self.stub.Subscribe(request):
yield Event.from_proto(event_msg)
[docs]
def on_event(
self,
topic: str,
callback: Callable[[Event], None],
filters: dict[str, str] | None = None,
) -> threading.Thread:
def _subscribe_thread():
try:
for event in self.subscribe(topic, filters):
if not self._running:
break
try:
callback(event)
except Exception:
pass
except grpc.RpcError:
pass
thread = threading.Thread(target=_subscribe_thread, daemon=True)
thread.start()
self._subscriptions.append(thread)
return thread
[docs]
def unsubscribe(self, topic: str) -> None:
if self.stub is None:
self.connect()
request = event_pb2.SubscribeRequest(topic=topic, subscriber_id=self.app_id)
self.stub.Unsubscribe(request)
[docs]
def list_topics(self) -> list[TopicInfo]:
if self.stub is None:
self.connect()
response = self.stub.ListTopics(event_pb2.Empty())
return [
TopicInfo(
topic=t.topic,
subscriber_count=t.subscriber_count,
total_messages=t.total_messages,
last_message_ts=t.last_message_ts,
)
for t in response.topics
]
[docs]
def get_topic_info(self, topic: str) -> TopicInfo | None:
if self.stub is None:
self.connect()
request = event_pb2.TopicInfo(topic=topic)
response = self.stub.GetTopicInfo(request)
if not response.topic:
return None
return TopicInfo(
topic=response.topic,
subscriber_count=response.subscriber_count,
total_messages=response.total_messages,
last_message_ts=response.last_message_ts,
)
[docs]
def get_stats(self) -> dict[str, Any]:
if self.stub is None:
self.connect()
response = self.stub.GetStats(event_pb2.Empty())
return {
"total_subscribers": response.total_subscribers,
"total_topics": response.total_topics,
"uptime_ms": response.uptime_ms,
"topic_stats": [
{
"topic": s.topic,
"published_count": s.published_count,
"delivered_count": s.delivered_count,
"dropped_count": s.dropped_count,
"avg_latency_us": s.avg_latency_us,
}
for s in response.topic_stats
],
}
[docs]
def get_topic_stats(self, topic: str) -> dict[str, Any]:
if self.stub is None:
self.connect()
request = event_pb2.TopicInfo(topic=topic)
response = self.stub.GetTopicStats(request)
return {
"topic": response.topic,
"published_count": response.published_count,
"delivered_count": response.delivered_count,
"dropped_count": response.dropped_count,
"avg_latency_us": response.avg_latency_us,
}