diff --git a/src/ezmsg/lsl/inlet.py b/src/ezmsg/lsl/inlet.py index b2b9850..1713e6d 100644 --- a/src/ezmsg/lsl/inlet.py +++ b/src/ezmsg/lsl/inlet.py @@ -118,6 +118,33 @@ class LSLInfo: channel_format: typing.Optional[str] = None +# Never let liblsl recover a lost stream on its own. Its recovery matches on +# source_id alone, so it ignores the `host` criterion and can silently re-attach +# to a same-named stream on a different machine; it also retries forever, so an +# upstream that returns with a different shape is never found and the inlet +# stalls silently. Reconnecting here instead costs no data -- a dropped +# connection discards the outlet's per-consumer queue either way -- and makes +# the loss observable. See `_produce`. +_RECOVER = False + +# Bound on fetching the full StreamInfo after opening. The description arrives +# over the wire (discovery results carry only the bare fields), and pylsl would +# otherwise wait forever, stranding a connect attempt that can no longer be +# cancelled. +_INFO_TIMEOUT = 5.0 + + +def _describe_target(info: LSLInfo) -> str: + """Human-readable form of what an inlet is looking for, for log messages.""" + named = (("name", info.name), ("type", info.type), ("host", info.host)) + parts = [f"{key}={value!r}" for key, value in named if value] + if info.channel_count is not None: + parts.append(f"channel_count={info.channel_count}") + if info.channel_format is not None: + parts.append(f"channel_format={info.channel_format!r}") + return ", ".join(parts) if parts else "any stream" + + def _sanitize_kwargs(kwargs: dict) -> dict: if "info" not in kwargs: replace_keys = set() @@ -183,6 +210,37 @@ class LSLInletSettings(ez.Settings): backlog, but may split that backlog across successive messages. """ + reconnect_grace_dur: float = 5.0 + """ + Seconds after losing a stream during which only the *same* stream (matching + ``source_id``) is accepted. Afterwards any stream matching ``info`` is taken. + + This reproduces liblsl's own recovery, which re-acquires by ``source_id`` + alone, but bounds it: liblsl retries indefinitely, so an upstream that comes + back with a different shape (and therefore a different ``source_id``) would + never be found. The grace period prefers the original stream while it might + still be coming back, then falls back to the resolver criteria so a + restarted, reconfigured upstream is picked up. Set to 0 to always take the + first match; the ``host`` criterion is honoured either way, which liblsl's + recovery does not do. + """ + + distinct_key_per_connection: bool = False + """ + Whether ``key`` gains a ``#`` suffix that increments each time the inlet + attaches to a *different outlet instance* (a changed StreamInfo ``uid``). + + Downstream processors key their state on ``(shape, rate, key)``, so a + restarted upstream that keeps the same name and shape is otherwise invisible + to them: filter state and partial windows carry across the discontinuity as + if no gap occurred. Enabling this forces those resets. + + Off by default because it changes the identity that ``key`` denotes: NWB + writers name containers by it and pipelines route on it, so a reconnect would + fork the recording. A dropped socket that re-attaches to the *same* outlet + instance never bumps the suffix, so a brief blip preserves state either way. + """ + @processor_state class LSLInletProducerState: @@ -219,8 +277,32 @@ class _PullSnapshot: class LSLInletProducer(BaseStatefulProducer[LSLInletSettings, typing.Optional[AxisArray], LSLInletProducerState]): def __init__(self, *args, settings: typing.Optional[LSLInletSettings] = None, **kwargs): kwargs = _sanitize_kwargs(kwargs) + # Also set in _reset_state, which only runs on the first __acall__. + self._logged_searching = False + self._logged_lost = False + self._lost = False + # Describes the *previous* connection, so unlike the flags above these + # deliberately survive _reset_state -- that reset is how a reconnect + # happens, and comparing against the old stream is the point. + self._connection_epoch = 0 + self._last_uid: typing.Optional[str] = None + self._last_source_id: typing.Optional[str] = None + self._last_signature: typing.Optional[tuple] = None + self._reconnect_source_id: typing.Optional[str] = None + self._reconnect_deadline = 0.0 super().__init__(*args, settings=settings, **kwargs) + def update_settings(self, new_settings: LSLInletSettings) -> None: + # New settings may retarget the inlet entirely, so holding out for the + # previous stream's source_id would delay connecting to the one just + # asked for. Both paths reach _reset_state, hence clearing it here. + self._clear_reconnect_preference() + super().update_settings(new_settings) + + def _clear_reconnect_preference(self) -> None: + self._reconnect_source_id = None + self._reconnect_deadline = 0.0 + def _reset_state(self) -> None: # Drop any existing connection and its derived state so a settings # change (e.g. a new target stream pushed via INPUT_SETTINGS) forces a @@ -233,6 +315,10 @@ def _reset_state(self) -> None: self._state.msg_template = None self._state.fetch_buffer = None self._warmed_up = False + # Log-once flags: both conditions are polled every tick, so log on transition only. + self._logged_searching = False + self._logged_lost = False + self._lost = False self._state.resolver = pylsl.ContinuousResolver(pred=None, forget_after=30.0) self._state.clock_sync = ClockSync() @@ -260,46 +346,75 @@ def _try_connect(self) -> None: channel_count=info.channel_count, channel_format=info.channel_format, ) - inlet = pylsl.StreamInlet(strm_info, max_chunklen=1, processing_flags=self.settings.processing_flags) + inlet = pylsl.StreamInlet( + strm_info, + max_chunklen=1, + recover=_RECOVER, + processing_flags=self.settings.processing_flags, + ) try: inlet.open_stream(timeout=2.0) except (pylsl.util.TimeoutError, pylsl.util.LostError): return self._state.inlet = inlet - self._setup_after_open() + if not self._setup_after_open(): + self._state.inlet = None return # Resolver-based path: match on whichever fields are provided. if self._state.resolver is None: return results: list[pylsl.StreamInfo] = self._state.resolver.results() - for strm_info in results: - b_match = True - b_match = b_match and ((not info.name) or strm_info.name() == info.name) - b_match = b_match and ((not info.type) or strm_info.type() == info.type) - b_match = b_match and ((not info.host) or strm_info.hostname() == info.host) - if info.channel_count is not None: - b_match = b_match and strm_info.channel_count() == info.channel_count - if info.channel_format is not None: - expected_cf = _string2cf.get(info.channel_format) - if expected_cf is not None: - b_match = b_match and strm_info.channel_format() == expected_cf - if b_match: - self._open_inlet(strm_info) - break + matches = [strm_info for strm_info in results if self._matches_criteria(strm_info)] + if not matches: + return + + # Within the grace window after a loss, hold out for the stream we were + # on. Matching on source_id is what liblsl's own recovery does; doing it + # here keeps the `host` criterion applied, which that recovery ignores. + if self._reconnect_source_id and time.monotonic() < self._reconnect_deadline: + same_source = [_ for _ in matches if _.source_id() == self._reconnect_source_id] + if not same_source: + return + matches = same_source + + self._open_inlet(matches[0]) + + def _matches_criteria(self, strm_info: pylsl.StreamInfo) -> bool: + """Whether a discovered stream satisfies every field set on ``settings.info``.""" + info = self.settings.info + if info.name and strm_info.name() != info.name: + return False + if info.type and strm_info.type() != info.type: + return False + if info.host and strm_info.hostname() != info.host: + return False + if info.channel_count is not None and strm_info.channel_count() != info.channel_count: + return False + if info.channel_format is not None: + expected_cf = _string2cf.get(info.channel_format) + if expected_cf is not None and strm_info.channel_format() != expected_cf: + return False + return True def _open_inlet(self, strm_info: pylsl.StreamInfo) -> None: """Create a StreamInlet from a discovered StreamInfo and set up buffers/template.""" self._state.inlet = pylsl.StreamInlet( strm_info, max_chunklen=1, + recover=_RECOVER, processing_flags=self.settings.processing_flags, ) self._state.inlet.open_stream(timeout=5.0) - self._setup_after_open() + if not self._setup_after_open(): + self._state.inlet = None + + def _setup_after_open(self) -> bool: + """Configure fetch buffer and message template after a stream is opened. - def _setup_after_open(self) -> None: - """Configure fetch buffer and message template after a stream is opened.""" + Returns False if the stream's full description could not be fetched, in + which case the caller drops the inlet and the connect is retried. + """ # Re-thread the first-data warmup on every (re)connect. self._warmed_up = False # Resolver is no longer needed once connected. Destroy it now (while we're @@ -307,12 +422,61 @@ def _setup_after_open(self) -> None: # run during shutdown. self._state.resolver = None - inlet_info = self._state.inlet.info() + self._logged_searching = False + self._logged_lost = False + + try: + inlet_info = self._state.inlet.info(timeout=_INFO_TIMEOUT) + except (pylsl.util.TimeoutError, pylsl.util.LostError) as exc: + ez.logger.warning("LSL inlet could not fetch the stream description (%s); retrying.", exc) + return False + # Fill in nominal_srate on settings (it may have been left at default). self.settings.info.nominal_srate = inlet_info.nominal_srate() # If possible, create a destination buffer for faster pulls. fmt = inlet_info.channel_format() n_ch = inlet_info.channel_count() + + # `uid` identifies the outlet *instance* and is regenerated whenever an + # outlet is constructed, so it -- not `source_id`, which is deliberately + # stable across restarts so consumers can re-acquire -- is what tells a + # dropped socket apart from a restarted upstream. + uid = inlet_info.uid() + source_id = inlet_info.source_id() + hostname = inlet_info.hostname() + if self._last_uid is not None and uid != self._last_uid: + self._connection_epoch += 1 + self._last_uid = uid + self._last_source_id = source_id + self._clear_reconnect_preference() + + # Name the resolved stream: resolution can cross machines and find the + # wrong one, which otherwise looks like a stream that is merely quiet. + ez.logger.info( + "LSL inlet connected to name=%r type=%r on host %r: %d ch @ %g Hz", + inlet_info.name(), + inlet_info.type(), + hostname, + n_ch, + inlet_info.nominal_srate(), + ) + + # A reconnect that lands on a different shape or a different machine is + # still a valid match on the configured criteria, so nothing else will + # complain -- but it silently changes what the data means. + signature = (n_ch, fmt, inlet_info.nominal_srate(), hostname) + if self._last_signature is not None and signature != self._last_signature: + changes = [ + f"{label}: {old!r} -> {new!r}" + for label, old, new in zip( + ("channel_count", "channel_format", "nominal_srate", "host"), + self._last_signature, + signature, + ) + if old != new + ] + ez.logger.warning("LSL inlet reconnected to a changed stream (%s).", "; ".join(changes)) + self._last_signature = signature if fmt in fmt2npdtype: dtype = fmt2npdtype[fmt] n_buff = int(self.settings.local_buffer_dur * inlet_info.nominal_srate()) or 1000 @@ -336,12 +500,22 @@ def _setup_after_open(self) -> None: time_ax = ( AxisArray.TimeAxis(fs=fs) if fs else AxisArray.CoordinateAxis(data=np.array([]), dims=["time"], unit="s") ) + # Epoch 0 renders as the bare name so the common case is unchanged. + key = inlet_info.name() + if self.settings.distinct_key_per_connection and self._connection_epoch: + key = f"{key}#{self._connection_epoch}" self._state.msg_template = AxisArray( data=np.empty((0, n_ch)), dims=["time", "ch"], axes={"time": time_ax, "ch": ch_ax}, - key=inlet_info.name(), + key=key, + attrs={ + "lsl_uid": uid, + "lsl_source_id": source_id, + "lsl_hostname": hostname, + }, ) + return True def _snapshot_pull_state(self) -> typing.Optional[_PullSnapshot]: """Capture strong references needed by a pull before entering a worker.""" @@ -392,8 +566,36 @@ def _pull(self, snapshot: _PullSnapshot, timeout: float = 0.0) -> typing.Optiona else: samples, timestamps = inlet.pull_chunk(timeout=timeout, min_samples=1) samples = np.array(samples) + except pylsl.util.LostError: + # Terminal for this connection: liblsl only raises this once the + # stream is gone for good (`recover=False`), and every later pull + # raises too. Flag it for `_produce` to act on rather than tearing + # down here -- this runs on a worker thread. + # + # Nothing is salvageable at this point. liblsl checks the lost state + # before draining, so buffered samples are unreachable even though + # `samples_available()` still counts them. + if self._state.inlet is inlet: + self._lost = True + if not self._logged_lost: + self._logged_lost = True + ez.logger.warning( + "LSL inlet lost the stream %r; will re-resolve.", + snapshot.msg_template.key, + ) + return None except Exception: - # The remote stream may have been lost or the inlet closed externally. + # Some other failure -- a closed handle, a malformed chunk. Not known + # to be terminal, so keep pulling. Log once per connection; stay quiet + # if this inlet is no longer the live one, which is shutdown or reset + # racing an in-flight pull. + if not self._logged_lost and self._state.inlet is inlet: + self._logged_lost = True + ez.logger.warning( + "LSL inlet pull failed for %r.", + snapshot.msg_template.key, + exc_info=True, + ) return None if not len(timestamps): @@ -447,9 +649,30 @@ async def _apull(self, timeout: float = 0.0) -> typing.Optional[AxisArray]: return result async def _produce(self) -> typing.Optional[AxisArray]: + if self._lost: + # `_reset_state` already drops the inlet, its buffers and template, + # and rebuilds the ContinuousResolver that `_setup_after_open` + # destroyed -- exactly the teardown a reconnect needs. Requesting a + # reset runs it at the top of the next `__acall__`, on this thread. + # The old inlet is released rather than closed, so an in-flight pull + # holding it via its snapshot stays valid until it returns. + self._lost = False + self._reconnect_source_id = self._last_source_id or None + self._reconnect_deadline = time.monotonic() + self.settings.reconnect_grace_dur + self._request_reset() + return None + if self._state.inlet is None: await asyncio.to_thread(self._try_connect) if self._state.inlet is None: + # Said once, on the first failed attempt. This line without a + # later "connected" line is the diagnosis. + if not self._logged_searching: + self._logged_searching = True + ez.logger.info( + "LSL inlet found no stream matching %s; still looking.", + _describe_target(self.settings.info), + ) await asyncio.sleep(0.01) return None diff --git a/tests/test_inlet.py b/tests/test_inlet.py index 43b6fc8..4c446e6 100644 --- a/tests/test_inlet.py +++ b/tests/test_inlet.py @@ -4,8 +4,10 @@ """ import asyncio +import os import tempfile import threading +import time import typing from pathlib import Path @@ -295,3 +297,415 @@ def test_inlet_comps_conns(rate: float): # We merely verify that the messages are being sent to the logger. assert len(messages) >= n_messages + + +class _FakeStreamInfo: + """Just enough of pylsl.StreamInfo for _setup_after_open.""" + + def __init__( + self, + name="CURSOR_PLAN", + stype="BCIPlan", + host="rpi5", + n_ch=2, + srate=50.0, + uid="uid-1", + source_id="src-1", + ): + self._name, self._type, self._host = name, stype, host + self._n_ch, self._srate = n_ch, srate + self._uid, self._source_id = uid, source_id + + def name(self): + return self._name + + def type(self): + return self._type + + def hostname(self): + return self._host + + def channel_count(self): + return self._n_ch + + def nominal_srate(self): + return self._srate + + def channel_format(self): + return pylsl.cf_float32 + + def uid(self): + return self._uid + + def source_id(self): + return self._source_id + + def desc(self): + return _FakeXML() + + +class _FakeXML: + def child(self, _name): + return self + + def empty(self): + return True + + +class _FakeInlet: + """Stands in for a connected pylsl.StreamInlet during _setup_after_open.""" + + def __init__(self, info=None): + self._info = info if info is not None else _FakeStreamInfo() + + def info(self, timeout=None): + return self._info + + +def _connect(producer, info=None): + """Drive _setup_after_open as though a stream had just been opened.""" + producer._state.inlet = _FakeInlet(info) + assert producer._setup_after_open() is True + + +def test_inlet_logs_the_stream_it_resolved(caplog): + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="CURSOR_PLAN", type="BCIPlan"))) + + with caplog.at_level("INFO"): + _connect(producer) + + assert "CURSOR_PLAN" in caplog.text + assert "BCIPlan" in caplog.text + assert "rpi5" in caplog.text + + +def test_inlet_stamps_stream_identity_onto_every_message(): + """`key` alone can't answer which outlet instance produced a segment.""" + producer = LSLInletProducer(settings=LSLInletSettings()) + _connect(producer, _FakeStreamInfo(uid="uid-abc", source_id="ezmsg-123", host="rpi5")) + + attrs = producer._state.msg_template.attrs + assert attrs["lsl_uid"] == "uid-abc" + assert attrs["lsl_source_id"] == "ezmsg-123" + assert attrs["lsl_hostname"] == "rpi5" + + +def test_setup_retries_when_the_description_cannot_be_fetched(caplog): + """The desc is fetched over the wire, so a connect can open but not complete.""" + + class TimingOutInlet: + def info(self, timeout=None): + raise pylsl.util.TimeoutError("no desc") + + producer = LSLInletProducer(settings=LSLInletSettings()) + producer._state.inlet = TimingOutInlet() + + with caplog.at_level("INFO"): + assert producer._setup_after_open() is False + + assert producer._state.msg_template is None + + +def test_inlet_logs_once_when_no_stream_matches(caplog): + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="CURSOR_PLAN", type="BCIPlan"))) + producer._state.clock_sync = None + + async def run_test(): + for _ in range(3): + assert await producer._produce() is None + + with caplog.at_level("INFO"): + asyncio.run(run_test()) + + searching = [r for r in caplog.records if "still looking" in r.message] + assert len(searching) == 1 + # Names what it wanted, so a typo'd stream name is legible from the log. + assert "CURSOR_PLAN" in searching[0].getMessage() + + +def test_a_reconnect_can_log_again(caplog): + """The log-once flags describe one connection, not the process lifetime.""" + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="CURSOR_PLAN", type="BCIPlan"))) + producer._logged_searching = True + producer._logged_lost = True + + _connect(producer) + assert producer._logged_searching is False + assert producer._logged_lost is False + + producer._state.clock_sync = None + producer._state.inlet = None + + async def run_test(): + assert await producer._produce() is None + + with caplog.at_level("INFO"): + asyncio.run(run_test()) + + assert any("still looking" in r.message for r in caplog.records) + + +def test_inlet_logs_a_lost_stream_once(caplog): + class RaisingInlet: + def pull_chunk(self, **_kwargs): + raise RuntimeError("stream lost") + + producer = LSLInletProducer(settings=LSLInletSettings()) + inlet = RaisingInlet() + producer._state.inlet = inlet + producer._state.clock_sync = object() + producer._state.msg_template = AxisArray( + data=np.empty((0, 2)), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=50.0)}, + key="CURSOR_PLAN", + ) + + with caplog.at_level("INFO"): + for _ in range(3): + assert asyncio.run(producer._apull(timeout=0.0)) is None + + lost = [r for r in caplog.records if r.levelname == "WARNING"] + assert len(lost) == 1 + assert "CURSOR_PLAN" in lost[0].getMessage() + + +def test_a_pull_racing_shutdown_is_not_reported_as_a_lost_stream(caplog): + """A pull raising on a handle shutdown() already dropped is teardown, not loss.""" + + class RaisingInlet: + def pull_chunk(self, **_kwargs): + raise RuntimeError("inlet closed") + + producer = LSLInletProducer(settings=LSLInletSettings()) + snapshot_inlet = RaisingInlet() + producer._state.inlet = snapshot_inlet + producer._state.clock_sync = object() + producer._state.msg_template = AxisArray( + data=np.empty((0, 2)), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=50.0)}, + key="CURSOR_PLAN", + ) + snapshot = producer._snapshot_pull_state() + producer.shutdown() + + with caplog.at_level("INFO"): + assert producer._pull(snapshot, timeout=0.0) is None + + assert [r for r in caplog.records if r.levelname == "WARNING"] == [] + + +def _template(key="CURSOR_PLAN", n_ch=2): + return AxisArray( + data=np.empty((0, n_ch)), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=50.0)}, + key=key, + ) + + +def test_a_lost_stream_drives_a_reconnect(caplog): + """LostError is terminal for the connection, so it must reach _produce.""" + + class LostInlet: + def pull_chunk(self, **_kwargs): + raise pylsl.util.LostError("the stream has been lost.") + + producer = LSLInletProducer(settings=LSLInletSettings()) + producer._state.inlet = LostInlet() + producer._state.clock_sync = object() + producer._state.msg_template = _template() + producer._last_source_id = "src-1" + + with caplog.at_level("INFO"): + assert asyncio.run(producer._apull(timeout=0.0)) is None + assert producer._lost is True + + assert asyncio.run(producer._produce()) is None + # Queued for the top of the next __acall__, where _reset_state tears the + # connection down and rebuilds the resolver. + assert producer._hash == -1 + assert producer._lost is False + assert producer._reconnect_source_id == "src-1" + assert producer._reconnect_deadline > time.monotonic() + + +def test_reconnect_holds_out_for_the_original_stream_then_gives_up(): + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="CURSOR_PLAN"))) + opened = [] + producer._open_inlet = opened.append + + replacement = _FakeStreamInfo(uid="uid-2", source_id="src-2") + producer._state.resolver = type("R", (), {"results": lambda _self: [replacement]})() + producer._reconnect_source_id = "src-1" + producer._reconnect_deadline = time.monotonic() + 30.0 + + producer._try_connect() + assert opened == [], "took a different stream while the original might return" + + producer._reconnect_deadline = time.monotonic() - 1.0 + producer._try_connect() + assert opened == [replacement], "never fell back to the resolver criteria" + + +def test_reconnect_takes_the_original_stream_when_it_is_back(): + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="CURSOR_PLAN"))) + opened = [] + producer._open_inlet = opened.append + + original = _FakeStreamInfo(uid="uid-9", source_id="src-1") + other = _FakeStreamInfo(uid="uid-2", source_id="src-2") + # Resolver order deliberately puts the impostor first. + producer._state.resolver = type("R", (), {"results": lambda _self: [other, original]})() + producer._reconnect_source_id = "src-1" + producer._reconnect_deadline = time.monotonic() + 30.0 + + producer._try_connect() + assert opened == [original] + + +def test_the_host_criterion_survives_a_reconnect(): + """liblsl's own recovery matches source_id alone and would ignore `host`.""" + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="CURSOR_PLAN", host="rpi5"))) + opened = [] + producer._open_inlet = opened.append + + elsewhere = _FakeStreamInfo(uid="uid-2", source_id="src-1", host="rpi6") + producer._state.resolver = type("R", (), {"results": lambda _self: [elsewhere]})() + producer._reconnect_source_id = "src-1" + producer._reconnect_deadline = time.monotonic() + 30.0 + + producer._try_connect() + assert opened == [], "re-attached to the right source_id on the wrong host" + + +def test_a_settings_change_drops_the_reconnect_preference(): + """New settings may retarget the inlet, so the old stream is no longer wanted.""" + producer = LSLInletProducer(settings=LSLInletSettings(info=LSLInfo(name="A"))) + producer._reconnect_source_id = "src-1" + producer._reconnect_deadline = time.monotonic() + 30.0 + + producer.update_settings(LSLInletSettings(info=LSLInfo(name="B"))) + + assert producer._reconnect_source_id is None + assert producer._reconnect_deadline == 0.0 + + +def test_a_restarted_upstream_bumps_the_key_epoch(): + producer = LSLInletProducer(settings=LSLInletSettings(distinct_key_per_connection=True)) + + _connect(producer, _FakeStreamInfo(uid="uid-1")) + assert producer._state.msg_template.key == "CURSOR_PLAN" + + # Same outlet instance: a dropped socket, not a restart. State stays valid. + _connect(producer, _FakeStreamInfo(uid="uid-1")) + assert producer._state.msg_template.key == "CURSOR_PLAN" + + # New outlet instance behind the same name and shape -- invisible downstream + # unless the key changes, since the hash is over (shape, rate, key). + _connect(producer, _FakeStreamInfo(uid="uid-2")) + assert producer._state.msg_template.key == "CURSOR_PLAN#1" + + +def test_key_is_stable_across_a_restart_by_default(): + """Off by default: NWB writers name containers by key and would fork one.""" + producer = LSLInletProducer(settings=LSLInletSettings()) + + _connect(producer, _FakeStreamInfo(uid="uid-1")) + _connect(producer, _FakeStreamInfo(uid="uid-2")) + + assert producer._state.msg_template.key == "CURSOR_PLAN" + assert producer._connection_epoch == 1, "epoch still tracked for attrs/provenance" + + +def test_reconnecting_to_a_changed_stream_warns(caplog): + """A different shape or machine still matches the criteria, so nothing else complains.""" + producer = LSLInletProducer(settings=LSLInletSettings()) + _connect(producer, _FakeStreamInfo(n_ch=64, host="rpi5")) + + with caplog.at_level("INFO"): + _connect(producer, _FakeStreamInfo(n_ch=32, host="rpi6")) + + warned = [r for r in caplog.records if r.levelname == "WARNING"] + assert len(warned) == 1 + message = warned[0].getMessage() + assert "channel_count" in message and "64" in message and "32" in message + assert "host" in message and "rpi6" in message + + +def test_reconnecting_to_an_identical_stream_does_not_warn(caplog): + producer = LSLInletProducer(settings=LSLInletSettings()) + _connect(producer, _FakeStreamInfo(uid="uid-1")) + + with caplog.at_level("INFO"): + _connect(producer, _FakeStreamInfo(uid="uid-2")) + + assert [r for r in caplog.records if r.levelname == "WARNING"] == [] + + +def test_a_lost_outlet_is_detected_and_replaced_by_a_restarted_one(caplog): + """End-to-end check of `recover=False`. + + With liblsl's own recovery enabled, a vanished outlet produces no error at + all -- the inlet retries the original source_id forever and simply stops + emitting, which is indistinguishable from a quiet stream. Here the loss must + surface, and a replacement advertising a different source_id and shape (a + reconfigured upstream, which liblsl would never re-acquire) must be picked up. + """ + name = f"TESTLOST_{os.getpid()}" + settings = LSLInletSettings( + info=LSLInfo(name=name, type="dummy"), + pull_timeout=0.01, + reconnect_grace_dur=0.5, + ) + producer = LSLInletProducer(settings=settings) + + def make_outlet(n_ch, source_id): + return pylsl.StreamOutlet( + pylsl.StreamInfo( + name=name, + type="dummy", + channel_count=n_ch, + nominal_srate=100.0, + channel_format=pylsl.cf_float32, + source_id=source_id, + ) + ) + + async def pump(outlet, n_ch, predicate, limit=400): + """Drive the producer until `predicate` holds, pushing if an outlet exists.""" + for _ in range(limit): + if outlet is not None: + outlet.push_chunk(np.zeros((10, n_ch), dtype=np.float32)) + msg = await producer.__acall__() + if predicate(msg): + return True + await asyncio.sleep(0.01) + return False + + with caplog.at_level("INFO"): + outlet = make_outlet(4, "src-original") + + connected = asyncio.run(pump(outlet, 4, lambda m: m is not None and np.prod(m.data.shape) > 0)) + assert connected, "never received data from the original outlet" + assert producer._state.msg_template.attrs["lsl_source_id"] == "src-original" + first_uid = producer._state.msg_template.attrs["lsl_uid"] + + del outlet + lost = asyncio.run(pump(None, 4, lambda _m: producer._reconnect_source_id is not None)) + assert lost, "a vanished outlet never surfaced as a lost stream" + assert producer._reconnect_source_id == "src-original" + + # Restarted upstream: same name/type, new instance, fewer channels. + replacement = make_outlet(2, "src-restarted") + back = asyncio.run(pump(replacement, 2, lambda m: m is not None and m.data.shape[1] == 2)) + assert back, "never reconnected to the replacement outlet" + + assert producer._state.msg_template.attrs["lsl_source_id"] == "src-restarted" + assert producer._state.msg_template.attrs["lsl_uid"] != first_uid + assert producer._connection_epoch == 1 + + assert any("lost the stream" in r.getMessage() for r in caplog.records) + changed = [r for r in caplog.records if "changed stream" in r.getMessage()] + assert changed and "channel_count" in changed[0].getMessage()