Skip to content

Support reading records from fsspec's memory:// #629

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ivorforce wants to merge 1 commit into
base: main
Choose a base branch
from
+222 −39
Open

Support reading records from fsspec's memory:// #629

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions changelog.d/memory-uris.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
32 changes: 32 additions & 0 deletions docs/io.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
-----------

Expand Down
53 changes: 52 additions & 1 deletion src/wfdb/io/_coreio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import os
import posixpath

import fsspec
Expand All @@ -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(
Expand Down
10 changes: 4 additions & 6 deletions src/wfdb/io/_signal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import io
import math
import os
import posixpath
import sys

import fsspec
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
63 changes: 31 additions & 32 deletions src/wfdb/io/record.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 ---------- #
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading