06 Python 分析事件数据

前置：先用 evt2_to_csv 或 MultiVision Studio 把 .raw 录像导出为 .csv（x, y, polarity, timestamp，见数据格式）。本页示例只读取 CSV、不调用任何 SDK，装好 pandas / numpy / matplotlib 即可运行。

没有相机？

可到资料下载 — 示例录像取一个 .raw，导出 CSV 后直接跑下面代码。

1. 读取与基本统计

import pandas as pd

df = pd.read_csv("events.csv")          # 列：x, y, polarity, timestamp(微秒)
print(df.head())

duration_us = df["timestamp"].iloc[-1] - df["timestamp"].iloc[0]
eps = len(df) / (duration_us / 1e6)
print(f"录制时长：{duration_us/1e6:.3f} s")
print(f"事件率 EPS：{eps:,.0f} events/s")
print(f"ON 事件占比：{(df['polarity']==1).mean():.1%}")

2. 用 numpy 把事件累加成一帧（简易 accumulation）

import numpy as np

H, W = 608, 768                 # 按你的传感器分辨率调整（CF-NRS1 EVS 为 768×608）
t0 = df["timestamp"].min()
window_us = 50_000              # 50 ms 累积窗

df_win = df[(df["timestamp"] - t0) < window_us]
frame = np.zeros((H, W), dtype=np.int16)
pol = np.where(df_win["polarity"].to_numpy() == 1, 1, -1)
np.add.at(frame, (df_win["y"].to_numpy(), df_win["x"].to_numpy()), pol)

调整 window_us 即可观察 accumulation time 对画面的影响：过小画面稀疏，过大运动拖尾。

3. 可视化

import matplotlib.pyplot as plt

plt.figure(figsize=(8, 6))
plt.imshow(frame, cmap="bwr", vmin=-5, vmax=5)
plt.title(f"accumulation {window_us/1000:.0f} ms")
plt.axis("off")
plt.colorbar(label="事件累加（红=ON，蓝=OFF）")
plt.savefig("frame.png", dpi=120)
plt.show()

也可画事件散点（XYT）观察运动轨迹：

sub = df_win.sample(min(len(df_win), 20000))      # 下采样避免点太密
plt.figure(figsize=(6, 6))
plt.scatter(sub["x"], sub["y"], c=sub["polarity"], s=1, cmap="bwr")
plt.gca().invert_yaxis()
plt.xlabel("x"); plt.ylabel("y"); plt.title("事件散点（XY）")
plt.savefig("scatter.png", dpi=120)
plt.show()

进阶：直接从相机实时采集

v2.0 的 Python 绑定（单一 hv_toolkit 模块）已支持直接从相机实时采集，同一套 API 通吃 USB（x86_64）与 MIPI HVS（S100/RK3588）。下面把上面离线分析的累加/统计思路接到实时流上：边采边解码、边累积成帧。

先构建绑定

Python 绑定默认关闭，构建加 --python：./run.sh --python build x86_64（USB）或 ./run.sh --python build s100（MIPI HVS）。完整步骤见第一个 Python 程序，接口见 Python API。

USB（x86_64）

import numpy as np
import hv_toolkit as hv

cfg = hv.DeviceConfig()
cfg.backend    = hv.Backend.Usb
cfg.vendor_id  = 0x1d6b
cfg.product_id = 0x0105

cam = hv.Camera()
cam.init(cfg)
cam.start_stream()

dec = hv.Evt2Decoder()          # USB 走 EVT2
H, W = 608, 768
f = hv.Frame()
window = []                     # 近窗事件缓存
for _ in range(100):
    if not cam.get_frame(f, 1000):
        continue
    ev = dec.decode(bytes(f.evs))      # → ndarray，字段 x/y/t/polarity
    window.append(ev)

# 与离线分析同样地累加成一帧
all_ev = np.concatenate(window)
frame = np.zeros((H, W), dtype=np.int16)
pol = np.where(all_ev['polarity'], 1, -1)
np.add.at(frame, (all_ev['y'], all_ev['x']), pol)

cam.stop_stream()
cam.destroy()

MIPI HVS（S100）

只换 backend 配置与解码器——Camera/Frame/get_frame 完全一致：

cfg = hv.DeviceConfig()
cfg.backend      = hv.Backend.MipiHvs
cfg.device_node  = "/dev/video0"
cfg.sensor_index = 0
cfg.i2c_bus      = 1

cam = hv.Camera(); cam.init(cfg); cam.start_stream()
dec = hv.MipiRaw8Decoder()       # MIPI HVS 的 Frame.evs 是 RAW8 子帧流，必须用它
f = hv.Frame()
while cam.get_frame(f, 1000):
    ev = dec.decode(bytes(f.evs))
    aps_bytes = f.aps.nbytes      # APS 通常为 NV12
    print(f"{len(ev)} events, aps={aps_bytes} bytes, fmt={f.format}")
cam.stop_stream(); cam.destroy()

板卡部署（LD_LIBRARY_PATH/PYTHONPATH 指到 /home/sunrise/build）见 Python API → 构建与部署。拿到实时事件后，可视化的画法与上文离线部分完全相同（matplotlib 累加图 / 散点图）。