Skip to content

Sensors and DVS Pipeline

Event camera (DVS) data loading, preprocessing, and spike encoding. The public package exports DVSLoader, events_to_spike_trains, and events_to_frames.

Python
from sc_neurocore.sensors import DVSLoader, events_to_spike_trains

loader = DVSLoader(width=128, height=128)
events = loader.from_numpy(raw_events)
spikes = events_to_spike_trains(events, width=128, height=128)

See Tutorial 45: DVS Pipeline.

API

sc_neurocore.sensors.dvs

DVS (Dynamic Vision Sensor) event processing pipeline.

Load event camera data, convert to spike trains or frames, and feed into SC-NeuroCore networks for processing and FPGA deployment.

Supports raw event arrays (structured numpy) and integration with the Tonic library for standard DVS datasets (N-MNIST, DVS-Gesture, N-Cars, etc.).

structured array with fields (x, y, t, p)

x: pixel x coordinate y: pixel y coordinate t: timestamp (microseconds) p: polarity (0=OFF, 1=ON)

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
Python
 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
 98
 99
100
@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 the total number of pixels in the DVS frame."""
        return self.width * self.height

    def from_numpy(self, events: np.ndarray[Any, Any]) -> np.ndarray[Any, Any]:
        """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) -> tuple[np.ndarray[Any, Any], int]:
        """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

n_pixels property

Return the total number of pixels in the DVS frame.

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
Python
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def from_numpy(self, events: np.ndarray[Any, Any]) -> np.ndarray[Any, Any]:
    """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
Python
 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
 98
 99
100
def from_tonic(self, dataset_name: str, index: int = 0) -> tuple[np.ndarray[Any, Any], int]:
    """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
Python
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
151
152
153
def events_to_spike_trains(
    events: np.ndarray[Any, Any],
    width: int,
    height: int,
    dt_us: float = 1000.0,
    duration_us: float | None = None,
) -> np.ndarray[Any, Any]:
    """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
Python
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
197
198
199
def events_to_frames(
    events: np.ndarray[Any, Any],
    width: int,
    height: int,
    dt_us: float = 10000.0,
    duration_us: float | None = None,
) -> np.ndarray[Any, Any]:
    """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