Skip to content

Sensors and DVS Pipeline

Event camera (DVS) data loading, preprocessing, and spike encoding. Supports AEDAT, HDF5, NumPy.

from sc_neurocore.sensors import DVSPipeline

pipeline = DVSPipeline(resolution=(240, 180))
events = pipeline.load("recording.aedat")

See Tutorial 45: DVS Pipeline.

sc_neurocore.sensors

Sensor interfaces: event cameras (DVS), audio, and other neuromorphic sensors.

DVSLoader dataclass

Load and preprocess DVS event camera data.

Parameters

width : int Sensor width in pixels. height : int Sensor height in pixels.

Source code in src/sc_neurocore/sensors/dvs.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@dataclass
class DVSLoader:
    """Load and preprocess DVS event camera data.

    Parameters
    ----------
    width : int
        Sensor width in pixels.
    height : int
        Sensor height in pixels.
    """

    width: int = 346
    height: int = 260

    @property
    def n_pixels(self) -> int:
        return self.width * self.height

    def from_numpy(self, events: np.ndarray) -> np.ndarray:
        """Load events from structured numpy array.

        Expected fields: 'x', 'y', 't', 'p' (or positional columns).
        Returns structured array with named fields.
        """
        if events.dtype.names is not None:
            return events
        if events.ndim == 2 and events.shape[1] >= 4:
            dtype = np.dtype([("x", np.int32), ("y", np.int32), ("t", np.int64), ("p", np.int8)])
            structured = np.zeros(events.shape[0], dtype=dtype)
            structured["x"] = events[:, 0].astype(np.int32)
            structured["y"] = events[:, 1].astype(np.int32)
            structured["t"] = events[:, 2].astype(np.int64)
            structured["p"] = events[:, 3].astype(np.int8)
            return structured
        raise ValueError("Events must be structured array or (N, 4+) array with x, y, t, p columns")

    def from_tonic(self, dataset_name: str, index: int = 0):
        """Load events from a Tonic dataset (requires tonic package).

        Parameters
        ----------
        dataset_name : str
            Tonic dataset name: 'nmnist', 'dvs_gesture', 'ncars', etc.
        index : int
            Sample index in the dataset.

        Returns
        -------
        (events, target) tuple
        """
        try:
            import tonic
        except ImportError:
            raise ImportError("pip install tonic") from None

        dataset_map = {  # pragma: no cover
            "nmnist": tonic.datasets.NMNIST,
            "dvs_gesture": tonic.datasets.DVSGesture,
        }
        cls = dataset_map.get(dataset_name)  # pragma: no cover
        if cls is None:  # pragma: no cover
            raise ValueError(f"Unknown dataset '{dataset_name}'. Options: {list(dataset_map)}")

        ds = cls(save_to="./data", train=True)  # pragma: no cover
        events, target = ds[index]  # pragma: no cover
        return self.from_numpy(events), target  # pragma: no cover

from_numpy(events)

Load events from structured numpy array.

Expected fields: 'x', 'y', 't', 'p' (or positional columns). Returns structured array with named fields.

Source code in src/sc_neurocore/sensors/dvs.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def from_numpy(self, events: np.ndarray) -> np.ndarray:
    """Load events from structured numpy array.

    Expected fields: 'x', 'y', 't', 'p' (or positional columns).
    Returns structured array with named fields.
    """
    if events.dtype.names is not None:
        return events
    if events.ndim == 2 and events.shape[1] >= 4:
        dtype = np.dtype([("x", np.int32), ("y", np.int32), ("t", np.int64), ("p", np.int8)])
        structured = np.zeros(events.shape[0], dtype=dtype)
        structured["x"] = events[:, 0].astype(np.int32)
        structured["y"] = events[:, 1].astype(np.int32)
        structured["t"] = events[:, 2].astype(np.int64)
        structured["p"] = events[:, 3].astype(np.int8)
        return structured
    raise ValueError("Events must be structured array or (N, 4+) array with x, y, t, p columns")

from_tonic(dataset_name, index=0)

Load events from a Tonic dataset (requires tonic package).

Parameters

dataset_name : str Tonic dataset name: 'nmnist', 'dvs_gesture', 'ncars', etc. index : int Sample index in the dataset.

Returns

(events, target) tuple

Source code in src/sc_neurocore/sensors/dvs.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def from_tonic(self, dataset_name: str, index: int = 0):
    """Load events from a Tonic dataset (requires tonic package).

    Parameters
    ----------
    dataset_name : str
        Tonic dataset name: 'nmnist', 'dvs_gesture', 'ncars', etc.
    index : int
        Sample index in the dataset.

    Returns
    -------
    (events, target) tuple
    """
    try:
        import tonic
    except ImportError:
        raise ImportError("pip install tonic") from None

    dataset_map = {  # pragma: no cover
        "nmnist": tonic.datasets.NMNIST,
        "dvs_gesture": tonic.datasets.DVSGesture,
    }
    cls = dataset_map.get(dataset_name)  # pragma: no cover
    if cls is None:  # pragma: no cover
        raise ValueError(f"Unknown dataset '{dataset_name}'. Options: {list(dataset_map)}")

    ds = cls(save_to="./data", train=True)  # pragma: no cover
    events, target = ds[index]  # pragma: no cover
    return self.from_numpy(events), target  # pragma: no cover

events_to_spike_trains(events, width, height, dt_us=1000.0, duration_us=None)

Convert DVS events to binary spike train matrix.

Parameters

events : structured ndarray with x, y, t, p fields width, height : int Sensor dimensions. dt_us : float Time bin width in microseconds (default 1000 = 1ms). duration_us : float, optional Total duration. If None, inferred from event timestamps.

Returns

ndarray of shape (n_bins, width * height * 2) Binary spike trains. Channels: [ON pixels, OFF pixels].

Source code in src/sc_neurocore/sensors/dvs.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def events_to_spike_trains(
    events: np.ndarray,
    width: int,
    height: int,
    dt_us: float = 1000.0,
    duration_us: float | None = None,
) -> np.ndarray:
    """Convert DVS events to binary spike train matrix.

    Parameters
    ----------
    events : structured ndarray with x, y, t, p fields
    width, height : int
        Sensor dimensions.
    dt_us : float
        Time bin width in microseconds (default 1000 = 1ms).
    duration_us : float, optional
        Total duration. If None, inferred from event timestamps.

    Returns
    -------
    ndarray of shape (n_bins, width * height * 2)
        Binary spike trains. Channels: [ON pixels, OFF pixels].
    """
    x = events["x"].astype(np.int64)
    y = events["y"].astype(np.int64)
    t = events["t"].astype(np.float64)
    p = events["p"].astype(np.int8)

    t_min = t.min()
    t_rel = t - t_min

    if duration_us is None:
        duration_us = t_rel.max() + dt_us

    n_bins = max(1, int(np.ceil(duration_us / dt_us)))
    n_channels = width * height * 2

    spikes = np.zeros((n_bins, n_channels), dtype=np.int8)

    for i in range(len(events)):
        bin_idx = min(int(t_rel[i] / dt_us), n_bins - 1)
        pixel_idx = int(y[i]) * width + int(x[i])
        if p[i] > 0:
            channel = pixel_idx
        else:
            channel = width * height + pixel_idx
        if 0 <= channel < n_channels:
            spikes[bin_idx, channel] = 1

    return spikes

events_to_frames(events, width, height, dt_us=10000.0, duration_us=None)

Convert DVS events to event count frames.

Parameters

events : structured ndarray width, height : int dt_us : float Frame duration in microseconds (default 10000 = 10ms). duration_us : float, optional

Returns

ndarray of shape (n_frames, 2, height, width) Event count frames with ON and OFF channels.

Source code in src/sc_neurocore/sensors/dvs.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def events_to_frames(
    events: np.ndarray,
    width: int,
    height: int,
    dt_us: float = 10000.0,
    duration_us: float | None = None,
) -> np.ndarray:
    """Convert DVS events to event count frames.

    Parameters
    ----------
    events : structured ndarray
    width, height : int
    dt_us : float
        Frame duration in microseconds (default 10000 = 10ms).
    duration_us : float, optional

    Returns
    -------
    ndarray of shape (n_frames, 2, height, width)
        Event count frames with ON and OFF channels.
    """
    x = events["x"].astype(np.int64)
    y = events["y"].astype(np.int64)
    t = events["t"].astype(np.float64)
    p = events["p"].astype(np.int8)

    t_min = t.min()
    t_rel = t - t_min

    if duration_us is None:
        duration_us = t_rel.max() + dt_us

    n_frames = max(1, int(np.ceil(duration_us / dt_us)))
    frames = np.zeros((n_frames, 2, height, width), dtype=np.float32)

    for i in range(len(events)):
        f = min(int(t_rel[i] / dt_us), n_frames - 1)
        yi = min(int(y[i]), height - 1)
        xi = min(int(x[i]), width - 1)
        ch = 1 if p[i] > 0 else 0
        frames[f, ch, yi, xi] += 1.0

    return frames