feat(inlet): add low-latency two-phase chunk pulls

.gitignore

@@ -22,3 +22,4 @@ liblsl.zip
 
 .claude
 CLAUDE.md
+.codegraph/

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,13 +228,61 @@ 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.
 
         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

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)]