API Reference: ecg_sdk.core.generator

ecg_sdk.core.generator

Dynamic Synthetic ECG Generator Module

Implements dynamical multi-lead synthetic ECG generation based on Gaussian wave decomposition along an angular cardiac phase trajectory (McSharry et al. model), with authentic electrophysiological morphology across all 12 standard clinical leads (I, II, III, aVR, aVL, aVF, V1-V6) and ambulatory leads (MLII).

Academic Literature Sources & Parameter Citations:

1. Seminal Dynamical Attractor Model (Lead II / Single Lead Formulation):

2. McSharry, P. E., Clifford, G. D., Tarassenko, L., & Smith, L. A. (2003). "A dynamical model for generating synthetic electrocardiogram signals." IEEE Transactions on Biomedical Engineering, 50(3), 289-294. DOI: 10.1109/TBME.2003.808805 [BibTeX: @McSharry2003-dynamical]

3. Multichannel ECG Modeling & 3D Vectorcardiographic Projection:

4. Sameni, R., Clifford, G. D., Jutten, C., & Shamsollahi, M. B. (2007). "Multichannel ECG and noise modeling: Application to maternal and fetal ECG signals." EURASIP Journal on Advances in Signal Processing, 2007, 1-14. DOI: 10.1155/2007/43507 [BibTeX: @Sameni2007-multichannel]
5. Clifford, G. D., Shoeb, A., McSharry, P. E., & Jané, R. (2005). "Model-based filtering, compression and classification of the ECG." Computers in Cardiology (CinC), 32, 159-162. DOI: 10.1109/CIC.2005.1588061 [BibTeX: @Clifford2005-CinC]

6. Sayadi, O., & Shamsollahi, M. B. (2008). "ECG synthetic generation using a modified dynamical model." Computers in Biology and Medicine, 38(3), 329-337. DOI: 10.1016/j.compbiomed.2007.12.003 [BibTeX: @Sayadi2008-modified]

7. Clinical 12-Lead Morphological Standards (Frontal & Precordial Norms):

8. Goldberger, A. L., Goldberger, Z. D., & Shvilkin, A. (2017). Goldberger's Clinical Electrocardiography: A Simplified Approach (9th Edition). Elsevier Health Sciences. ISBN: 978-0323401692. [BibTeX: @Goldberger2017-clinical]

CANONICAL_LEAD_PARAMS = {'I': [{'name': 'P', 'theta': -1.2, 'a': 0.15, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.1, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 1.1, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.2, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.3, 'b': 0.4}], 'II': [{'name': 'P', 'theta': -1.2, 'a': 0.22, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.15, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 1.65, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.3, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.42, 'b': 0.4}], 'III': [{'name': 'P', 'theta': -1.2, 'a': 0.08, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.12, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 0.85, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.25, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.18, 'b': 0.4}], 'aVR': [{'name': 'P', 'theta': -1.2, 'a': -0.18, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': 0.1, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 0.15, 'b': 0.1}, {'name': 'S', 'theta': 0.15, 'a': -1.35, 'b': 0.14}, {'name': 'T', 'theta': 1.1, 'a': -0.35, 'b': 0.4}], 'aVL': [{'name': 'P', 'theta': -1.2, 'a': 0.06, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.08, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 0.65, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.18, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.15, 'b': 0.4}], 'aVF': [{'name': 'P', 'theta': -1.2, 'a': 0.16, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.12, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 1.25, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.25, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.3, 'b': 0.4}], 'V1': [{'name': 'P', 'theta': -1.2, 'a': 0.08, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': 0.0, 'b': 0.1}, {'name': 'R', 'theta': -0.1, 'a': 0.25, 'b': 0.1}, {'name': 'S', 'theta': 0.15, 'a': -1.2, 'b': 0.14}, {'name': 'T', 'theta': 1.1, 'a': 0.2, 'b': 0.4}], 'V2': [{'name': 'P', 'theta': -1.2, 'a': 0.12, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': 0.0, 'b': 0.1}, {'name': 'R', 'theta': -0.08, 'a': 0.45, 'b': 0.1}, {'name': 'S', 'theta': 0.15, 'a': -1.65, 'b': 0.14}, {'name': 'T', 'theta': 1.1, 'a': 0.38, 'b': 0.4}], 'V3': [{'name': 'P', 'theta': -1.2, 'a': 0.15, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.08, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 0.95, 'b': 0.12}, {'name': 'S', 'theta': 0.25, 'a': -1.0, 'b': 0.13}, {'name': 'T', 'theta': 1.1, 'a': 0.45, 'b': 0.4}], 'V4': [{'name': 'P', 'theta': -1.2, 'a': 0.16, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.12, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 1.75, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.45, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.45, 'b': 0.4}], 'V5': [{'name': 'P', 'theta': -1.2, 'a': 0.16, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.15, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 1.85, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.2, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.4, 'b': 0.4}], 'V6': [{'name': 'P', 'theta': -1.2, 'a': 0.14, 'b': 0.25}, {'name': 'Q', 'theta': -0.35, 'a': -0.12, 'b': 0.1}, {'name': 'R', 'theta': 0.0, 'a': 1.45, 'b': 0.12}, {'name': 'S', 'theta': 0.3, 'a': -0.12, 'b': 0.12}, {'name': 'T', 'theta': 1.1, 'a': 0.32, 'b': 0.4}]} module-attribute

LEAD_ALIASES = {'LEAD_I': 'I', 'LEAD_1': 'I', 'LEAD1': 'I', 'LEAD_II': 'II', 'LEAD_2': 'II', 'LEAD2': 'II', 'MLII': 'II', 'LEAD_III': 'III', 'LEAD_3': 'III', 'LEAD3': 'III', 'AVR': 'aVR', 'LEAD_AVR': 'aVR', 'AVL': 'aVL', 'LEAD_AVL': 'aVL', 'AVF': 'aVF', 'LEAD_AVF': 'aVF', 'LEAD_V1': 'V1', 'LEAD_V2': 'V2', 'LEAD_V3': 'V3', 'LEAD_V4': 'V4', 'LEAD_V5': 'V5', 'LEAD_V6': 'V6'} module-attribute

generate_synthetic_ecg(duration=10.0, sampling_rate=250.0, fs=None, heart_rate=60.0, noise_level=0.03, powerline_hz=50.0, powerline_noise=None, powerline_freq=None, baseline_wander=True, lead_names=None, leads=None, anomaly=None, random_seed=42, seed=None, **kwargs)

Generate realistic multi-lead synthetic ECG signals with authentic electrophysiological morphologies across all 12 standard clinical leads following the McSharry-Clifford-Sameni dynamical models and Goldberger clinical norms.

Mathematical Model:

The cardiac angular phase trajectory theta(t) rotates at the instantaneous heart rate: d(theta)/dt = 2*pi * f_HR(t) where f_HR(t) includes Respiratory Sinus Arrhythmia (RSA / HRV modulation at ~0.25 Hz).

The synthetic voltage v_l(t) for lead l is the superposition of Gaussian wavelets: v_l(t) = sum_{i in {P,Q,R,S,T}} a_{l,i} * exp( -0.5 * ( (theta(t) - theta_{l,i}) / b_{l,i} )^2 )

Parameters:

Name	Type	Description	Default
duration	float	Duration of recording in seconds.	10.0
sampling_rate	float	Sampling frequency in Hz (\(f_s\)). Also accepts alias fs.	250.0
fs	Optional[float]	Optional alias for sampling_rate.	None
heart_rate	float	Baseline heart rate in beats per minute (bpm).	60.0
noise_level	float	Amplitude of Gaussian EMG noise in mV.	0.03
powerline_hz	Optional[float]	Frequency of powerline interference (50.0 or 60.0 Hz, or None). Also accepts alias powerline_freq.	50.0
powerline_noise	Optional[float]	Amplitude of powerline interference in mV (default 0.035 mV).	None
powerline_freq	Optional[float]	Optional alias for powerline_hz.	None
baseline_wander	Union[bool, float]	If True or float amplitude, adds realistic respiration baseline drift (~0.15 Hz).	True
lead_names	Optional[List[str]]	List of lead names (e.g. ['I', 'II', 'V1', 'V5'] or standard 12 leads). Also accepts alias leads.	None
leads	Optional[List[str]]	Optional alias for lead_names.	None
anomaly	Optional[str]	Optional anomaly type to inject ('normal', 'bradycardia', 'tachycardia', 'pvc', 'pause', 'noisy').	None
random_seed	Optional[int]	Seed for reproducibility. Also accepts alias seed.	42
seed	Optional[int]	Optional alias for random_seed.	None

Returns:

Type	Description
ECGSignal	ECGSignal instance containing the synthetic multi-lead recording.

Source code in ecg_sdk/core/generator.py

def generate_synthetic_ecg(
    duration: float = 10.0,
    sampling_rate: float = 250.0,
    fs: Optional[float] = None,
    heart_rate: float = 60.0,
    noise_level: float = 0.03,
    powerline_hz: Optional[float] = 50.0,
    powerline_noise: Optional[float] = None,
    powerline_freq: Optional[float] = None,
    baseline_wander: Union[bool, float] = True,
    lead_names: Optional[List[str]] = None,
    leads: Optional[List[str]] = None,
    anomaly: Optional[str] = None,
    random_seed: Optional[int] = 42,
    seed: Optional[int] = None,
    **kwargs: Any,
) -> ECGSignal:
    """
    Generate realistic multi-lead synthetic ECG signals with authentic
    electrophysiological morphologies across all 12 standard clinical leads
    following the McSharry-Clifford-Sameni dynamical models and Goldberger clinical norms.

    Mathematical Model:
    -------------------
    The cardiac angular phase trajectory theta(t) rotates at the instantaneous heart rate:
        d(theta)/dt = 2*pi * f_HR(t)
    where f_HR(t) includes Respiratory Sinus Arrhythmia (RSA / HRV modulation at ~0.25 Hz).

    The synthetic voltage v_l(t) for lead l is the superposition of Gaussian wavelets:
        v_l(t) = sum_{i in {P,Q,R,S,T}} a_{l,i} * exp( -0.5 * ( (theta(t) - theta_{l,i}) / b_{l,i} )^2 )

    Args:
        duration: Duration of recording in seconds.
        sampling_rate: Sampling frequency in Hz ($f_s$). Also accepts alias `fs`.
        fs: Optional alias for `sampling_rate`.
        heart_rate: Baseline heart rate in beats per minute (bpm).
        noise_level: Amplitude of Gaussian EMG noise in mV.
        powerline_hz: Frequency of powerline interference (50.0 or 60.0 Hz, or None). Also accepts alias `powerline_freq`.
        powerline_noise: Amplitude of powerline interference in mV (default 0.035 mV).
        powerline_freq: Optional alias for `powerline_hz`.
        baseline_wander: If True or float amplitude, adds realistic respiration baseline drift (~0.15 Hz).
        lead_names: List of lead names (e.g. ['I', 'II', 'V1', 'V5'] or standard 12 leads). Also accepts alias `leads`.
        leads: Optional alias for `lead_names`.
        anomaly: Optional anomaly type to inject ('normal', 'bradycardia', 'tachycardia', 'pvc', 'pause', 'noisy').
        random_seed: Seed for reproducibility. Also accepts alias `seed`.
        seed: Optional alias for `random_seed`.

    Returns:
        ECGSignal instance containing the synthetic multi-lead recording.

    """
    # Resolve aliases
    if fs is not None:
        sampling_rate = float(fs)
    if leads is not None:
        lead_names = leads
    if seed is not None:
        random_seed = seed
    if powerline_freq is not None:
        powerline_hz = powerline_freq

    if random_seed is not None:
        rng = np.random.RandomState(random_seed)
    else:
        rng = np.random.RandomState()

    if lead_names is None:
        lead_names = ["Lead_I", "Lead_II"]

    total_samples = int(round(duration * sampling_rate))
    t = np.arange(total_samples) / sampling_rate

    # Adjust heart rate for specific rhythm anomalies
    effective_hr = heart_rate
    if anomaly == "bradycardia":
        effective_hr = 38.0
    elif anomaly == "tachycardia":
        effective_hr = 150.0

    # Cardiac cycle period in seconds (RR interval)
    rr_mean = 60.0 / effective_hr

    # Add Heart Rate Variability (RSA - Respiratory Sinus Arrhythmia at ~0.25 Hz, McSharry 2003)
    rsa_mod = 0.04 * np.sin(2 * np.pi * 0.25 * t)
    instantaneous_freq = (1.0 / rr_mean) + rsa_mod

    # Phase angle trajectory theta(t) mapped to [-pi, pi]
    phase = 2 * np.pi * np.cumsum(instantaneous_freq) / sampling_rate
    theta = np.mod(phase + np.pi, 2 * np.pi) - np.pi

    def synthesize_lead_signal(params: List[Dict[str, Any]], lead_phase: np.ndarray) -> np.ndarray:
        v = np.zeros(total_samples)
        for p in params:
            d_theta = np.mod(lead_phase - p["theta"] + np.pi, 2 * np.pi) - np.pi
            v += p["a"] * np.exp(-0.5 * (d_theta / p["b"]) ** 2)
        return v

    lead_data_list = []
    for i, lead_name in enumerate(lead_names):
        # Fetch authentic canonical parameters for this specific lead
        params = get_lead_parameters(lead_name)
        v_clean = synthesize_lead_signal(params, theta)

        # Apply PVC anomaly if requested (lead-dependent polarity)
        if anomaly in ["pvc", "ventricular_ectopic"]:
            pvc_interval_samples = int(3.5 * sampling_rate)
            pvc_duration_samples = int(0.25 * sampling_rate)

            curr_sample = int(1.5 * sampling_rate)
            while curr_sample + pvc_duration_samples < total_samples:
                # Inverted in lateral leads, positive in V1 (Goldberger 2017)
                is_inverted_lead = "aVR" in lead_name or "V1" in lead_name
                polarity = 1.0 if is_inverted_lead else -1.0

                pvc_t = np.linspace(-1, 1, pvc_duration_samples)
                pvc_wave = polarity * (
                    -1.5 * np.exp(-0.5 * (pvc_t / 0.3) ** 2)
                    + 0.6 * np.exp(-0.5 * ((pvc_t - 0.5) / 0.4) ** 2)
                )
                v_clean[curr_sample : curr_sample + pvc_duration_samples] = pvc_wave
                curr_sample += pvc_interval_samples

        # Apply Pause / Asystole anomaly
        if anomaly in ["pause", "asystole"]:
            pause_start = int(3.0 * sampling_rate)
            pause_end = int(6.5 * sampling_rate)
            if pause_start < total_samples:
                v_clean[pause_start : min(pause_end, total_samples)] *= 0.02

        # Add realistic noise components (EMG muscle tremor, powerline, baseline wander)
        noise = rng.normal(0, noise_level, total_samples)

        if powerline_hz is not None:
            # 50 Hz / 60 Hz powerline hum with lead-dependent spatial phase shift (Sameni 2007)
            pl_amp = float(powerline_noise) if powerline_noise is not None else 0.035
            powerline = pl_amp * np.sin(2 * np.pi * powerline_hz * t + (i * np.pi / 6))
            noise += powerline

        if baseline_wander:
            # Respiration wander ~0.15 Hz + slow postural drift ~0.05 Hz (Moody et al. 1992)
            bw_amp = (
                float(baseline_wander)
                if isinstance(baseline_wander, (int, float))
                and not isinstance(baseline_wander, bool)
                else 1.0
            )
            wander = bw_amp * (
                0.10 * np.sin(2 * np.pi * 0.15 * t + (i * 0.2))
                + 0.05 * np.cos(2 * np.pi * 0.05 * t)
            )
            noise += wander

        if anomaly == "noisy":
            # Add heavy electrode motion artifacts
            burst_noise = rng.normal(0, 0.4, total_samples) * (np.sin(2 * np.pi * 1.2 * t) > 0.5)
            noise += burst_noise

        lead_signal = v_clean + noise
        lead_data_list.append(lead_signal)

    data = np.column_stack(lead_data_list)

    metadata = {
        "generator": "McSharry Dynamical Model (12-Lead Anatomical)",
        "sources": [
            "McSharry et al. (2003) IEEE TBME",
            "Sameni et al. (2007) EURASIP JASP",
            "Clifford et al. (2005) CinC",
            "Goldberger et al. (2017) Clinical ECG 9th Ed.",
        ],
        "heart_rate_bpm": effective_hr,
        "anomaly": anomaly if anomaly else "none",
        "noise_level": noise_level,
        "powerline_hz": powerline_hz,
        "baseline_wander": baseline_wander,
    }

    return ECGSignal(
        data=data,
        sampling_rate=sampling_rate,
        lead_names=lead_names,
        record_name=f"SYNTH_{anomaly.upper() if anomaly else 'NORMAL'}",
        patient_id="SYNTH_PATIENT_01",
        units="mV",
        metadata=metadata,
    )

get_lead_parameters(lead_name)

Look up canonical electrophysiological Gaussian parameters for a given lead name.

Parameters:

Name	Type	Description	Default
lead_name	str	String name of the lead (e.g. 'I', 'II', 'aVR', 'V1', 'MLII').	required

Returns:

Type	Description
List[Dict[str, Any]]	List of dictionaries with keys 'name', 'theta', 'a', 'b'.

Source code in ecg_sdk/core/generator.py

def get_lead_parameters(lead_name: str) -> List[Dict[str, Any]]:
    """
    Look up canonical electrophysiological Gaussian parameters for a given lead name.

    Args:
        lead_name: String name of the lead (e.g. 'I', 'II', 'aVR', 'V1', 'MLII').

    Returns:
        List of dictionaries with keys 'name', 'theta', 'a', 'b'.
    """
    clean_name = lead_name.strip()
    upper_name = clean_name.upper()

    # 1. Direct canonical lookup
    if clean_name in CANONICAL_LEAD_PARAMS:
        return CANONICAL_LEAD_PARAMS[clean_name]

    # 2. Alias mapping lookup
    if upper_name in LEAD_ALIASES:
        canonical_key = LEAD_ALIASES[upper_name]
        return CANONICAL_LEAD_PARAMS[canonical_key]

    # 3. Case-insensitive lookup
    for k, v in CANONICAL_LEAD_PARAMS.items():
        if k.upper() == upper_name:
            return v

    # Fallback to Lead II parameters if unknown
    return CANONICAL_LEAD_PARAMS["II"]