Skip to content

Basic processor

Provides base classes for rPPG signal computation.

Processor

Base rPPG processor, extracting the average green channel from the ROI.

Source code in src\yarppg\processors\processor.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class Processor:
    """Base rPPG processor, extracting the average green channel from the ROI."""

    def process(self, roi: RegionOfInterest) -> RppgResult:
        """Calculate average green channel in the roi area."""
        avg = masked_average(roi.baseimg, roi.mask)
        bg_mean = Color.null()
        if roi.bg_mask is not None:
            bg_mean = masked_average(roi.baseimg, roi.bg_mask)

        return RppgResult(avg.g, roi, roi_mean=avg, bg_mean=bg_mean)

    def reset(self) -> None:
        """Reset internal state and intermediate values."""
        pass  # no persistent values in base class

process(roi)

Calculate average green channel in the roi area.

Source code in src\yarppg\processors\processor.py 
13
14
15
16
17
18
19
20
def process(self, roi: RegionOfInterest) -> RppgResult:
    """Calculate average green channel in the roi area."""
    avg = masked_average(roi.baseimg, roi.mask)
    bg_mean = Color.null()
    if roi.bg_mask is not None:
        bg_mean = masked_average(roi.baseimg, roi.bg_mask)

    return RppgResult(avg.g, roi, roi_mean=avg, bg_mean=bg_mean)

reset()

Reset internal state and intermediate values.

Source code in src\yarppg\processors\processor.py 
22
23
24
def reset(self) -> None:
    """Reset internal state and intermediate values."""
    pass  # no persistent values in base class

FilteredProcessor

Bases: Processor

Processor with temporal filtering of the extracted signal.

Source code in src\yarppg\processors\processor.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class FilteredProcessor(Processor):
    """Processor with temporal filtering of the extracted signal."""

    def __init__(self, processor: Processor, livefilter: DigitalFilter | None = None):
        self.processor = processor
        self.livefilter = livefilter

    def process(self, roi: RegionOfInterest) -> RppgResult:
        """Calculate processor output and apply digital filter."""
        result = self.processor.process(roi)
        if self.livefilter is not None and np.isfinite(result.value):
            # only calculate filter update if not NaN
            result.value = self.livefilter.process(result.value)
        return result

    def reset(self) -> None:
        """Reset internal state and intermediate values."""
        self.processor.reset()
        if self.livefilter is not None:
            self.livefilter.reset()

process(roi)

Calculate processor output and apply digital filter.

Source code in src\yarppg\processors\processor.py 
34
35
36
37
38
39
40
def process(self, roi: RegionOfInterest) -> RppgResult:
    """Calculate processor output and apply digital filter."""
    result = self.processor.process(roi)
    if self.livefilter is not None and np.isfinite(result.value):
        # only calculate filter update if not NaN
        result.value = self.livefilter.process(result.value)
    return result

reset()

Reset internal state and intermediate values.

Source code in src\yarppg\processors\processor.py 
42
43
44
45
46
def reset(self) -> None:
    """Reset internal state and intermediate values."""
    self.processor.reset()
    if self.livefilter is not None:
        self.livefilter.reset()