Coverage for src/evutils/types.py: 98%
135 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Core data types for event streams.
3Defines the structured NumPy dtypes used throughout evutils — ``Events``
4(timestamp, x, y, polarity) and ``Triggers`` (timestamp, polarity, id) —
5together with small helpers for checking event arrays.
6"""
8from typing import TypeVar
10import numpy as np
12__all__ = ['Event_dtype', 'Trigger_dtype', 'EventArray', 'TriggerArray', 'DataBatch', 'is_monotonically_increasing', 'EventsChecker']
14from dataclasses import dataclass
16#: A structured numpy dtype for event data.
17#:
18#: Fields:
19#:
20#: - `t` (np.int64): Timestamp of the event (us).
21#: - `x` (np.uint16): X-coordinate.
22#: - `y` (np.uint16): Y-coordinate.
23#: - `p` (np.uint8): Polarity (0: off, 1: on).
24Event_dtype = np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)])
26#: A structured numpy dtype for trigger data.
27#:
28#: Fields:
29#:
30#: - `t` (np.int64): Timestamp of the event (us).
31#: - `p` (np.uint8): Polarity (0: off, 1: on).
32#: - `id` (np.uint8): Identifier.
33Trigger_dtype = np.dtype([('t', np.int64), ('p', np.uint8), ('id', np.uint8)])
35def is_monotonically_increasing(events: np.ndarray) -> bool:
36 """Checks if the event ts is monotonically increasing.
38 Parameters
39 ----------
40 events : np.ndarray
41 Array of events with a 't' field for timestamps.
43 Returns
44 -------
45 bool
46 True if timestamps are monotonically increasing, False otherwise.
48 """
49 return bool(np.all(np.diff(events['t']) >= 0))
51_S = TypeVar("_S", bound="SoaArray")
53class SoaArray:
54 """Abstract base class for struct-of-arrays (SoA) layout.
56 Examples
57 --------
58 >>> import numpy as np
59 >>> from evutils.types import EventArray
60 >>> events = EventArray(t=[1, 2], x=[10, 20], y=[30, 40], p=[1, 0])
61 >>> float(np.mean(events.x)) # SoA layout allows fast operations on single columns
62 15.0
63 >>> events.to_numpy() # doctest: +SKIP
64 array([(1, 10, 30, 1), (2, 20, 40, 0)],
65 dtype=[('t', '<i8'), ('x', '<u2'), ('y', '<u2'), ('p', 'u1')])
66 """
68 __slots__ = ('_metadata',)
69 _aos_dtype: np.dtype
70 _fields: tuple[str, ...]
72 @property
73 def metadata(self) -> 'dict | None':
74 """Optional lightweight, per-array metadata (e.g. ``sensor_size``).
76 Defaults to ``None`` and is never part of the AoS/NumPy view. It is
77 carried through slicing and :meth:`copy` (shallow), so a sensor size
78 stamped on read survives most event processing. Transforms that change
79 the meaning of a field (e.g. spatial cropping) are responsible for
80 updating it.
81 """
82 return getattr(self, '_metadata', None)
84 @metadata.setter
85 def metadata(self, value: 'dict | None') -> None:
86 self._metadata = value
88 @property
89 def sensor_size(self) -> "tuple[int, int] | None":
90 """Convenience accessor for ``metadata['sensor_size']`` (or ``None``).
92 By convention a ``(width, height)`` tuple, matching the ``(W, H)`` order
93 that the spatial transforms and the io layer use.
94 """
95 m = self.metadata
96 return m.get('sensor_size') if m else None
98 @sensor_size.setter
99 def sensor_size(self, value: "tuple[int, int] | None") -> None:
100 m = self.metadata
101 if m is None:
102 self.metadata = {'sensor_size': value}
103 else:
104 m['sensor_size'] = value
106 def __getitem__(self, key: "str | slice | np.ndarray") -> "np.ndarray | SoaArray":
107 if isinstance(key, str):
108 return getattr(self, key)
110 if isinstance(key, (list, tuple)) and all(isinstance(k, str) for k in key):
111 if len(key) == 0:
112 raise ValueError("Cannot index with an empty list of fields.")
113 fields = tuple(key)
114 class DynamicSoaArray(SoaArray):
115 __slots__ = fields
116 _aos_dtype = np.dtype([(f, self._aos_dtype[f]) for f in fields])
117 _fields = fields
118 def __init__(self, **kwargs):
119 for k, v in kwargs.items():
120 setattr(self, k, v)
121 subset = DynamicSoaArray(**{f: getattr(self, f) for f in fields})
122 # x/y are preserved, so a sensor size stays valid on the subset.
123 subset.metadata = self.metadata.copy() if self.metadata is not None else None
124 return subset
126 # When indexing a single element, return a NumPy void record to match AoS behaviour exactly.
127 if isinstance(key, (int, np.integer)):
128 record = np.empty((), dtype=self._aos_dtype)
129 for f in self._fields:
130 record[f] = getattr(self, f)[key]
131 return record[()] # Returns a scalar np.void
133 # Otherwise, slice all columns and return a new SoA array
134 sliced_args = {f: getattr(self, f)[key] for f in self._fields}
135 sliced = self.__class__(**sliced_args)
136 sliced.metadata = self.metadata.copy() if self.metadata is not None else None
137 return sliced
139 def __len__(self) -> int:
140 return len(getattr(self, self._fields[0]))
142 def __repr__(self) -> str:
143 n = len(self)
144 name = self.__class__.__name__
145 if n == 0:
146 return f"{name}(empty)"
148 if n <= 10:
149 return f"{name}(n={n}):\n{self.to_aos()}"
151 # Slice before AoS conversion for speed
152 head_str = str(self[:3].to_aos()).rstrip(']')
153 tail_str = str(self[-3:].to_aos()).lstrip('[')
155 return f"{name}(n={n}):\n{head_str}\n ...\n {tail_str}]"
157 def copy(self: _S) -> _S:
158 """Return a deep copy with independent column arrays."""
159 copied_args = {f: getattr(self, f).copy() for f in self._fields}
160 new = self.__class__(**copied_args)
161 # Shallow-copy the metadata dict so mutating one array's metadata does
162 # not leak into the copy.
163 new.metadata = None if self.metadata is None else dict(self.metadata)
164 return new
166 @classmethod
167 def empty(cls: 'type[_S]', metadata: 'dict | None' = None) -> _S:
168 """Return an empty SoA array with correctly-typed (zero-length) columns."""
169 args = {f: np.empty(0, dtype=cls._aos_dtype[f]) for f in cls._fields}
170 new = cls(**args)
171 new.metadata = metadata
172 return new
174 @classmethod
175 def from_aos(cls: 'type[_S]', aos_array: np.ndarray, metadata: 'dict | None' = None) -> _S:
176 """Constructs a SoA array from an array of structures (AoS) numpy array."""
177 args = {f: np.ascontiguousarray(aos_array[f]) for f in cls._fields}
178 new = cls(**args)
179 new.metadata = metadata
180 return new
182 def to_aos(self) -> np.ndarray:
183 """Converts the SoA array to an array of structures (AoS) numpy array."""
184 aos_array = np.empty(len(self), dtype=self._aos_dtype)
185 for f in self._fields:
186 aos_array[f] = getattr(self, f)
187 return aos_array
189 def to_numpy(self) -> np.ndarray:
190 """Converts the SoA array to a structured numpy array. Alias for to_aos."""
191 return self.to_aos()
193 def __array__(self, dtype: "np.dtype | type | None" = None, copy: bool | None = None) -> np.ndarray:
194 """Numpy interop: ``np.asarray(arr)`` returns the AoS structured array."""
195 aos = self.to_aos()
196 if dtype is not None:
197 return aos.astype(dtype)
198 return aos
200class EventArray(SoaArray):
201 """A container for storing event data in a struct-of-arrays (SoA) layout.
203 The four fields ``t``, ``x``, ``y`` and ``p`` are kept as separate
204 contiguous numpy arrays. This is the native layout of the C parser and
205 avoids the padding of the packed :data:`Event_dtype` struct.
207 Both attribute access (``events.t``) and key access (``events['t']``) return
208 the underlying column, so most code written for structured arrays keeps
209 working. ``np.asarray(events)`` yields the array-of-structures form (see
210 :meth:`__array__`), which lets EventArray flow into code that still expects
211 :data:`Event_dtype`.
213 Examples
214 --------
215 >>> from evutils.types import EventArray
216 >>> events = EventArray(t=[100, 150], x=[10, 20], y=[30, 40], p=[1, 0])
217 >>> events.t
218 array([100, 150])
219 >>> events.x
220 array([10, 20], dtype=uint16)
221 >>> events.y
222 array([30, 40], dtype=uint16)
223 >>> events.p
224 array([1, 0], dtype=uint8)
225 >>> events[:10]
226 EventArray(n=2):
227 [(100, 10, 30, 1) (150, 20, 40, 0)]
228 >>> events[0] # doctest: +SKIP
229 np.void((100, 10, 30, 1), dtype=[('t', '<i8'), ('x', '<u2'), ('y', '<u2'), ('p', 'u1')])
230 """
232 __slots__ = ['t', 'x', 'y', 'p']
233 _aos_dtype = Event_dtype
234 _fields = ('t', 'x', 'y', 'p')
236 def __init__(self, t: "np.typing.ArrayLike", x: "np.typing.ArrayLike", y: "np.typing.ArrayLike", p: "np.typing.ArrayLike", metadata: 'dict | None' = None) -> None:
237 t_arr = np.atleast_1d(np.asarray(t, dtype=np.int64))
238 x_arr = np.atleast_1d(np.asarray(x, dtype=np.uint16))
239 y_arr = np.atleast_1d(np.asarray(y, dtype=np.uint16))
240 p_arr = np.atleast_1d(np.asarray(p, dtype=np.uint8))
242 if not (t_arr.ndim == 1 and x_arr.ndim == 1 and y_arr.ndim == 1 and p_arr.ndim == 1):
243 t_arr, x_arr, y_arr, p_arr = t_arr.ravel(), x_arr.ravel(), y_arr.ravel(), p_arr.ravel()
245 if not (len(t_arr) == len(x_arr) == len(y_arr) == len(p_arr)):
246 raise ValueError(f"Length mismatch: t({len(t_arr)}), x({len(x_arr)}), y({len(y_arr)}), p({len(p_arr)})")
248 self.t = t_arr
249 self.x = x_arr
250 self.y = y_arr
251 self.p = p_arr
252 self.metadata = metadata
254class TriggerArray(SoaArray):
255 """A container for storing trigger data in a struct-of-arrays (SoA) layout.
257 Examples
258 --------
259 >>> from evutils.types import TriggerArray
260 >>> triggers = TriggerArray(t=[1000, 2000], p=[1, 0], id=[0, 1])
261 >>> triggers.t
262 array([1000, 2000])
263 >>> triggers.p
264 array([1, 0], dtype=uint8)
265 >>> triggers.id
266 array([0, 1], dtype=uint8)
267 """
269 __slots__ = ['t', 'p', 'id']
270 _aos_dtype = Trigger_dtype
271 _fields = ('t', 'p', 'id')
273 def __init__(self, t: "np.typing.ArrayLike", p: "np.typing.ArrayLike", id: "np.typing.ArrayLike", metadata: 'dict | None' = None) -> None:
274 t_arr = np.atleast_1d(np.asarray(t, dtype=np.int64))
275 p_arr = np.atleast_1d(np.asarray(p, dtype=np.uint8))
276 id_arr = np.atleast_1d(np.asarray(id, dtype=np.uint8))
278 if not (t_arr.ndim == 1 and p_arr.ndim == 1 and id_arr.ndim == 1):
279 t_arr, p_arr, id_arr = t_arr.ravel(), p_arr.ravel(), id_arr.ravel()
281 if not (len(t_arr) == len(p_arr) == len(id_arr)):
282 raise ValueError(f"Length mismatch: t({len(t_arr)}), p({len(p_arr)}), id({len(id_arr)})")
284 self.t = t_arr
285 self.p = p_arr
286 self.id = id_arr
287 self.metadata = metadata
289@dataclass
290class DataBatch:
291 """A single batch of multiplexed data containing events, triggers, and potentially other modalities."""
292 events: EventArray
293 triggers: TriggerArray
295 # We leave room for future modalities here
296 # imu: ImuArray | None = None
297 # frames: FrameArray | None = None
299# EventsChecker validates the event types defined above; imported here (at the
300# end, after the types exist) so it is reachable as ``evutils.types.EventsChecker``
301# without a circular import.
302from ._checker import EventsChecker # noqa: E402