Features¶

Extractor¶

FeatureExtractor ¶

FeatureExtractor(lead_for_intervals: str = 'II', rpeak_method: Optional[str] = None, delineation_method: Optional[str] = None, features: Union[None, FeatureSpec, Iterable[FeatureSpec]] = None, backend: Union[str, DelineationBackend, None] = None, backend_kwargs: Optional[Dict[str, Any]] = None)

Extract clinically meaningful ECG features from an ECGDataset.

Parameters:

Name	Type	Description	Default
`lead_for_intervals`	`str`	Lead whose R peaks define the beats (RR/HR/HRV). Defaults to "II"; if not present the first available lead is used.	`'II'`
`rpeak_method`	`str or None`	`nk.ecg_peaks` method (or `"scipy"`) given to the backend. None keeps the backend's default.	`None`
`delineation_method`	`str or None`	`nk.ecg_delineate` method given to the backend (`hybrid` default `"prominence"`, `neurokit2` default `"dwt"`). None keeps the backend's default.	`None`
`features`	`None, str, FeatureGroup, callable or a list of them`	What to compute: group names (`"rhythm"`, `"intervals"`, `"amplitude"`, `"hrv"`), individual column names (e.g. `"qtc_bazett_ms"`), `FeatureGroup` instances or `func(ctx) -> dict` callables. None / `"all"` computes every registered group.	`None`
`backend`	`str or DelineationBackend`	Delineation library: `"hybrid"` (default), `"neurokit2"`, `"scipy"`, any name added with `register_backend`, or a backend instance.	`None`
`backend_kwargs`	`dict`	Extra constructor arguments for a backend given by name.	`None`

Outputs

beat_df : pd.DataFrame One row per (sample_idx, beat_idx, lead). Always has r_sample and rr_ms; plus the per-beat columns (intervals) and per-lead columns (amplitudes) of the selected groups.

summary_df : pd.DataFrame One row per recording: sample_idx plus the selected feature columns.

Source code in ecgen_eval/features/extractor.py

def __init__(
    self,
    lead_for_intervals: str = "II",
    rpeak_method: Optional[str] = None,
    delineation_method: Optional[str] = None,
    features: Union[None, FeatureSpec, Iterable[FeatureSpec]] = None,
    backend: Union[str, DelineationBackend, None] = None,
    backend_kwargs: Optional[Dict[str, Any]] = None,
) -> None:
    self.lead_for_intervals = lead_for_intervals
    kwargs = dict(backend_kwargs or {})
    if rpeak_method is not None:
        kwargs["rpeak_method"] = rpeak_method
    if delineation_method is not None:
        kwargs["method"] = delineation_method
    self.backend = get_backend(backend, **kwargs)
    self.selection: List[Tuple[FeatureGroup, Optional[List[str]]]] = resolve_features(
        features
    )
    self._warned: set = set()

columns `property` ¶

columns: List[str]

Summary columns this extractor produces (unknown for groups without columns).

extract ¶

extract(dataset: ECGDataset) -> Tuple[DataFrame, DataFrame]

Run the feature extraction pipeline.

Parameters:

Name	Type	Description	Default
`dataset`	`ECGDataset — shape (N, L, T).`		required

Returns:

Type	Description
`(beat_df, summary_df)`

Source code in ecgen_eval/features/extractor.py

def extract(self, dataset: ECGDataset) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Run the feature extraction pipeline.

    Parameters
    ----------
    dataset : ECGDataset  — shape (N, L, T).

    Returns
    -------
    (beat_df, summary_df)
    """
    beat_rows: List[dict] = []
    summary_rows: List[dict] = []

    primary_lead_idx = self._primary_lead_idx(dataset.lead_names)

    for sample_idx in range(dataset.n_samples):
        ctx = self._delineate(dataset.data[sample_idx], dataset, primary_lead_idx)
        summary_rows.append(self._summary_row(ctx, sample_idx))
        beat_rows.extend(self._beat_rows(ctx, sample_idx))

    beat_df = pd.DataFrame(beat_rows)
    summary_df = pd.DataFrame(summary_rows)
    return beat_df, summary_df

extract_features ¶

extract_features(dataset: ECGDataset, lead_for_intervals: str = 'II', features: Union[None, FeatureSpec, Iterable[FeatureSpec]] = None, backend: Union[str, DelineationBackend, None] = None, **kwargs) -> Tuple[DataFrame, DataFrame]

Convenience wrapper around :class:FeatureExtractor.

Extra keyword arguments (rpeak_method, delineation_method, backend_kwargs) are passed to :class:FeatureExtractor.

Returns:

Type	Description
`(beat_df, summary_df)`

Source code in ecgen_eval/features/extractor.py

def extract_features(
    dataset: ECGDataset,
    lead_for_intervals: str = "II",
    features: Union[None, FeatureSpec, Iterable[FeatureSpec]] = None,
    backend: Union[str, DelineationBackend, None] = None,
    **kwargs,
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Convenience wrapper around :class:`FeatureExtractor`.

    Extra keyword arguments (``rpeak_method``, ``delineation_method``, ``backend_kwargs``)
    are passed to :class:`FeatureExtractor`.

    Returns
    -------
    (beat_df, summary_df)
    """
    extractor = FeatureExtractor(
        lead_for_intervals=lead_for_intervals, features=features, backend=backend, **kwargs
    )
    return extractor.extract(dataset)

Feature groups¶

groups ¶

Feature groups: what is measured from a delineated recording.

A feature group computes a few related summary_df columns from a :class:RecordingContext (the recording plus the R peaks and fiducials found by the delineation backend). Groups can also add per-beat or per-lead columns to beat_df.

Built-in groups

============== ============================================================================== rhythm rr_mean_ms, rr_std_ms, hr_bpm intervals pr_interval_ms, qrs_duration_ms, qt_interval_ms, qtc_bazett_ms, p_wave_duration_ms hrv hrv_sdnn_ms, hrv_rmssd_ms, hrv_pnn50 amplitude st_deviation_mv, t_amplitude_mv ============== ==============================================================================

Add your own by subclassing :class:FeatureGroup, or decorate a function with :func:feature_group::

@feature_group("qrs_energy", columns=["qrs_energy"])
def qrs_energy(ctx):
    return {"qrs_energy": float(np.sum(ctx.primary_signal ** 2))}

RecordingContext `dataclass` ¶

RecordingContext(recording: ndarray, lead_names: List[str], fs: float, primary_lead_idx: int, rpeaks: ndarray, waves: Optional[Dict[str, ndarray]], cache: dict = dict())

Everything a feature group may use for one recording.

FeatureGroup ¶

Base class for a group of related features.

Subclasses set name, columns and description and implement :meth:compute. :meth:per_beat / :meth:per_lead optionally add columns to beat_df.

compute ¶

compute(ctx: RecordingContext) -> Dict[str, float]

Summary values for one recording, keyed by column name.

Source code in ecgen_eval/features/groups.py

def compute(self, ctx: RecordingContext) -> Dict[str, float]:
    """Summary values for one recording, keyed by column name."""
    raise NotImplementedError

per_beat ¶

per_beat(ctx: RecordingContext) -> Dict[str, ndarray]

Arrays of length ctx.n_beats (one value per beat, shared by all leads).

Source code in ecgen_eval/features/groups.py

def per_beat(self, ctx: RecordingContext) -> Dict[str, np.ndarray]:
    """Arrays of length ``ctx.n_beats`` (one value per beat, shared by all leads)."""
    return {}

per_lead ¶

per_lead(ctx: RecordingContext) -> Dict[str, ndarray]

Arrays of length len(ctx.lead_names) (one value per lead, shared by beats).

Source code in ecgen_eval/features/groups.py

def per_lead(self, ctx: RecordingContext) -> Dict[str, np.ndarray]:
    """Arrays of length ``len(ctx.lead_names)`` (one value per lead, shared by beats)."""
    return {}

FunctionFeatureGroup ¶

FunctionFeatureGroup(name: str, func: Callable[[RecordingContext], Dict[str, float]], columns: Optional[Sequence[str]] = None, description: str = '')

Bases: FeatureGroup

Wrap a plain function func(ctx) -> dict as a feature group.

columns may be omitted; individual columns of such a group can then not be selected by name, but the group can.

Source code in ecgen_eval/features/groups.py

def __init__(
    self,
    name: str,
    func: Callable[[RecordingContext], Dict[str, float]],
    columns: Optional[Sequence[str]] = None,
    description: str = "",
) -> None:
    self.name = name
    self.func = func
    self.columns = tuple(columns or ())
    self.description = description or (func.__doc__ or "").strip().split("\n")[0]

register_feature_group ¶

register_feature_group(group: FeatureGroup, overwrite: bool = False) -> FeatureGroup

Register a :class:FeatureGroup instance so it can be selected by name.

Source code in ecgen_eval/features/groups.py

def register_feature_group(group: FeatureGroup, overwrite: bool = False) -> FeatureGroup:
    """Register a :class:`FeatureGroup` instance so it can be selected by name."""
    if not group.name:
        raise ValueError("Feature group needs a name.")
    if group.name in AVAILABLE_FEATURE_GROUPS and not overwrite:
        raise ValueError(f"Feature group {group.name!r} already registered "
                         "(pass overwrite=True).")
    clash = set(group.columns) & {c for g in AVAILABLE_FEATURE_GROUPS.values()
                                  if g.name != group.name for c in g.columns}
    if clash:
        raise ValueError(f"Columns {sorted(clash)} already belong to another feature group.")
    AVAILABLE_FEATURE_GROUPS[group.name] = group
    return group

feature_group ¶

feature_group(name: str, columns: Optional[Sequence[str]] = None, description: str = '', overwrite: bool = False)

Decorator registering func(ctx) -> dict as a feature group.

Source code in ecgen_eval/features/groups.py

def feature_group(
    name: str, columns: Optional[Sequence[str]] = None, description: str = "",
    overwrite: bool = False,
):
    """Decorator registering ``func(ctx) -> dict`` as a feature group."""
    def deco(func):
        register_feature_group(FunctionFeatureGroup(name, func, columns, description),
                               overwrite=overwrite)
        return func
    return deco

available_features ¶

available_features() -> Dict[str, Tuple[str, ...]]

Mapping of registered group name → its column names.

Source code in ecgen_eval/features/groups.py

def available_features() -> Dict[str, Tuple[str, ...]]:
    """Mapping of registered group name → its column names."""
    return {name: g.columns for name, g in AVAILABLE_FEATURE_GROUPS.items()}

resolve_features ¶

resolve_features(features: Union[None, FeatureSpec, Iterable[FeatureSpec]]) -> List[Tuple[FeatureGroup, Optional[List[str]]]]

Turn a feature selection into the groups to run.

features may be None / "all" (every registered group), or any mix of group names, column names, :class:FeatureGroup instances and func(ctx) -> dict callables.

Returns:

Type	Description
list of ``(group, columns)`` — ``columns`` lists the columns of that group to keep, or is
`None to keep all of them.`

Source code in ecgen_eval/features/groups.py

def resolve_features(
    features: Union[None, FeatureSpec, Iterable[FeatureSpec]],
) -> List[Tuple[FeatureGroup, Optional[List[str]]]]:
    """
    Turn a feature selection into the groups to run.

    ``features`` may be ``None`` / ``"all"`` (every registered group), or any mix of group
    names, column names, :class:`FeatureGroup` instances and ``func(ctx) -> dict`` callables.

    Returns
    -------
    list of ``(group, columns)`` — ``columns`` lists the columns of that group to keep, or is
    None to keep all of them.
    """
    if features is None or (isinstance(features, str) and features == "all"):
        return [(g, None) for g in AVAILABLE_FEATURE_GROUPS.values()]
    if isinstance(features, (str, FeatureGroup)) or callable(features):
        features = [features]

    column_owner = {c: g for g in AVAILABLE_FEATURE_GROUPS.values() for c in g.columns}
    selected: Dict[str, Tuple[FeatureGroup, Optional[List[str]]]] = {}

    def add(group: FeatureGroup, column: Optional[str] = None) -> None:
        if column is None:
            selected[group.name] = (group, None)
            return
        _, cols = selected.setdefault(group.name, (group, []))
        if cols is not None and column not in cols:
            cols.append(column)

    for spec in features:
        if isinstance(spec, FeatureGroup):
            add(spec)
        elif isinstance(spec, str) and spec == "all":
            for g in AVAILABLE_FEATURE_GROUPS.values():
                add(g)
        elif isinstance(spec, str) and spec in AVAILABLE_FEATURE_GROUPS:
            add(AVAILABLE_FEATURE_GROUPS[spec])
        elif isinstance(spec, str) and spec in column_owner:
            add(column_owner[spec], spec)
        elif callable(spec):
            add(FunctionFeatureGroup(spec.__name__, spec))
        else:
            raise ValueError(
                f"Unknown feature {spec!r}. Groups: {sorted(AVAILABLE_FEATURE_GROUPS)}; "
                f"columns: {sorted(column_owner)}"
            )
    return list(selected.values())

Backends¶

backends ¶

Delineation backends: the library that finds the beats and their fiducial points.

A backend turns one recording into R peaks and a dict of fiducial arrays aligned with them (element i belongs to beat i, NaN when missing). Feature groups (:mod:ecgen_eval.features.groups) only read these, so any backend works with any feature.

Fiducial keys understood by the built-in feature groups:

ECG_QRS_Onsets, ECG_QRS_Offsets, ECG_P_Onsets, ECG_P_Offsets, ECG_T_Peaks, ECG_T_Offsets.

Built-in backends

hybrid (default): NeuroKit2 R peaks, multi-lead spatial-velocity QRS boundaries, NeuroKit2 prominence P/T on every lead with tangent-method T offset (validated against GE 12SL, see :mod:ecgen_eval.features.delineation).
neurokit2: plain NeuroKit2 ecg_peaks + ecg_delineate (dwt, cwt, peak or prominence) on the primary lead, or every lead combined with the median.
scipy: no NeuroKit2. SciPy Pan-Tompkins R peaks + spatial-velocity QRS boundaries; P and T fiducials are absent, so PR, QT, QTc and P duration are NaN.

Add your own with :func:register_backend.

DelineationBackend ¶

Base class for delineation backends.

Subclasses set name (and optionally description / requires) and implement :meth:delineate. Constructors must be callable with keyword arguments only.

delineate ¶

delineate(recording: ndarray, fs: float, primary_lead_idx: int) -> Tuple[ndarray, Optional[Dict[str, ndarray]]]

Parameters:

Name	Type	Default
`recording`	`(L, T) ECG.`	required
`fs`	`float`	required
`primary_lead_idx`	`int — lead whose R peaks define the beats.`	required

Returns:

Type	Description
`(rpeaks, waves) — sorted R-peak sample indices, and fiducial arrays aligned with them`
(``None`` when the recording could not be delineated).

Source code in ecgen_eval/features/backends.py

def delineate(
    self, recording: np.ndarray, fs: float, primary_lead_idx: int
) -> Tuple[np.ndarray, Optional[Dict[str, np.ndarray]]]:
    """
    Parameters
    ----------
    recording : (L, T) ECG.
    fs : float
    primary_lead_idx : int — lead whose R peaks define the beats.

    Returns
    -------
    (rpeaks, waves) — sorted R-peak sample indices, and fiducial arrays aligned with them
    (``None`` when the recording could not be delineated).
    """
    raise NotImplementedError

HybridBackend ¶

HybridBackend(rpeak_method: str = 'neurokit', method: str = 'prominence', pt_waves: bool = True)

Bases: DelineationBackend

Default backend: NeuroKit2 R peaks, spatial-velocity QRS, NeuroKit2 P/T on every lead.

Parameters:

Name	Type	Description	Default
`rpeak_method`	`str`	`nk.ecg_peaks` method, or `"scipy"` for the SciPy Pan-Tompkins detector.	`'neurokit'`
`method`	`str`	`nk.ecg_delineate` method for the P and T waves (default `"prominence"`).	`'prominence'`
`pt_waves`	`bool`	Delineate P and T waves (needs NeuroKit2).	`True`

Source code in ecgen_eval/features/backends.py

def __init__(
    self, rpeak_method: str = "neurokit", method: str = "prominence", pt_waves: bool = True
) -> None:
    self.rpeak_method = rpeak_method
    self.method = method
    self.pt_waves = pt_waves

ScipyBackend ¶

ScipyBackend(rpeak_method: str = 'scipy')

Bases: HybridBackend

SciPy-only backend: Pan-Tompkins R peaks and spatial-velocity QRS, no P/T waves.

Source code in ecgen_eval/features/backends.py

def __init__(self, rpeak_method: str = "scipy") -> None:
    super().__init__(rpeak_method=rpeak_method, pt_waves=False)

NeuroKitBackend ¶

NeuroKitBackend(method: str = 'dwt', rpeak_method: str = 'neurokit', leads: str = 'primary')

Bases: DelineationBackend

Plain NeuroKit2 pipeline: nk.ecg_clean → nk.ecg_peaks → nk.ecg_delineate.

Parameters:

Name	Type	Description	Default
`method`	`str`	`nk.ecg_delineate` method: `"dwt"` (default), `"cwt"`, `"peak"` or `"prominence"`. `"peak"` gives no QRS boundaries, so QRS duration and the QRS onset-based intervals (PR, QT) are NaN.	`'dwt'`
`rpeak_method`	`str`	`nk.ecg_peaks` method.	`'neurokit'`
`leads`	`str`	`"primary"` delineates the primary lead only; `"all"` delineates every lead at the primary lead's R peaks and takes the median across leads.	`'primary'`

Source code in ecgen_eval/features/backends.py

def __init__(
    self, method: str = "dwt", rpeak_method: str = "neurokit", leads: str = "primary"
) -> None:
    if leads not in ("primary", "all"):
        raise ValueError(f"leads must be 'primary' or 'all', got {leads!r}")
    self.method = method
    self.rpeak_method = rpeak_method
    self.leads = leads

register_backend ¶

register_backend(cls: Type[DelineationBackend], name: Optional[str] = None, overwrite: bool = False) -> Type[DelineationBackend]

Register a :class:DelineationBackend subclass so it can be selected by name.

Usable as a decorator::

@register_backend
class MyBackend(DelineationBackend):
    name = "mylib"
    def delineate(self, recording, fs, primary_lead_idx): ...

Source code in ecgen_eval/features/backends.py

def register_backend(
    cls: Type[DelineationBackend], name: Optional[str] = None, overwrite: bool = False
) -> Type[DelineationBackend]:
    """
    Register a :class:`DelineationBackend` subclass so it can be selected by name.

    Usable as a decorator::

        @register_backend
        class MyBackend(DelineationBackend):
            name = "mylib"
            def delineate(self, recording, fs, primary_lead_idx): ...
    """
    key = name or cls.name
    if not key:
        raise ValueError("Backend needs a name (set the `name` class attribute).")
    if key in AVAILABLE_BACKENDS and not overwrite:
        raise ValueError(f"Backend {key!r} already registered (pass overwrite=True).")
    AVAILABLE_BACKENDS[key] = cls
    return cls

get_backend ¶

get_backend(backend: Union[str, DelineationBackend, None] = None, **kwargs) -> DelineationBackend

Return a backend instance from a registered name (kwargs go to its constructor).

Source code in ecgen_eval/features/backends.py

def get_backend(
    backend: Union[str, DelineationBackend, None] = None, **kwargs
) -> DelineationBackend:
    """Return a backend instance from a registered name (kwargs go to its constructor)."""
    if isinstance(backend, DelineationBackend):
        if kwargs:
            raise ValueError("Backend options can only be given with a backend name.")
        return backend
    name = backend or DEFAULT_BACKEND
    if name not in AVAILABLE_BACKENDS:
        raise ValueError(
            f"Unknown feature backend {name!r}. Available: {sorted(AVAILABLE_BACKENDS)}"
        )
    return AVAILABLE_BACKENDS[name](**kwargs)

R-peak detection and delineation¶

qrs_detection ¶

R-peak detection for ECG signals.

Primary: neurokit2.ecg_clean() + neurokit2.ecg_peaks() Fallback: scipy-based implementation in ecgen_eval.features.utils

detect_rpeaks ¶

detect_rpeaks(signal: ndarray, fs: float, method: str = 'neurokit') -> ndarray

Detect R-peak sample indices in a single-lead ECG signal.

Parameters:

Name	Type	Description	Default
`signal`	`np.ndarray shape (T,)`	Single-lead ECG time series.	required
`fs`	`float`	Sampling frequency in Hz.	required
`method`	`str`	Passed to neurokit2.ecg_peaks() (e.g. `"neurokit"`, `"pantompkins1985"`, `"hamilton2002"`, `"elgendi2010"`). `"scipy"` forces the SciPy Pan-Tompkins detector, which is also the fallback when neurokit2 is unavailable.	`'neurokit'`

Returns:

Type	Description
`np.ndarray of int — R-peak sample indices, sorted ascending.`

Source code in ecgen_eval/features/qrs_detection.py

def detect_rpeaks(signal: np.ndarray, fs: float, method: str = "neurokit") -> np.ndarray:
    """
    Detect R-peak sample indices in a single-lead ECG signal.

    Parameters
    ----------
    signal : np.ndarray  shape (T,)
        Single-lead ECG time series.
    fs : float
        Sampling frequency in Hz.
    method : str
        Passed to neurokit2.ecg_peaks() (e.g. ``"neurokit"``, ``"pantompkins1985"``,
        ``"hamilton2002"``, ``"elgendi2010"``). ``"scipy"`` forces the SciPy Pan-Tompkins
        detector, which is also the fallback when neurokit2 is unavailable.

    Returns
    -------
    np.ndarray of int  — R-peak sample indices, sorted ascending.
    """
    if signal.ndim != 1:
        raise ValueError(f"Expected 1-D signal, got shape {signal.shape}")

    if method == "scipy":
        from ecgen_eval.features.utils import find_r_peaks_scipy

        return find_r_peaks_scipy(signal, fs)

    try:
        import neurokit2 as nk  # optional dependency

        clean = nk.ecg_clean(signal.astype(float), sampling_rate=int(fs))
        _, info = nk.ecg_peaks(clean, sampling_rate=int(fs), method=method)
        peaks = np.asarray(info["ECG_R_Peaks"], dtype=int)
        return peaks[peaks >= 0]
    except ImportError:
        from ecgen_eval.features.utils import find_r_peaks_scipy

        return find_r_peaks_scipy(signal, fs)
    except Exception:
        # neurokit2 present but failed (e.g. signal too short/flat)
        from ecgen_eval.features.utils import find_r_peaks_scipy

        return find_r_peaks_scipy(signal, fs)

delineate_ecg ¶

delineate_ecg(signal: ndarray, rpeaks: ndarray, fs: float, method: str = 'prominence') -> Optional[dict]

Delineate P, QRS and T boundaries of a single-lead ECG.

Thin wrapper around :func:ecgen_eval.features.delineation.delineate_recording; returns fiducial arrays aligned with rpeaks, or None when there are fewer than 2 R peaks.

Source code in ecgen_eval/features/qrs_detection.py

def delineate_ecg(
    signal: np.ndarray, rpeaks: np.ndarray, fs: float, method: str = "prominence"
) -> Optional[dict]:
    """
    Delineate P, QRS and T boundaries of a single-lead ECG.

    Thin wrapper around :func:`ecgen_eval.features.delineation.delineate_recording`; returns
    fiducial arrays aligned with ``rpeaks``, or None when there are fewer than 2 R peaks.
    """
    from ecgen_eval.features.delineation import delineate_recording

    try:
        return delineate_recording(signal, rpeaks, fs, method=method)
    except Exception:
        return None

Interval features¶

interval_features ¶

Interval feature extraction: RR, HR, PR, QRS, QT, QTc, P-wave duration.

Intervals are computed per beat from the aligned fiducial points of :func:ecgen_eval.features.delineation.delineate_recording, checked against plausible physiological ranges, and summarised with the median over beats:

PR = QRS onset - P onset
QRS = QRS offset - QRS onset
QT = T offset - QRS onset; QTc (Bazett) = QT / sqrt(mean RR in s)
P duration = P offset - P onset

QRS onset/offset need only NumPy/SciPy; the P and T points need NeuroKit2. Features whose points are missing are NaN (there are no fixed placeholder values).

extract_rr_hr ¶

extract_rr_hr(rpeaks: ndarray, fs: float) -> Dict[str, float]

RR intervals (ms) and heart rate (bpm) from R-peak positions.

Source code in ecgen_eval/features/interval_features.py

def extract_rr_hr(rpeaks: np.ndarray, fs: float) -> Dict[str, float]:
    """RR intervals (ms) and heart rate (bpm) from R-peak positions."""
    if len(rpeaks) < 2:
        return {
            "rr_mean_ms": np.nan,
            "rr_std_ms": np.nan,
            "hr_bpm": np.nan,
        }
    rr_samples = np.diff(rpeaks)
    rr_ms = samples_to_ms(rr_samples, fs)
    hr = 60_000.0 / np.mean(rr_ms) if np.mean(rr_ms) > 0 else np.nan
    return {
        "rr_mean_ms": float(np.mean(rr_ms)),
        "rr_std_ms": float(np.std(rr_ms)),
        "hr_bpm": float(hr),
    }

beat_intervals ¶

beat_intervals(waves: Optional[dict], fs: float) -> Dict[str, ndarray]

Per-beat intervals (ms, NaN where missing or implausible) from aligned fiducials.

Source code in ecgen_eval/features/interval_features.py

def beat_intervals(waves: Optional[dict], fs: float) -> Dict[str, np.ndarray]:
    """Per-beat intervals (ms, NaN where missing or implausible) from aligned fiducials."""
    out = {}
    for name, (start, end) in _INTERVAL_POINTS.items():
        if not waves or start not in waves or end not in waves:
            out[name] = np.array([])
            continue
        a = np.asarray(waves[start], dtype=float)
        b = np.asarray(waves[end], dtype=float)
        ms = (b - a) / fs * 1000.0
        lo, hi = INTERVAL_BOUNDS_MS[name]
        with np.errstate(invalid="ignore"):
            out[name] = np.where((ms >= lo) & (ms <= hi), ms, np.nan)
    return out

extract_interval_features ¶

extract_interval_features(signal: ndarray, fs: float, rpeaks: ndarray, waves: Optional[dict] = None) -> Dict[str, float]

Extract PR, QRS, QT, QTc, and P-wave duration.

Parameters:

Name	Type	Description	Default
`signal`	(T,) or (L, T) ECG; only used to delineate when ``waves`` is None.		required
`fs`	`float`		required
`rpeaks`	`(M,) R-peak sample indices.`		required
`waves`	`dict or None`	Fiducials aligned with `rpeaks` (from `delineate_recording`). When None, the signal is delineated here.	`None`

Returns:

Type	Description
`dict with keys: pr_interval_ms, qrs_duration_ms, qt_interval_ms,`	qtc_bazett_ms, p_wave_duration_ms

Source code in ecgen_eval/features/interval_features.py

def extract_interval_features(
    signal: np.ndarray,
    fs: float,
    rpeaks: np.ndarray,
    waves: Optional[dict] = None,
) -> Dict[str, float]:
    """
    Extract PR, QRS, QT, QTc, and P-wave duration.

    Parameters
    ----------
    signal : (T,) or (L, T) ECG; only used to delineate when ``waves`` is None.
    fs : float
    rpeaks : (M,) R-peak sample indices.
    waves : dict or None
        Fiducials aligned with ``rpeaks`` (from ``delineate_recording``). When None, the
        signal is delineated here.

    Returns
    -------
    dict with keys: pr_interval_ms, qrs_duration_ms, qt_interval_ms,
                    qtc_bazett_ms, p_wave_duration_ms
    """
    if waves is None and len(rpeaks) >= 2:
        from ecgen_eval.features.delineation import delineate_recording

        waves = delineate_recording(signal, rpeaks, fs)

    per_beat = beat_intervals(waves, fs)
    feats = {name: _median(v) for name, v in per_beat.items()}

    rr = extract_rr_hr(rpeaks, fs)
    rr_s = rr["rr_mean_ms"] / 1000.0
    qt = feats["qt_interval_ms"]
    qtc = qt / np.sqrt(rr_s) if np.isfinite(qt) and np.isfinite(rr_s) and rr_s > 0 else np.nan

    return {
        "pr_interval_ms": feats["pr_interval_ms"],
        "qrs_duration_ms": feats["qrs_duration_ms"],
        "qt_interval_ms": qt,
        "qtc_bazett_ms": float(qtc),
        "p_wave_duration_ms": feats["p_wave_duration_ms"],
    }

Amplitude features¶

amplitude_features ¶

Amplitude feature extraction: ST deviation, T-wave amplitude.

Both are measured per beat against the beat's baseline, the median level of the PR segment just before the QRS onset ([QRS onset - 40 ms, QRS onset - 10 ms]):

ST deviation = signal at the J point (QRS offset) + 80 ms - baseline.
T amplitude = signed extreme of the T wave (the sample furthest from the baseline) between J + 80 ms and the T offset (or QRS onset + min(0.6 RR, 550 ms) when the T offset is unknown).

The recording value is the median over beats. Fiducials come from :func:ecgen_eval.features.delineation.delineate_recording, so leads of a recording share the same beat timing.

extract_amplitude_features ¶

extract_amplitude_features(signal: ndarray, fs: float, rpeaks: ndarray, waves: Optional[dict] = None) -> Dict[str, float]

Extract ST deviation and T-wave amplitude for one lead.

Parameters:

Name	Type	Description	Default
`signal`	`(T,) Single-lead ECG (mV).`		required
`fs`	`float`		required
`rpeaks`	`(M,) R-peak sample indices.`		required
`waves`	`dict or None`	Fiducials aligned with `rpeaks` (from `delineate_recording`); computed from this lead alone when None.	`None`

Returns:

Type	Description
`dict with keys: st_deviation_mv, t_amplitude_mv`

Source code in ecgen_eval/features/amplitude_features.py

def extract_amplitude_features(
    signal: np.ndarray,
    fs: float,
    rpeaks: np.ndarray,
    waves: Optional[dict] = None,
) -> Dict[str, float]:
    """
    Extract ST deviation and T-wave amplitude for one lead.

    Parameters
    ----------
    signal : (T,) Single-lead ECG (mV).
    fs : float
    rpeaks : (M,) R-peak sample indices.
    waves : dict or None
        Fiducials aligned with ``rpeaks`` (from ``delineate_recording``); computed from this
        lead alone when None.

    Returns
    -------
    dict with keys: st_deviation_mv, t_amplitude_mv
    """
    nan = {"st_deviation_mv": np.nan, "t_amplitude_mv": np.nan}
    rpeaks = np.asarray(rpeaks, dtype=int)
    if len(rpeaks) < 2:
        return nan
    if waves is None:
        from ecgen_eval.features.delineation import delineate_recording

        waves = delineate_recording(signal, rpeaks, fs)
    if not waves:
        return nan

    sig = np.asarray(signal, dtype=float)
    n = len(sig)
    on = np.asarray(waves["ECG_QRS_Onsets"], dtype=float)
    off = np.asarray(waves["ECG_QRS_Offsets"], dtype=float)
    t_off = np.asarray(waves.get("ECG_T_Offsets", np.full(len(rpeaks), np.nan)), dtype=float)
    rr = np.diff(rpeaks)
    rr = np.append(rr, rr[-1])

    b0, b1 = ms_to_samples(_BASELINE_START_MS, fs), ms_to_samples(_BASELINE_END_MS, fs)
    st_gap = ms_to_samples(_ST_AFTER_J_MS, fs)
    t_cap = ms_to_samples(_T_MAX_AFTER_ONSET_MS, fs)

    st_vals, t_vals = [], []
    for i in range(len(rpeaks)):
        if not (np.isfinite(on[i]) and np.isfinite(off[i])):
            continue
        qo, j = int(on[i]), int(off[i])
        if qo - b0 < 0:
            continue
        baseline = float(np.median(sig[qo - b0: qo - b1 + 1]))
        st_pt = j + st_gap
        if st_pt < n:
            st_vals.append(sig[st_pt] - baseline)
        end = qo + min(t_cap, int(_T_MAX_RR_FRACTION * rr[i]))
        if np.isfinite(t_off[i]) and st_pt < t_off[i] <= end + t_cap:
            end = int(t_off[i])
        end = min(end, n - 1)
        if end > st_pt:
            seg = sig[st_pt: end + 1] - baseline
            t_vals.append(float(seg[np.argmax(np.abs(seg))]))

    return {
        "st_deviation_mv": float(np.median(st_vals)) if st_vals else np.nan,
        "t_amplitude_mv": float(np.median(t_vals)) if t_vals else np.nan,
    }

HRV features¶

hrv_features ¶

HRV feature extraction: SDNN, RMSSD, pNN50.

Computed directly from RR interval series; does not require neurokit2. When neurokit2 is installed it is used to cross-check, but the same manual computation is the primary path to keep the dependency optional.

extract_hrv_features ¶

extract_hrv_features(rpeaks: ndarray, fs: float) -> Dict[str, float]

Compute time-domain HRV indices from R-peak positions.

Parameters:

Name	Type	Description	Default
`rpeaks`	`np.ndarray Shape (M,) — sorted R-peak sample indices.`		required
`fs`	`float — sampling frequency in Hz.`		required

Returns:

Type	Description
`dict with keys: hrv_sdnn_ms, hrv_rmssd_ms, hrv_pnn50`

Source code in ecgen_eval/features/hrv_features.py

def extract_hrv_features(rpeaks: np.ndarray, fs: float) -> Dict[str, float]:
    """
    Compute time-domain HRV indices from R-peak positions.

    Parameters
    ----------
    rpeaks : np.ndarray  Shape (M,) — sorted R-peak sample indices.
    fs : float — sampling frequency in Hz.

    Returns
    -------
    dict with keys: hrv_sdnn_ms, hrv_rmssd_ms, hrv_pnn50
    """
    nan = {"hrv_sdnn_ms": np.nan, "hrv_rmssd_ms": np.nan, "hrv_pnn50": np.nan}

    if len(rpeaks) < 3:
        return nan

    rr_ms = samples_to_ms(np.diff(rpeaks.astype(float)), fs)

    # Filter physiologically implausible RR (< 300 ms or > 2000 ms)
    rr_ms = rr_ms[(rr_ms >= 300) & (rr_ms <= 2000)]
    if len(rr_ms) < 2:
        return nan

    sdnn = float(np.std(rr_ms, ddof=1))

    successive_diff = np.diff(rr_ms)
    rmssd = float(np.sqrt(np.mean(successive_diff ** 2)))
    pnn50 = float(100.0 * np.sum(np.abs(successive_diff) > 50.0) / len(successive_diff))

    return {
        "hrv_sdnn_ms": sdnn,
        "hrv_rmssd_ms": rmssd,
        "hrv_pnn50": pnn50,
    }

Utilities¶

utils ¶

Shared helpers for ECG feature extraction (filtering, peak refinement).

bandpass_filter ¶

bandpass_filter(signal: ndarray, fs: float, low: float = 0.5, high: float = 40.0) -> ndarray

4th-order Butterworth bandpass filter. Falls back to identity if scipy absent.

Source code in ecgen_eval/features/utils.py

def bandpass_filter(signal: np.ndarray, fs: float, low: float = 0.5, high: float = 40.0) -> np.ndarray:
    """4th-order Butterworth bandpass filter. Falls back to identity if scipy absent."""
    try:
        from scipy.signal import butter, filtfilt

        nyq = fs / 2.0
        lo = max(low / nyq, 1e-6)
        hi = min(high / nyq, 1.0 - 1e-6)
        b, a = butter(4, [lo, hi], btype="band")
        return filtfilt(b, a, signal.astype(np.float64)).astype(signal.dtype)
    except ImportError:
        return signal

find_r_peaks_scipy ¶

find_r_peaks_scipy(signal: ndarray, fs: float) -> ndarray

Pan-Tompkins-inspired R-peak detector using scipy.signal.find_peaks.

Steps: bandpass (5–15 Hz) → derivative → square → moving-window integrate → find_peaks with minimum RR distance of 300 ms → refine on filtered signal. Returns array of sample indices.

Source code in ecgen_eval/features/utils.py

def find_r_peaks_scipy(signal: np.ndarray, fs: float) -> np.ndarray:
    """
    Pan-Tompkins-inspired R-peak detector using scipy.signal.find_peaks.

    Steps: bandpass (5–15 Hz) → derivative → square → moving-window integrate
    → find_peaks with minimum RR distance of 300 ms → refine on filtered signal.
    Returns array of sample indices.
    """
    try:
        from scipy.signal import butter, filtfilt, find_peaks

        sig = signal.astype(np.float64)
        nyq = fs / 2.0
        b, a = butter(4, [5.0 / nyq, 15.0 / nyq], btype="band")
        filt = filtfilt(b, a, sig)

        diff = np.gradient(filt)
        squared = diff ** 2
        win = max(1, int(0.15 * fs))
        integrated = np.convolve(squared, np.ones(win) / win, mode="same")

        min_dist = max(1, int(0.3 * fs))  # 300 ms
        peaks, _ = find_peaks(integrated, distance=min_dist, height=0.1 * np.max(integrated))

        # Refine: snap to max of |filt| within ±50 ms
        refine = max(1, int(0.05 * fs))
        refined = []
        for p in peaks:
            lo = max(0, p - refine)
            hi = min(len(filt), p + refine + 1)
            refined.append(lo + int(np.argmax(np.abs(filt[lo:hi]))))
        return np.array(refined, dtype=int)

    except ImportError:
        # Minimal fallback: threshold at 60% of max, spaced ≥300 ms apart
        threshold = 0.6 * float(np.max(np.abs(signal)))
        above = (np.abs(signal) > threshold).astype(int)
        crossings = np.where(np.diff(above) == 1)[0]
        if len(crossings) == 0:
            return np.array([], dtype=int)
        min_dist = int(0.3 * fs)
        peaks = [int(crossings[0])]
        for c in crossings[1:]:
            if c - peaks[-1] >= min_dist:
                peaks.append(int(c))
        return np.array(peaks, dtype=int)

Features¶

Extractor¶

FeatureExtractor ¶

columns property ¶

extract ¶

extract_features ¶

Feature groups¶

groups ¶

RecordingContext dataclass ¶

FeatureGroup ¶

compute ¶

per_beat ¶

per_lead ¶

FunctionFeatureGroup ¶

register_feature_group ¶

feature_group ¶

available_features ¶

resolve_features ¶

Backends¶

backends ¶

DelineationBackend ¶

delineate ¶

HybridBackend ¶

ScipyBackend ¶

NeuroKitBackend ¶

register_backend ¶

get_backend ¶

R-peak detection and delineation¶

qrs_detection ¶

detect_rpeaks ¶

delineate_ecg ¶

Interval features¶

interval_features ¶

extract_rr_hr ¶

beat_intervals ¶

extract_interval_features ¶

Amplitude features¶

amplitude_features ¶

extract_amplitude_features ¶

HRV features¶

hrv_features ¶

extract_hrv_features ¶

Utilities¶

utils ¶

bandpass_filter ¶

find_r_peaks_scipy ¶

columns `property` ¶

RecordingContext `dataclass` ¶