From 5574959128f207c9523b2aeeee913c5c7b8e8097 Mon Sep 17 00:00:00 2001 From: Lukas Tenbrink Date: Sun, 8 Nov 2026 14:39:58 +0100 Subject: [PATCH] Support reading records from fsspec's memory filesystem Add memory:// to the protocols handled through fsspec, so that records can be read without touching the disk, for example from an upload form. Build signal and segment paths with forward slashes for URIs. They used to contain backslashes on Windows, which also affected cloud URIs. Co-Authored-By: Claude Opus 5.5 --- changelog.d/memory-uris.rst | 2 + docs/io.rst | 32 ++++++++++++ src/wfdb/io/_coreio.py | 53 ++++++++++++++++++- src/wfdb/io/_signal.py | 10 ++-- src/wfdb/io/record.py | 63 +++++++++++----------- tests/test_memory.py | 101 ++++++++++++++++++++++++++++++++++++ 6 files changed, 222 insertions(+), 39 deletions(-) create mode 100644 changelog.d/memory-uris.rst create mode 100644 tests/test_memory.py diff --git a/changelog.d/memory-uris.rst b/changelog.d/memory-uris.rst new file mode 100644 index 00000000..91acb973 --- /dev/null +++ b/changelog.d/memory-uris.rst @@ -0,0 +1,2 @@ +**Read records from memory** + ``rdrecord``, ``rdsamp`` and ``rdheader`` now accept ``memory://`` URIs, which read from the in-memory filesystem of ``fsspec``, as ``rdann`` already did. This also makes cloud URIs work for signal files and multi-segment records on Windows, where the file paths used to be built with backslashes. diff --git a/docs/io.rst b/docs/io.rst index 0c435577..10595198 100644 --- a/docs/io.rst +++ b/docs/io.rst @@ -136,6 +136,38 @@ handles credential discovery automatically. against the configured PhysioNet index URL. +Reading From Memory +------------------- + +Records and annotations can also be read from the in-memory filesystem of +``fsspec``, for example when the files come from an upload form instead of +from disk. Store all files of the record in the same directory, and pass a +``memory://`` URI as the ``record_name`` argument: + +.. code-block:: python + + import fsspec + import wfdb + + fs = fsspec.filesystem("memory") + fs.pipe("/upload/100.hea", hea_bytes) + fs.pipe("/upload/100.dat", dat_bytes) + fs.pipe("/upload/100.atr", atr_bytes) + + record = wfdb.rdrecord("memory://upload/100") + ann = wfdb.rdann("memory://upload/100", "atr") + +The files stay in memory until they are removed, for example with +``fs.rm("/upload", recursive=True)``. + +.. warning:: + + The in-memory filesystem is shared by the whole process, and everything + that opens a file shares a single file position. Use a unique directory + for each upload, and do not read the same record from several threads at + once, as this can return wrong data without raising an error. + + Downloading ----------- diff --git a/src/wfdb/io/_coreio.py b/src/wfdb/io/_coreio.py index 74f76822..c8040c4c 100644 --- a/src/wfdb/io/_coreio.py +++ b/src/wfdb/io/_coreio.py @@ -1,3 +1,4 @@ +import os import posixpath import fsspec @@ -6,7 +7,57 @@ from wfdb.io.download import config # Cloud protocols -CLOUD_PROTOCOLS = ["az://", "azureml://", "s3://", "gs://"] +CLOUD_PROTOCOLS = ["az://", "azureml://", "s3://", "gs://", "memory://"] + + +def _split_path(record_name): + """Split a local path or cloud URI into its directory and base name. + + Unlike `os.path.split`, this keeps the protocol of a URI intact if the URI + has no directory, so that "memory://100" gives "memory://" and "100". + + Parameters + ---------- + record_name : str or os.PathLike + The local path or cloud URI. + + Returns + ------- + dir_name : str + The local directory or cloud URI of the directory. + base_name : str + The last component of the path. + + """ + path = os.fspath(record_name) + if any(path.startswith(proto) for proto in CLOUD_PROTOCOLS): + protocol, separator, location = path.partition("://") + dir_name, base_name = posixpath.split(location) + return protocol + separator + dir_name, base_name + return os.path.split(path) + + +def _join_path(dir_name, file_name): + """Join a local directory or cloud URI with a file name. + + URIs always use forward slashes, regardless of the operating system. + + Parameters + ---------- + dir_name : str + The local directory or cloud URI. + file_name : str + The name of the file. + + Returns + ------- + str + The path or URI of the file. + + """ + if any(dir_name.startswith(proto) for proto in CLOUD_PROTOCOLS): + return posixpath.join(dir_name, file_name) + return os.path.join(dir_name, file_name) def _open_file( diff --git a/src/wfdb/io/_signal.py b/src/wfdb/io/_signal.py index 2b671c31..9b5e4dc8 100644 --- a/src/wfdb/io/_signal.py +++ b/src/wfdb/io/_signal.py @@ -1,7 +1,6 @@ import io import math import os -import posixpath import sys import fsspec @@ -1711,7 +1710,7 @@ def _rd_dat_file( # Local or cloud dat file elif pn_dir is None: - with fsspec.open(os.path.join(dir_name, file_name), "rb") as fp: + with fsspec.open(_coreio._join_path(dir_name, file_name), "rb") as fp: fp.seek(start_byte) sig_data = util.fromfile( fp, dtype=np.dtype(DATA_LOAD_TYPES[fmt]), count=element_count @@ -1916,7 +1915,7 @@ def _rd_compressed_file( open_file = wfdb_archive.open(file_name) else: if pn_dir is None: - file_name = os.path.join(dir_name, file_name) + file_name = _coreio._join_path(dir_name, file_name) open_file = _coreio._open_file(pn_dir, file_name, "rb") with open_file as fp: @@ -2600,10 +2599,9 @@ def _infer_sig_len( if wfdb_archive is not None: file_size = wfdb_archive.file_size(file_name) - # If this is a cloud path, use posixpath to construct the path and fsspec to open - # file + # If this is a cloud path, use fsspec to open the file elif any(dir_name.startswith(proto) for proto in CLOUD_PROTOCOLS): - with fsspec.open(posixpath.join(dir_name, file_name), mode="rb") as f: + with fsspec.open(_coreio._join_path(dir_name, file_name), mode="rb") as f: file_size = f.seek(0, os.SEEK_END) # If the PhysioNet database path is provided, construct the download path using the diff --git a/src/wfdb/io/record.py b/src/wfdb/io/record.py index 2d280710..5b8afe44 100644 --- a/src/wfdb/io/record.py +++ b/src/wfdb/io/record.py @@ -10,7 +10,7 @@ import pandas as pd from wfdb.io import _header, _signal, _url, download, header, util -from wfdb.io._coreio import CLOUD_PROTOCOLS +from wfdb.io._coreio import CLOUD_PROTOCOLS, _join_path, _split_path from wfdb.io.archive import WFDBArchive, archive_for_writing, is_archive_path # -------------- WFDB Signal Calibration and Classification ---------- # @@ -169,8 +169,8 @@ class BaseRecord: files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). n_sig : int, optional Total number of signals. fs : int, float, optional @@ -700,8 +700,8 @@ class Record(BaseRecord, _header.HeaderMixin, _signal.SignalMixin): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). n_sig : int, optional Total number of signals. fs : float, optional @@ -1080,8 +1080,8 @@ class MultiRecord(BaseRecord, _header.MultiHeaderMixin): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). n_sig : int, optional Total number of signals. fs : int, float, optional @@ -1837,10 +1837,10 @@ def rdheader(record_name, pn_dir=None, rd_segments=False): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). To read the header from a - local .wfdb archive, pass the path of the archive including the - .wfdb extension. + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). + To read the header from a local .wfdb archive, pass the path of + the archive including the .wfdb extension. pn_dir : str, optional Option used to stream data from Physionet. The Physionet database directory from which to find the required record files. @@ -1861,7 +1861,7 @@ def rdheader(record_name, pn_dir=None, rd_segments=False): >>> ecg_record = wfdb.rdheader('100', pn_dir='mitdb') """ - dir_name, base_record_name = os.path.split(record_name) + dir_name, base_record_name = _split_path(record_name) file_name = f"{base_record_name}.hea" # Read the header, and any segment headers, from a .wfdb archive @@ -1873,10 +1873,9 @@ def rdheader(record_name, pn_dir=None, rd_segments=False): _read_segment_headers(record, dir_name, pn_dir, archive) return record - # If this is a cloud path, use posixpath to construct the path and fsspec to open - # file + # If this is a cloud path, use fsspec to open the file if any(dir_name.startswith(proto) for proto in CLOUD_PROTOCOLS): - with fsspec.open(posixpath.join(dir_name, file_name), mode="r") as f: + with fsspec.open(_join_path(dir_name, file_name), mode="r") as f: header_content = f.read() # If the PhysioNet database path is provided, construct the download path using the @@ -1942,7 +1941,7 @@ def _read_record_header(record_name, dir_name, pn_dir, wfdb_archive): """ if wfdb_archive is not None: return _parse_header_content(wfdb_archive.read_header(record_name)) - return rdheader(os.path.join(dir_name, record_name), pn_dir=pn_dir) + return rdheader(_join_path(dir_name, record_name), pn_dir=pn_dir) def _read_segment_headers(record, dir_name, pn_dir, wfdb_archive): @@ -2069,10 +2068,10 @@ def rdrecord( files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). To read a record from a local - .wfdb archive, pass the path of the archive including the .wfdb - extension. + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). + To read a record from a local .wfdb archive, pass the path of + the archive including the .wfdb extension. sampfrom : int, optional The starting sample number to read for all channels. sampto : int, 'end', optional @@ -2201,7 +2200,7 @@ def _rdrecord( archive. """ - dir_name, base_record_name = os.path.split(record_name) + dir_name, base_record_name = _split_path(record_name) # Update the dir_name using abspath unless it is a cloud path if not any(dir_name.startswith(proto) for proto in CLOUD_PROTOCOLS): dir_name = os.path.abspath(dir_name) @@ -2409,7 +2408,7 @@ def _rdrecord( ) else: record.segments[seg_num] = rdrecord( - os.path.join(dir_name, record.seg_name[seg_num]), + _join_path(dir_name, record.seg_name[seg_num]), sampfrom=seg_ranges[i][0], sampto=seg_ranges[i][1], channels=seg_channels[i], @@ -2470,8 +2469,8 @@ def rdsamp( files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). sampfrom : int, optional The starting sample number to read for all channels. sampto : int, 'end', optional @@ -2590,8 +2589,8 @@ def sampfreq(record_name, pn_dir=None): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). pn_dir : str, optional Option used to stream data from Physionet. The Physionet database directory from which to find the required record files. @@ -2640,8 +2639,8 @@ def signame(record_name, pn_dir=None, sig_nums=[]): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). pn_dir : str, optional Option used to stream data from Physionet. The Physionet database directory from which to find the required record files. @@ -2699,8 +2698,8 @@ def wfdbdesc(record_name, pn_dir=None): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). pn_dir : str, optional Option used to stream data from Physionet. The Physionet database directory from which to find the required record files. @@ -2833,8 +2832,8 @@ def wfdbtime(record_name, input_times, pn_dir=None): files are searched for locally, or when `record_name` is a supported cloud URI, they are read directly from cloud storage via `fsspec`. Supported URIs: `s3://`, `gs://`, `az://`, - `azureml://`. Install the appropriate `fsspec` backend - (e.g., `s3fs`, `gcsfs`, `adlfs`). + `azureml://`, `memory://`. Other than for `memory://`, install the + appropriate `fsspec` backend (e.g., `s3fs`, `gcsfs`, `adlfs`). input_times : str, list The desired times (or samples) to be converted and displayed for the chosen record. This can either be a single string, or a list of diff --git a/tests/test_memory.py b/tests/test_memory.py new file mode 100644 index 00000000..68890bc4 --- /dev/null +++ b/tests/test_memory.py @@ -0,0 +1,101 @@ +from pathlib import Path + +import fsspec +import numpy as np +import pytest + +import wfdb + +SAMPLE_DATA = Path("sample-data") + + +@pytest.fixture +def memory_fs(): + """Provide the in-memory filesystem and remove the test files afterwards.""" + fs = fsspec.filesystem("memory") + yield fs + if fs.exists("/wfdb-test"): + fs.rm("/wfdb-test", recursive=True) + + +def upload(fs, pattern): + """Copy the sample files that match a pattern to the in-memory filesystem.""" + for path in SAMPLE_DATA.glob(pattern): + if path.is_file(): + name = path.relative_to(SAMPLE_DATA).as_posix() + fs.pipe(f"/wfdb-test/{name}", path.read_bytes()) + + +@pytest.mark.parametrize( + ("record_name", "pattern", "kwargs"), + [ + pytest.param("100", "100.*", {"sampto": 1000}, id="format-212"), + pytest.param("100", "100.*", {"sampfrom": 500, "sampto": 1000}, id="sampfrom"), + pytest.param("flacformats", "flacformats.*", {}, id="flac"), + pytest.param("s0010_re", "s0010_re.*", {"sampto": 100}, id="multiple-dat"), + pytest.param( + "multi-segment/fixed1/v102s", + "multi-segment/fixed1/*", + {"sampto": 1000}, + id="multi-segment-fixed", + ), + pytest.param( + "multi-segment/041s/041s", + "multi-segment/041s/*", + {"sampto": 1000}, + id="multi-segment-variable", + ), + ], +) +def test_rdrecord(memory_fs, record_name, pattern, kwargs): + upload(memory_fs, pattern) + + record = wfdb.rdrecord(f"memory://wfdb-test/{record_name}", **kwargs) + + expected = wfdb.rdrecord(str(SAMPLE_DATA / record_name), **kwargs) + assert record.sig_name == expected.sig_name + np.testing.assert_array_equal(record.p_signal, expected.p_signal) + + +def test_rdheader(memory_fs): + upload(memory_fs, "100.*") + + header = wfdb.rdheader("memory://wfdb-test/100") + + assert header.sig_name == wfdb.rdheader("sample-data/100").sig_name + + +def test_rdann(memory_fs): + upload(memory_fs, "100.*") + + annotation = wfdb.rdann("memory://wfdb-test/100", "atr") + + expected = wfdb.rdann("sample-data/100", "atr") + np.testing.assert_array_equal(annotation.sample, expected.sample) + assert annotation.symbol == expected.symbol + + +@pytest.mark.parametrize("uri", ["memory://100", "memory:///100"]) +def test_record_without_directory(uri): + fs = fsspec.filesystem("memory") + names = ["100.hea", "100.dat"] + for name in names: + fs.pipe(f"/{name}", (SAMPLE_DATA / name).read_bytes()) + + try: + header = wfdb.rdheader(uri) + record = wfdb.rdrecord(uri, sampto=1000) + finally: + for name in names: + fs.rm(f"/{name}") + + expected = wfdb.rdrecord("sample-data/100", sampto=1000) + assert header.sig_name == expected.sig_name + np.testing.assert_array_equal(record.p_signal, expected.p_signal) + + +def test_missing_file(memory_fs): + upload(memory_fs, "100.*") + + with pytest.raises(FileNotFoundError): + wfdb.rdrecord("memory://wfdb-test/nonexistent")