core/sensorium/vision_event/packet.py
2026-06-06 12:37:57 -07:00

67 lines
2.2 KiB
Python

"""Canonical construction for synthetic event-stream vision packets."""
from __future__ import annotations
from collections.abc import Iterable
from sensorium.vision.checksum import sha256_json
from sensorium.vision_event.types import Event, EventPacket
def _event_from_tuple(raw: Event | tuple[int, int, int, int]) -> Event:
if isinstance(raw, Event):
return raw
x, y, polarity, t_q = raw
return Event(int(x), int(y), int(polarity), int(t_q))
def build_event_packet(
*,
grid_w: int,
grid_h: int,
packet_tick: int,
events: Iterable[Event | tuple[int, int, int, int]],
source_id: str = "synthetic-event-fixture",
) -> EventPacket:
if grid_w <= 0 or grid_h <= 0:
raise ValueError("event packet grid dimensions must be positive")
if packet_tick < 0:
raise ValueError("event packet tick must be non-negative")
canonical_events = tuple(sorted(_event_from_tuple(event) for event in events))
if not canonical_events:
raise ValueError("event packet requires at least one event")
for event in canonical_events:
if not (0 <= event.x < grid_w and 0 <= event.y < grid_h):
raise ValueError("event coordinate outside canonical sensor grid")
if event.polarity not in {-1, 1}:
raise ValueError("event polarity must be -1 or 1")
if event.t_q < 0:
raise ValueError("event time quantum must be non-negative")
source_sha256 = sha256_json({
"kind": "EventPacket.source",
"source_id": str(source_id),
"grid_w": int(grid_w),
"grid_h": int(grid_h),
})
payload = {
"kind": "EventPacket",
"grid_w": int(grid_w),
"grid_h": int(grid_h),
"packet_tick": int(packet_tick),
"source_sha256": source_sha256,
"events": [
{"x": e.x, "y": e.y, "polarity": e.polarity, "t_q": e.t_q}
for e in canonical_events
],
}
return EventPacket(
grid_w=int(grid_w),
grid_h=int(grid_h),
packet_tick=int(packet_tick),
events=canonical_events,
source_sha256=source_sha256,
canonical_sha256=sha256_json(payload),
)
__all__ = ["build_event_packet"]