PepExtractionPipelineReferenceQPeaks#

class pepbench.pipelines.PepExtractionPipelineReferenceQPeaks(*, heartbeat_segmentation_algo: BaseHeartbeatSegmentation, q_peak_algo: BaseEcgExtraction, b_point_algo: BaseBPointExtraction, c_point_algo: BaseCPointExtraction = cf(CPointExtractionScipyFindPeaks(handle_missing_events='warn', window_c_correction=3)), outlier_correction_algo: BaseBPointOutlierCorrection | None = None, handle_negative_pep: Literal['nan', 'zero', 'keep'] = 'nan', handle_missing_events: Literal['raise', 'warn', 'ignore'] | None = None)[source]#

tpcp Pipeline for PEP extraction that uses reference Q-peaks for Q-peak detection.

This pipeline validates different B-point extraction algorithms by computing the pre-ejection period (PEP) using reference Q-peak annotations from labeled datasets. It performs heartbeat segmentation, heartbeat matching against reference annotations, Q-peak alignment using reference annotations, runs C-point and B-point extraction, applies outlier correction for B-point samples, and computes final PEP values.

Parameters:
heartbeat_segmentation_algoBaseHeartbeatSegmentation

Algorithm for heartbeat segmentation.

q_peak_algoBaseEcgExtraction

Algorithm for Q-peak extraction.

b_point_algoBaseBPointExtraction

Algorithm for B-point extraction.

c_point_algoBaseCPointExtraction

Algorithm for C-point extraction, necessary for most subsequent B-point extraction algorithms.

outlier_correction_algoBaseOutlierCorrection

Algorithm for outlier correction of B-point data (optional).

handle_negative_pepone of {"nan", "zero", "keep"}
How to handle negative PEP values. Possible values are:

  • "nan": Set negative PEP values to NaN

  • "zero": Set negative PEP values to 0

  • "keep": Keep negative PEP values as is

handle_missing_eventsone of {"warn", "ignore", "raise"}
How to handle missing events. Possible values are:

  • "warn": Issue a warning if missing events are detected

  • "ignore": Ignore missing events

  • "raise": Raise an error if missing events are detected

Other Parameters:
datapointBaseUnifiedPepExtractionDataset

The data to run the pipeline on. This needs to be a valid datapoint (i.e. a dataset with just a single row). The Dataset should be a child class of BaseUnifiedPepExtractionDataset or implement all the same parameters and methods. This means that it must also implement methods to get the reference heartbeats and reference PEP.

Attributes:
``heartbeat_segmentation_results_`` :class:`~biopsykit.signals.ecg.segmentation.HeartbeatSegmentationDataFrame`

Results from the heartbeat segmentation step.

``q_peak_results_`` :class:`~biopsykit.signals.ecg.event_extraction.QPeakDataFrame`

Results from the Q-peak extraction step.

``c_point_results_`` :class:`~biopsykit.signals.icg.event_extraction.CPointDataFrame`

Results from the C-point extraction step.

``b_point_results_`` :class:`~biopsykit.signals.icg.event_extraction.BPointDataFrame`

Results from the B-point extraction step.

``b_point_after_outlier_correction_results_`` :class:`~biopsykit.signals.icg.event_extraction.BPointDataFrame`

Results from the B-point extraction step after outlier correction.

``pep_results_`` :class:`~biopsykit.signals.pep.PepResultDataFrame`

Results from the PEP extraction step.

Methods

clone()

Create a new instance of the class with all parameters copied over.

get_params([deep])

Get parameters for this algorithm.

run(datapoint)

Run the pipeline on the given datapoint.

safe_run(datapoint)

Run the pipeline with some additional checks.

set_params(**params)

Set the parameters of this Algorithm.
__init__(*, heartbeat_segmentation_algo: BaseHeartbeatSegmentation, q_peak_algo: BaseEcgExtraction, b_point_algo: BaseBPointExtraction, c_point_algo: BaseCPointExtraction = cf(CPointExtractionScipyFindPeaks(handle_missing_events='warn', window_c_correction=3)), outlier_correction_algo: BaseBPointOutlierCorrection | None = None, handle_negative_pep: Literal['nan', 'zero', 'keep'] = 'nan', handle_missing_events: Literal['raise', 'warn', 'ignore'] | None = None) → None[source]#

Initialize a BasePepExtractionPipeline.

Parameters:
heartbeat_segmentation_algoBaseHeartbeatSegmentation

Algorithm instance used to segment ECG into heartbeats.

q_peak_algoBaseEcgExtraction

Algorithm instance used to detect Q-peaks in the ECG.

b_point_algoBaseBPointExtraction

Algorithm instance used to detect B-points in the ICG.

c_point_algoBaseCPointExtraction, optional

Algorithm used to detect C-points in the ICG. Required by many B-point extractors. Defaults to a scipy-based peak finder clone.

outlier_correction_algoBaseBPointOutlierCorrection or None, optional

Algorithm for outlier correction applied to B-point results. If None, a dummy no-op outlier corrector is used.

handle_negative_pep{‘nan’, ‘zero’, ‘keep’}, optional
Strategy to handle negative PEP values:

  • 'nan': set negative PEP to NaN

  • 'zero': set negative PEP to 0

  • 'keep': keep negative values as-is

Default is 'nan'.

handle_missing_events{‘warn’, ‘ignore’, ‘raise’} or None, optional

Strategy to handle missing events during extraction. If None, defaults to 'warn'.

run(datapoint: DatasetT) → Self[source]#

Run the pipeline on the given datapoint.

The pipeline executes the following steps:

  • Validate configuration values (for example, that handle_negative_pep is valid).

  • Clone configured algorithms and optionally set handle_missing_events on algorithms that support missing-event handling.

  • Extract heartbeats from the ECG signal.

  • Match extracted heartbeats to the reference heartbeat annotations (tolerance 100 ms).

  • Select true-positive matches and align reference Q-peaks with extracted heartbeat indices.

  • Run C-point and B-point extraction for matched heartbeats, add nan_reason column to aligned Q-peak table.

  • Mark and prepare B-point samples for outlier correction, then apply the outlier correction algorithm.

  • Compute PEP values using aligned Q- and B-points and store results on the pipeline.

Parameters:
datapointBasePepDatasetWithAnnotations

Labeled datapoint that provides ECG data, sampling rates, reference heartbeat annotations, and reference PEP/Q-peak annotations.

Returns:
Self

The pipeline instance with result attributes populated (see class attributes).

Raises:
ValueError

If handle_negative_pep is not a valid value from the allowed options.

Notes

  • Heartbeat matching uses a default tolerance of 100 ms.

  • False positives and negatives in heartbeat matching are currently not corrected automatically by this pipeline; only true-positive matches are used to compute PEP.

clone() → Self[source]#

Create a new instance of the class with all parameters copied over.

This will create a new instance of the class itself and all nested objects

get_params(deep: bool = True) → dict[str, Any][source]#

Get parameters for this algorithm.

Parameters:
deep

Only relevant if object contains nested algorithm objects. If this is the case and deep is True, the params of these nested objects are included in the output using a prefix like nested_object_name__ (Note the two “_” at the end)

Returns:
params

Parameter names mapped to their values.

safe_run(datapoint: DatasetT) → Self[source]#

Run the pipeline with some additional checks.

It is preferred to use this method over run, as it can catch some simple implementation errors of custom pipelines.

The following things are checked:

  • The run method must return self (or at least an instance of the pipeline)

  • The run method must set result attributes on the pipeline

  • All result attributes must have a trailing _ in their name

  • The run method must not modify the input parameters of the pipeline

Parameters:
datapoint

An instance of a tpcp.Dataset containing only a single datapoint. The structure of the data will depend on the dataset.

Returns:
self

The class instance with all result attributes populated

set_params(**params: Any) → Self[source]#

Set the parameters of this Algorithm.

To set parameters of nested objects use nested_object_name__para_name=.