tessl/pypi-wfdb

Python package for reading, writing, and processing physiologic signals and annotations in WFDB format.

Overview

Eval results

Files

I/O Operations

Name: tessl/pypi-wfdb
Author: tessl

Core functionality for reading and writing WFDB records and annotations, including database access, data source management, and file utilities. This module provides the fundamental data structures and functions needed to work with physiological signal data in WFDB format.

Capabilities

Record Reading

Read WFDB record files with flexible options for sample ranges, channel selection, and data format.

def rdheader(record_name: str, pn_dir: str = None, rd_segments: bool = False) -> Union[Record, MultiRecord]:
    """
    Read WFDB header file and return Record or MultiRecord object.
    
    Parameters:
    - record_name: str, record name without extension
    - pn_dir: str, optional PhysioNet database directory
    - rd_segments: bool, read segment headers for multi-segment records
    
    Returns:
    Record or MultiRecord object with metadata only
    """

def rdrecord(record_name: str, sampfrom: int = 0, sampto: Union[int, str] = None, 
             channels: List[int] = None, physical: bool = True, pn_dir: str = None,
             m2s: bool = True, smooth_frames: bool = True, ignore_skew: bool = False,
             return_res: int = 64, force_channels: bool = True, 
             channel_names: List[str] = None, warn_empty: bool = False) -> Union[Record, MultiRecord]:
    """
    Read WFDB record with signal data.
    
    Parameters:
    - record_name: str, record name without extension
    - sampfrom: int, starting sample number (0-indexed)
    - sampto: int or 'end', ending sample number
    - channels: list of int, channel indices to read
    - physical: bool, return physical units (True) or digital (False)
    - pn_dir: str, PhysioNet database directory
    - m2s: bool, convert MultiRecord to Record
    - smooth_frames: bool, smooth multi-sample per frame signals
    - ignore_skew: bool, ignore signal skew
    - return_res: int, return resolution (8, 16, 32, 64)
    - force_channels: bool, force channel matching for variable layout
    - channel_names: list of str, channel names to read
    - warn_empty: bool, warn if no signals returned
    
    Returns:
    Record or MultiRecord object with signal data
    """

def rdsamp(record_name: str, sampfrom: int = 0, sampto: Union[int, str] = None,
           channels: List[int] = None, pn_dir: str = None, 
           channel_names: List[str] = None, warn_empty: bool = False,
           return_res: int = 64) -> Tuple[np.ndarray, Dict[str, Any]]:
    """
    Read WFDB record signals and basic metadata.
    
    Parameters:
    - record_name: str, record name
    - sampfrom: int, starting sample number
    - sampto: int or 'end', ending sample number
    - channels: list of int, channel indices to read
    - pn_dir: str, PhysioNet database directory
    - channel_names: list of str, channel names to read
    - warn_empty: bool, warn if no signals returned
    - return_res: int, return resolution
    
    Returns:
    - signals: ndarray, physical signal array (MxN)
    - fields: dict, metadata including fs, units, sig_name, comments
    """

Record Writing

Write single-segment WFDB records with comprehensive metadata support.

def wrsamp(record_name: str, fs: float, units: List[str], sig_name: List[str],
           p_signal: np.ndarray = None, d_signal: np.ndarray = None,
           e_p_signal: Union[np.ndarray, List] = None, e_d_signal: Union[np.ndarray, List] = None,
           samps_per_frame: List[int] = None, fmt: List[str] = None,
           adc_gain: List[float] = None, baseline: List[int] = None,
           comments: List[str] = None, base_time: datetime.time = None,
           base_date: datetime.date = None, base_datetime: datetime.datetime = None,
           write_dir: str = "") -> None:
    """
    Write single-segment WFDB record.
    
    Parameters:
    - record_name: str, record name
    - fs: float, sampling frequency in Hz
    - units: list of str, signal units for each channel
    - sig_name: list of str, signal names for each channel
    - p_signal: ndarray, physical signals (MxN array)
    - d_signal: ndarray, digital signals (MxN array)
    - e_p_signal: ndarray or list, expanded physical signals
    - e_d_signal: ndarray or list, expanded digital signals
    - samps_per_frame: list of int, samples per frame for each channel
    - fmt: list of str, WFDB format for each channel
    - adc_gain: list of float, ADC gain values
    - baseline: list of int, digital baseline values
    - comments: list of str, header comments
    - base_time: datetime.time, record start time
    - base_date: datetime.date, record start date
    - base_datetime: datetime.datetime, combined date/time
    - write_dir: str, output directory
    """

Annotation I/O

Read and write WFDB annotation files with support for various annotation types and metadata.

def rdann(record_name: str, extension: str, sampfrom: int = 0, 
          sampto: Union[int, str] = 'end', shift_samps: bool = False,
          pn_dir: str = None, summarize_labels: bool = False,
          return_label_elements: List[str] = ['symbol'], 
          encoding: str = None) -> Annotation:
    """
    Read WFDB annotation file.
    
    Parameters:
    - record_name: str, record name without extension
    - extension: str, annotation file extension (e.g., 'atr', 'qrs')
    - sampfrom: int, starting sample number
    - sampto: int or 'end', ending sample number
    - shift_samps: bool, shift sample numbers by sampfrom
    - pn_dir: str, PhysioNet database directory
    - summarize_labels: bool, summarize contained labels
    - return_label_elements: list of str, label elements to return
    - encoding: str, file encoding
    
    Returns:
    Annotation object
    """

def wrann(record_name: str, extension: str, sample: Union[List[int], np.ndarray],
          symbol: List[str] = None, subtype: Union[List[int], np.ndarray] = None,
          chan: Union[List[int], np.ndarray] = None, 
          num: Union[List[int], np.ndarray] = None,
          aux_note: List[str] = None, fs: float = None, 
          write_dir: str = "") -> None:
    """
    Write WFDB annotation file.
    
    Parameters:
    - record_name: str, record name
    - extension: str, annotation file extension
    - sample: array-like, sample locations for annotations
    - symbol: list of str, annotation symbols
    - subtype: array-like, annotation subtypes
    - chan: array-like, signal channels for annotations
    - num: array-like, annotation numbers
    - aux_note: list of str, auxiliary notes
    - fs: float, sampling frequency
    - write_dir: str, output directory
    """

def show_ann_labels() -> None:
    """Display standard WFDB annotation labels and their meanings."""

def show_ann_classes() -> None:
    """Display standard WFDB annotation classes and categories."""

def mrgann(record_name: str, extension: str, ann_1: Annotation, ann_2: Annotation,
           fs: float, write_dir: str = "") -> None:
    """
    Merge two annotation objects and write to file.
    
    Parameters:
    - record_name: str, output record name
    - extension: str, output annotation extension
    - ann_1: Annotation, first annotation object
    - ann_2: Annotation, second annotation object  
    - fs: float, sampling frequency
    - write_dir: str, output directory
    """

Database Access

Download and access PhysioNet databases and records.

def dl_database(db_dir: str, dl_dir: str, records: Union[List[str], str] = "all",
                annotators: Union[List[str], str, None] = "all", 
                keep_subdirs: bool = True, overwrite: bool = False) -> None:
    """
    Download complete PhysioNet database.
    
    Parameters:
    - db_dir: str, PhysioNet database directory name
    - dl_dir: str, local download directory path
    - records: list of str or 'all', specific records to download
    - annotators: list of str, 'all', or None, annotation types to download
    - keep_subdirs: bool, preserve directory structure
    - overwrite: bool, overwrite existing files
    """

def dl_files(db: str, dl_dir: str, files: List[str], 
             keep_subdirs: bool = True, overwrite: bool = False) -> None:
    """
    Download specific files from PhysioNet database.
    
    Parameters:
    - db: str, database name
    - dl_dir: str, local download directory
    - files: list of str, specific files to download
    - keep_subdirs: bool, preserve directory structure
    - overwrite: bool, overwrite existing files
    """

def get_dbs() -> List[str]:
    """
    Get list of available PhysioNet databases.
    
    Returns:
    List of database names and descriptions
    """

def get_record_list(db_dir: str, records: Union[List[str], str] = "all") -> List[str]:
    """
    Get record list for a PhysioNet database.
    
    Parameters:
    - db_dir: str, database directory name
    - records: list of str or 'all', specific records or all
    
    Returns:
    List of record names in the database
    """

def set_db_index_url(db_index_url: str) -> None:
    """
    Set custom database index URL for PhysioNet access.
    
    Parameters:
    - db_index_url: str, URL to database index
    """

Data Source Management

Configure and manage data sources for accessing local and remote databases.

def show_data_sources() -> None:
    """Display currently configured data sources."""

def add_data_source(ds: DataSource) -> None:
    """
    Add new data source configuration.
    
    Parameters:
    - ds: DataSource, data source object to add
    """

def remove_data_source(ds_name: str) -> None:
    """
    Remove data source by name.
    
    Parameters:
    - ds_name: str, name of data source to remove
    """

def reset_data_sources(keep_pn: bool = False) -> None:
    """
    Reset data sources to default configuration.
    
    Parameters:
    - keep_pn: bool, keep PhysioNet data source
    """

Utility Functions

Additional utilities for record information and time conversion.

def sampfreq(record_name: str, pn_dir: str = None) -> None:
    """
    Display sampling frequency for each signal in a record.
    Prints signal names and frequencies to stdout.
    
    Parameters:
    - record_name: str, record name
    - pn_dir: str, PhysioNet database directory
    
    Returns:
    None - prints output to stdout
    """

def signame(record_name: str, pn_dir: str = None, sig_nums: List[int] = []) -> None:
    """
    Display signal names for a record.
    Prints signal names to stdout.
    
    Parameters:
    - record_name: str, record name
    - pn_dir: str, PhysioNet database directory
    - sig_nums: list of int, specific signal numbers to display
    
    Returns:
    None - prints output to stdout
    """

def wfdbdesc(record_name: str, pn_dir: str = None) -> None:
    """
    Display detailed record information.
    Prints comprehensive record metadata to stdout.
    
    Parameters:
    - record_name: str, record name
    - pn_dir: str, PhysioNet database directory
    
    Returns:
    None - prints output to stdout
    """

def wfdbtime(record_name: str, input_times: List[Union[int, str]], 
             pn_dir: str = None) -> None:
    """
    Convert and display time formats for a record.
    Prints formatted time information to stdout.
    
    Parameters:
    - record_name: str, record name
    - input_times: list, time values to convert
    - pn_dir: str, PhysioNet database directory
    
    Returns:
    None - prints output to stdout
    """

Types

class Record:
    """Single-segment WFDB record representation."""
    record_name: str          # Record name without extension
    n_sig: int               # Number of signals
    fs: float                # Sampling frequency in Hz  
    sig_len: int             # Signal length in samples
    p_signal: np.ndarray     # Physical signal values (MxN array)
    d_signal: np.ndarray     # Digital signal values (MxN array)
    e_p_signal: Union[np.ndarray, List]  # Expanded physical signals
    e_d_signal: Union[np.ndarray, List]  # Expanded digital signals
    sig_name: List[str]      # Signal names for each channel
    units: List[str]         # Units for each channel
    file_name: List[str]     # Data file names
    fmt: List[str]           # WFDB format for each channel
    adc_gain: List[float]    # ADC gain values
    baseline: List[int]      # Digital baseline values
    comments: List[str]      # Header comments
    base_time: datetime.time # Record start time
    base_date: datetime.date # Record start date
    base_datetime: datetime.datetime  # Combined date/time
    
    def get_frame_number(self, time_value: Union[int, float, str]) -> int: ...
    def get_elapsed_time(self, time_value: Union[int, float, str]) -> float: ...
    def get_absolute_time(self, time_value: Union[int, float, str]) -> datetime.datetime: ...
    def wrsamp(self, expanded: bool = False, write_dir: str = "") -> None: ...
    def to_dataframe(self) -> pd.DataFrame: ...
    def check_field(self, field: str, required_channels: Union[str, List[int]] = "all") -> bool: ...

class MultiRecord:
    """Multi-segment WFDB record representation."""
    segments: List[Union[Record, None]]  # List of Record objects or None for empty segments
    layout: str                         # "fixed" or "variable" layout
    seg_name: List[str]                 # Segment names
    seg_len: List[int]                  # Length of each segment
    sig_segments: List[List[int]]       # Signal segments
    
    def multi_to_single(self, physical: bool, return_res: int = 64, 
                       expanded: bool = False) -> Record: ...
    def wrsamp(self, write_dir: str = "") -> None: ...

class Annotation:
    """WFDB annotation representation."""
    record_name: str         # Associated record name
    extension: str           # Annotation file extension
    sample: np.ndarray       # Annotation sample locations
    symbol: List[str]        # Annotation symbols
    subtype: np.ndarray      # Annotation subtypes
    chan: np.ndarray         # Signal channels for annotations
    num: np.ndarray          # Annotation numbers
    aux_note: List[str]      # Auxiliary notes
    fs: float               # Sampling frequency
    label_store: np.ndarray  # Integer label encoding
    description: List[str]   # Label descriptions
    custom_labels: pd.DataFrame     # Custom annotation labels
    contained_labels: pd.DataFrame  # Labels present in annotation
    
    def apply_range(self, sampfrom: int = 0, sampto: int = None) -> Annotation: ...
    def wrann(self, write_fs: bool = False, write_dir: str = "") -> None: ...

class DataSource:
    """Data source configuration."""
    name: str                # Data source name
    ds_type: DataSourceType  # LOCAL or HTTP
    uri: str                # Data source URI

DataSourceType = Literal["LOCAL", "HTTP"]

SIGNAL_CLASSES: pd.DataFrame  # Signal classification information

Usage Examples

Basic Record Operations

import wfdb

# Read header only
record = wfdb.rdheader('100', pn_dir='mitdb')
print(f"Record has {record.n_sig} signals at {record.fs} Hz")

# Read specific channels and time range
record = wfdb.rdrecord('100', pn_dir='mitdb', 
                       sampfrom=0, sampto=3600,  # First 10 seconds
                       channels=[0, 1])           # First two channels

# Read signals without full record object
signals, fields = wfdb.rdsamp('100', pn_dir='mitdb', channels=[0])
print(f"Signal shape: {signals.shape}")
print(f"Sampling frequency: {fields['fs']} Hz")

Annotation Handling

import wfdb

# Read annotations
ann = wfdb.rdann('100', 'atr', pn_dir='mitdb')
print(f"Found {len(ann.sample)} annotations")
print(f"First 5 symbols: {ann.symbol[:5]}")

# Filter annotations by time range
ann_subset = ann.apply_range(sampfrom=1000, sampto=5000)

# Write annotations
wfdb.wrann('output', 'atr', 
           sample=[100, 200, 300],
           symbol=['N', 'V', 'N'])

Database Access

import wfdb

# List available databases
dbs = wfdb.get_dbs()
print(f"Found {len(dbs)} databases")

# Download specific record
wfdb.dl_files('mitdb', './data', ['100.hea', '100.dat', '100.atr'])

# Download entire database  
wfdb.dl_database('mitdb', './mitdb_data', records=['100', '101', '102'])

Install with Tessl CLI