# -*- coding: utf-8 -*-
import pandas as pd
from ..misc import as_vector
from ..misc.report import create_report
from ..signal import signal_rate
from .rsp_amplitude import rsp_amplitude
from .rsp_clean import rsp_clean
from .rsp_methods import rsp_methods
from .rsp_peaks import rsp_peaks
from .rsp_phase import rsp_phase
from .rsp_plot import rsp_plot
from .rsp_rvt import rsp_rvt
from .rsp_symmetry import rsp_symmetry
[docs]
def rsp_process(
rsp_signal,
sampling_rate=1000,
method="khodadad2018",
method_rvt="harrison2021",
report=None,
**kwargs
):
"""**Process a respiration (RSP) signal**
Convenience function that automatically processes a respiration signal with one of the
following methods:
* `Khodadad et al. (2018) <https://iopscience.iop.org/article/10.1088/1361-6579/aad7e6/meta>`_
* `BioSPPy <https://github.com/PIA-Group/BioSPPy/blob/master/biosppy/signals/resp.py>`_
Parameters
----------
rsp_signal : Union[list, np.array, pd.Series]
The raw respiration channel (as measured, for instance, by a respiration belt).
sampling_rate : int
The sampling frequency of :func:`.rsp_signal` (in Hz, i.e., samples/second).
method : str
The processing pipeline to apply. Can be one of ``"khodadad2018"`` (default)
or ``"biosppy"``.
method_rvt : str
The rvt method to apply. Can be one of ``"harrison2021"`` (default), ``"birn2006"``
or ``"power2020"``.
report : str
The filename of a report containing description and figures of processing
(e.g. ``"myreport.html"``). Needs to be supplied if a report file
should be generated. Defaults to ``None``. Can also be ``"text"`` to
just print the text in the console without saving anything.
**kwargs
Other arguments to be passed to specific methods. For more information,
see :func:`.rsp_methods`.
Returns
-------
signals : DataFrame
A DataFrame of same length as :func:`.rsp_signal` containing the following columns:
* ``"RSP_Raw"``: the raw signal.
* ``"RSP_Clean"``: the cleaned signal.
* ``"RSP_Peaks"``: the respiratory peaks (exhalation onsets) marked as "1" in a list of
zeros.
* ``"RSP_Troughs"``: the respiratory troughs (inhalation onsets) marked as "1" in a list of
zeros.
* ``"RSP_Rate"``: breathing rate interpolated between inhalation peaks.
* ``"RSP_Amplitude"``: breathing amplitude interpolated between inhalation peaks.
* ``"RSP_Phase"``: breathing phase, marked by "1" for inspiration and "0" for expiration.
* ``"RSP_Phase_Completion"``: breathing phase completion, expressed in percentage (from 0 to
1), representing the stage of the current respiratory phase.
* ``"RSP_RVT"``: respiratory volume per time (RVT).
info : dict
A dictionary containing the samples at which inhalation peaks and exhalation troughs occur,
accessible with the keys ``"RSP_Peaks"``, and ``"RSP_Troughs"`` respectively, as well as the
signals' sampling rate.
See Also
--------
rsp_clean, rsp_findpeaks, .signal_rate, rsp_amplitude, rsp_plot, rsp_phase, rsp_rvt, rsp_symmetry
Examples
--------
.. ipython:: python
import neurokit2 as nk
rsp = nk.rsp_simulate(duration=90, respiratory_rate=15)
signals, info = nk.rsp_process(rsp, sampling_rate=1000, report="text")
@savefig p_rsp_process_1.png scale=100%
fig = nk.rsp_plot(signals, info)
@suppress
plt.close()
"""
# Sanitize input
rsp_signal = as_vector(rsp_signal)
methods = rsp_methods(
sampling_rate=sampling_rate, method=method, method_rvt=method_rvt, **kwargs
)
# Clean signal
rsp_cleaned = rsp_clean(
rsp_signal,
sampling_rate=sampling_rate,
method=methods["method_cleaning"],
**methods["kwargs_cleaning"],
)
# Extract, fix and format peaks
peak_signal, info = rsp_peaks(
rsp_cleaned,
sampling_rate=sampling_rate,
method=methods["method_peaks"],
amplitude_min=0.3,
**methods["kwargs_peaks"],
)
info["sampling_rate"] = sampling_rate # Add sampling rate in dict info
# Get additional parameters
phase = rsp_phase(peak_signal, desired_length=len(rsp_signal))
amplitude = rsp_amplitude(rsp_cleaned, peak_signal)
rate = signal_rate(
info["RSP_Troughs"], sampling_rate=sampling_rate, desired_length=len(rsp_signal)
)
symmetry = rsp_symmetry(rsp_cleaned, peak_signal)
rvt = rsp_rvt(
rsp_cleaned,
method=methods["method_rvt"],
sampling_rate=sampling_rate,
silent=True,
)
# Prepare output
signals = pd.DataFrame(
{
"RSP_Raw": rsp_signal,
"RSP_Clean": rsp_cleaned,
"RSP_Amplitude": amplitude,
"RSP_Rate": rate,
"RSP_RVT": rvt,
}
)
signals = pd.concat([signals, phase, symmetry, peak_signal], axis=1)
if report is not None:
# Generate report containing description and figures of processing
if ".html" in str(report):
fig = rsp_plot(signals, info)
else:
fig = None
create_report(file=report, signals=signals, info=methods, fig=fig)
return signals, info
