BasePepExtractionPipeline#

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

Base class for PEP extraction pipelines.

This class provides all the necessary methods to extract PEP from ECG and ICG data using the specified algorithms. For usage, it is recommended to use the derived pipelines (e.g., PepExtractionPipeline).

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

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.

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

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.

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

Run the pipeline.

Note

It is usually preferred to use safe_run on custom pipelines instead of run, as safe_run can catch certain implementation errors of the run method.

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

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