Help customizing PPG pipeline

[MalloryJfeldman]

Hey all, I'm trying to customize a PPG pipeline. I'm relatively new to python and cannot figure out what is going wrong. Even when I get my code to work, I can't seem to get the custom functions to replicate the normal preprocessing function output. Does anyone have code for custom defined functions that you're willing to share? I think it would be really helpful to see an example!

[DominiqueMakowski]

Hi, would you mind sharing what you have so far and maybe we can troubleshoot your issue

[MalloryJfeldman]

As of right now-- i've literally copied over the code for the functions:

from neurokit2.misc import as_vector
from neurokit2.signal import signal_filter
from neurokit2.signal import signal_rate
from neurokit2.signal.signal_formatpeaks import _signal_from_indices
from neurokit2.misc import as_vector
from neurokit2.ppg.ppg_findpeaks import ppg_findpeaks

def unc_ppg_clean(ppg_signal, sampling_rate=SR, method="elgendi"):

    ppg_signal = as_vector(ppg_signal)

    method = method.lower()
    if method in ["elgendi"]:
        clean = _ppg_clean_elgendi(ppg_signal, sampling_rate)
    else:
        raise ValueError("Neurokit error: Please use one of the following methods: 'elgendi'.")

    return clean

def _ppg_clean_elgendi(ppg_signal, sampling_rate):

    filtered = signal_filter(
        ppg_signal, sampling_rate=sampling_rate, lowcut=0.5, highcut=5, order=3, method="butter_ba"
    )
    return filtered

def unc_ppg_process(ppg_signal, sampling_rate=SR, **kwargs):

    # Sanitize input
    ppg_signal = as_vector(ppg_signal)

    # Clean signal
    ppg_cleaned = unc_ppg_clean(ppg_signal, sampling_rate=sampling_rate)

    # Find peaks
    info = ppg_findpeaks(ppg_cleaned, sampling_rate=sampling_rate, **kwargs)

    # Mark peaks
    peaks_signal = _signal_from_indices(info['PPG_Peaks'], desired_length=len(ppg_cleaned))

    # Rate computation
    rate = signal_rate(info["PPG_Peaks"], sampling_rate=sampling_rate, desired_length=len(ppg_cleaned))

    # Prepare output
    signals = pd.DataFrame({"PPG_Raw": ppg_signal, "PPG_Clean": ppg_cleaned,
                            "PPG_Rate": rate, "PPG_Peaks": peaks_signal})

    return signals, info

But then when I run:

signals, info = unc_ppg_process(ppg_signal, sampling_rate=SR) signals, info = nk.ppg_process(ppg_signal, sampling_rate=SR)

The two lines give me different output. Ultimately I want to be able to change the filtering settings but I don't want to change anything until I can at least replicate the output from the default functions.

[MalloryJfeldman]

[DerAndereJohannes]

I might be understanding something wrong, but comparing your code to the ppg default function for me shows a difference, but not as large as you show there. Here is my figure:

I generate this using the following code:

# signal
ppg_signal = nk.ppg_simulate(duration=40, sampling_rate=SR, heart_rate=75, random_state=42)

# processing
signals, info = nk.ppg_process(ppg_signal, sampling_rate=SR)
signals2, info2 = unc_ppg_process(ppg_signal, sampling_rate=SR)

# graphing
fig, axs = plt.subplots(2, 1)
axs[0].set_title("Neurokit Default")
axs[0].plot(signals["PPG_Clean"])
axs[0].scatter(x=info["PPG_Peaks"], y=[signals["PPG_Clean"][i] for i in info["PPG_Peaks"]], color="red")
axs[0].legend(["Clean signal", "Peaks"])
axs[1].set_title("unc_ppg_process Custom Function")
axs[1].plot(signals2["PPG_Clean"])
axs[1].scatter(x=info2["PPG_Peaks"], y=[signals2["PPG_Clean"][i] for i in info2["PPG_Peaks"]], color="red")
axs[1].legend(["Clean signal", "Peaks"])
plt.tight_layout()
plt.show()