Source code for neurokit2.rsp.rsp_phase

# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd

from ..signal import signal_phase
from .rsp_fixpeaks import _rsp_fixpeaks_retrieve




def rsp_phase(peaks, troughs=None, desired_length=None):
    """**Compute respiratory phase (inspiration and expiration)**

    Finds the respiratory phase, labelled as 1 for inspiration and 0 for expiration.

    Parameters
    ----------
    peaks : list or array or DataFrame or Series or dict
        The samples at which the inhalation peaks occur. If a dict or a DataFrame is passed, it is
        assumed that these containers were obtained with :func:`.rsp_findpeaks`.
    troughs : list or array or DataFrame or Series or dict
        The samples at which the inhalation troughs occur. If a dict or a DataFrame is passed, it is
        assumed that these containers were obtained with :func:`.rsp_findpeaks`.
    desired_length : int
        By default, the returned respiration rate has the same number of elements as ``peaks``. If
        set to an integer, the returned rate will be interpolated between ``peaks`` over
        ``desired_length`` samples. Has no effect if a DataFrame is passed in as the ``peaks``
        argument.

    Returns
    -------
    pd.DataFrame
        A DataFrame of same length as :func:`.rsp_signal` containing the following columns:

        * ``"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.

    See Also
    --------
    rsp_clean, rsp_peaks, rsp_amplitude, rsp_process, rsp_plot

    Examples
    --------
    .. ipython:: python

      import neurokit2 as nk

      rsp = nk.rsp_simulate(duration=30, respiratory_rate=15)
      cleaned = nk.rsp_clean(rsp, sampling_rate=1000)
      peak_signal, info = nk.rsp_peaks(cleaned)

      phase = nk.rsp_phase(peak_signal, desired_length=len(cleaned))
      @savefig p_rsp_phase1.png scale=100%
      fig = nk.signal_plot([rsp, phase], standardize=True)
      @suppress
      plt.close()

    """
    # Format input.
    peaks, troughs = _rsp_fixpeaks_retrieve(peaks, troughs)

    # Phase
    inspiration = np.full(desired_length, np.nan)
    inspiration[peaks] = 0.0
    inspiration[troughs] = 1.0

    last_element = np.where(~np.isnan(inspiration))[0][
        -1
    ]  # Avoid filling beyond the last peak/trough
    inspiration[0:last_element] = pd.Series(inspiration).ffill().values[0:last_element]

    # Phase Completion
    completion = signal_phase(inspiration, method="percent")

    return pd.DataFrame({"RSP_Phase": inspiration, "RSP_Phase_Completion": completion})