From 9a0a4077c5ecd043e282b27daa6b5063a43879b8 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 22 Jul 2026 13:14:29 -0400 Subject: [PATCH 1/2] ignore .codegraph/ --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index f741155..1a50e3e 100644 --- a/.gitignore +++ b/.gitignore @@ -22,3 +22,4 @@ liblsl.zip .claude CLAUDE.md +.codegraph/ \ No newline at end of file From 4e3cd1b771b5ddc6330c83b009807d3c1c8a94a6 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 22 Jul 2026 14:04:12 -0400 Subject: [PATCH 2/2] feat(inlet): add low-latency chunk pulls --- src/pylsl/inlet.py | 53 ++++++++++++++- test/test_pull_chunk.py | 139 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 191 insertions(+), 1 deletion(-) diff --git a/src/pylsl/inlet.py b/src/pylsl/inlet.py index 6dbfdd7..6e1069e 100644 --- a/src/pylsl/inlet.py +++ b/src/pylsl/inlet.py @@ -1,4 +1,5 @@ import ctypes +import operator from .lib import lib, fmt2type, fmt2pull_sample, fmt2pull_chunk, cf_string from .util import handle_error, FOREVER @@ -209,7 +210,9 @@ def pull_sample(self, timeout=FOREVER, sample=None): else: return None, None - def pull_chunk(self, timeout=0.0, max_samples=1024, dest_obj=None): + def pull_chunk( + self, timeout=0.0, max_samples=1024, dest_obj=None, min_samples=None + ): """Pull a chunk of samples from the inlet. Keyword arguments: @@ -225,6 +228,14 @@ def pull_chunk(self, timeout=0.0, max_samples=1024, dest_obj=None): number of samples. A numpy buffer must be order='C' (default None) + min_samples -- Minimum number of samples to wait for before returning + the samples that are immediately available, up to + max_samples. Set this to 1 to wait up to timeout for the + first sample and then return without waiting for the + remainder of the chunk. If the timeout expires first, + fewer samples may be returned. The default of None + preserves the original behavior, which waits until + max_samples is reached or the timeout expires. Returns a tuple (samples,timestamps) where samples is a list of samples (each itself a list of values), and timestamps is a list of time-stamps. @@ -232,6 +243,46 @@ def pull_chunk(self, timeout=0.0, max_samples=1024, dest_obj=None): Throws a LostError if the stream source has been lost. """ + if min_samples is None: + return self._pull_chunk_once(timeout, max_samples, dest_obj) + + try: + min_samples = operator.index(min_samples) + except TypeError: + raise TypeError("min_samples must be an integer or None") from None + if not 1 <= min_samples <= max_samples: + raise ValueError("min_samples must be between 1 and max_samples") + + # liblsl's fixed-buffer pull waits for max_samples or timeout. First + # make that target min_samples, then drain whatever else is already + # available without blocking. This retains the existing behavior when + # min_samples is omitted while providing a low-latency mode when it is 1. + samples, timestamps = self._pull_chunk_once( + timeout, min_samples, dest_obj + ) + num_samples = len(timestamps) + remaining = max_samples - num_samples + if num_samples == 0 or remaining == 0: + return samples, timestamps + + if dest_obj is not None: + bytes_per_sample = ctypes.sizeof(self.value_type) * self.channel_count + dest_view = memoryview(dest_obj).cast("B")[ + num_samples * bytes_per_sample : + ] + else: + dest_view = None + + more_samples, more_timestamps = self._pull_chunk_once( + 0.0, remaining, dest_view + ) + if samples is not None: + samples.extend(more_samples) + timestamps.extend(more_timestamps) + return samples, timestamps + + def _pull_chunk_once(self, timeout, max_samples, dest_obj): + """Perform one fixed-buffer liblsl chunk pull.""" # look up a pre-allocated buffer of appropriate length num_channels = self.channel_count max_values = max_samples * num_channels diff --git a/test/test_pull_chunk.py b/test/test_pull_chunk.py index 0b3e751..af05270 100644 --- a/test/test_pull_chunk.py +++ b/test/test_pull_chunk.py @@ -6,8 +6,10 @@ variable-length string ("Markers"-style) chunk. """ +import ctypes import time +import numpy as np import pytest import pylsl @@ -65,3 +67,140 @@ def test_pull_chunk_roundtrip(channel_format: int, samples: list): assert got_samples == samples assert len(got_ts) == n_samples + + +def test_pull_chunk_min_samples_returns_available_data_without_waiting_for_max(): + samples = [[1.0], [2.0], [3.0]] + source_id = "test_pull_chunk_min_samples_id" + info = pylsl.StreamInfo( + name="test_pull_chunk_min_samples", + type="test", + channel_count=1, + nominal_srate=0, + channel_format=pylsl.cf_float32, + source_id=source_id, + ) + outlet = pylsl.StreamOutlet(info) + streams = pylsl.resolve_byprop("source_id", source_id, timeout=2) + assert streams, "outlet was not discovered" + inlet = pylsl.StreamInlet(streams[0]) + inlet.open_stream(timeout=2) + time.sleep(0.5) + outlet.push_chunk(samples) + + deadline = time.monotonic() + 2 + while inlet.samples_available() < len(samples) and time.monotonic() < deadline: + time.sleep(0.01) + assert inlet.samples_available() >= len(samples) + + started = time.monotonic() + got_samples, timestamps = inlet.pull_chunk( + timeout=1.0, max_samples=16, min_samples=1 + ) + elapsed = time.monotonic() - started + + assert got_samples == samples + assert len(timestamps) == len(samples) + assert elapsed < 0.5 + + +def test_pull_chunk_min_samples_drains_without_blocking(monkeypatch): + inlet = object.__new__(pylsl.StreamInlet) + calls = [] + responses = [ + ([[1.0], [2.0]], [1.0, 2.0]), + ([[3.0], [4.0]], [3.0, 4.0]), + ] + + def pull_once(timeout, max_samples, dest_obj): + calls.append((timeout, max_samples, dest_obj)) + return responses.pop(0) + + monkeypatch.setattr(inlet, "_pull_chunk_once", pull_once) + + samples, timestamps = inlet.pull_chunk( + timeout=0.5, max_samples=5, min_samples=2 + ) + + assert samples == [[1.0], [2.0], [3.0], [4.0]] + assert timestamps == [1.0, 2.0, 3.0, 4.0] + assert calls == [(0.5, 2, None), (0.0, 3, None)] + + +def test_pull_chunk_min_samples_timeout_does_not_drain(monkeypatch): + inlet = object.__new__(pylsl.StreamInlet) + calls = [] + + def pull_once(timeout, max_samples, dest_obj): + calls.append((timeout, max_samples, dest_obj)) + return [], [] + + monkeypatch.setattr(inlet, "_pull_chunk_once", pull_once) + + samples, timestamps = inlet.pull_chunk( + timeout=0.5, max_samples=5, min_samples=1 + ) + + assert samples == [] + assert timestamps == [] + assert calls == [(0.5, 1, None)] + + +def test_pull_chunk_min_samples_appends_into_dest_obj(monkeypatch): + inlet = object.__new__(pylsl.StreamInlet) + inlet.value_type = ctypes.c_float + inlet.channel_count = 2 + calls = [] + + def pull_once(timeout, max_samples, dest_obj): + calls.append((timeout, max_samples, dest_obj)) + view = np.frombuffer(dest_obj, dtype=np.float32).reshape(-1, 2) + if timeout: + view[0] = [1.0, 2.0] + return None, [1.0] + view[:2] = [[3.0, 4.0], [5.0, 6.0]] + return None, [2.0, 3.0] + + monkeypatch.setattr(inlet, "_pull_chunk_once", pull_once) + dest = np.zeros((4, 2), dtype=np.float32) + + samples, timestamps = inlet.pull_chunk( + timeout=0.5, + max_samples=len(dest), + dest_obj=dest, + min_samples=1, + ) + + assert samples is None + assert timestamps == [1.0, 2.0, 3.0] + np.testing.assert_array_equal( + dest, [[1.0, 2.0], [3.0, 4.0], [5.0, 6.0], [0.0, 0.0]] + ) + assert calls[0][:2] == (0.5, 1) + assert calls[0][2] is dest + assert calls[1][:2] == (0.0, 3) + assert isinstance(calls[1][2], memoryview) + + +@pytest.mark.parametrize("min_samples", [0, 5]) +def test_pull_chunk_rejects_invalid_min_samples(min_samples): + inlet = object.__new__(pylsl.StreamInlet) + + with pytest.raises(ValueError, match="between 1 and max_samples"): + inlet.pull_chunk(max_samples=4, min_samples=min_samples) + + +def test_pull_chunk_default_retains_single_call(monkeypatch): + inlet = object.__new__(pylsl.StreamInlet) + calls = [] + + def pull_once(timeout, max_samples, dest_obj): + calls.append((timeout, max_samples, dest_obj)) + return [[1.0]], [1.0] + + monkeypatch.setattr(inlet, "_pull_chunk_once", pull_once) + + result = inlet.pull_chunk(timeout=0.25, max_samples=7) + + assert result == ([[1.0]], [1.0]) + assert calls == [(0.25, 7, None)]