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