Skip to content

Heart rate estimation

Heart rate calculation utilities.

HrCalculator

Base class for heart rate calculation.

Source code in src\yarppg\hr_calculator.py 
 9
10
11
12
13
14
15
16
17
18
class HrCalculator:
    """Base class for heart rate calculation."""

    def update(self, value: float) -> float:  # noqa: ARG002
        """Process the new data and update HR estimate."""
        return np.nan

    def reset(self) -> None:
        """Clear the the internal state."""
        pass

reset()

Clear the the internal state.

Source code in src\yarppg\hr_calculator.py 
16
17
18
def reset(self) -> None:
    """Clear the the internal state."""
    pass

update(value)

Process the new data and update HR estimate.

Source code in src\yarppg\hr_calculator.py 
12
13
14
def update(self, value: float) -> float:  # noqa: ARG002
    """Process the new data and update HR estimate."""
    return np.nan

PeakBasedHrCalculator

Bases: HrCalculator

Peak-based heart rate calculation.

Source code in src\yarppg\hr_calculator.py 
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
class PeakBasedHrCalculator(HrCalculator):
    """Peak-based heart rate calculation."""

    def __init__(
        self,
        fs: float,
        window_seconds: float = 10,
        distance: float = 0.5,
        update_interval: int = 10,
    ):
        self.winsize = int(fs * window_seconds)
        self.values = deque(maxlen=self.winsize)
        self.mindist = int(fs * distance)

        self.update_interval = update_interval
        self.frames_seen = 0
        self.last_hr = np.nan

    def update(self, value: float) -> float:
        """Process the new data and update HR estimate in frames per beat."""
        self.frames_seen += 1
        self.values.append(value)
        if (
            len(self.values) < self.winsize
            or self.frames_seen % self.update_interval != 0
        ):
            return self.last_hr
        peaks, _ = scipy.signal.find_peaks(self.values, distance=self.mindist)
        self.last_hr = np.diff(peaks).mean()
        return self.last_hr

    def reset(self) -> None:
        """Clear the internal buffer and intermediate values."""
        self.frames_seen = 0
        self.values.clear()
        self.last_hr = np.nan

reset()

Clear the internal buffer and intermediate values.

Source code in src\yarppg\hr_calculator.py 
52
53
54
55
56
def reset(self) -> None:
    """Clear the internal buffer and intermediate values."""
    self.frames_seen = 0
    self.values.clear()
    self.last_hr = np.nan

update(value)

Process the new data and update HR estimate in frames per beat.

Source code in src\yarppg\hr_calculator.py 
39
40
41
42
43
44
45
46
47
48
49
50
def update(self, value: float) -> float:
    """Process the new data and update HR estimate in frames per beat."""
    self.frames_seen += 1
    self.values.append(value)
    if (
        len(self.values) < self.winsize
        or self.frames_seen % self.update_interval != 0
    ):
        return self.last_hr
    peaks, _ = scipy.signal.find_peaks(self.values, distance=self.mindist)
    self.last_hr = np.diff(peaks).mean()
    return self.last_hr