Skip to content

Preprocessing & SQI API Reference

ecg_sdk.preprocessing.filters

Digital Filtering Module for Biopotential ECG Signals

Implements zero-phase forward-backward Butterworth bandpass, highpass, lowpass, and IIR Notch filters (50 Hz / 60 Hz powerline cancellation) using Second-Order Sections (SOS) for numerical stability.

References
  • Oppenheim, A. V., & Schafer, R. W. (2009). Discrete-Time Signal Processing.
  • Clifford, G. D., Azuaje, F., & McSharry, P. (2006). Advanced Methods and Tools for ECG Data Analysis. Artech House.

butter_bandpass_filter(data, lowcut=0.5, highcut=45.0, fs=250.0, order=4)

Apply a zero-phase Butterworth bandpass filter using Second-Order Sections (SOS).

Parameters

data : np.ndarray 1D array (N,) or 2D array (N, L) of ECG samples. lowcut : float, default=0.5 Lower cutoff frequency in Hertz (Hz). highcut : float, default=45.0 Upper cutoff frequency in Hertz (Hz). fs : float, default=250.0 Sampling frequency in Hertz (Hz). order : int, default=4 Filter order (effective order is 2*order due to bidirectional filtering).

Returns

np.ndarray Filtered zero-phase ECG biopotential signal with identical shape.

Source code in ecg_sdk/preprocessing/filters.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def butter_bandpass_filter(
    data: np.ndarray,
    lowcut: float = 0.5,
    highcut: float = 45.0,
    fs: float = 250.0,
    order: int = 4,
) -> np.ndarray:
    """
    Apply a zero-phase Butterworth bandpass filter using Second-Order Sections (SOS).

    Parameters
    ----------
    data : np.ndarray
        1D array (N,) or 2D array (N, L) of ECG samples.
    lowcut : float, default=0.5
        Lower cutoff frequency in Hertz (Hz).
    highcut : float, default=45.0
        Upper cutoff frequency in Hertz (Hz).
    fs : float, default=250.0
        Sampling frequency in Hertz (Hz).
    order : int, default=4
        Filter order (effective order is 2*order due to bidirectional filtering).

    Returns
    -------
    np.ndarray
        Filtered zero-phase ECG biopotential signal with identical shape.
    """
    nyquist = 0.5 * fs
    if lowcut <= 0 or highcut >= nyquist:
        raise ValueError(
            f"Cutoff frequencies ({lowcut}, {highcut}) must satisfy 0 < lowcut < highcut < Nyquist ({nyquist} Hz)."
        )

    low = lowcut / nyquist
    high = highcut / nyquist
    sos = scipy_signal.butter(order, [low, high], btype="bandpass", output="sos")

    # Apply zero-phase filter along time axis (axis 0)
    if data.ndim == 1:
        return scipy_signal.sosfiltfilt(sos, data)
    elif data.ndim == 2:
        return scipy_signal.sosfiltfilt(sos, data, axis=0)
    else:
        raise ValueError(f"Expected 1D or 2D array, got ndim={data.ndim}")

butter_highpass_filter(data, cutoff=0.5, fs=250.0, order=4)

Apply a zero-phase Butterworth highpass filter using Second-Order Sections (SOS).

Source code in ecg_sdk/preprocessing/filters.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def butter_highpass_filter(
    data: np.ndarray,
    cutoff: float = 0.5,
    fs: float = 250.0,
    order: int = 4,
) -> np.ndarray:
    """
    Apply a zero-phase Butterworth highpass filter using Second-Order Sections (SOS).
    """
    nyquist = 0.5 * fs
    if cutoff <= 0 or cutoff >= nyquist:
        raise ValueError(
            f"Cutoff frequency ({cutoff} Hz) must satisfy 0 < cutoff < Nyquist ({nyquist} Hz)."
        )

    normal_cutoff = cutoff / nyquist
    sos = scipy_signal.butter(order, normal_cutoff, btype="highpass", output="sos")

    if data.ndim == 1:
        return scipy_signal.sosfiltfilt(sos, data)
    elif data.ndim == 2:
        return scipy_signal.sosfiltfilt(sos, data, axis=0)
    else:
        raise ValueError(f"Expected 1D or 2D array, got ndim={data.ndim}")

butter_lowpass_filter(data, cutoff=45.0, fs=250.0, order=4)

Apply a zero-phase Butterworth lowpass filter using Second-Order Sections (SOS).

Source code in ecg_sdk/preprocessing/filters.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def butter_lowpass_filter(
    data: np.ndarray,
    cutoff: float = 45.0,
    fs: float = 250.0,
    order: int = 4,
) -> np.ndarray:
    """
    Apply a zero-phase Butterworth lowpass filter using Second-Order Sections (SOS).
    """
    nyquist = 0.5 * fs
    if cutoff <= 0 or cutoff >= nyquist:
        raise ValueError(
            f"Cutoff frequency ({cutoff} Hz) must satisfy 0 < cutoff < Nyquist ({nyquist} Hz)."
        )

    normal_cutoff = cutoff / nyquist
    sos = scipy_signal.butter(order, normal_cutoff, btype="lowpass", output="sos")

    if data.ndim == 1:
        return scipy_signal.sosfiltfilt(sos, data)
    elif data.ndim == 2:
        return scipy_signal.sosfiltfilt(sos, data, axis=0)
    else:
        raise ValueError(f"Expected 1D or 2D array, got ndim={data.ndim}")

iir_notch_filter(data, freq=50.0, fs=250.0, quality_factor=30.0)

Apply a zero-phase Infinite Impulse Response (IIR) Notch filter to remove powerline interference.

Parameters

data : np.ndarray 1D array (N,) or 2D array (N, L) of ECG samples. freq : float, default=50.0 Center frequency to notch out in Hertz (50.0 Hz Europe/Asia, 60.0 Hz Americas). fs : float, default=250.0 Sampling frequency in Hertz (Hz). quality_factor : float, default=30.0 Quality factor Q = freq / bandwidth. Higher Q creates a narrower notch.

Returns

np.ndarray Notch-filtered signal with identical shape.

Source code in ecg_sdk/preprocessing/filters.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def iir_notch_filter(
    data: np.ndarray,
    freq: float = 50.0,
    fs: float = 250.0,
    quality_factor: float = 30.0,
) -> np.ndarray:
    """
    Apply a zero-phase Infinite Impulse Response (IIR) Notch filter to remove powerline interference.

    Parameters
    ----------
    data : np.ndarray
        1D array (N,) or 2D array (N, L) of ECG samples.
    freq : float, default=50.0
        Center frequency to notch out in Hertz (50.0 Hz Europe/Asia, 60.0 Hz Americas).
    fs : float, default=250.0
        Sampling frequency in Hertz (Hz).
    quality_factor : float, default=30.0
        Quality factor Q = freq / bandwidth. Higher Q creates a narrower notch.

    Returns
    -------
    np.ndarray
        Notch-filtered signal with identical shape.
    """
    nyquist = 0.5 * fs
    if freq <= 0 or freq >= nyquist:
        # If notch frequency is outside the signal Nyquist bandwidth, return data unchanged
        return data.copy()

    b, a = scipy_signal.iirnotch(freq, quality_factor, fs=fs)

    if data.ndim == 1:
        return scipy_signal.filtfilt(b, a, data)
    elif data.ndim == 2:
        return scipy_signal.filtfilt(b, a, data, axis=0)
    else:
        raise ValueError(f"Expected 1D or 2D array, got ndim={data.ndim}")

filter_ecg(signal_or_data, fs=None, lowcut=0.5, highcut=45.0, notch_freq=50.0, notch_q=30.0, order=4)

Comprehensive ECG signal conditioning pipeline applying zero-phase bandpass and powerline notch filtering.

Parameters

signal_or_data : ECGSignal or np.ndarray Input ECG signal container or raw numpy biopotential array. fs : float, optional Sampling frequency in Hz (mandatory if passing raw numpy array; inferred if ECGSignal). lowcut : float, default=0.5 Bandpass lower frequency cutoff (Hz). Suppresses respiratory baseline wander. highcut : float, default=45.0 Bandpass upper frequency cutoff (Hz). Suppresses high-frequency EMG muscle noise. notch_freq : float, optional, default=50.0 Powerline notch frequency (50.0 Hz or 60.0 Hz). Set to None to skip notch filtering. notch_q : float, default=30.0 Quality factor for the IIR notch filter. order : int, default=4 Butterworth filter order.

Returns

ECGSignal or np.ndarray Filtered signal with matched container type and preserved metadata.

Source code in ecg_sdk/preprocessing/filters.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def filter_ecg(
    signal_or_data: Union[ECGSignal, np.ndarray],
    fs: float | None = None,
    lowcut: float = 0.5,
    highcut: float = 45.0,
    notch_freq: float | None = 50.0,
    notch_q: float = 30.0,
    order: int = 4,
) -> Union[ECGSignal, np.ndarray]:
    """
    Comprehensive ECG signal conditioning pipeline applying zero-phase bandpass and powerline notch filtering.

    Parameters
    ----------
    signal_or_data : ECGSignal or np.ndarray
        Input ECG signal container or raw numpy biopotential array.
    fs : float, optional
        Sampling frequency in Hz (mandatory if passing raw numpy array; inferred if ECGSignal).
    lowcut : float, default=0.5
        Bandpass lower frequency cutoff (Hz). Suppresses respiratory baseline wander.
    highcut : float, default=45.0
        Bandpass upper frequency cutoff (Hz). Suppresses high-frequency EMG muscle noise.
    notch_freq : float, optional, default=50.0
        Powerline notch frequency (50.0 Hz or 60.0 Hz). Set to None to skip notch filtering.
    notch_q : float, default=30.0
        Quality factor for the IIR notch filter.
    order : int, default=4
        Butterworth filter order.

    Returns
    -------
    ECGSignal or np.ndarray
        Filtered signal with matched container type and preserved metadata.
    """
    # Check if input is an ECGSignal instance
    is_ecg_signal = hasattr(signal_or_data, "data") and (
        hasattr(signal_or_data, "sampling_rate") or hasattr(signal_or_data, "fs")
    )

    if is_ecg_signal:
        raw_data = signal_or_data.data
        sampling_rate = float(
            signal_or_data.sampling_rate
            if hasattr(signal_or_data, "sampling_rate")
            else signal_or_data.fs
        )
    else:
        raw_data = np.asarray(signal_or_data, dtype=np.float64)
        if fs is None:
            raise ValueError(
                "Sampling frequency `fs` must be specified when passing a raw NumPy array."
            )
        sampling_rate = float(fs)

    # 1. Bandpass filter
    filtered = butter_bandpass_filter(
        raw_data,
        lowcut=lowcut,
        highcut=highcut,
        fs=sampling_rate,
        order=order,
    )

    # 2. Notch filter (if requested)
    if notch_freq is not None:
        filtered = iir_notch_filter(
            filtered,
            freq=notch_freq,
            fs=sampling_rate,
            quality_factor=notch_q,
        )

    if is_ecg_signal:
        # Return new ECGSignal container with updated data and history in metadata
        new_metadata = dict(signal_or_data.metadata)
        new_metadata["preprocessing_filters"] = {
            "bandpass": (lowcut, highcut),
            "notch_freq": notch_freq,
            "order": order,
        }
        return type(signal_or_data)(
            data=filtered,
            sampling_rate=sampling_rate,
            lead_names=signal_or_data.lead_names,
            units=signal_or_data.units,
            patient_id=signal_or_data.patient_id,
            record_name=signal_or_data.record_name,
            metadata=new_metadata,
        )

    return filtered

ecg_sdk.preprocessing.baseline

Baseline Wander Removal Module

Removes low-frequency baseline drift (caused by patient respiration, perspiration, and electrode motion) using the standard clinical two-stage cascaded median filter or polynomial detrending.

References
  • de Chazal, P., O'Dwyer, M., & Reilly, R. B. (2004). Automatic classification of heartbeats using ECG morphology and heartbeat interval features. IEEE TBME.
  • Clifford, G. D., Azuaje, F., & McSharry, P. (2006). Advanced Methods and Tools for ECG Data Analysis. Artech House.

cascaded_median_baseline_wander(data, fs=250.0, window1_ms=200.0, window2_ms=600.0)

Estimate and remove baseline wander using a two-stage cascaded 1D median filter.

The first stage applies a ~200 ms median filter to suppress the P-wave and QRS complex. The second stage applies a ~600 ms median filter to suppress the T-wave, yielding the estimated low-frequency baseline drift b(t).

Parameters

data : np.ndarray 1D array (N,) or 2D array (N, L) of ECG samples. fs : float, default=250.0 Sampling frequency in Hertz (Hz). window1_ms : float, default=200.0 First median filter window in milliseconds (to remove QRS and P waves). window2_ms : float, default=600.0 Second median filter window in milliseconds (to remove T wave).

Returns

cleaned_data : np.ndarray ECG signal with baseline wander subtracted. baseline : np.ndarray Estimated baseline drift component b(t).

Source code in ecg_sdk/preprocessing/baseline.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def cascaded_median_baseline_wander(
    data: np.ndarray,
    fs: float = 250.0,
    window1_ms: float = 200.0,
    window2_ms: float = 600.0,
) -> tuple[np.ndarray, np.ndarray]:
    """
    Estimate and remove baseline wander using a two-stage cascaded 1D median filter.

    The first stage applies a ~200 ms median filter to suppress the P-wave and QRS complex.
    The second stage applies a ~600 ms median filter to suppress the T-wave, yielding
    the estimated low-frequency baseline drift b(t).

    Parameters
    ----------
    data : np.ndarray
        1D array (N,) or 2D array (N, L) of ECG samples.
    fs : float, default=250.0
        Sampling frequency in Hertz (Hz).
    window1_ms : float, default=200.0
        First median filter window in milliseconds (to remove QRS and P waves).
    window2_ms : float, default=600.0
        Second median filter window in milliseconds (to remove T wave).

    Returns
    -------
    cleaned_data : np.ndarray
        ECG signal with baseline wander subtracted.
    baseline : np.ndarray
        Estimated baseline drift component b(t).
    """
    # Calculate filter kernel sizes in samples (must be odd integers)
    w1 = int(round((window1_ms / 1000.0) * fs))
    if w1 % 2 == 0:
        w1 += 1
    w1 = max(3, w1)

    w2 = int(round((window2_ms / 1000.0) * fs))
    if w2 % 2 == 0:
        w2 += 1
    w2 = max(w1, w2)

    is_1d = data.ndim == 1
    arr = data[:, np.newaxis] if is_1d else data

    N, L = arr.shape
    baseline = np.zeros_like(arr)

    for ch in range(L):
        # Stage 1: remove QRS and P
        stage1 = ndimage.median_filter(arr[:, ch], size=w1, mode="reflect")
        # Stage 2: remove T wave to obtain smooth baseline
        stage2 = ndimage.median_filter(stage1, size=w2, mode="reflect")
        baseline[:, ch] = stage2

    cleaned = arr - baseline

    if is_1d:
        return cleaned.squeeze(axis=-1), baseline.squeeze(axis=-1)
    return cleaned, baseline

polynomial_detrend(data, order=3)

Fit and subtract an n-th degree polynomial baseline from the signal.

Parameters

data : np.ndarray 1D (N,) or 2D (N, L) ECG array. order : int, default=3 Polynomial degree.

Returns

cleaned_data : np.ndarray baseline : np.ndarray

Source code in ecg_sdk/preprocessing/baseline.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def polynomial_detrend(
    data: np.ndarray,
    order: int = 3,
) -> tuple[np.ndarray, np.ndarray]:
    """
    Fit and subtract an n-th degree polynomial baseline from the signal.

    Parameters
    ----------
    data : np.ndarray
        1D (N,) or 2D (N, L) ECG array.
    order : int, default=3
        Polynomial degree.

    Returns
    -------
    cleaned_data : np.ndarray
    baseline : np.ndarray
    """
    is_1d = data.ndim == 1
    arr = data[:, np.newaxis] if is_1d else data
    N, L = arr.shape
    t = np.linspace(-1.0, 1.0, N)

    baseline = np.zeros_like(arr)
    for ch in range(L):
        poly_coeffs = np.polyfit(t, arr[:, ch], deg=order)
        baseline[:, ch] = np.polyval(poly_coeffs, t)

    cleaned = arr - baseline
    if is_1d:
        return cleaned.squeeze(axis=-1), baseline.squeeze(axis=-1)
    return cleaned, baseline

remove_baseline_wander(signal_or_data, fs=None, method='cascaded_median', window1_ms=200.0, window2_ms=600.0, poly_order=3, return_baseline=False)

High-level baseline drift removal dispatcher supporting ECGSignal or numpy arrays.

Parameters

signal_or_data : ECGSignal or np.ndarray Input ECG biopotential signal. fs : float, optional Sampling frequency in Hz (required if passing np.ndarray). method : {"cascaded_median", "polynomial"}, default="cascaded_median" Baseline estimation algorithm. window1_ms : float, default=200.0 Stage-1 median filter window (for cascaded_median). window2_ms : float, default=600.0 Stage-2 median filter window (for cascaded_median). poly_order : int, default=3 Polynomial degree (for polynomial method). return_baseline : bool, default=False If True, also returns the extracted baseline array.

Returns

cleaned_signal : ECGSignal or np.ndarray baseline : np.ndarray (optional, if return_baseline=True)

Source code in ecg_sdk/preprocessing/baseline.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def remove_baseline_wander(
    signal_or_data: Union[ECGSignal, np.ndarray],
    fs: float | None = None,
    method: str = "cascaded_median",
    window1_ms: float = 200.0,
    window2_ms: float = 600.0,
    poly_order: int = 3,
    return_baseline: bool = False,
) -> Union[ECGSignal, np.ndarray, tuple[Union[ECGSignal, np.ndarray], np.ndarray]]:
    """
    High-level baseline drift removal dispatcher supporting ECGSignal or numpy arrays.

    Parameters
    ----------
    signal_or_data : ECGSignal or np.ndarray
        Input ECG biopotential signal.
    fs : float, optional
        Sampling frequency in Hz (required if passing np.ndarray).
    method : {"cascaded_median", "polynomial"}, default="cascaded_median"
        Baseline estimation algorithm.
    window1_ms : float, default=200.0
        Stage-1 median filter window (for cascaded_median).
    window2_ms : float, default=600.0
        Stage-2 median filter window (for cascaded_median).
    poly_order : int, default=3
        Polynomial degree (for polynomial method).
    return_baseline : bool, default=False
        If True, also returns the extracted baseline array.

    Returns
    -------
    cleaned_signal : ECGSignal or np.ndarray
    baseline : np.ndarray (optional, if return_baseline=True)
    """
    is_ecg_signal = hasattr(signal_or_data, "data") and (
        hasattr(signal_or_data, "sampling_rate") or hasattr(signal_or_data, "fs")
    )

    if is_ecg_signal:
        raw_data = signal_or_data.data
        sampling_rate = float(
            signal_or_data.sampling_rate
            if hasattr(signal_or_data, "sampling_rate")
            else signal_or_data.fs
        )
    else:
        raw_data = np.asarray(signal_or_data, dtype=np.float64)
        if fs is None:
            raise ValueError(
                "Sampling frequency `fs` must be specified when passing a raw NumPy array."
            )
        sampling_rate = float(fs)

    if method == "cascaded_median":
        cleaned, baseline = cascaded_median_baseline_wander(
            raw_data,
            fs=sampling_rate,
            window1_ms=window1_ms,
            window2_ms=window2_ms,
        )
    elif method == "polynomial":
        cleaned, baseline = polynomial_detrend(raw_data, order=poly_order)
    else:
        raise ValueError(
            f"Unknown baseline removal method '{method}'. Choose 'cascaded_median' or 'polynomial'."
        )

    if is_ecg_signal:
        new_meta = dict(signal_or_data.metadata)
        new_meta["baseline_removal"] = {"method": method}
        res_sig = type(signal_or_data)(
            data=cleaned,
            sampling_rate=sampling_rate,
            lead_names=signal_or_data.lead_names,
            units=signal_or_data.units,
            patient_id=signal_or_data.patient_id,
            record_name=signal_or_data.record_name,
            metadata=new_meta,
        )
        if return_baseline:
            return res_sig, baseline
        return res_sig

    if return_baseline:
        return cleaned, baseline
    return cleaned

ecg_sdk.preprocessing.quality

Signal Quality Index (SQI) Engine

Implements multi-metric electrophysiological signal quality assessment to detect noise, muscle artifact (EMG), electrode detachment, powerline interference, and severe respiratory baseline wander.

References
  • Clifford, G. D., Behar, J., Li, Q., & Rezek, I. (2012). Signal quality indices and data fusion for determining clinical acceptability of electrocardiograms. Physiological Measurement, 33(9), 1419–1433.
  • Orphanidou, C., et al. (2015). Signal-quality indices for the electrocardiogram and photoplethysmogram: derivation and applications to mobile monitoring. IEEE JBHI, 19(3), 832–838.

SignalQualityResult dataclass

Comprehensive multi-lead Signal Quality Index report.

Source code in ecg_sdk/preprocessing/quality.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(frozen=True)
class SignalQualityResult:
    """Comprehensive multi-lead Signal Quality Index report."""

    overall_sqi: float
    overall_grade: str
    is_acceptable: bool
    lead_metrics: Dict[str, LeadQuality] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)

    def summary(self) -> str:
        """Render a clean text table of lead quality grades."""
        lines = [
            "=" * 68,
            "SIGNAL QUALITY ASSESSMENT REPORT (SQI)",
            "=" * 68,
            f"Overall Composite SQI : {self.overall_sqi:.3f} / 1.000",
            f"Overall Quality Grade : {self.overall_grade.upper()}",
            f"Acceptable for AI/ML  : {'YES [✓]' if self.is_acceptable else 'NO [✗]'}",
            "-" * 68,
            f"{'Lead':<10} {'pSQI':<8} {'kSQI':<8} {'basSQI':<9} {'hfSQI':<8} {'Score':<8} {'Grade'}",
            "-" * 68,
        ]
        for lead, m in self.lead_metrics.items():
            lines.append(
                f"{lead:<10} {m.psqi:<8.3f} {m.ksqi:<8.2f} {m.bas_sqi:<9.3f} {m.hf_sqi:<8.3f} {m.composite_sqi:<8.3f} {m.grade}"
            )
        lines.append("=" * 68)
        return "\n".join(lines)

summary()

Render a clean text table of lead quality grades.

Source code in ecg_sdk/preprocessing/quality.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def summary(self) -> str:
    """Render a clean text table of lead quality grades."""
    lines = [
        "=" * 68,
        "SIGNAL QUALITY ASSESSMENT REPORT (SQI)",
        "=" * 68,
        f"Overall Composite SQI : {self.overall_sqi:.3f} / 1.000",
        f"Overall Quality Grade : {self.overall_grade.upper()}",
        f"Acceptable for AI/ML  : {'YES [✓]' if self.is_acceptable else 'NO [✗]'}",
        "-" * 68,
        f"{'Lead':<10} {'pSQI':<8} {'kSQI':<8} {'basSQI':<9} {'hfSQI':<8} {'Score':<8} {'Grade'}",
        "-" * 68,
    ]
    for lead, m in self.lead_metrics.items():
        lines.append(
            f"{lead:<10} {m.psqi:<8.3f} {m.ksqi:<8.2f} {m.bas_sqi:<9.3f} {m.hf_sqi:<8.3f} {m.composite_sqi:<8.3f} {m.grade}"
        )
    lines.append("=" * 68)
    return "\n".join(lines)

LeadQuality dataclass

Quality metrics for a single ECG lead.

Source code in ecg_sdk/preprocessing/quality.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
@dataclass(frozen=True)
class LeadQuality:
    """Quality metrics for a single ECG lead."""

    lead_name: str
    psqi: float  # Power spectrum QRS concentration (5-15 Hz / 5-40 Hz)
    ksqi: float  # Kurtosis (clean ECG > 5.0)
    ssqi: float  # Skewness
    bas_sqi: float  # Baseline wander power ratio (<0.5 Hz / 0.5-40 Hz)
    hf_sqi: float  # High-frequency noise ratio (>45 Hz / 0.5-40 Hz)
    composite_sqi: float  # Overall normalized score [0.0, 1.0]
    grade: str  # "excellent", "acceptable", "unusable"
    is_acceptable: bool  # True if composite_sqi >= threshold

assess_signal_quality(signal_or_data, fs=None, lead_names=None, acceptance_threshold=0.6)

Evaluate multi-lead Signal Quality Index (SQI) across all channels.

Parameters

signal_or_data : ECGSignal or np.ndarray Input ECG biopotential signal. fs : float, optional Sampling frequency in Hz (required if passing np.ndarray). lead_names : list of str, optional Lead labels corresponding to channels. acceptance_threshold : float, default=0.60 Minimum composite SQI threshold for acceptable clinical quality.

Returns

SignalQualityResult Structured report containing overall score, clinical grade, and per-lead metrics.

Source code in ecg_sdk/preprocessing/quality.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def assess_signal_quality(
    signal_or_data: Union[ECGSignal, np.ndarray],
    fs: float | None = None,
    lead_names: list[str] | None = None,
    acceptance_threshold: float = 0.60,
) -> SignalQualityResult:
    """
    Evaluate multi-lead Signal Quality Index (SQI) across all channels.

    Parameters
    ----------
    signal_or_data : ECGSignal or np.ndarray
        Input ECG biopotential signal.
    fs : float, optional
        Sampling frequency in Hz (required if passing np.ndarray).
    lead_names : list of str, optional
        Lead labels corresponding to channels.
    acceptance_threshold : float, default=0.60
        Minimum composite SQI threshold for acceptable clinical quality.

    Returns
    -------
    SignalQualityResult
        Structured report containing overall score, clinical grade, and per-lead metrics.
    """
    is_ecg_signal = hasattr(signal_or_data, "data") and (
        hasattr(signal_or_data, "sampling_rate") or hasattr(signal_or_data, "fs")
    )

    if is_ecg_signal:
        arr = signal_or_data.data
        sampling_rate = float(
            signal_or_data.sampling_rate
            if hasattr(signal_or_data, "sampling_rate")
            else signal_or_data.fs
        )
        labels = signal_or_data.lead_names
    else:
        raw_arr = np.asarray(signal_or_data, dtype=np.float64)
        if raw_arr.ndim == 1:
            arr = raw_arr[:, np.newaxis]
        elif raw_arr.ndim == 2:
            arr = raw_arr
        else:
            raise ValueError(f"Expected 1D or 2D array, got ndim={raw_arr.ndim}")

        if fs is None:
            raise ValueError(
                "Sampling frequency `fs` must be provided when passing a raw NumPy array."
            )
        sampling_rate = float(fs)
        N, L = arr.shape
        labels = (
            lead_names
            if lead_names is not None and len(lead_names) == L
            else [f"Lead_{i + 1}" for i in range(L)]
        )

    N, L = arr.shape
    lead_results: Dict[str, LeadQuality] = {}

    scores = []
    for ch in range(L):
        lead_name = labels[ch] if ch < len(labels) else f"Lead_{ch + 1}"
        lead_q = evaluate_lead_quality(
            arr[:, ch],
            fs=sampling_rate,
            lead_name=lead_name,
            acceptance_threshold=acceptance_threshold,
        )
        lead_results[lead_name] = lead_q
        scores.append(lead_q.composite_sqi)

    overall_score = float(np.mean(scores)) if scores else 0.0
    if overall_score >= 0.75:
        overall_grade = "excellent"
    elif overall_score >= acceptance_threshold:
        overall_grade = "acceptable"
    else:
        overall_grade = "unusable"

    return SignalQualityResult(
        overall_sqi=overall_score,
        overall_grade=overall_grade,
        is_acceptable=(overall_score >= acceptance_threshold),
        lead_metrics=lead_results,
        metadata={"num_leads": L, "duration_s": N / sampling_rate, "fs": sampling_rate},
    )

evaluate_lead_quality(data_1d, fs=250.0, lead_name='Lead', acceptance_threshold=0.6)

Assess quality indices for a single 1D lead signal and calculate composite score.

Source code in ecg_sdk/preprocessing/quality.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def evaluate_lead_quality(
    data_1d: np.ndarray,
    fs: float = 250.0,
    lead_name: str = "Lead",
    acceptance_threshold: float = 0.60,
) -> LeadQuality:
    """
    Assess quality indices for a single 1D lead signal and calculate composite score.
    """
    # Check for flatline or NaN/Inf
    if len(data_1d) == 0 or not np.all(np.isfinite(data_1d)) or np.std(data_1d) < 1e-6:
        return LeadQuality(
            lead_name=lead_name,
            psqi=0.0,
            ksqi=0.0,
            ssqi=0.0,
            bas_sqi=1.0,
            hf_sqi=1.0,
            composite_sqi=0.0,
            grade="unusable",
            is_acceptable=False,
        )

    psqi = calculate_psqi(data_1d, fs=fs)
    ksqi = calculate_ksqi(data_1d)
    ssqi = calculate_ssqi(data_1d)
    bas_sqi = calculate_baseline_sqi(data_1d, fs=fs)
    hf_sqi = calculate_hf_sqi(data_1d, fs=fs)

    # Normalized component scoring [0, 1]
    # 1. pSQI score: ideal between 0.4 and 0.9
    score_psqi = np.clip(psqi / 0.6, 0.0, 1.0)

    # 2. kSQI score: clean ECG > 5.0, gaussian noise ≈ 3.0
    score_ksqi = np.clip((ksqi - 2.5) / 5.0, 0.0, 1.0)

    # 3. basSQI score: penalty for high baseline drift
    score_bas = np.clip(1.0 - (bas_sqi / 0.5), 0.0, 1.0)

    # 4. hfSQI score: penalty for high high-frequency noise
    score_hf = np.clip(1.0 - (hf_sqi / 0.5), 0.0, 1.0)

    # Composite weighted score
    composite = 0.35 * score_psqi + 0.30 * score_ksqi + 0.20 * score_bas + 0.15 * score_hf
    composite = float(np.clip(composite, 0.0, 1.0))

    if composite >= 0.75:
        grade = "excellent"
    elif composite >= acceptance_threshold:
        grade = "acceptable"
    else:
        grade = "unusable"

    return LeadQuality(
        lead_name=lead_name,
        psqi=psqi,
        ksqi=ksqi,
        ssqi=ssqi,
        bas_sqi=bas_sqi,
        hf_sqi=hf_sqi,
        composite_sqi=composite,
        grade=grade,
        is_acceptable=(composite >= acceptance_threshold),
    )

calculate_psqi(data_1d, fs=250.0, qrs_band=(5.0, 15.0), total_band=(5.0, 40.0))

Compute Power Spectrum SQI (pSQI): Ratio of power in the QRS band (5-15 Hz) relative to the total ECG diagnostic frequency band (5-40 Hz).

Higher pSQI indicates well-defined QRS complexes with minimal baseline or high-frequency distortion.

Source code in ecg_sdk/preprocessing/quality.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def calculate_psqi(
    data_1d: np.ndarray,
    fs: float = 250.0,
    qrs_band: tuple[float, float] = (5.0, 15.0),
    total_band: tuple[float, float] = (5.0, 40.0),
) -> float:
    """
    Compute Power Spectrum SQI (pSQI): Ratio of power in the QRS band (5-15 Hz)
    relative to the total ECG diagnostic frequency band (5-40 Hz).

    Higher pSQI indicates well-defined QRS complexes with minimal baseline or high-frequency distortion.
    """
    freqs, psd = scipy_signal.welch(data_1d, fs=fs, nperseg=min(len(data_1d), int(2 * fs)))
    if len(freqs) == 0 or np.sum(psd) == 0:
        return 0.0

    # Integrate PSD in QRS band
    qrs_mask = (freqs >= qrs_band[0]) & (freqs <= qrs_band[1])
    total_mask = (freqs >= total_band[0]) & (freqs <= total_band[1])

    p_qrs = _trapezoid(psd[qrs_mask], freqs[qrs_mask]) if np.any(qrs_mask) else 0.0
    p_total = _trapezoid(psd[total_mask], freqs[total_mask]) if np.any(total_mask) else 0.0

    if p_total <= 0:
        return 0.0
    return float(np.clip(p_qrs / p_total, 0.0, 1.0))

calculate_ksqi(data_1d)

Compute Kurtosis SQI (kSQI): 4th standardized statistical moment. Clean ECG with tall R-peaks has kSQI > 5.0. Gaussian noise has kSQI ≈ 3.0.

Source code in ecg_sdk/preprocessing/quality.py 
112
113
114
115
116
117
118
119
120
def calculate_ksqi(data_1d: np.ndarray) -> float:
    """
    Compute Kurtosis SQI (kSQI): 4th standardized statistical moment.
    Clean ECG with tall R-peaks has kSQI > 5.0. Gaussian noise has kSQI ≈ 3.0.
    """
    if len(data_1d) < 4:
        return 0.0
    k = float(stats.kurtosis(data_1d, fisher=False))  # Pearson definition (normal = 3.0)
    return float(np.nan_to_num(k, nan=0.0, posinf=100.0, neginf=0.0))

calculate_ssqi(data_1d)

Compute Skewness SQI (sSQI): 3rd standardized statistical moment.

Source code in ecg_sdk/preprocessing/quality.py 
123
124
125
126
127
128
129
130
def calculate_ssqi(data_1d: np.ndarray) -> float:
    """
    Compute Skewness SQI (sSQI): 3rd standardized statistical moment.
    """
    if len(data_1d) < 3:
        return 0.0
    s = float(stats.skew(data_1d))
    return float(np.nan_to_num(s, nan=0.0))

calculate_baseline_sqi(data_1d, fs=250.0)

Compute Baseline Drift Ratio (basSQI): Power below 0.5 Hz / power in 0.5-40 Hz. Lower values indicate less baseline drift contamination.

Source code in ecg_sdk/preprocessing/quality.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def calculate_baseline_sqi(data_1d: np.ndarray, fs: float = 250.0) -> float:
    """
    Compute Baseline Drift Ratio (basSQI): Power below 0.5 Hz / power in 0.5-40 Hz.
    Lower values indicate less baseline drift contamination.
    """
    freqs, psd = scipy_signal.welch(data_1d, fs=fs, nperseg=min(len(data_1d), int(2 * fs)))
    if len(freqs) == 0 or np.sum(psd) == 0:
        return 1.0

    drift_mask = (freqs >= 0.0) & (freqs < 0.5)
    ecg_mask = (freqs >= 0.5) & (freqs <= 40.0)

    p_drift = _trapezoid(psd[drift_mask], freqs[drift_mask]) if np.any(drift_mask) else 0.0
    p_ecg = _trapezoid(psd[ecg_mask], freqs[ecg_mask]) if np.any(ecg_mask) else 0.0

    if p_ecg <= 0:
        return 1.0
    return float(np.nan_to_num(p_drift / p_ecg, nan=1.0))

calculate_hf_sqi(data_1d, fs=250.0)

Compute High-Frequency Noise Ratio (hfSQI): Power above 45 Hz / power in 0.5-40 Hz. Lower values indicate less EMG muscle or powerline noise.

Source code in ecg_sdk/preprocessing/quality.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def calculate_hf_sqi(data_1d: np.ndarray, fs: float = 250.0) -> float:
    """
    Compute High-Frequency Noise Ratio (hfSQI): Power above 45 Hz / power in 0.5-40 Hz.
    Lower values indicate less EMG muscle or powerline noise.
    """
    freqs, psd = scipy_signal.welch(data_1d, fs=fs, nperseg=min(len(data_1d), int(2 * fs)))
    if len(freqs) == 0 or np.sum(psd) == 0:
        return 1.0

    hf_mask = (freqs > 45.0) & (freqs <= (fs * 0.5))
    ecg_mask = (freqs >= 0.5) & (freqs <= 40.0)

    p_hf = _trapezoid(psd[hf_mask], freqs[hf_mask]) if np.any(hf_mask) else 0.0
    p_ecg = _trapezoid(psd[ecg_mask], freqs[ecg_mask]) if np.any(ecg_mask) else 0.0

    if p_ecg <= 0:
        return 1.0
    return float(np.nan_to_num(p_hf / p_ecg, nan=1.0))