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 PEP extraction from ECG and ICG data.

This pipeline can be used to run the PEP extraction algorithms on a given dataset. It will use the algorithms passed to the __init__ method to extract PEP from the data. The results will be stored in the attributes of the class.

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_HeartbeatSegmentationDataFrame

Results from the heartbeat segmentation step.

q_peak_results_QPeakDataFrame

Results from the Q-peak extraction step.

c_point_results_CPointDataFrame

Results from the C-point extraction step.

b_point_results_BPointDataFrame

Results from the B-point extraction step.

b_point_after_outlier_correction_results_BPointDataFrame

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

pep_results_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]#
run(datapoint: BasePepDatasetT) → Self[source]#

Run the pipeline on the given datapoint.

The pipeline will extract PEP from the given datapoint using the specified algorithms. The results will be stored in the attributes of the class.

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 instance of the class with the results set.

Raises:
ValueError

If the handle_negative_pep parameter is not one of the allowed values.

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