API Reference: ecg_sdk.core.loaders

ecg_sdk.core.loaders

ECG Data Loaders Module

Provides a unified high-level factory function load_ecg and specialized loaders for clinical, benchmark, tabular, and synthetic ECG datasets: - Universal multi-source dispatcher: load_ecg(...) - PhysioNet WFDB records (local files or streaming from physionet.org) - Automatic sampling frequency standardization (target_fs) and scaling of annotations - Automatic voltage normalization (normalize='zscore'|'minmax'|'robust') - MIT-BIH Arrhythmia Database (mitdb) with expert cardiologist annotations - St. Petersburg INCART 12-lead Arrhythmia Database (incartdb) - PTB-XL 12-lead diagnostic ECG database (ptb-xl) - European ST-T Database (edb), Creighton University (cudb), Fantasia, etc. - CSV / JSON tabular formats and NumPy / Pandas in-memory representations.

PHYSIONET_DATABASES = {'mitdb': {'name': 'MIT-BIH Arrhythmia Database', 'pn_dir': 'mitdb', 'nominal_fs': 360.0, 'leads': ['MLII', 'V1/V5'], 'description': 'Standard inter-patient arrhythmia benchmark with cardiologist annotations.', 'records': ['100', '101', '102', '103', '104', '105', '106', '107', '108', '109', '111', '112', '113', '114', '115', '116', '117', '118', '119', '121', '122', '123', '124', '200', '201', '202', '203', '205', '207', '208', '209', '210', '212', '213', '214', '215', '217', '219', '220', '221', '222', '223', '228', '230', '231', '232', '233', '234']}, 'incartdb': {'name': 'St. Petersburg INCART 12-lead Arrhythmia Database', 'pn_dir': 'incartdb', 'nominal_fs': 257.0, 'leads': ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6'], 'description': '75 12-lead recordings with detailed arrhythmia and ischemic annotations.', 'records': [f'I{i:02d}' for i in range(1, 76)]}, 'ptb-xl': {'name': 'PTB-XL 12-Lead Diagnostic ECG Database', 'pn_dir': 'ptb-xl/1.0.3/', 'nominal_fs': 500.0, 'leads': ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6'], 'description': '21,837 clinical 12-lead records with SCP-ECG diagnostic statements.'}, 'nstdb': {'name': 'MIT-BIH Noise Stress Test Database', 'pn_dir': 'nstdb', 'nominal_fs': 360.0, 'leads': ['MLII', 'V1'], 'description': 'ECG signals with calibrated baseline wander, muscle tremor, and electrode noise.'}, 'afdb': {'name': 'MIT-BIH Atrial Fibrillation Database', 'pn_dir': 'afdb', 'nominal_fs': 250.0, 'leads': ['ECG1', 'ECG2'], 'description': '25 long-term recordings of human subjects with paroxysmal atrial fibrillation.'}, 'cudb': {'name': 'Creighton University Ventricular Tachyarrhythmia Database', 'pn_dir': 'cudb', 'nominal_fs': 250.0, 'leads': ['ECG'], 'description': '35 recordings of subjects experiencing ventricular tachycardia/fibrillation.'}, 'svdb': {'name': 'MIT-BIH Supraventricular Arrhythmia Database', 'pn_dir': 'svdb', 'nominal_fs': 128.0, 'leads': ['ECG1', 'ECG2'], 'description': '78 recordings of patients with supraventricular arrhythmias.'}, 'edb': {'name': 'European ST-T Database', 'pn_dir': 'edb', 'nominal_fs': 250.0, 'leads': ['Lead_0', 'Lead_1'], 'description': '90 ambulatory recordings for myocardial ischemia evaluation.'}, 'ludb': {'name': 'Lobachevsky University Electrocardiography Database', 'pn_dir': 'ludb', 'nominal_fs': 500.0, 'leads': ['I', 'II', 'III', 'aVR', 'aVL', 'aVF', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6'], 'description': '200 12-lead ECGs with delineations of P, QRS, and T wave boundaries.'}, 'fantasia': {'name': 'Fantasia Database (Healthy Aging)', 'pn_dir': 'fantasia', 'nominal_fs': 250.0, 'leads': ['RESP', 'ECG'], 'description': 'Simultaneous ECG and respiration recordings from young and elderly subjects.'}} module-attribute

DATABASE_ALIASES = {'mit-bih': 'mitdb', 'mit_bih': 'mitdb', 'mitbih': 'mitdb', 'mit': 'mitdb', 'mitdb': 'mitdb', 'incart': 'incartdb', 'incartdb': 'incartdb', 'ptb-xl': 'ptb-xl', 'ptbxl': 'ptb-xl', 'ptb': 'ptb-xl', 'nst': 'nstdb', 'nstdb': 'nstdb', 'noise_stress': 'nstdb', 'afdb': 'afdb', 'atrial_fib': 'afdb', 'cudb': 'cudb', 'svdb': 'svdb', 'edb': 'edb', 'ludb': 'ludb', 'fantasia': 'fantasia'} module-attribute

load_ecg(source, database=None, record=None, start_time=0.0, duration=None, sampling_rate=None, target_fs=None, normalize=None, lead_names=None, channels=None, load_annotations=True, **kwargs)

Universal High-Level ECG Loader and Normalization Factory.

Intelligently ingests, streams, parses, and constructs a standardized ECGSignal from diverse biomedical sources, with automatic sampling rate standardization (target_fs) and amplitude normalization (normalize):

1. PhysioNet URIs: e.g. source="physionet:mitdb/100", source="mitdb/100".
2. PhysioNet Database + Record: e.g. load_ecg("100", database="mit-bih"), load_ecg("I01", database="incartdb").
3. Local Files: Automatically detected via file extension (.csv, .hea, .dat, .parquet).
4. Synthetic Generator Schemes: e.g. source="synthetic:pvc", source="synthetic:normal", source="synthetic:bradycardia".
5. In-Memory Arrays: Raw numpy.ndarray or pandas.DataFrame.
6. ECGSignal Objects: Passthrough with optional temporal slicing (start_time, duration).

Parameters:

Name	Type	Description	Default
source	Union[str, ndarray, DataFrame, ECGSignal]	URI string, file path, database identifier, array, or DataFrame.	required
database	Optional[str]	Optional database alias (e.g. 'mitdb', 'mit-bih', 'incartdb', 'ptb-xl', 'fantasia').	None
record	Optional[Union[str, int]]	Optional record identifier when database is specified.	None
start_time	float	Start time in seconds (default 0.0).	0.0
duration	Optional[float]	Window length in seconds (None loads entire record).	None
sampling_rate	Optional[float]	Sampling frequency in Hz (if required for CSV / numpy arrays).	None
target_fs	Optional[float]	Target sampling frequency in Hz (e.g. 250.0). If specified, automatically resamples the signal and scales companion ground-truth annotations proportionally.	None
normalize	Optional[str]	Amplitude normalization method: 'zscore' (mean=0, std=1), 'minmax' ([0, 1]), or 'robust' (median/IQR). Default is None (raw physical units).	None
lead_names	Optional[List[str]]	Optional list of lead names.	None
channels	Optional[List[int]]	Optional list of channel indices to load.	None
load_annotations	bool	If True, automatically fetches companion clinical annotations.	True
**kwargs	Any	Additional parameters forwarded to specific loaders or generators.	{}

Returns:

Type	Description
ECGSignal	Standardized ECGSignal instance.

Examples:

>>> # 1. Stream MIT-BIH (360 Hz) and standardize to 250 Hz with Z-Score normalization:
>>> sig = load_ecg("physionet:mitdb/100", duration=10.0, target_fs=250.0, normalize="zscore")
>>>
>>> # 2. Stream St. Petersburg INCART 12-lead (257 Hz) normalized to 250 Hz:
>>> sig = load_ecg("I01", database="incartdb", duration=5.0, target_fs=250.0)

Source code in ecg_sdk/core/loaders.py

def load_ecg(
    source: Union[str, np.ndarray, pd.DataFrame, ECGSignal],
    database: Optional[str] = None,
    record: Optional[Union[str, int]] = None,
    start_time: float = 0.0,
    duration: Optional[float] = None,
    sampling_rate: Optional[float] = None,
    target_fs: Optional[float] = None,
    normalize: Optional[str] = None,
    lead_names: Optional[List[str]] = None,
    channels: Optional[List[int]] = None,
    load_annotations: bool = True,
    **kwargs: Any,
) -> ECGSignal:
    """
    Universal High-Level ECG Loader and Normalization Factory.

    Intelligently ingests, streams, parses, and constructs a standardized `ECGSignal`
    from diverse biomedical sources, with automatic sampling rate standardization
    (`target_fs`) and amplitude normalization (`normalize`):

    1. **PhysioNet URIs:** e.g. `source="physionet:mitdb/100"`, `source="mitdb/100"`.
    2. **PhysioNet Database + Record:** e.g. `load_ecg("100", database="mit-bih")`, `load_ecg("I01", database="incartdb")`.
    3. **Local Files:** Automatically detected via file extension (`.csv`, `.hea`, `.dat`, `.parquet`).
    4. **Synthetic Generator Schemes:** e.g. `source="synthetic:pvc"`, `source="synthetic:normal"`, `source="synthetic:bradycardia"`.
    5. **In-Memory Arrays:** Raw `numpy.ndarray` or `pandas.DataFrame`.
    6. **ECGSignal Objects:** Passthrough with optional temporal slicing (`start_time`, `duration`).

    Args:
        source: URI string, file path, database identifier, array, or DataFrame.
        database: Optional database alias (e.g. 'mitdb', 'mit-bih', 'incartdb', 'ptb-xl', 'fantasia').
        record: Optional record identifier when database is specified.
        start_time: Start time in seconds (default 0.0).
        duration: Window length in seconds (None loads entire record).
        sampling_rate: Sampling frequency in Hz (if required for CSV / numpy arrays).
        target_fs: Target sampling frequency in Hz (e.g. 250.0). If specified, automatically
                   resamples the signal and scales companion ground-truth annotations proportionally.
        normalize: Amplitude normalization method: 'zscore' (mean=0, std=1),
                   'minmax' ([0, 1]), or 'robust' (median/IQR). Default is None (raw physical units).
        lead_names: Optional list of lead names.
        channels: Optional list of channel indices to load.
        load_annotations: If True, automatically fetches companion clinical annotations.
        **kwargs: Additional parameters forwarded to specific loaders or generators.

    Returns:
        Standardized `ECGSignal` instance.

    Examples:
        >>> # 1. Stream MIT-BIH (360 Hz) and standardize to 250 Hz with Z-Score normalization:
        >>> sig = load_ecg("physionet:mitdb/100", duration=10.0, target_fs=250.0, normalize="zscore")
        >>>
        >>> # 2. Stream St. Petersburg INCART 12-lead (257 Hz) normalized to 250 Hz:
        >>> sig = load_ecg("I01", database="incartdb", duration=5.0, target_fs=250.0)
    """
    sig: Optional[ECGSignal] = None

    # -------------------------------------------------------------------------
    # Case 1: In-Memory ECGSignal Instance
    # -------------------------------------------------------------------------
    if isinstance(source, ECGSignal):
        if duration is not None or start_time > 0.0:
            end_time = min(source.duration, start_time + duration) if duration else source.duration
            sig = source.slice(start_time=start_time, end_time=end_time)
        else:
            sig = source.copy()

    # -------------------------------------------------------------------------
    # Case 2: In-Memory Pandas DataFrame
    # -------------------------------------------------------------------------
    elif isinstance(source, pd.DataFrame):
        time_col = kwargs.get("time_column", "time_sec")
        fs = sampling_rate if sampling_rate is not None else 250.0
        if time_col in source.columns:
            t = source[time_col].values
            if len(t) > 1 and np.mean(np.diff(t)) > 0:
                fs = float(1.0 / np.mean(np.diff(t)))
            df_leads = source.drop(columns=[time_col])
        else:
            df_leads = source
        df_num = df_leads.select_dtypes(include=[np.number])
        constructed = ECGSignal(
            data=df_num.values,
            sampling_rate=fs,
            lead_names=list(df_num.columns),
            record_name=kwargs.get("record_name", "DATAFRAME_RECORD"),
        )
        if duration is not None or start_time > 0.0:
            end_time = (
                min(constructed.duration, start_time + duration)
                if duration
                else constructed.duration
            )
            sig = constructed.slice(start_time=start_time, end_time=end_time)
        else:
            sig = constructed

    # -------------------------------------------------------------------------
    # Case 3: In-Memory NumPy Array
    # -------------------------------------------------------------------------
    elif isinstance(source, np.ndarray):
        fs = sampling_rate if sampling_rate is not None else 250.0
        constructed = ECGSignal(
            data=source,
            sampling_rate=fs,
            lead_names=lead_names,
            record_name=kwargs.get("record_name", "ARRAY_RECORD"),
        )
        if duration is not None or start_time > 0.0:
            end_time = (
                min(constructed.duration, start_time + duration)
                if duration
                else constructed.duration
            )
            sig = constructed.slice(start_time=start_time, end_time=end_time)
        else:
            sig = constructed

    # -------------------------------------------------------------------------
    # Case 4: String Identifier Parsing
    # -------------------------------------------------------------------------
    elif isinstance(source, str):
        src_str = source.strip()

        # 4.1 Synthetic Generator Schemes ("synthetic:pvc", "synth:bradycardia", "synthetic:normal")
        if src_str.startswith("synthetic:") or src_str.startswith("synth:"):
            from ecg_sdk.core.generator import generate_synthetic_ecg

            parts = src_str.split(":", 1)
            anomaly_type = parts[1].strip() if len(parts) > 1 else None
            gen_duration = duration if duration is not None else 10.0
            gen_fs = (
                target_fs
                if target_fs is not None
                else (sampling_rate if sampling_rate is not None else 250.0)
            )
            sig = generate_synthetic_ecg(
                duration=gen_duration,
                sampling_rate=gen_fs,
                anomaly=anomaly_type,
                lead_names=lead_names,
                **kwargs,
            )

        # 4.2 Explicit PhysioNet URI Schemes ("physionet:mitdb/100", "pn:incartdb/I01")
        elif src_str.startswith("physionet:") or src_str.startswith("pn:"):
            clean_uri = src_str.split(":", 1)[1].strip()
            if "/" in clean_uri:
                db_part, rec_part = clean_uri.split("/", 1)
                db_clean = db_part.lower().strip()
                pn_dir = DATABASE_ALIASES.get(db_clean, db_clean)
                sig = _load_physionet_record_by_pn_dir(
                    pn_dir=pn_dir,
                    record_id=rec_part,
                    start_time=start_time,
                    duration=duration,
                    channels=channels,
                    load_annotations=load_annotations,
                )

        # 4.3 Database Argument Specified (e.g. database="mit-bih", source="100")
        elif database is not None:
            db_key = str(database).lower().strip()
            pn_dir = DATABASE_ALIASES.get(db_key, db_key)
            target_rec = str(record) if record is not None else src_str
            sig = _load_physionet_record_by_pn_dir(
                pn_dir=pn_dir,
                record_id=target_rec,
                start_time=start_time,
                duration=duration,
                channels=channels,
                load_annotations=load_annotations,
            )

        # 4.4 Slash-delimited PhysioNet Identifier (e.g. "mitdb/100", "incartdb/I01", "ptb-xl/1.0.3/...")
        elif "/" in src_str and not os.path.exists(src_str):
            db_part, rec_part = src_str.split("/", 1)
            db_clean = db_part.lower().strip()
            if db_clean in DATABASE_ALIASES or db_clean in PHYSIONET_DATABASES:
                pn_dir = DATABASE_ALIASES.get(db_clean, db_clean)
                sig = _load_physionet_record_by_pn_dir(
                    pn_dir=pn_dir,
                    record_id=rec_part,
                    start_time=start_time,
                    duration=duration,
                    channels=channels,
                    load_annotations=load_annotations,
                )

        # 4.5 Local CSV File
        elif src_str.endswith(".csv") or src_str.endswith(".tsv"):
            fs = sampling_rate if sampling_rate is not None else 250.0
            constructed = load_csv_record(
                src_str, sampling_rate=fs, lead_columns=lead_names, **kwargs
            )
            if duration is not None or start_time > 0.0:
                end_time = (
                    min(constructed.duration, start_time + duration)
                    if duration
                    else constructed.duration
                )
                sig = constructed.slice(start_time=start_time, end_time=end_time)
            else:
                sig = constructed

        # 4.6 Local WFDB Record (if file exists or has .hea/.dat extension)
        elif (
            os.path.exists(src_str)
            or os.path.exists(f"{src_str}.hea")
            or src_str.endswith(".hea")
            or src_str.endswith(".dat")
        ):
            fs_nominal = sampling_rate if sampling_rate is not None else 360.0
            start_samp = int(round(start_time * fs_nominal))
            end_samp = (
                int(round((start_time + duration) * fs_nominal)) if duration is not None else None
            )
            sig = load_wfdb_record(
                record_name_or_path=src_str,
                pn_dir=None,
                start_sample=start_samp,
                end_sample=end_samp,
                channels=channels,
                load_annotations=load_annotations,
            )

        # 4.7 Pure Numerical Record ID fallback (assume MIT-BIH if 3-digit number e.g. "100", "200")
        elif (
            src_str.isdigit()
            and len(src_str) == 3
            and src_str in PHYSIONET_DATABASES["mitdb"]["records"]
        ):
            sig = load_mitbih_record(
                record=src_str,
                start_time=start_time,
                duration=duration,
                channels=channels,
                load_annotations=load_annotations,
            )

    if sig is None:
        raise ValueError(
            f"Unable to resolve ECG source '{source}'. Provide a valid local file path, "
            f"PhysioNet URI (e.g. 'physionet:mitdb/100'), database name (database='mit-bih'), "
            f"or synthetic scheme (e.g. 'synthetic:pvc')."
        )

    # -------------------------------------------------------------------------
    # Post-Processing 1: Automatic Sampling Rate Standardization (Resampling)
    # -------------------------------------------------------------------------
    if target_fs is not None and target_fs > 0:
        if abs(sig.sampling_rate - target_fs) > 1e-3:
            sig = sig.resample(target_fs=float(target_fs))

    # -------------------------------------------------------------------------
    # Post-Processing 2: Automatic Voltage Normalization (zscore / minmax / robust)
    # -------------------------------------------------------------------------
    if normalize is not None:
        sig = sig.normalize(method=str(normalize).lower().strip())

    return sig

list_physionet_databases()

List all supported benchmark and clinical PhysioNet databases with metadata.

Source code in ecg_sdk/core/loaders.py

def list_physionet_databases() -> Dict[str, Dict[str, Any]]:
    """
    List all supported benchmark and clinical PhysioNet databases with metadata.
    """
    return dict(PHYSIONET_DATABASES)

list_mitbih_records()

List all 48 standard record IDs from the MIT-BIH Arrhythmia Database.

Source code in ecg_sdk/core/loaders.py

def list_mitbih_records() -> List[str]:
    """
    List all 48 standard record IDs from the MIT-BIH Arrhythmia Database.
    """
    return list(PHYSIONET_DATABASES["mitdb"]["records"])

load_wfdb_record(record_name_or_path, pn_dir=None, start_sample=0, end_sample=None, channels=None, load_annotations=True, annotator='atr', target_fs=None, normalize=None)

Load a PhysioNet WFDB record from local disk or streamed directly from PhysioNet.

Source code in ecg_sdk/core/loaders.py

def load_wfdb_record(
    record_name_or_path: str,
    pn_dir: Optional[str] = None,
    start_sample: int = 0,
    end_sample: Optional[int] = None,
    channels: Optional[List[int]] = None,
    load_annotations: bool = True,
    annotator: str = "atr",
    target_fs: Optional[float] = None,
    normalize: Optional[str] = None,
) -> ECGSignal:
    """
    Load a PhysioNet WFDB record from local disk or streamed directly from PhysioNet.
    """
    import wfdb

    # Clean path (strip .hea or .dat if provided for local files)
    if not pn_dir and (
        record_name_or_path.endswith(".hea") or record_name_or_path.endswith(".dat")
    ):
        base_path = os.path.splitext(record_name_or_path)[0]
    else:
        base_path = str(record_name_or_path)

    # Read the WFDB signal
    record = wfdb.rdrecord(
        base_path,
        pn_dir=pn_dir,
        sampfrom=start_sample,
        sampto=end_sample,
        channels=channels,
        physical=True,
    )

    lead_names = (
        record.sig_name if record.sig_name else [f"Lead_{i + 1}" for i in range(record.n_sig)]
    )
    units = record.units[0] if (record.units and len(record.units) > 0) else "mV"

    # Attempt to load expert annotations (R-peaks and beat classes)
    annotation_dict: Optional[Dict[str, Any]] = None
    if load_annotations:
        try:
            ann = wfdb.rdann(
                base_path,
                annotator,
                pn_dir=pn_dir,
                sampfrom=start_sample,
                sampto=end_sample,
            )
            adjusted_samples = np.array(ann.sample) - start_sample
            annotation_dict = {
                "r_peaks": adjusted_samples,
                "symbols": list(ann.symbol),
                "subtypes": list(ann.subtype) if hasattr(ann, "subtype") else None,
                "chan": list(ann.chan) if hasattr(ann, "chan") else None,
                "num": list(ann.num) if hasattr(ann, "num") else None,
                "aux_note": list(ann.aux_note) if hasattr(ann, "aux_note") else None,
            }
        except Exception:
            annotation_dict = None

    metadata = {
        "record_name": record.record_name,
        "comments": record.comments,
        "base_date": str(record.base_date) if record.base_date else None,
        "base_time": str(record.base_time) if record.base_time else None,
        "adc_gain": record.adc_gain,
        "baseline": record.baseline,
        "source": f"PhysioNet ({pn_dir})" if pn_dir else "Local WFDB",
    }
    if annotation_dict is not None:
        metadata["annotations"] = annotation_dict

    sig = ECGSignal(
        data=record.p_signal,
        sampling_rate=float(record.fs),
        lead_names=lead_names,
        record_name=record.record_name,
        patient_id=f"PN_{record.record_name}",
        units=units,
        metadata=metadata,
    )

    if target_fs is not None and abs(sig.sampling_rate - target_fs) > 1e-3:
        sig = sig.resample(target_fs=float(target_fs))

    if normalize is not None:
        sig = sig.normalize(method=str(normalize).lower().strip())

    return sig

load_mitbih_record(record='100', start_time=0.0, duration=10.0, channels=None, target_fs=None, normalize=None, load_annotations=True, pn_dir='mitdb')

Convenience loader for the MIT-BIH Arrhythmia Database from PhysioNet.

Standard MIT-BIH sampling rate is 360 Hz, typically featuring MLII and V1/V5 leads.

Parameters:

Name	Type	Description	Default
record	Union[str, int]	MIT-BIH record number (e.g. '100', '106', '200', 100).	'100'
start_time	float	Starting offset in seconds.	0.0
duration	Optional[float]	Window length in seconds (e.g. 10.0s). If None, loads entire ~30 min recording.	10.0
channels	Optional[List[int]]	Channel indices to load (default None = all channels).	None
target_fs	Optional[float]	Target sampling frequency in Hz (e.g. 250.0). Resamples both signal and annotations.	None
normalize	Optional[str]	'zscore', 'minmax', or 'robust'.	None
load_annotations	bool	If True, fetches expert cardiologist annotations (.atr).	True
pn_dir	str	PhysioNet database identifier (default 'mitdb').	'mitdb'

Returns:

Type	Description
ECGSignal	Standardized ECGSignal instance.

Source code in ecg_sdk/core/loaders.py

def load_mitbih_record(
    record: Union[str, int] = "100",
    start_time: float = 0.0,
    duration: Optional[float] = 10.0,
    channels: Optional[List[int]] = None,
    target_fs: Optional[float] = None,
    normalize: Optional[str] = None,
    load_annotations: bool = True,
    pn_dir: str = "mitdb",
) -> ECGSignal:
    """
    Convenience loader for the MIT-BIH Arrhythmia Database from PhysioNet.

    Standard MIT-BIH sampling rate is 360 Hz, typically featuring MLII and V1/V5 leads.

    Args:
        record: MIT-BIH record number (e.g. '100', '106', '200', 100).
        start_time: Starting offset in seconds.
        duration: Window length in seconds (e.g. 10.0s). If None, loads entire ~30 min recording.
        channels: Channel indices to load (default None = all channels).
        target_fs: Target sampling frequency in Hz (e.g. 250.0). Resamples both signal and annotations.
        normalize: 'zscore', 'minmax', or 'robust'.
        load_annotations: If True, fetches expert cardiologist annotations (.atr).
        pn_dir: PhysioNet database identifier (default 'mitdb').

    Returns:
        Standardized ECGSignal instance.
    """
    rec_str = str(record).strip()
    fs_nominal = 360.0

    start_sample = int(round(start_time * fs_nominal))
    end_sample = int(round((start_time + duration) * fs_nominal)) if duration is not None else None

    return load_wfdb_record(
        record_name_or_path=rec_str,
        pn_dir=pn_dir,
        start_sample=start_sample,
        end_sample=end_sample,
        channels=channels,
        load_annotations=load_annotations,
        annotator="atr",
        target_fs=target_fs,
        normalize=normalize,
    )

load_ptbxl_record(record_path, pn_dir=None, start_sample=0, end_sample=None, target_fs=None, normalize=None)

Load a 12-lead PTB-XL diagnostic record formatted with standard 12-lead naming.

Source code in ecg_sdk/core/loaders.py

def load_ptbxl_record(
    record_path: str,
    pn_dir: Optional[str] = None,
    start_sample: int = 0,
    end_sample: Optional[int] = None,
    target_fs: Optional[float] = None,
    normalize: Optional[str] = None,
) -> ECGSignal:
    """
    Load a 12-lead PTB-XL diagnostic record formatted with standard 12-lead naming.
    """
    signal = load_wfdb_record(
        record_name_or_path=record_path,
        pn_dir=pn_dir,
        start_sample=start_sample,
        end_sample=end_sample,
        load_annotations=False,
        target_fs=target_fs,
        normalize=normalize,
    )
    canonical_12_leads = ["I", "II", "III", "aVR", "aVL", "aVF", "V1", "V2", "V3", "V4", "V5", "V6"]
    if signal.n_leads == 12:
        signal.lead_names = canonical_12_leads
    return signal

load_csv_record(file_path, sampling_rate=250.0, time_column='time_sec', lead_columns=None, target_fs=None, normalize=None)

Load an ECG recording from a CSV file.

Source code in ecg_sdk/core/loaders.py

def load_csv_record(
    file_path: str,
    sampling_rate: float = 250.0,
    time_column: Optional[str] = "time_sec",
    lead_columns: Optional[List[str]] = None,
    target_fs: Optional[float] = None,
    normalize: Optional[str] = None,
) -> ECGSignal:
    """
    Load an ECG recording from a CSV file.
    """
    df = pd.read_csv(file_path)

    if time_column and time_column in df.columns:
        t = df[time_column].values
        if len(t) > 1:
            dt = np.mean(np.diff(t))
            if dt > 0:
                sampling_rate = float(1.0 / dt)
        df_leads = df.drop(columns=[time_column])
    else:
        df_leads = df

    if lead_columns is not None:
        selected_cols = [c for c in lead_columns if c in df_leads.columns]
        if not selected_cols:
            raise ValueError(f"None of specified lead columns {lead_columns} found in CSV.")
        df_leads = df_leads[selected_cols]
    else:
        df_leads = df_leads.select_dtypes(include=[np.number])

    data = df_leads.values
    lead_names = list(df_leads.columns)
    record_name = os.path.splitext(os.path.basename(file_path))[0]

    sig = ECGSignal(
        data=data,
        sampling_rate=sampling_rate,
        lead_names=lead_names,
        record_name=record_name,
        patient_id=f"PATIENT_{record_name}",
        units="mV",
        metadata={"source_file": file_path},
    )

    if target_fs is not None and abs(sig.sampling_rate - target_fs) > 1e-3:
        sig = sig.resample(target_fs=float(target_fs))

    if normalize is not None:
        sig = sig.normalize(method=str(normalize).lower().strip())

    return sig

save_to_csv(signal, file_path)

Save an ECGSignal object to a CSV file.

Source code in ecg_sdk/core/loaders.py

def save_to_csv(signal: ECGSignal, file_path: str) -> None:
    """
    Save an ECGSignal object to a CSV file.
    """
    os.makedirs(os.path.dirname(os.path.abspath(file_path)), exist_ok=True)
    df = signal.to_dataframe()
    df.to_csv(file_path, index=False)