WFDB Python Package

import wfdb

# Core I/O operations
from wfdb import rdrecord, rdann, wrsamp, wrann

# Signal processing (must use module path)
import wfdb.processing
# Then use: wfdb.processing.xqrs_detect, wfdb.processing.find_peaks

# Plotting
from wfdb.plot import plot_wfdb, plot_items

# Format conversion (must use module path)
import wfdb.io.convert
# Then use: wfdb.io.convert.read_edf, wfdb.io.convert.wfdb_to_mat

import wfdb
import numpy as np

# Read a WFDB record from PhysioNet
record = wfdb.rdrecord('100', pn_dir='mitdb')
print(f"Record: {record.record_name}")
print(f"Signals: {record.sig_name}")
print(f"Length: {record.sig_len} samples at {record.fs} Hz")

# Read annotations
annotation = wfdb.rdann('100', 'atr', pn_dir='mitdb')
print(f"Found {len(annotation.sample)} annotations")

# Plot the first 3600 samples (10 seconds at 360 Hz)
wfdb.plot_wfdb(record=record, annotation=annotation, 
               time_units='seconds', title='ECG Record 100')

# Detect QRS complexes
qrs_inds = wfdb.processing.xqrs_detect(record.p_signal[:, 0], fs=record.fs)
print(f"Detected {len(qrs_inds)} QRS complexes")

# Calculate heart rate
hr = wfdb.processing.compute_hr(record.sig_len, qrs_inds, record.fs)
print(f"Mean heart rate: {np.mean(hr):.1f} BPM")

def rdrecord(record_name: str, sampfrom: int = 0, sampto: Union[int, str] = None, 
             channels: List[int] = None, physical: bool = True, 
             pn_dir: str = None) -> Union[Record, MultiRecord]: ...

def rdann(record_name: str, extension: str, sampfrom: int = 0, 
          sampto: Union[int, str] = 'end', pn_dir: str = None) -> Annotation: ...

def wrsamp(record_name: str, fs: float, units: List[str], sig_name: List[str],
           p_signal: np.ndarray = None, **kwargs) -> None: ...

def sampfreq(record_name: str, pn_dir: str = None) -> float: ...

def signame(record_name: str, pn_dir: str = None, sig_nums: List[int] = []) -> List[str]: ...

def wfdbdesc(record_name: str, pn_dir: str = None) -> List[str]: ...

def wfdbtime(record_name: str, input_times: Union[List, np.ndarray], pn_dir: str = None) -> List[str]: ...

def show_ann_labels() -> None: ...

def show_ann_classes() -> None: ...

def mrgann(record_name: str, extensions: List[str], pn_dir: str = None) -> None: ...

def dl_files(db: str, dl_dir: str, files: List[str], keep_subdirs: bool = True, overwrite: bool = False) -> None: ...

def get_dbs() -> List[str]: ...

def get_record_list(db_dir: str, records: str = "all") -> List[str]: ...

def set_db_index_url(db_index_url: str = None) -> None: ...

def show_data_sources() -> None: ...

def add_data_source(data_source: DataSource) -> None: ...

def remove_data_source(name: str) -> None: ...

def reset_data_sources() -> None: ...

class Record:
    record_name: str
    n_sig: int
    fs: float
    sig_len: int
    p_signal: np.ndarray
    sig_name: List[str]
    units: List[str]

class Annotation:
    record_name: str
    sample: np.ndarray
    symbol: List[str]
    fs: float

def xqrs_detect(sig: np.ndarray, fs: float, sampfrom: int = 0, sampto: str = "end", 
                conf: str = None, learn: bool = True, verbose: bool = True) -> np.ndarray: ...

def gqrs_detect(sig: np.ndarray, fs: float, **kwargs) -> np.ndarray: ...

def find_peaks(sig: np.ndarray) -> Tuple[np.ndarray, np.ndarray]: ...

def compute_hr(sig_len: int, qrs_inds: np.ndarray, fs: float) -> np.ndarray: ...

def resample_sig(x: np.ndarray, fs: float, fs_target: float) -> Tuple[np.ndarray, np.ndarray]: ...

class XQRS:
    def detect(self, sig: np.ndarray, verbose: bool = False) -> np.ndarray: ...

def plot_wfdb(record: Record = None, annotation: Annotation = None, 
              plot_sym: bool = False, time_units: str = 'seconds', **kwargs) -> None: ...

def plot_items(signal: np.ndarray = None, ann_samp: List[int] = None, 
               fs: float = None, time_units: str = "samples", **kwargs) -> None: ...

def plot_all_records(records: List[str], annotation: List[str] = None, 
                     **kwargs) -> None: ...

def read_edf(file_name: str, pn_dir: str = None, **kwargs) -> Record: ...

def wfdb_to_edf(record_name: str, pn_dir: str = None, **kwargs) -> None: ...

def wfdb_to_mat(record_name: str, pn_dir: str = None, **kwargs) -> None: ...

def csv_to_wfdb(file_name: str, fs: float, units: List[str], 
                sig_name: List[str], **kwargs) -> None: ...

class Record:
    """Single-segment WFDB record representation."""
    record_name: str
    n_sig: int
    fs: float
    sig_len: int
    p_signal: np.ndarray  # Physical signal values (MxN array)
    d_signal: np.ndarray  # Digital signal values (MxN array)
    sig_name: List[str]   # Signal names for each channel
    units: List[str]      # Units for each channel
    comments: List[str]   # Header comments
    base_time: datetime.time
    base_date: datetime.date
    
    def wrsamp(self, expanded: bool = False, write_dir: str = "") -> None: ...
    def to_dataframe(self) -> pd.DataFrame: ...

class MultiRecord:
    """Multi-segment WFDB record representation."""
    segments: List[Union[Record, None]]
    layout: str  # "fixed" or "variable"
    seg_name: List[str]
    seg_len: List[int]
    
    def multi_to_single(self, physical: bool, return_res: int = 64) -> Record: ...

class Annotation:
    """WFDB annotation representation."""
    record_name: str
    extension: str
    sample: np.ndarray      # Annotation sample locations
    symbol: List[str]       # Annotation symbols
    fs: float              # Sampling frequency
    aux_note: List[str]    # Auxiliary notes
    
    def wrann(self, write_fs: bool = False, write_dir: str = "") -> None: ...

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

class XQRS:
    """Configurable QRS detector."""
    def detect(self, sig: np.ndarray, verbose: bool = False) -> np.ndarray: ...

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