PepExtractionPipeline#

class pepbench.pipelines.PepExtractionPipeline(*, 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]#

Standard tpcp Pipeline for pre-ejection period (PEP) extraction from ECG and ICG data.

The PepExtractionPipeline orchestrates a full extraction workflow: heartbeat segmentation (ECG), Q-peak detection (ECG), C- and B-point extraction (ICG), outlier correction and final PEP calculation. Algorithms provided to the pipeline are cloned before execution so original instances are not modified.

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:
datapointBasePepDataset

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 BasePepDataset or implement all the same parameters and methods.

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.

Notes

The pipeline sets the handle_missing_events parameter on algorithms that implement CanHandleMissingEventsMixin when the pipeline’s handle_missing_events is not None.

Negative PEP handling is performed by the outlier correction algorithm and controlled via handle_negative_pep which must correspond to one of the values defined in biopsykit.signals.pep._pep_extraction.NEGATIVE_PEP_HANDLING.

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 a single 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: BasePepDatasetT) → Self[source]#

Run the pipeline on a single datapoint.

Executes the extraction sequence and stores results on the pipeline instance (e.g. heartbeat_segmentation_results_, q_peak_results_, c_point_results_, b_point_results_, b_point_after_outlier_correction_results_, pep_results_).

Parameters:
datapointBasePepDataset

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 BasePepDataset or implement all the same parameters and methods.

Returns:
Self

The pipeline instance with extraction results stored as attributes.

Raises:
ValueError

If handle_negative_pep is not one of the allowed values defined in biopsykit.signals.pep._pep_extraction.NEGATIVE_PEP_HANDLING.

Notes

Sampling rates are taken from the datapoint (sampling_rate_ecg and sampling_rate_icg). Outlier correction is applied to B/C points, and the final PEP computation uses the B-points after outlier correction.

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=.