Skip to content

synthesize

Classes

Functions

synthesize(signal_length=10000, sample_rate=1000, heart_rate=60, frequency_modulation=0.3, ibi_randomness=0.1)

Generate synthetic PPG signal. Utilize pk.signal.noise methods to make more realistic.

Parameters:

  • signal_length (int, default: 10000 ) –

    Length of signal in samples. Defaults to 10000.

  • sample_rate (float, default: 1000 ) –

    Sample rate in Hz. Defaults to 1000 Hz.

  • heart_rate (float, default: 60 ) –

    Heart rate in BPM. Defaults to 60 BPM.

  • frequency_modulation (float, default: 0.3 ) –

    Frequency modulation strength [0,1]. Defaults to 0.3.

  • ibi_randomness (float, default: 0.1 ) –

    IBI randomness in range [0,1]. Defaults to 0.1.

Returns:

Source code in physiokit/ppg/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
 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
 69
 70
 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
100
def synthesize(
    signal_length: int = 10000,
    sample_rate: float = 1000,
    heart_rate: float = 60,
    frequency_modulation: float = 0.3,
    ibi_randomness: float = 0.1,
) -> tuple[npt.NDArray, npt.NDArray, npt.NDArray]:
    """Generate synthetic PPG signal. Utilize pk.signal.noise methods to make more realistic.

    Args:
        signal_length (int, optional): Length of signal in samples. Defaults to 10000.
        sample_rate (float, optional): Sample rate in Hz. Defaults to 1000 Hz.
        heart_rate (float, optional): Heart rate in BPM. Defaults to 60 BPM.
        frequency_modulation (float, optional): Frequency modulation strength [0,1]. Defaults to 0.3.
        ibi_randomness (float, optional): IBI randomness in range [0,1]. Defaults to 0.1.

    Returns:
        npt.NDArray: Synthetic PPG, segmentation mask, fiducial mask
    """
    duration = signal_length / sample_rate
    period = 60 / heart_rate  # in seconds
    n_period = int(np.rint(duration / period) + 1)
    periods = np.ones(n_period) * period

    # Mark onset of each wave in seconds
    x_onset = np.cumsum(periods)
    x_onset -= x_onset[0]  # make sure seconds start at zero

    # Add respiratory sinus arrythmia (RSA)
    periods, x_onset = _frequency_modulation(
        periods,
        x_onset,
        modulation_frequency=0.05,
        modulation_strength=frequency_modulation,
    )

    # Modulate onset of each wave randomly ~[0, ibi_randomness]
    x_onset = _random_x_offset(x_onset, ibi_randomness)
    y_onset = np.random.normal(0, 0.1, n_period)

    # Create systolic peaks within the waves in seconds
    x_sys = x_onset + np.random.normal(0.175, 0.01, n_period) * periods
    y_sys = y_onset + np.random.normal(1.5, 0.15, n_period)

    # Create dicrotic notches within the waves in seconds
    x_notch = x_onset + np.random.normal(0.4, 0.001, n_period) * periods
    y_notch = y_sys * np.random.normal(0.49, 0.01, n_period)

    # Create diastolic peaks within the waves in seconds
    x_dia = x_onset + np.random.normal(0.45, 0.001, n_period) * periods
    y_dia = y_sys * np.random.normal(0.51, 0.01, n_period)

    # Convert seconds to sample
    x_onset_n = np.ceil(x_onset * sample_rate).astype(int)
    x_sys_n = np.ceil(x_sys * sample_rate).astype(int)
    x_notch_n = np.ceil(x_notch * sample_rate).astype(int)
    x_dia_n = np.ceil(x_dia * sample_rate).astype(int)

    # Concatenate all landmarks and sort them
    x_all = np.concatenate((x_onset_n, x_sys_n, x_notch_n, x_dia_n))
    x_all.sort(kind="mergesort")

    y_all = np.zeros(n_period * 4)
    y_all[0::4] = y_onset
    y_all[1::4] = y_sys
    y_all[2::4] = y_notch
    y_all[3::4] = y_dia

    # Interpolate a continuous signal between the landmarks (i.e., Cartesian coordinates).
    samples = np.arange(int(np.ceil(duration * sample_rate)))

    # Create fiducial mask
    fids = np.zeros(len(samples), dtype=np.int32)
    fids[x_sys_n[x_sys_n < fids.size]] = PpgFiducial.systolic_peak
    fids[x_notch_n[x_notch_n < fids.size]] = PpgFiducial.dicrotic_notch
    fids[x_dia_n[x_dia_n < fids.size]] = PpgFiducial.diastolic_peak

    # Create segmentation mask
    x_sys_seg = np.concatenate((x_onset_n, x_dia_n - 1))
    x_sys_seg.sort(kind="mergesort")
    segs = np.full(len(samples), fill_value=PpgSegment.diastolic, dtype=np.int32)
    for i in range(len(x_sys_seg) // 2):
        segs[x_sys_seg[2 * i] : x_sys_seg[2 * i + 1]] = PpgSegment.systolic

    # Interpolate
    interp_function = scipy.interpolate.Akima1DInterpolator(x_all, y_all)
    ppg = interp_function(samples)

    ppg = ppg[:signal_length]
    segs = segs[:signal_length]
    fids = fids[:signal_length]

    return ppg, segs, fids