Preprocessing

Transform the signal

pyPCG.preprocessing.slice_signal(sig: pcg_signal, time_len: float = 60, overlap: float = 0) → list[pcg_signal]

Slice long signal into a list of shorter signals

Parameters:

• sig (pcg_signal) – input long signal

• time_len (float, optional) – desired short timelength [seconds]. Defaults to 60.

• overlap (float, optional) – overlap percentage. Defaults to 0.

Returns:

list of shorter signals

Return type:

list[pcg_signal]

pyPCG.preprocessing.resample(sig: pcg_signal, target_fs: int) → pcg_signal

Resample signal to target samplerate

Parameters:

• sig (pcg_signal) – input signal

• target_fs (int) – target samplerate

Returns:

resampled signal

Return type:

pcg_signal

Envelope calculation

pyPCG.preprocessing.envelope(sig: pcg_signal) → pcg_signal

Calculates the envelope of the signal based on Hilbert transformation

Parameters:

sig (pcg_signal) – input signal

Returns:

envelope

Return type:

pcg_signal

pyPCG.preprocessing.homomorphic(sig: pcg_signal, filt_ord: int = 6, filt_cutfreq: float = 8) → pcg_signal

Calculate the homomoprphic envelope of the signal

Parameters:

• sig (pcg_signal) – input signal

• filt_ord (int, optional) – lowpass filter order. Defaults to 6.

• filt_cutfreq (float, optional) – lowpass filter cutoff frequency. Defaults to 8.

Raises:

ValueError – The cutoff frequency exceeds the Nyquist limit

Returns:

homomoprhic envelope

Return type:

pcg_signal

Denoising functions

pyPCG.preprocessing.wt_denoise(sig: pcg_signal, th: float = 0.2, wt_family: str = 'coif4', wt_level: int = 5) → pcg_signal

Denoise the signal with a wavelet thresholding method

Parameters:

• sig (pcg_signal) – input noisy signal

• th (float, optional) – threshold value given as a percentage of maximum. Defaults to 0.2.

• wt_family (str, optional) – wavelet family. Defaults to “coif4”.

• wt_level (int, optional) – wavelet decomposition level. Defaults to 5.

Returns:

denoised signal

Return type:

pcg_signal

pyPCG.preprocessing.wt_denoise_sth(sig: pcg_signal, wt_family: str = 'coif4', wt_level: int = 5) → pcg_signal

Denoise the signal with automatic wavelet thresholding method. Threshold is calculated automatically.

Based on: D.L. Donoho, and I.M. Johnstone, Ideal spatial adaptation by wavelet shrinkage, Biometrika, vol. 81, no. 3, pp. 425-455, 1994

Parameters:

• sig (pcg_signal) – input noisy signal

• wt_family (str, optional) – wavelet family. Defaults to “coif4”.

• wt_level (int, optional) – wavelet decomposition level. Defaults to 5.

Returns:

denoised signal

Return type:

pcg_signal

pyPCG.preprocessing.emd_denoise_sth(sig: pcg_signal) → pcg_signal

EMD based denoising with soft threshold method.

Based on: Boudraa, Abdel-O & Cexus, Jean-Christophe & Saidi, Zazia. (2005). EMD-Based Signal Noise Reduction. Signal Processing. 1.

Note: Tau parameter calculation modified from the original

Parameters:

sig (pcg_signal) – input signal

Returns:

denoised signal

Return type:

pcg_signal

pyPCG.preprocessing.emd_denoise_savgol(sig: pcg_signal, window: int = 10, poly: int = 3) → pcg_signal

EMD based denoising method with Savoy-Golatzky filter method.

Based on: Boudraa, Abdel-O & Cexus, Jean-Christophe & Saidi, Zazia. (2005). EMD-Based Signal Noise Reduction. Signal Processing. 1.

Parameters:

• sig (pcg_signal) – input signal

• window (int, optional) – savgol filter window size [samples]. Defaults to 10.

• poly (int, optional) – savgol polynomial degree to fit. Defaults to 3.

Returns:

denoised signal

Return type:

pcg_signal

Filtering

pyPCG.preprocessing.filter(sig: pcg_signal, filt_ord: int, filt_cutfreq: float, filt_type: str = 'LP') → pcg_signal

Filters the signal based on the input parameters with a Butterworth filter design

Parameters:

• sig (pcg_signal) – input signal

• filt_ord (int) – filter order

• filt_cutfreq (float) – filter cutoff frequency

• filt_type (str, optional) – filter type: “LP” (low-pass) or “HP” (high-pass). Defaults to “LP”.

Raises:

• NotImplementedError – Other filter type

• ValueError – Filter cutoff exceeds Nyquist limit

Returns:

filtered signal

Return type:

pcg_signal

Processing pipeline

typeddict pyPCG.preprocessing.process_config

Type to hold processing calculation configs

Required Keys:

• step (Callable) – function for the calculation

• params (dict[str, Union[int, float, str]]) – parameters to pass to the function as keyword arguments

class pyPCG.preprocessing.process_pipeline(*configs: Callable | process_config)

Processing pipeline. One step’s input is the previous step’s output

steps

List of steps as functions or function and parameters as keyword dictionary

Type:

list[Callable | process_config]

Example

Creating a simple pipeline

>>> import pyPCG
>>> my_pipeline = pyPCG.process_pipeline(pyPCG.zero_center, pyPCG.unit_scale)
>>> print(my_pipeline)
PCG processing pipeline [2 steps]

Creating a pipeline with parameters:

Option 1: Create a dictionary with the appropriate function with the parameters passed as keyword arguments

For an easier experience, use the process_config type

>>> import pyPCG
>>> import pyPCG.preprocessing as preproc
>>> step_1 = {"step":preproc.filter,"params":{"filt_ord":6,"filt_cutfreq":100,"filt_type":"LP"}}
>>> step_2 = {"step":preproc.filter,"params":{"filt_ord":6,"filt_cutfreq":20,"filt_type":"HP"}}
>>> my_pipeline = pyPCG.process_pipeline(step_1,step_2)

Option 2: Using functools.partial

>>> import pyPCG
>>> import pyPCG.preprocessing as preproc
>>> from functools import partial
>>> step_1 = partial(preproc.filter, filt_ord=6, filt_cutfreq=100, filt_type="LP")
>>> step_2 = partial(preproc.filter, filt_ord=6, filt_cutfreq=20, filt_type="HP")
>>> my_pipeline = pyPCG.process_pipeline(step_1,step_2)

Use the above pipeline: >>> import pyPCG.io as pcg_io >>> data, fs = pcg_io.read_signal_file(“example.wav”,”wav”) >>> example = pyPCG.pcg_signal(data,fs) >>> processed = my_pipeline.run(example)

run(input: pcg_signal) → pcg_signal

Run the processing pipeline

Parameters:

input (pcg_signal) – input signal

Returns:

processed signal

Return type:

pcg_signal