Transforms & representations

EventCV transforms return a new stream, so they chain like a pipeline. Every transform and representation is available both as a method (stream.flip_x()) and as an OpenCV-style free function (ecv.flip_x(stream)).

import eventcv as ecv
import numpy as np
from pathlib import Path

# Point PATH at your own recording (.npz / .hdf5 / .bag / .txt / .aedat / .dat).
# Here we locate the small test fixture bundled with the EventCV source.
def _fixture(rel="data/test/example.npz"):
    for base in (Path.cwd(), *Path.cwd().parents):
        if (base / rel).exists():
            return str(base / rel)
    raise FileNotFoundError(rel)

PATH = _fixture()

stream = ecv.load(PATH)
len(stream), stream.sensor_size

(106295, (640, 480))

Chaining transforms

Geometry, temporal, and polarity ops compose; each returns a new stream.

pipe = (stream
        .crop(0, 0, 320, 240)   # x, y, width, height
        .hot_pixel_filter()     # drop noisy always-on pixels
        .flip_x())
print("events:", len(stream), "->", len(pipe))
print("sensor size:", pipe.sensor_size)

events: 106295 -> 19859
sensor size: (320, 240)

Two spellings, one operation

The functional form mirrors the methods exactly, so pick whichever reads better.

a = stream.flip_x().voxel(bins=5).numpy()
b = ecv.voxel(ecv.flip_x(stream), bins=5).numpy()
np.array_equal(a, b)

True

Representations

Render the pipeline output as dense tensors ready for NumPy / PyTorch.

for name, frame in [("count", pipe.count()),
                    ("voxel(bins=5)", pipe.voxel(bins=5)),
                    ("time surface", pipe.tsurf())]:
    arr = frame.numpy()
    print(f"{name:16} {str(arr.shape):18} {arr.dtype}")

count            (1, 240, 320)      uint64
voxel(bins=5)    (5, 240, 320)      float32
time surface     (2, 240, 320)      float32

As a PyTorch dataset

open(..., repr=…) gives the reader a default representation. Every slice remembers it, so reader.slice(n).view() renders that representation instead of raw polarity (pass a name to view(...) to override). It also makes the reader a map-style dataset: len(ds) frames, ds[i] a [C, H, W] array, and ds.batch(idx) a [B, C, H, W] stack — no PyTorch needed to produce the arrays.

ds = ecv.open(PATH, dt_ms=20, repr="count")
print("frames:", len(ds))
print("ds[0]:", ds[0].shape, ds[0].dtype)
print("batch([0, 1]):", ds.batch([0, 1]).shape)

frames: 3
ds[0]: (1, 480, 640) uint64
batch([0, 1]): (2, 1, 480, 640)

With PyTorch installed, hand the reader straight to a DataLoader. With a repr set, slices are dense arrays that collate into a [B, C, H, W] tensor automatically:

import torch

ds = ecv.open("recording.hdf5", dt_ms=30, repr="count")
loader = torch.utils.data.DataLoader(ds, batch_size=32, shuffle=True)
for batch in loader:      # batch: [B, C, H, W]
    ...

Without a repr, reader[i] is a raw EventStream. Those are sparse and variable-length, so they can’t stack into a tensor — pass eventcv.collate to batch them as a list instead:

reader = ecv.open("recording.hdf5", dt_ms=30)
loader = torch.utils.data.DataLoader(reader, batch_size=32, collate_fn=ecv.collate)
for batch in loader:      # batch: list[EventStream]
    batch[0].view()