from signal_processor import SignalProcessor import numpy as np def _make_frame(r, g, b, motion=0.0, luminance=140.0): return { "rois": { "forehead": {"r": r, "g": g, "b": b}, "left_cheek": {"r": r + 0.5, "g": g + 0.3, "b": b + 0.2}, "right_cheek": {"r": r - 0.3, "g": g - 0.2, "b": b - 0.1}, }, "motion": motion, "luminance": luminance, } def _run_synthetic_session( hr_bpm, drift_bpm=48.0, drift_amp=0.0, pulse_amp=2.0, fps=30, duration=20, include_background=False, ): sp = SignalProcessor(fps=fps) rng = np.random.default_rng(1) result = None for i in range(fps * duration): t = i / fps pulse = np.sin(2 * np.pi * (hr_bpm / 60.0) * t) drift = drift_amp * np.sin(2 * np.pi * (drift_bpm / 60.0) * t + 0.5) r = 140.0 + drift + 0.8 * pulse_amp * pulse + 0.2 * rng.normal() g = ( 142.0 + drift + pulse_amp * np.sin(2 * np.pi * (hr_bpm / 60.0) * t + 0.3) + 0.2 * rng.normal() ) b = ( 138.0 + drift + 0.4 * pulse_amp * np.sin(2 * np.pi * (hr_bpm / 60.0) * t + 0.6) + 0.2 * rng.normal() ) frame = _make_frame(r, g, b, motion=0.1, luminance=(r + g + b) / 3.0) if include_background: frame["background"] = { "r": 100.0 + drift + 0.2 * rng.normal(), "g": 102.0 + drift + 0.2 * rng.normal(), "b": 98.0 + drift + 0.2 * rng.normal(), } sp.last_compute_time = -999 result = sp.process(frame) return result def test_buffering_status_before_enough_samples(): sp = SignalProcessor(fps=30) for i in range(100): result = sp.process(_make_frame(140.0, 142.0, 138.0)) assert result["status"] == "buffering" assert result["bpm"] == 0 def test_bpm_detection_72bpm(sine_wave_72bpm): sp = SignalProcessor(fps=30) result = None for r, g, b in sine_wave_72bpm: sp.last_compute_time = -999 res = sp.process(_make_frame(float(r), float(g), float(b))) result = res assert result is not None assert 65 <= result["bpm"] <= 80 def test_bpm_detection_90bpm(sine_wave_90bpm): sp = SignalProcessor(fps=30) result = None for r, g, b in sine_wave_90bpm: sp.last_compute_time = -999 res = sp.process(_make_frame(float(r), float(g), float(b))) result = res assert result is not None assert 83 <= result["bpm"] <= 100 def test_bpm_detection_resists_common_illumination_drift(): result = _run_synthetic_session(hr_bpm=90.0, drift_amp=4.0) assert result is not None assert result["status"] == "measuring" assert 83 <= result["bpm"] <= 100 def test_bpm_detection_resists_moderate_drift_across_common_heart_rates(): for hr_bpm in (80.0, 90.0, 100.0): result = _run_synthetic_session(hr_bpm=hr_bpm, drift_amp=4.0) assert result is not None assert result["status"] == "measuring" assert hr_bpm - 7 <= result["bpm"] <= hr_bpm + 7 def test_extreme_drift_returns_poor_signal_instead_of_false_low_bpm(): result = _run_synthetic_session(hr_bpm=90.0, drift_amp=8.0) assert result is not None assert result["status"] == "poor_signal" assert result["bpm"] == 0 def test_background_reference_cancels_extreme_common_drift(): result = _run_synthetic_session(hr_bpm=90.0, drift_amp=8.0, include_background=True) assert result is not None assert result["status"] == "measuring" assert 83 <= result["bpm"] <= 100 def test_response_includes_methods_dict(): sp = SignalProcessor(fps=30) fps = 30 t = np.linspace(0, 15, fps * 15) freq = 72.0 / 60.0 for i in range(fps * 15): r = 140.0 + 2.0 * np.sin(2 * np.pi * freq * t[i]) g = 142.0 + 4.0 * np.sin(2 * np.pi * freq * t[i] + 0.3) b = 138.0 + 1.0 * np.sin(2 * np.pi * freq * t[i] + 0.6) sp.last_compute_time = -999 result = sp.process(_make_frame(r, g, b)) assert "methods" in result for method_name in ("green", "chrom", "pos", "gr", "grgb"): assert method_name in result["methods"] assert "bpm" in result["methods"][method_name] assert "snr" in result["methods"][method_name] def test_response_includes_best_method(): sp = SignalProcessor(fps=30) fps = 30 t = np.linspace(0, 15, fps * 15) freq = 72.0 / 60.0 for i in range(fps * 15): r = 140.0 + 2.0 * np.sin(2 * np.pi * freq * t[i]) g = 142.0 + 4.0 * np.sin(2 * np.pi * freq * t[i] + 0.3) b = 138.0 + 1.0 * np.sin(2 * np.pi * freq * t[i] + 0.6) sp.last_compute_time = -999 result = sp.process(_make_frame(r, g, b)) assert "best_method" in result assert result["best_method"] in ("green", "chrom", "pos", "gr", "grgb") def test_response_includes_diagnostics(): result = _run_synthetic_session(hr_bpm=90.0, drift_amp=4.0) assert result is not None assert "diagnostics" in result assert result["diagnostics"]["effective_fps"] > 0 assert result["diagnostics"]["artifact_bpm"] >= 0 assert result["diagnostics"]["artifact_score"] >= 0 assert result["diagnostics"]["usable_methods"] >= 0 def test_response_includes_motion(): sp = SignalProcessor(fps=30) sp.last_compute_time = -999 result = sp.process(_make_frame(140.0, 142.0, 138.0, motion=0.5)) assert "motion" in result def test_high_motion_suppresses_reading(): sp = SignalProcessor(fps=30) fps = 30 t = np.linspace(0, 15, fps * 15) freq = 72.0 / 60.0 for i in range(fps * 15): r = 140.0 + 2.0 * np.sin(2 * np.pi * freq * t[i]) g = 142.0 + 4.0 * np.sin(2 * np.pi * freq * t[i] + 0.3) b = 138.0 + 1.0 * np.sin(2 * np.pi * freq * t[i] + 0.6) motion = 5.0 if i > fps * 10 else 0.1 sp.last_compute_time = -999 result = sp.process(_make_frame(r, g, b, motion=motion)) assert result["confidence"] < 0.3 or result["status"] == "poor_signal" def test_waveform_length_capped(): sp = SignalProcessor(fps=30) fps = 30 t = np.linspace(0, 20, fps * 20) freq = 72.0 / 60.0 for i in range(fps * 20): r = 140.0 + 2.0 * np.sin(2 * np.pi * freq * t[i]) g = 142.0 + 4.0 * np.sin(2 * np.pi * freq * t[i] + 0.3) b = 138.0 + 1.0 * np.sin(2 * np.pi * freq * t[i] + 0.6) sp.last_compute_time = -999 sp.process(_make_frame(r, g, b)) sp.last_compute_time = -999 last = sp.process(_make_frame(r, g, b)) assert len(last["waveform"]) <= 450