michaelsthr/pyvse

# pyvse Python SDK for IFM VSE100 / VSA004 acceleration sensors. Reverse-engineered communication protocol. Not affiliated with or endorsed by IFM. ## Install pip install pyvse **Dependencies:** `numpy` (required), `pandas` + `matplotlib` (optional, for export/visualization). ## Quick start ### One-shot collect import pyvse with pyvse.connect("192.168.0.1") as client: data = client.raw("50K").collect(seconds=5.0) data.to_csv("run1.csv") df = data.to_dataframe() # pandas DataFrame: time_s, accel_g arr = data.to_numpy() # np.ndarray shape (N,), unit: g ### Live stream import pyvse with pyvse.connect("192.168.0.1") as client: for frame in client.raw("50K").stream(batch_size=700): print(frame.g_values.mean()) # np.ndarray shape (700,), unit: g ### Stream to CSV (memory-efficient) import pyvse with pyvse.connect("192.168.0.1") as client: with pyvse.CSVSink("run1.csv", sample_rate=50_000) as sink: for frame in client.raw("50K").stream(): sink.push(frame.g_values) ### Stream to PlotJuggler import pyvse with pyvse.connect("192.168.0.1") as client: with pyvse.PlotJugglerSink() as pj: # default: 127.0.0.1:9870 UDP for frame in client.raw("50K").stream(): pj.push_all(frame.g_values) # one UDP message per sample ## API reference ### `pyvse.connect(host, port=3321, timeout=5.0) → VSEClient` Returns a context manager. Use `with` to ensure clean disconnect. ### `VSEClient` | Method | Returns | Description | |--------|---------|-------------| | `raw(rate="50K")` | `RawCommand` | rate: `"50K"` or `"100K"` | ### `RawCommand` | Method | Returns | Description | |--------|---------|-------------| | `collect(seconds)` | `RawData` | Collect fixed-length capture | | `stream(batch_size=700)` | `Iterator[RawFrame]` | Live stream; runs until interrupted | ### `RawData` | Attribute / Method | Description | |--------------------|-------------| | `g_values` | `np.ndarray` shape `(N,)`, float64, unit: g | | `sample_rate` | `int`, Hz | | `time_axis` | `np.ndarray` of timestamps in seconds | | `to_numpy()` | Returns `g_values` | | `to_dataframe()` | pandas DataFrame with columns `time_s`, `accel_g` | | `to_csv(path)` | Write CSV with `time_s`, `accel_g` columns | ### `RawFrame` | Attribute | Description | |-----------|-------------| | `g_values` | `np.ndarray` shape `(batch_size,)`, float64, unit: g | | `sample_rate` | `int`, Hz | ### `CSVSink(path, sample_rate)` Stream-friendly CSV writer. Use as context manager, call `.push(g_values)` per frame. ### `PlotJugglerSink(host="127.0.0.1", port=9870)` Pushes data to PlotJuggler via UDP JSON streamer. | Method | Description | |--------|-------------| | `push(g_values, channel="accel_g")` | Send batch mean as single message | | `push_all(g_values, channel="accel_g")` | Send each sample as individual message | ### Exceptions | Exception | Raised when | |-----------|-------------| | `VSEConnectionError` | TCP connect fails | | `VSETimeoutError` | Socket read times out | | `VSEProtocolError` | Unexpected response from device | | `VSEError` | Base class for all pyvse errors | ## Disclaimer Independent implementation developed through observation of device communication. Not affiliated with or endorsed by IFM.