Skip to content

Respiratory (RSP)

Respiratory rate is often measured on the chest using a respiration belt or a respiratory inductance plethysmography (RIP) sensor. PhysioKit provides a set of functions to process RSP signals. The functions can be used to generate synthetic RSP signals, clean noisy RSP signals, extract respiratory peaks, compute respiratory rate, and compute dual band metrics.

physiokit.rsp.defines

RspDualMetrics dataclass

Respiratory dual band metrics.

Source code in physiokit/rsp/defines.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass
class RspDualMetrics:
    """Respiratory dual band metrics."""

    rc_rr: float = 0  # RC respiratory rate (BPM)
    ab_rr: float = 0  # AB respiratory rate (BPM)
    vt_rr: float = 0  # VT respiratory rate (BPM)
    phase: float = 0  # Phase angle (degrees)
    lbi: float = 0  # Labored breathing index
    rc_lead: bool = False  # RC leads AB
    rc_percent: float = 0  # Percent RC contribution
    qos: float = 0  # Quality of signal (0-1)

    rc_pk_freq: float = 0  # RC peak frequency (Hz)
    rc_pk_pwr: float = 0  # RC peak power
    ab_pk_freq: float = 0  # AB peak frequency (Hz)
    ab_pk_pwr: float = 0  # AB peak power
    vt_pk_freq: float = 0  # VT peak frequency (Hz)
    vt_pk_pwr: float = 0  # VT peak power

RspFiducial

Bases: IntEnum

RSP fiducials labels

Source code in physiokit/rsp/defines.py 
13
14
15
16
17
class RspFiducial(IntEnum):
    """RSP fiducials labels"""

    inhale_peak = 1
    exhale_trough = 2

RspSegment

Bases: IntEnum

RSP Segment labels

Source code in physiokit/rsp/defines.py 
 5
 6
 7
 8
 9
10
class RspSegment(IntEnum):
    """RSP Segment labels"""

    background = 0
    inhale = 1
    exhale = 2

physiokit.rsp.clean

clean(data, lowcut=0.05, highcut=3, sample_rate=1000, order=3, axis=-1, forward_backward=True)

Clean respiratory signal using biquad filter.

By default, applies a 3rd order Butterworth filter between 0.05 and 3 Hz.

Parameters:

  • data (NDArray) –

    Signal

  • lowcut (float, default: 0.05 ) –

    Lower cutoff in Hz. Defaults to 0.05 Hz (3 bpm).

  • highcut (float, default: 3 ) –

    Upper cutoff in Hz. Defaults to 3 Hz (180 bpm).

  • sample_rate (float, default: 1000 ) –

    Sample rate in Hz. Defaults to 1000 Hz.

  • order (int, default: 3 ) –

    Filter order. Defaults to 3 (3rd order Butterworth filter).

  • axis (int, default: -1 ) –

    Axis to apply against. Defaults to -1.

  • forward_backward (bool, default: True ) –

    Apply filter forward and backward. Defaults to True.

Returns:

  • NDArray –

    npt.NDArray: Cleaned signal

Source code in physiokit/rsp/clean.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def clean(
    data: npt.NDArray,
    lowcut: float = 0.05,
    highcut: float = 3,
    sample_rate: float = 1000,
    order: int = 3,
    axis: int = -1,
    forward_backward: bool = True,
) -> npt.NDArray:
    """Clean respiratory signal using biquad filter.

    By default, applies a 3rd order Butterworth filter between 0.05 and 3 Hz.

    Args:
        data (npt.NDArray): Signal
        lowcut (float, optional): Lower cutoff in Hz. Defaults to 0.05 Hz (3 bpm).
        highcut (float, optional): Upper cutoff in Hz. Defaults to 3 Hz (180 bpm).
        sample_rate (float, optional): Sample rate in Hz. Defaults to 1000 Hz.
        order (int, optional): Filter order. Defaults to 3 (3rd order Butterworth filter).
        axis (int, optional): Axis to apply against. Defaults to -1.
        forward_backward (bool, optional): Apply filter forward and backward. Defaults to True.

    Returns:
        npt.NDArray: Cleaned signal
    """

    # Bandpass filter
    return filter_signal(
        data=data,
        lowcut=lowcut,
        highcut=highcut,
        sample_rate=sample_rate,
        order=order,
        axis=axis,
        forward_backward=forward_backward,
    )

physiokit.rsp.metrics

compute_dual_band_metrics(rc, ab, sample_rate=1000, lowcut=0.05, highcut=3.0, fft_len=None, pwr_threshold=0.8)

Compute respiratory dual band metrics.

Parameters:

  • rc (array) –

    Ribcage band.

  • ab (array) –

    Abdominal band.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • lowcut (float, default: 0.05 ) –

    Lowcut frequency in Hz. Defaults to 0.05 Hz.

  • highcut (float, default: 3.0 ) –

    Highcut frequency in Hz. Defaults to 3.0 Hz.

  • fft_len (int, default: None ) –

    FFT length. Defaults to None.

  • pwr_threshold (float, default: 0.8 ) –

    Power threshold. Defaults to 0.80.

Returns:

  • RspDualMetrics ( RspDualMetrics ) –

    Respiratory dual band metrics.

Source code in physiokit/rsp/metrics.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def compute_dual_band_metrics(
    rc: npt.NDArray,
    ab: npt.NDArray,
    sample_rate: float = 1000,
    lowcut: float = 0.05,
    highcut: float = 3.0,
    fft_len: int | None = None,
    pwr_threshold: float = 0.80,
) -> RspDualMetrics:
    """Compute respiratory dual band metrics.

    Args:
        rc (array): Ribcage band.
        ab (array): Abdominal band.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        lowcut (float, optional): Lowcut frequency in Hz. Defaults to 0.05 Hz.
        highcut (float, optional): Highcut frequency in Hz. Defaults to 3.0 Hz.
        fft_len (int, optional): FFT length. Defaults to None.
        pwr_threshold (float, optional): Power threshold. Defaults to 0.80.

    Returns:
        RspDualMetrics: Respiratory dual band metrics.
    """

    # Remove DC
    rc = rc - rc.mean()
    ab = ab - ab.mean()

    # Compute Vt
    vt = rc + ab

    # Compute FFT
    freqs, rc_sp = compute_fft(rc, sample_rate=sample_rate, fft_len=fft_len, window="blackman")
    freqs, ab_sp = compute_fft(ab, sample_rate=sample_rate, fft_len=fft_len, window="blackman")
    freqs, vt_sp = compute_fft(vt, sample_rate=sample_rate, fft_len=fft_len, window="blackman")

    # Clip to frequency band of interest
    l_idx = np.where(freqs >= lowcut)[0][0]
    r_idx = np.where(freqs >= highcut)[0][0]
    freqs = freqs[l_idx:r_idx]
    rc_sp = rc_sp[l_idx:r_idx]
    ab_sp = ab_sp[l_idx:r_idx]
    vt_sp = vt_sp[l_idx:r_idx]

    # Compute power spectrum
    rc_ps = 2 * np.abs(rc_sp)
    ab_ps = 2 * np.abs(ab_sp)
    vt_ps = 2 * np.abs(vt_sp)

    # Compute Vtc (corrected Vt)
    vtc_ps = rc_ps + ab_ps

    # Find dominant frequency
    rc_pk_idx = np.argmax(rc_ps)
    ab_pk_idx = np.argmax(ab_ps)
    # vt_pk_idx = np.argmax(vt_ps)
    vtc_pk_idx = np.argmax(vtc_ps)

    # Find all peaks above threshold
    rc_pk_idxs = np.where(rc_ps >= pwr_threshold * rc_ps[rc_pk_idx])[0]
    rc_pk_vals = rc_ps[rc_pk_idxs]
    ab_pk_idxs = np.where(ab_ps >= pwr_threshold * ab_ps[ab_pk_idx])[0]
    ab_pk_vals = ab_ps[ab_pk_idxs]

    vtc_pk_idxs = np.where(vtc_ps >= pwr_threshold * vtc_ps[vtc_pk_idx])[0]
    vtc_pk_vals = vtc_ps[vtc_pk_idxs]

    # Compute respiratory rates
    rc_rr = 60 * np.sum(freqs[rc_pk_idxs] * rc_pk_vals) / np.sum(rc_pk_vals)
    ab_rr = 60 * np.sum(freqs[ab_pk_idxs] * ab_pk_vals) / np.sum(ab_pk_vals)
    vtc_rr = 60 * np.sum(freqs[vtc_pk_idxs] * vtc_pk_vals) / np.sum(vtc_pk_vals)

    # Compute phase angle
    vtc_dom_div = rc_sp[vtc_pk_idxs] / ab_sp[vtc_pk_idxs]
    vtc_dom_angle = np.arctan2(np.imag(vtc_dom_div), np.real(vtc_dom_div))
    angles = (180 / np.pi) * np.sum(vtc_dom_angle * vtc_pk_vals) / np.sum(vtc_pk_vals)
    phase = np.abs(angles)
    rc_lead = angles > 0

    # Compute LBI: 𝚫 Vtc / 𝚫 Vt
    lbi = np.clip(np.sum(vtc_ps[vtc_pk_idxs]) / np.sum(vt_ps[vtc_pk_idxs]), 1, 10)

    # Compute %RC
    rc_percent = 100 * np.mean(rc_ps[vtc_pk_idxs] / (rc_ps[vtc_pk_idxs] + ab_ps[vtc_pk_idxs]))

    # Compute QoS
    qos = np.sum(rc_ps * ab_ps) / (np.sum(rc_ps) + np.sum(ab_ps))

    return RspDualMetrics(
        rc_rr=rc_rr,
        ab_rr=ab_rr,
        vt_rr=vtc_rr,
        phase=phase,
        lbi=lbi,
        rc_lead=rc_lead,
        rc_percent=rc_percent,
        qos=qos,
        rc_pk_freq=freqs[rc_pk_idx],
        rc_pk_pwr=rc_ps[rc_pk_idx],
        ab_pk_freq=freqs[ab_pk_idx],
        ab_pk_pwr=ab_ps[ab_pk_idx],
        vt_pk_freq=freqs[vtc_pk_idx],
        vt_pk_pwr=vtc_ps[vtc_pk_idx],
    )

compute_respiratory_rate(data, sample_rate=1000, method='fft', **kwargs)

Compute respiratory rate in BPM from signal.

Parameters:

  • data (array) –

    RSP signal.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • method (str, default: 'fft' ) –

    Method to compute respiratory rate. Defaults to 'fft'.

  • **kwargs (dict, default: {} ) –

    Keyword arguments to pass to method.

Returns:

  • tuple[float, float] –

    tuple[float, float]: Respiratory rate (BPM) and qos metric.

Source code in physiokit/rsp/metrics.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def compute_respiratory_rate(
    data: npt.NDArray, sample_rate: float = 1000, method: str = "fft", **kwargs: dict
) -> tuple[float, float]:
    """Compute respiratory rate in BPM from signal.

    Args:
        data (array): RSP signal.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        method (str, optional): Method to compute respiratory rate. Defaults to 'fft'.
        **kwargs (dict): Keyword arguments to pass to method.

    Returns:
        tuple[float, float]: Respiratory rate (BPM) and qos metric.
    """
    match method:
        case "fft":
            bpm, qos = compute_respiratory_rate_from_fft(data=data, sample_rate=sample_rate, **kwargs)
        case "peak":
            bpm, qos = compute_respiratory_rate_from_peaks(data=data, sample_rate=sample_rate, **kwargs)
        case _:
            raise NotImplementedError(f"Respiratory rate computation method {method} not implemented.")
    # END MATCH
    return bpm, qos

compute_respiratory_rate_from_fft(data, sample_rate=1000, lowcut=0.05, highcut=3.0)

Compute respiratory rate in BPM from FFT of respiratory signal.

Parameters:

  • data (array) –

    RSP signal.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • lowcut (float, default: 0.05 ) –

    Lowcut frequency in Hz. Defaults to 0.05 Hz.

  • highcut (float, default: 3.0 ) –

    Highcut frequency in Hz. Defaults to 3.0 Hz.

Returns:

Source code in physiokit/rsp/metrics.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def compute_respiratory_rate_from_fft(
    data: npt.NDArray,
    sample_rate: float = 1000,
    lowcut: float = 0.05,
    highcut: float = 3.0,
) -> tuple[float, float]:
    """Compute respiratory rate in BPM from FFT of respiratory signal.

    Args:
        data (array): RSP signal.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        lowcut (float, optional): Lowcut frequency in Hz. Defaults to 0.05 Hz.
        highcut (float, optional): Highcut frequency in Hz. Defaults to 3.0 Hz.

    Returns:
        float: Respiratory rate (BPM).
    """
    freqs, sp = compute_fft(data, sample_rate=sample_rate, window="blackman")
    l_idx = np.where(freqs >= lowcut)[0][0]
    r_idx = np.where(freqs >= highcut)[0][0]
    freqs = freqs[l_idx:r_idx]
    ps = 2 * np.abs(sp[l_idx:r_idx])
    fft_pk_idx = np.argmax(ps)
    bpm = freqs[fft_pk_idx] * 60
    qos = ps[fft_pk_idx] / np.sum(ps)
    return bpm, qos

compute_respiratory_rate_from_peaks(data, sample_rate=1000, min_rr=0.5, max_rr=20, min_delta=0.5)

Compute respiratory rate in BPM from peaks of PPG signal.

Parameters:

  • data (array) –

    RSP signal.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • min_rr (float, default: 0.5 ) –

    Minimum RR interval in seconds. Defaults to 0.5 s.

  • max_rr (float, default: 20 ) –

    Maximum RR interval in seconds. Defaults to 20 s.

  • min_delta (float, default: 0.5 ) –

    Minimum delta between RR intervals in seconds. Defaults to 0.5 s.

Returns:

Source code in physiokit/rsp/metrics.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def compute_respiratory_rate_from_peaks(
    data: npt.NDArray, sample_rate: float = 1000, min_rr: float = 0.5, max_rr: float = 20, min_delta: float = 0.5
) -> tuple[float, float]:
    """Compute respiratory rate in BPM from peaks of PPG signal.

    Args:
        data (array): RSP signal.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        min_rr (float, optional): Minimum RR interval in seconds. Defaults to 0.5 s.
        max_rr (float, optional): Maximum RR interval in seconds. Defaults to 20 s.
        min_delta (float, optional): Minimum delta between RR intervals in seconds. Defaults to 0.5 s.

    Returns:
        float: Respiratory rate (BPM).
    """
    peaks = find_peaks(data=data, sample_rate=sample_rate)
    rri = compute_rr_intervals(peaks=peaks)
    rmask = filter_rr_intervals(rr_ints=rri, sample_rate=sample_rate, min_rr=min_rr, max_rr=max_rr, min_delta=min_delta)

    bpm = 60 / (np.nanmean(rri[rmask == 0]) / sample_rate)
    qos = rmask[rmask == 0].size / rmask.size
    return bpm, qos

physiokit.rsp.peaks

compute_rr_intervals(peaks)

Compute RR intervals from resp peaks.

Parameters:

  • peaks (array) –

    R peaks.

Returns:

  • NDArray –

    npt.NDArray: RR intervals.

Source code in physiokit/rsp/peaks.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def compute_rr_intervals(
    peaks: npt.NDArray,
) -> npt.NDArray:
    """Compute RR intervals from resp peaks.

    Args:
        peaks (array): R peaks.

    Returns:
        npt.NDArray: RR intervals.
    """

    rr_ints = np.diff(peaks)
    if rr_ints.size == 0:
        return rr_ints
    rr_ints = np.hstack((rr_ints[0], rr_ints))
    return rr_ints

filter_peaks(peaks, sample_rate=1000, min_rr=0.5, max_rr=20, min_delta=0.5)

Filter out peaks with RR intervals outside of normal range.

Parameters:

  • peaks (array) –

    Respiratory peaks.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • min_rr (float, default: 0.5 ) –

    Minimum RR interval in seconds. Defaults to 0.5 s.

  • max_rr (float, default: 20 ) –

    Maximum RR interval in seconds. Defaults to 20 s.

  • min_delta (float, default: 0.5 ) –

    Minimum RR interval delta. Defaults to 0.5.

Returns:

  • NDArray –

    npt.NDArray: Filtered peaks.

Source code in physiokit/rsp/peaks.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def filter_peaks(
    peaks: npt.NDArray, sample_rate: float = 1000, min_rr: float = 0.5, max_rr: float = 20, min_delta: float = 0.5
) -> npt.NDArray:
    """Filter out peaks with RR intervals outside of normal range.

    Args:
        peaks (array): Respiratory peaks.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        min_rr (float, optional): Minimum RR interval in seconds. Defaults to 0.5 s.
        max_rr (float, optional): Maximum RR interval in seconds. Defaults to 20 s.
        min_delta (float, optional): Minimum RR interval delta. Defaults to 0.5.

    Returns:
        npt.NDArray: Filtered peaks.
    """
    if peaks.size <= 1:
        return peaks
    # Capture RR intervals
    rr_ints = np.diff(peaks)
    rr_ints = np.hstack((rr_ints[0], rr_ints))

    # Filter out peaks with RR intervals outside of normal range
    rr_mask = np.where((rr_ints < min_rr * sample_rate) | (rr_ints > max_rr * sample_rate), 1, 0)

    # Filter out peaks that deviate more than delta
    if min_delta is not None:
        rr_mask = quotient_filter_mask(rr_ints, mask=rr_mask, lowcut=1 - min_delta, highcut=1 + min_delta)
    filt_peaks = peaks[np.where(rr_mask == 0)[0]]
    return filt_peaks

filter_rr_intervals(rr_ints, sample_rate=1000, min_rr=0.5, max_rr=20, min_delta=0.5)

Filter out peaks with RR intervals outside of normal range.

Parameters:

  • rr_ints (array) –

    RR intervals.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • min_rr (float, default: 0.5 ) –

    Minimum RR interval in seconds. Defaults to 0.5 s.

  • max_rr (float, default: 20 ) –

    Maximum RR interval in seconds. Defaults to 20 s.

  • min_delta (float, default: 0.5 ) –

    Minimum RR interval delta. Defaults to 0.5.

Returns:

  • NDArray –

    npt.NDArray: Filtered RR intervals.

Source code in physiokit/rsp/peaks.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def filter_rr_intervals(
    rr_ints: npt.NDArray, sample_rate: float = 1000, min_rr: float = 0.5, max_rr: float = 20, min_delta: float = 0.5
) -> npt.NDArray:
    """Filter out peaks with RR intervals outside of normal range.

    Args:
        rr_ints (array): RR intervals.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        min_rr (float, optional): Minimum RR interval in seconds. Defaults to 0.5 s.
        max_rr (float, optional): Maximum RR interval in seconds. Defaults to 20 s.
        min_delta (float, optional): Minimum RR interval delta. Defaults to 0.5.

    Returns:
        npt.NDArray: Filtered RR intervals.
    """
    if rr_ints.size == 0:
        return np.array([])

    # Filter out peaks with RR intervals outside of normal range
    rr_mask = np.where((rr_ints < min_rr * sample_rate) | (rr_ints > max_rr * sample_rate), 1, 0)

    # Filter out peaks that deviate more than delta
    rr_mask = quotient_filter_mask(rr_ints, mask=rr_mask, lowcut=1 - min_delta, highcut=1 + min_delta)

    return rr_mask

find_peaks(data, sample_rate=1000, peak_window=0.5, breath_window=2.0, breath_offset=0.05, peak_delay=0.3)

Find peaks in RSP signal.

Assumes input data is bandpass filtered with a lowcut of .05 Hz and a highcut of 3 Hz.

Parameters:

  • data (array) –

    RSP signal.

  • sample_rate (float, default: 1000 ) –

    Sampling rate in Hz. Defaults to 1000 Hz.

  • peak_window (float, default: 0.5 ) –

    Peak window in seconds. Defaults to 0.5 s.

  • breath_window (float, default: 2.0 ) –

    Breath window in seconds. Defaults to 2.0 s.

  • breath_offset (float, default: 0.05 ) –

    Breath offset in seconds. Defaults to 0.05 s.

  • peak_delay (float, default: 0.3 ) –

    Peak delay in seconds. Defaults to 0.3 s.

Returns:

  • NDArray –

    npt.NDArray: Peak locations.

Source code in physiokit/rsp/peaks.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def find_peaks(
    data: npt.NDArray,
    sample_rate: float = 1000,
    peak_window: float = 0.5,
    breath_window: float = 2.0,
    breath_offset: float = 0.05,
    peak_delay: float = 0.3,
) -> npt.NDArray:
    """Find peaks in RSP signal.

    Assumes input data is bandpass filtered with a lowcut of .05 Hz and a highcut of 3 Hz.

    Args:
        data (array): RSP signal.
        sample_rate (float, optional): Sampling rate in Hz. Defaults to 1000 Hz.
        peak_window (float, optional): Peak window in seconds. Defaults to 0.5 s.
        breath_window (float, optional): Breath window in seconds. Defaults to 2.0 s.
        breath_offset (float, optional): Breath offset in seconds. Defaults to 0.05 s.
        peak_delay (float, optional): Peak delay in seconds. Defaults to 0.3 s.

    Returns:
        npt.NDArray: Peak locations.
    """

    # Clip negative values and square the signal
    sqrd = np.where(data > 0, data**2, 0)

    # Apply 1st moving average filter
    ma_peak_kernel = int(np.rint(peak_window * sample_rate))
    ma_peak = spn.uniform_filter1d(sqrd, ma_peak_kernel, mode="nearest")

    # Apply 2nd moving average filter
    ma_breath_kernel = int(np.rint(breath_window * sample_rate))
    ma_breath = spn.uniform_filter1d(sqrd, ma_breath_kernel, mode="nearest")

    # Thresholds
    min_height = ma_breath + breath_offset * np.mean(sqrd)
    min_width = int(np.rint(peak_window * sample_rate))
    min_delay = int(np.rint(peak_delay * sample_rate))

    # Identify wave boundaries
    waves = ma_peak > min_height
    beg_waves = np.where(np.logical_and(np.logical_not(waves[0:-1]), waves[1:]))[0]
    end_waves = np.where(np.logical_and(waves[0:-1], np.logical_not(waves[1:])))[0]
    end_waves = end_waves[end_waves > beg_waves[0]]

    # Identify peaks
    peaks = []
    for i in range(min(beg_waves.size, end_waves.size)):
        beg, end = beg_waves[i], end_waves[i]
        peak = beg + np.argmax(data[beg:end])
        peak_width = end - beg
        peak_delay = peak - peaks[-1] if peaks else min_delay

        # Enforce minimum length and delay between peaks
        if (peak_width < min_width) or (peak_delay < min_delay):
            continue
        peaks.append(peak)
    # END FOR

    return np.array(peaks, dtype=int)

physiokit.rsp.synthesize

synthesize(signal_length=10000, sample_rate=1000, respiratory_rate=15)

Generate synthetic respiratory signal using breathmetrics method.

Utilize pk.signal.noise methods to make more realistic.

Parameters:

  • signal_length (int, default: 10000 ) –

    Signal length in samples. Defaults to 10000.

  • sample_rate (float, default: 1000 ) –

    Sample rate in Hz. Defaults to 1000 Hz.

  • respiratory_rate (float, default: 15 ) –

    Respiratory rate in breaths per minute. Defaults to 15 bpm.

Returns:

  • tuple[NDArray, NDArray, NDArray] –

    tuple[npt.NDArray, npt.NDArray, npt.NDArray]: Synthetic respiratory signal, segmentation mask, fiducial mask

Source code in physiokit/rsp/synthesize.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def synthesize(
    signal_length: int = 10000,
    sample_rate: float = 1000,
    respiratory_rate: float = 15,
) -> tuple[npt.NDArray, npt.NDArray, npt.NDArray]:
    """Generate synthetic respiratory signal using breathmetrics method.

    Utilize pk.signal.noise methods to make more realistic.

    Args:
        signal_length (int, optional): Signal length in samples. Defaults to 10000.
        sample_rate (float, optional): Sample rate in Hz. Defaults to 1000 Hz.
        respiratory_rate (float, optional): Respiratory rate in breaths per minute. Defaults to 15 bpm.

    Returns:
        tuple[npt.NDArray, npt.NDArray, npt.NDArray]: Synthetic respiratory signal, segmentation mask, fiducial mask
    """

    rsp, segs, fids = _simulate_breathmetrics_core(
        signal_length=signal_length,
        sample_rate=sample_rate,
        breathing_rate=respiratory_rate / 60,
        signal_noise=0,
    )
    return rsp, segs, fids