From 6ac8296221831e03843c420b1abc870be030ced7 Mon Sep 17 00:00:00 2001 From: Stefan Appelhoff Date: Sat, 8 Aug 2026 12:04:38 +0200 Subject: [PATCH 1/4] Scope the log level set by verbose to the call `load_xdf(verbose=...)` set the level on the module logger and never restored it, so a single call reconfigured logging for the rest of the process. It also broke the documented meaning of `verbose=None` ("use root logger level") for every later call: once the logger has a level of its own, it no longer inherits. An application that had configured `pyxdf` to its liking would silently lose that configuration to any library that happened to pass `verbose`. --- CHANGELOG.md | 1 + src/pyxdf/pyxdf.py | 25 +++++++++++++++++++++- test/test_logging.py | 49 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 74 insertions(+), 1 deletion(-) create mode 100644 test/test_logging.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a247427..285f962 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ ## [UNRELEASED] - YYYY-MM-DD ### Fixed - Fix conversion to `int` for `sample_count` in XDF footer ([#173](https://github.com/xdf-modules/pyxdf/pull/173) by [Clemens Brunner](https://github.com/cbrnr)) +- Restore the logger level after `load_xdf` returns, so passing `verbose` no longer overrides the calling application's logging configuration for the rest of the process ([#175](https://github.com/xdf-modules/pyxdf/pull/175) by [Stefan Appelhoff](https://github.com/sappelhoff)) ## [1.17.5] - 2026-06-15 ### Changed diff --git a/src/pyxdf/pyxdf.py b/src/pyxdf/pyxdf.py index abbba6a..df73bf3 100644 --- a/src/pyxdf/pyxdf.py +++ b/src/pyxdf/pyxdf.py @@ -10,6 +10,7 @@ This function is closely following the load_xdf reference implementation. """ +import functools import gzip import io import itertools @@ -75,6 +76,26 @@ def __init__(self, xml): self.samplebytes = self.nchns * self.dtype.itemsize +def _scoped_log_level(func): + """Restore the logger level afterwards, so ``verbose`` only affects the call. + + Without this, passing ``verbose`` leaves the level set on the module logger for + the rest of the process, overriding whatever the application configured and + breaking the documented meaning of ``verbose=None`` for every later call. + """ + + @functools.wraps(func) + def wrapper(*args, **kwargs): + level = logger.level + try: + return func(*args, **kwargs) + finally: + logger.setLevel(level) + + return wrapper + + +@_scoped_log_level def load_xdf( filename, select_streams=None, @@ -121,7 +142,9 @@ def load_xdf( - None: load all streams (default). verbose : Passing True will set logging level to DEBUG, False will set it to - WARNING, and None will use root logger level. (default: None) + WARNING, and None will use root logger level. (default: None) The level is + restored when the call returns, so it does not affect later calls or the + rest of the application. synchronize_clocks : Whether to enable clock synchronization based on ClockOffset chunks. (default: true) diff --git a/test/test_logging.py b/test/test_logging.py new file mode 100644 index 0000000..7a0613d --- /dev/null +++ b/test/test_logging.py @@ -0,0 +1,49 @@ +import logging + +import pytest + +from pyxdf import load_xdf + +logger = logging.getLogger("pyxdf.pyxdf") + + +@pytest.fixture(autouse=True) +def restore_level(): + """Guard the test session against the very leak these tests check for.""" + level = logger.level + yield + logger.setLevel(level) + + +@pytest.mark.parametrize("verbose", [None, True, False]) +def test_verbose_does_not_leak_log_level(tmp_path, verbose): + """The level set by verbose is scoped to the call, including on failure.""" + logger.setLevel(logging.NOTSET) + + with pytest.raises(Exception, match="does not exist"): + load_xdf(tmp_path / "missing.xdf", verbose=verbose) + + assert logger.level == logging.NOTSET + + +def test_verbose_does_not_override_application_level(tmp_path): + """A level the application set is still in place after a call.""" + logger.setLevel(logging.ERROR) + + with pytest.raises(Exception, match="does not exist"): + load_xdf(tmp_path / "missing.xdf", verbose=True) + + assert logger.level == logging.ERROR + + +@pytest.mark.parametrize( + ("verbose", "expected"), [(True, True), (False, False), (None, False)] +) +def test_verbose_applies_during_the_call(tmp_path, caplog, verbose, expected): + """verbose still decides whether the import message is emitted.""" + logger.setLevel(logging.NOTSET) + + with pytest.raises(Exception, match="does not exist"): + load_xdf(tmp_path / "missing.xdf", verbose=verbose) + + assert ("Importing XDF file" in caplog.text) is expected From efb05cb27efae6a17c073cb31ca9f00a65711885 Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Sat, 8 Aug 2026 14:14:25 +0200 Subject: [PATCH 2/4] Tweaks --- CHANGELOG.md | 2 +- src/pyxdf/pyxdf.py | 22 +++++++++++----------- test/test_logging.py | 17 ++++++++++++++--- 3 files changed, 26 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 285f962..de1c26a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,7 @@ ## [UNRELEASED] - YYYY-MM-DD ### Fixed - Fix conversion to `int` for `sample_count` in XDF footer ([#173](https://github.com/xdf-modules/pyxdf/pull/173) by [Clemens Brunner](https://github.com/cbrnr)) -- Restore the logger level after `load_xdf` returns, so passing `verbose` no longer overrides the calling application's logging configuration for the rest of the process ([#175](https://github.com/xdf-modules/pyxdf/pull/175) by [Stefan Appelhoff](https://github.com/sappelhoff)) +- Scope the log level set by `verbose` to the `load_xdf` call, so it no longer overrides the calling application's logging configuration for the rest of the process ([#175](https://github.com/xdf-modules/pyxdf/pull/175) by [Stefan Appelhoff](https://github.com/sappelhoff)) ## [1.17.5] - 2026-06-15 ### Changed diff --git a/src/pyxdf/pyxdf.py b/src/pyxdf/pyxdf.py index df73bf3..775ff49 100644 --- a/src/pyxdf/pyxdf.py +++ b/src/pyxdf/pyxdf.py @@ -76,8 +76,8 @@ def __init__(self, xml): self.samplebytes = self.nchns * self.dtype.itemsize -def _scoped_log_level(func): - """Restore the logger level afterwards, so ``verbose`` only affects the call. +def verbose(func): + """Apply the ``verbose`` argument of ``func`` to the logger for that call only. Without this, passing ``verbose`` leaves the level set on the module logger for the rest of the process, overriding whatever the application configured and @@ -85,17 +85,20 @@ def _scoped_log_level(func): """ @functools.wraps(func) - def wrapper(*args, **kwargs): + def wrapper(*args, verbose=None, **kwargs): + if verbose is None: + return func(*args, verbose=verbose, **kwargs) level = logger.level + logger.setLevel(logging.DEBUG if verbose else logging.WARNING) try: - return func(*args, **kwargs) + return func(*args, verbose=verbose, **kwargs) finally: logger.setLevel(level) return wrapper -@_scoped_log_level +@verbose def load_xdf( filename, select_streams=None, @@ -142,9 +145,9 @@ def load_xdf( - None: load all streams (default). verbose : Passing True will set logging level to DEBUG, False will set it to - WARNING, and None will use root logger level. (default: None) The level is - restored when the call returns, so it does not affect later calls or the - rest of the application. + WARNING, and None will use root logger level. The previous level is restored + before this function returns, so the setting does not affect later calls or + the rest of the application. (default: None) synchronize_clocks : Whether to enable clock synchronization based on ClockOffset chunks. (default: true) @@ -223,9 +226,6 @@ def load_xdf( Examples: >>> streams, fileheader = load_xdf('myrecording.xdf') """ - if verbose is not None: - logger.setLevel(logging.DEBUG if verbose else logging.WARNING) - logger.info("Importing XDF file %s..." % filename) # if select_streams is an int or a list of int, load only streams associated with diff --git a/test/test_logging.py b/test/test_logging.py index 7a0613d..d207cf7 100644 --- a/test/test_logging.py +++ b/test/test_logging.py @@ -26,6 +26,17 @@ def test_verbose_does_not_leak_log_level(tmp_path, verbose): assert logger.level == logging.NOTSET +def test_verbose_none_leaves_the_logger_untouched(tmp_path, monkeypatch): + """With verbose=None the level is never set, so nothing is restored over it.""" + calls = [] + monkeypatch.setattr(logger, "setLevel", calls.append) + + with pytest.raises(Exception, match="does not exist"): + load_xdf(tmp_path / "missing.xdf", verbose=None) + + assert calls == [] + + def test_verbose_does_not_override_application_level(tmp_path): """A level the application set is still in place after a call.""" logger.setLevel(logging.ERROR) @@ -36,12 +47,12 @@ def test_verbose_does_not_override_application_level(tmp_path): assert logger.level == logging.ERROR -@pytest.mark.parametrize( - ("verbose", "expected"), [(True, True), (False, False), (None, False)] -) +@pytest.mark.parametrize(("verbose", "expected"), [(True, True), (False, False)]) def test_verbose_applies_during_the_call(tmp_path, caplog, verbose, expected): """verbose still decides whether the import message is emitted.""" logger.setLevel(logging.NOTSET) + # Capture whatever the logger lets through, whichever level pytest was invoked with + caplog.set_level(logging.NOTSET) with pytest.raises(Exception, match="does not exist"): load_xdf(tmp_path / "missing.xdf", verbose=verbose) From 1217b5817180a3f4223ad8c09fd35f6ab041b232 Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Sat, 8 Aug 2026 14:30:58 +0200 Subject: [PATCH 3/4] Fix Ruff --- src/pyxdf/__init__.py | 2 +- src/pyxdf/cli/playback_lsl.py | 9 ++- src/pyxdf/cli/print_metadata.py | 4 +- src/pyxdf/pyxdf.py | 96 ++++++++++++++++---------------- test/mock_data_stream.py | 26 +++++---- test/test_clock_sync.py | 8 +-- test/test_data.py | 16 +++--- test/test_detect_breaks.py | 4 +- test/test_detect_clock_resets.py | 6 +- test/test_library_basic.py | 7 +-- test/test_mock_data_stream.py | 13 ++--- test/test_segment_indices.py | 5 +- 12 files changed, 97 insertions(+), 99 deletions(-) diff --git a/src/pyxdf/__init__.py b/src/pyxdf/__init__.py index cbb9ff5..a207fce 100644 --- a/src/pyxdf/__init__.py +++ b/src/pyxdf/__init__.py @@ -1,4 +1,4 @@ from .__version__ import __version__ as __version__ from .pyxdf import load_xdf, match_streaminfos, resolve_streams -__all__ = ["load_xdf", "resolve_streams", "match_streaminfos"] +__all__ = ["load_xdf", "match_streaminfos", "resolve_streams"] diff --git a/src/pyxdf/cli/playback_lsl.py b/src/pyxdf/cli/playback_lsl.py index cc7dd7b..3fa42b6 100644 --- a/src/pyxdf/cli/playback_lsl.py +++ b/src/pyxdf/cli/playback_lsl.py @@ -2,7 +2,6 @@ import sys import time from dataclasses import dataclass -from typing import List, Optional import numpy as np import pylsl @@ -58,7 +57,7 @@ def __init__( self, rate: float = 1.0, loop_time: float = 0.0, - max_sample_rate: Optional[float] = None, + max_sample_rate: float | None = None, ): """ Create an object that tracks file playback time at optional non-realtime rate. @@ -126,7 +125,7 @@ def step_range(self) -> tuple[float, float]: def t0(self) -> float: return self._wall_start + self._n_loop * self._boundary - def sleep(self, duration: Optional[float] = None) -> None: + def sleep(self, duration: float | None = None) -> None: if duration is None: if self._max_srate <= 0: duration = 0.005 @@ -163,7 +162,7 @@ def main( wrap_dur = max(wrap_dur, tvec[-1] - tvec[0] + 1 / srate) # Create list of Streamer objects - streamers: List[Streamer] = [] + streamers: list[Streamer] = [] for strm_ix, strm in enumerate(streams): tvec = strm["time_stamps"] srate = float(strm["info"]["nominal_srate"][0]) @@ -222,7 +221,7 @@ def main( read_heads[streamer.name] = stop_idx if not loop and all( - [t_stop >= streamer.tvec[-1] for streamer in streamers] + t_stop >= streamer.tvec[-1] for streamer in streamers ): print("Playback finished.") break diff --git a/src/pyxdf/cli/print_metadata.py b/src/pyxdf/cli/print_metadata.py index 90caf5f..3a151c6 100644 --- a/src/pyxdf/cli/print_metadata.py +++ b/src/pyxdf/cli/print_metadata.py @@ -14,7 +14,7 @@ def main(fname: str): logging.basicConfig(level=logging.DEBUG) # Use logging.INFO to reduce output streams, _ = pyxdf.load_xdf(fname) - print("Found {} streams:".format(len(streams))) + print(f"Found {len(streams)} streams:") for ix, stream in enumerate(streams): msg = ( "Stream {}: {} - type {} - uid {} - shape {} in {} segments at {} " @@ -34,7 +34,7 @@ def main(fname: str): ) if any(stream["time_stamps"]): duration = stream["time_stamps"][-1] - stream["time_stamps"][0] - print("\tDuration: {} s".format(duration)) + print(f"\tDuration: {duration} s") print("Done.") diff --git a/src/pyxdf/pyxdf.py b/src/pyxdf/pyxdf.py index 775ff49..143f244 100644 --- a/src/pyxdf/pyxdf.py +++ b/src/pyxdf/pyxdf.py @@ -32,15 +32,15 @@ class StreamData: def __init__(self, xml): """Init a new StreamData object from a stream header.""" - fmts = dict( - double64=np.float64, - float32=np.float32, - string=object, - int32=np.int32, - int16=np.int16, - int8=np.int8, - int64=np.int64, - ) + fmts = { + "double64": np.float64, + "float32": np.float32, + "string": object, + "int32": np.int32, + "int16": np.int16, + "int8": np.int8, + "int64": np.int64, + } # number of channels self.nchns = int(xml["info"]["channel_count"][0]) # nominal sampling rate in Hz @@ -226,7 +226,7 @@ def load_xdf( Examples: >>> streams, fileheader = load_xdf('myrecording.xdf') """ - logger.info("Importing XDF file %s..." % filename) + logger.info(f"Importing XDF file {filename}...") # if select_streams is an int or a list of int, load only streams associated with # the corresponding stream IDs @@ -236,11 +236,11 @@ def load_xdf( pass elif isinstance(select_streams, int): select_streams = [select_streams] - elif all([isinstance(elem, dict) for elem in select_streams]): + elif all(isinstance(elem, dict) for elem in select_streams): select_streams = match_streaminfos(resolve_streams(filename), select_streams) if not select_streams: # no streams found raise ValueError("No matching streams found.") - elif not all([isinstance(elem, int) for elem in select_streams]): + elif not all(isinstance(elem, int) for elem in select_streams): raise ValueError( "Argument 'select_streams' must be an int, a list of ints, or a list of " "dicts." @@ -302,13 +302,16 @@ def load_xdf( continue else: # to be executed if no exception was raised - log_str += ", StreamId={}".format(StreamId) + log_str += f", StreamId={StreamId}" logger.debug(log_str) - if StreamId is not None and select_streams is not None: - if StreamId not in select_streams: - f.read(chunklen - 2 - 4) # skip remaining chunk contents - continue + if ( + StreamId is not None + and select_streams is not None + and StreamId not in select_streams + ): + f.read(chunklen - 2 - 4) # skip remaining chunk contents + continue # read the chunk's [Content]... if tag == 1: @@ -339,7 +342,7 @@ def load_xdf( # append to the time series... temp[StreamId].time_series.append(values) temp[StreamId].time_stamps.append(stamps) - except Exception as e: + except Exception as e: # noqa: BLE001 (tolerate any corruption in the chunk) # an error occurred (perhaps a chopped-off file): emit a warning and # scan forward to the next recognized chunk logger.error( @@ -418,7 +421,7 @@ def load_xdf( if len(stream.time_stamps) > 0: stream.segments.append((0, len(stream.time_stamps) - 1)) # inclusive - for k in streams.keys(): + for k in streams: stream = streams[k] tmp = temp[k] if "stream_id" in stream["info"]: # this is non-standard @@ -427,9 +430,8 @@ def load_xdf( "Using the 'stream_id' value {} from the beginning of the StreamHeader " "chunk instead.".format(stream["info"]["stream_id"], k) ) - if synchronize_clocks: - if tmp.segments != tmp.clock_segments: - logger.warning(f"Stream {k}: Segments and clock-segments differ") + if synchronize_clocks and tmp.segments != tmp.clock_segments: + logger.warning(f"Stream {k}: Segments and clock-segments differ") stream["info"]["stream_id"] = k stream["info"]["effective_srate"] = tmp.effective_srate stream["info"]["segments"] = tmp.segments @@ -451,20 +453,20 @@ def open_xdf(file): if isinstance(file, (io.RawIOBase, io.BufferedIOBase)): if isinstance(file, io.TextIOBase): - raise ValueError("file has to be opened in binary mode") + raise TypeError("file has to be opened in binary mode") f = file else: filename = Path(file) # ensure convert to pathlib object # check absolute path after following symlinks if not filename.resolve().exists(): - raise Exception("file %s does not exist." % filename) + raise FileNotFoundError(f"file {filename} does not exist.") if filename.suffix == ".xdfz" or filename.suffixes == [".xdf", ".gz"]: - f = gzip.open(str(filename), "rb") + f = gzip.open(str(filename), "rb") # noqa: SIM115 (returned open for caller) else: - f = open(str(filename), "rb") + f = open(str(filename), "rb") # noqa: SIM115 (returned open for caller) if f.read(4) != b"XDF:": # magic bytes - raise IOError("Invalid XDF file {}".format(file)) + raise OSError(f"Invalid XDF file {file}") return f @@ -892,10 +894,8 @@ def _clock_sync( if ts_start == ts_stop: logger.warning( - ( - f"Stream {stream_id}: " - f"No samples in clock offsets {range_i}, skipping..." - ) + f"Stream {stream_id}: " + f"No samples in clock offsets {range_i}, skipping..." ) else: stream.clock_segments.append((ts_start, ts_stop - 1)) @@ -1078,7 +1078,7 @@ def match_streaminfos(stream_infos, parameters, *, case_sensitive=True): match = False for request in parameters: for info in stream_infos: - for key in request.keys(): + for key in request: if case_sensitive: match = info[key] == request[key] else: @@ -1120,10 +1120,8 @@ def parse_xdf(fname): chunks : list List of all chunks contained in the XDF file. """ - chunks = [] with open_xdf(fname) as f: - for chunk in _read_chunks(f): - chunks.append(chunk) + chunks = list(_read_chunks(f)) return chunks @@ -1134,19 +1132,19 @@ def parse_chunks(chunks): if chunk["tag"] == 2: # stream header chunk # if you edit, check for consistency with parsing in load_xdf streams.append( - dict( - stream_id=chunk["stream_id"], - name=chunk.get("name"), # optional - type=chunk.get("type"), # optional - source_id=chunk.get("source_id"), # optional - created_at=chunk.get("created_at"), # optional - uid=chunk.get("uid"), # optional - session_id=chunk.get("session_id"), # optional - hostname=chunk.get("hostname"), # optional - channel_count=int(chunk["channel_count"]), - channel_format=chunk["channel_format"], - nominal_srate=float(chunk["nominal_srate"]), - ) + { + "stream_id": chunk["stream_id"], + "name": chunk.get("name"), # optional + "type": chunk.get("type"), # optional + "source_id": chunk.get("source_id"), # optional + "created_at": chunk.get("created_at"), # optional + "uid": chunk.get("uid"), # optional + "session_id": chunk.get("session_id"), # optional + "hostname": chunk.get("hostname"), # optional + "channel_count": int(chunk["channel_count"]), + "channel_format": chunk["channel_format"], + "nominal_srate": float(chunk["nominal_srate"]), + } ) return streams @@ -1165,7 +1163,7 @@ def _read_chunks(f): XDF chunk. """ while True: - chunk = dict() + chunk = {} try: chunk["nbytes"] = _read_varlen_int(f) except EOFError: diff --git a/test/mock_data_stream.py b/test/mock_data_stream.py index 72cd67c..ff227de 100644 --- a/test/mock_data_stream.py +++ b/test/mock_data_stream.py @@ -32,7 +32,7 @@ def __init__( srate=None, tdiff=None, fmt="float64", - time_stamps=[], + time_stamps=None, time_series=None, clock_times=None, clock_values=None, @@ -50,6 +50,8 @@ def __init__( self.srate = srate self.tdiff = tdiff self.fmt = fmt + if time_stamps is None: + time_stamps = [] # Ensure time_stamps is always a float64 array self.time_stamps = np.array(time_stamps, dtype="float64") # Use time_stamps as time_series data if no time_series data are provided @@ -67,16 +69,16 @@ def __init__( else: # Validate time_series data if fmt == "string": - if not all([isinstance(sample, list) for sample in time_series]): + if not all(isinstance(sample, list) for sample in time_series): raise ValueError("All string samples must be lists of strings") if not all( - [len(sample) == len(time_series[0]) for sample in time_series] + len(sample) == len(time_series[0]) for sample in time_series ): raise ValueError( "All samples must have the same number of channels" ) if not all( - [isinstance(x, str) for sample in time_series for x in sample] + isinstance(x, str) for sample in time_series for x in sample ): raise ValueError("All string sample values must be strings") self.time_series = time_series @@ -91,7 +93,7 @@ def __init__( if clock_times is not None: if not isinstance(clock_times, list): raise ValueError("Clock times must be a list") - if not all([isinstance(time, numbers.Number) for time in clock_times]): + if not all(isinstance(time, numbers.Number) for time in clock_times): raise ValueError("All clock times must be numeric") self.clock_times = clock_times else: @@ -99,15 +101,17 @@ def __init__( if clock_values is not None: if not isinstance(clock_values, list): raise ValueError("Clock values must be a list") - if not all([isinstance(value, numbers.Number) for value in clock_values]): + if not all(isinstance(value, numbers.Number) for value in clock_values): raise ValueError("All clock values must be numeric") self.clock_values = clock_values else: self.clock_values = [] - if clock_times is not None and clock_values is not None: - # If both clock_times and clock_values are provided they must be the same - # length - if len(clock_times) != len(clock_values): - raise ValueError("Clock times and values must be the same length") + # If both clock_times and clock_values are provided they must be the same length + if ( + clock_times is not None + and clock_values is not None + and len(clock_times) != len(clock_values) + ): + raise ValueError("Clock times and values must be the same length") self.segments = [] self.clock_segments = [] diff --git a/test/test_clock_sync.py b/test/test_clock_sync.py index 3096f3a..060114a 100644 --- a/test/test_clock_sync.py +++ b/test/test_clock_sync.py @@ -2,14 +2,14 @@ import numpy as np import pytest -from pyxdf.pyxdf import _clock_sync - from mock_data_stream import MockStreamData +from pyxdf.pyxdf import _clock_sync + # No clock resets -@pytest.mark.parametrize("n_clock_offsets", list(range(0, 5))) +@pytest.mark.parametrize("n_clock_offsets", list(range(5))) @pytest.mark.parametrize("handle_clock_resets", [True, False]) def test_sync_empty_stream(n_clock_offsets, handle_clock_resets): time_stamps = [] @@ -35,7 +35,7 @@ def test_sync_empty_stream(n_clock_offsets, handle_clock_resets): np.testing.assert_equal(streams[1].clock_segments, []) -@pytest.mark.parametrize("n_time_stamps", list(range(0, 5))) +@pytest.mark.parametrize("n_time_stamps", list(range(5))) @pytest.mark.parametrize("handle_clock_resets", [True, False]) def test_sync_empty_offsets(n_time_stamps, handle_clock_resets): tdiff = 1 diff --git a/test/test_data.py b/test/test_data.py index f50f635..4774f5d 100644 --- a/test/test_data.py +++ b/test/test_data.py @@ -130,12 +130,14 @@ def test_minimal_file(synchronize_clocks): # Time-series data s = [ [ - 'LabRecorder xdfwriter' - "5.1" - "5.99" - "-.01" - "-.02" - "" + ( + 'LabRecorder xdfwriter' + "5.1" + "5.99" + "-.01" + "-.02" + "" + ) ], ["Hello"], ["World"], @@ -165,7 +167,7 @@ def test_minimal_file(synchronize_clocks): @pytest.mark.skipif("minimal" not in files, reason="File not found.") def test_minimal_file_segments(jitter_break_threshold_seconds): path = files["minimal"] - streams, header = load_xdf( + streams, _ = load_xdf( path, dejitter_timestamps=True, jitter_break_threshold_seconds=jitter_break_threshold_seconds, diff --git a/test/test_detect_breaks.py b/test/test_detect_breaks.py index 3e5ccec..6886173 100644 --- a/test/test_detect_breaks.py +++ b/test/test_detect_breaks.py @@ -1,9 +1,9 @@ import numpy as np import pytest -from pyxdf.pyxdf import _detect_breaks - from mock_data_stream import MockStreamData +from pyxdf.pyxdf import _detect_breaks + # Segment at absolute time-intervals exceeding a given threshold. Sequences of # identical absolute intervals should be segmented the same regardless of # whether they are positive or negative. diff --git a/test/test_detect_clock_resets.py b/test/test_detect_clock_resets.py index 70296e4..5a43648 100644 --- a/test/test_detect_clock_resets.py +++ b/test/test_detect_clock_resets.py @@ -1,15 +1,15 @@ import numpy as np import pytest -from pyxdf.pyxdf import _detect_clock_resets - from mock_data_stream import MockStreamData +from pyxdf.pyxdf import _detect_clock_resets + # Test error condition @pytest.mark.parametrize("n_clock_offsets", [0, 1]) def test_detect_resets_length_error(n_clock_offsets): - clock_times = list(range(0, n_clock_offsets)) + clock_times = list(range(n_clock_offsets)) clock_values = [0] * n_clock_offsets stream = MockStreamData( clock_times=clock_times, diff --git a/test/test_library_basic.py b/test/test_library_basic.py index d9896ee..570b061 100644 --- a/test/test_library_basic.py +++ b/test/test_library_basic.py @@ -8,15 +8,12 @@ # %% test def test_load_xdf_present(): - """ - Check that pyxdf has the all important load_xdf. - This is nothing more than a placeholder so the CI system has a test to pass. - """ + """Check that pyxdf has the all important load_xdf.""" assert hasattr(pyxdf, "load_xdf") def test_read_varlen_int(): - """""" + """Check that _read_varlen_int decodes 1-, 4-, and 8-byte integers.""" def vla(data: bytes): return pyxdf.pyxdf._read_varlen_int(io.BytesIO(data)) diff --git a/test/test_mock_data_stream.py b/test/test_mock_data_stream.py index b648070..01c826d 100644 --- a/test/test_mock_data_stream.py +++ b/test/test_mock_data_stream.py @@ -2,7 +2,6 @@ import numpy as np import pytest - from mock_data_stream import MockStreamData @@ -153,7 +152,7 @@ def test_mock_stream_timeseries_defaults(fmt): srate = 1 tdiff = 1 # Time-stamps should be returned as a float64 array - time_stamps = list(range(0, 10)) + time_stamps = list(range(10)) clock_times = [0] clock_values = [-10] stream = MockStreamData( @@ -171,13 +170,11 @@ def test_mock_stream_timeseries_defaults(fmt): assert np.isdtype(stream.time_stamps.dtype, np.float64) np.testing.assert_equal(stream.time_stamps, time_stamps) if fmt == "string": - assert all([isinstance(sample, list) for sample in stream.time_series]) + assert all(isinstance(sample, list) for sample in stream.time_series) assert all( - [ - isinstance(value, str) - for sample in stream.time_series - for value in sample - ] + isinstance(value, str) + for sample in stream.time_series + for value in sample ) np.testing.assert_allclose( np.array(stream.time_series, dtype=np.float64)[:, 0], time_stamps diff --git a/test/test_segment_indices.py b/test/test_segment_indices.py index 5c2c364..f8f596c 100644 --- a/test/test_segment_indices.py +++ b/test/test_segment_indices.py @@ -2,6 +2,7 @@ import numpy as np import pytest + from pyxdf.pyxdf import _find_segment_indices @@ -35,12 +36,12 @@ def test_segments(b_breaks, expected): ( size, n_breaks, - list(itertools.combinations(range(0, size), n_breaks)), + list(itertools.combinations(range(size), n_breaks)), ) # All sequences of length 3 to 5, continuing from the tests above. for size in range(3, 6) # All numbers of breaks between 0 and size - for n_breaks in range(0, size + 1) + for n_breaks in range(size + 1) ] for breaks in [np.array(breaks, dtype=int) for breaks in breaks_list] ], From 6e173ddb140b699d072c8c2d291fb027cbdbd9af Mon Sep 17 00:00:00 2001 From: Clemens Brunner Date: Sat, 8 Aug 2026 14:32:29 +0200 Subject: [PATCH 4/4] Ruff format --- src/pyxdf/cli/playback_lsl.py | 4 +--- test/test_mock_data_stream.py | 4 +--- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/src/pyxdf/cli/playback_lsl.py b/src/pyxdf/cli/playback_lsl.py index 3fa42b6..53054d6 100644 --- a/src/pyxdf/cli/playback_lsl.py +++ b/src/pyxdf/cli/playback_lsl.py @@ -220,9 +220,7 @@ def main( ) read_heads[streamer.name] = stop_idx - if not loop and all( - t_stop >= streamer.tvec[-1] for streamer in streamers - ): + if not loop and all(t_stop >= streamer.tvec[-1] for streamer in streamers): print("Playback finished.") break timer.sleep() diff --git a/test/test_mock_data_stream.py b/test/test_mock_data_stream.py index 01c826d..aec9f8a 100644 --- a/test/test_mock_data_stream.py +++ b/test/test_mock_data_stream.py @@ -172,9 +172,7 @@ def test_mock_stream_timeseries_defaults(fmt): if fmt == "string": assert all(isinstance(sample, list) for sample in stream.time_series) assert all( - isinstance(value, str) - for sample in stream.time_series - for value in sample + isinstance(value, str) for sample in stream.time_series for value in sample ) np.testing.assert_allclose( np.array(stream.time_series, dtype=np.float64)[:, 0], time_stamps