Impulse Response Fatigue Model, EMA Smoothing, and Acoustic Feature Extraction

.github/workflows/ci.yml

@@ -0,0 +1,42 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y libportaudio2 ffmpeg
+
+      - name: Install project dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e .
+          pip install praat-parselmouth
+          pip install pytest pytest-cov
+
+      - name: Run tests with coverage
+        run: |
+          pytest --cov=voiceMonitor --cov-report=term-missing --cov-report=xml
+
+      - name: Upload coverage report
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-report
+          path: coverage.xml

requirements.txt

@@ -3,4 +3,5 @@ soundfile>=0.12
 pydub>=0.25
 ffmpeg
 tqdm>=4.65
-sounddevice>=0.4
+sounddevice>=0.4
+praat-parselmouth

tests/test_acoustic_features.py

@@ -0,0 +1,21 @@
+import logging
+
+from voiceMonitor.audio_stream import extract_acoustic_features
+
+
+def test_extraction_returns_empty_dict_on_missing_file(caplog):
+    """
+    A nonexistent/invalid wav path should trigger the failure path and
+    return an empty dict rather than raising, and the failure should be
+    logged as a warning rather than silently swallowed.
+    """
+    with caplog.at_level(logging.WARNING):
+        result = extract_acoustic_features("/tmp/this_file_does_not_exist.wav")
+
+    assert result == {}
+    assert any("Acoustic feature extraction failed" in message for message in caplog.messages)
+
+
+def test_extraction_returns_dict_type_even_on_failure():
+    result = extract_acoustic_features("/tmp/another_missing_file.wav")
+    assert isinstance(result, dict)

tests/test_analytics_impulse.py

@@ -0,0 +1,93 @@
+import math
+
+from voiceMonitor.analytics import EMAFatigue, ImpulseResponseFatigue, SessionAnalytics
+
+
+def test_ema_first_value_passthrough():
+    ema = EMAFatigue(alpha=0.3)
+    assert ema.update(50) == 50
+
+
+def test_ema_smooths_toward_new_value():
+    ema = EMAFatigue(alpha=0.5)
+    ema.update(0)
+    result = ema.update(100)
+    assert result == 50  # 0.5 * 100 + 0.5 * 0
+
+
+def test_ema_default_alpha_from_config():
+    ema = EMAFatigue()
+    assert ema.alpha > 0 and ema.alpha <= 1
+
+
+def test_impulse_accumulates_with_no_elapsed_time():
+    model = ImpulseResponseFatigue(tau_fast=90, tau_slow=1500)
+    state1 = model.update(20, elapsed_seconds=0)
+    state2 = model.update(20, elapsed_seconds=0)
+    assert state2["acute"] == 40
+    assert state2["chronic"] == 40
+
+
+def test_impulse_acute_decays_faster_than_chronic():
+    model = ImpulseResponseFatigue(tau_fast=10, tau_slow=1000)
+    model.update(100, elapsed_seconds=0)
+    # advance a long time relative to tau_fast but short relative to tau_slow
+    state = model.update(0, elapsed_seconds=50)
+    assert state["acute"] < state["chronic"]
+
+
+def test_impulse_readiness_is_chronic_minus_acute():
+    model = ImpulseResponseFatigue(tau_fast=10, tau_slow=1000)
+    model.update(100, elapsed_seconds=0)
+    state = model.update(0, elapsed_seconds=50)
+    assert math.isclose(
+        state["readiness_experimental"], state["chronic"] - state["acute"]
+    )
+
+
+def test_recovery_eta_zero_when_already_safe():
+    model = ImpulseResponseFatigue(tau_fast=90, tau_slow=1500)
+    model.update(10, elapsed_seconds=0)
+    eta = model.recovery_eta_seconds(safe_level=40)
+    assert eta == 0.0
+
+
+def test_recovery_eta_positive_when_above_safe_level():
+    model = ImpulseResponseFatigue(tau_fast=90, tau_slow=1500)
+    model.update(100, elapsed_seconds=0)
+    eta = model.recovery_eta_seconds(safe_level=40)
+    assert eta > 0
+
+
+def test_recovery_eta_matches_manual_formula():
+    model = ImpulseResponseFatigue(tau_fast=90, tau_slow=1500)
+    model.update(200, elapsed_seconds=0)
+    expected = 90 * math.log(200 / 40)
+    assert math.isclose(model.recovery_eta_seconds(safe_level=40), expected)
+
+
+def test_session_analytics_summary_shape():
+    sa = SessionAnalytics()
+    sa.add(20, "ts0", 0)
+    sa.add(80, "ts1", 5)
+    summary = sa.summary
+    for key in [
+        "average_fatigue", "max_fatigue", "readings", "smoothed_fatigue",
+        "acute_load", "chronic_load", "readiness_experimental", "recovery_eta_sec",
+    ]:
+        assert key in summary
+
+
+def test_session_analytics_average_and_max():
+    sa = SessionAnalytics()
+    sa.add(20, "ts0", 0)
+    sa.add(80, "ts1", 5)
+    assert sa.average == 50
+    assert sa.maximum == 80
+
+
+def test_session_analytics_empty_defaults():
+    sa = SessionAnalytics()
+    assert sa.average == 0
+    assert sa.maximum == 0
+    assert sa.summary["readings"] == 0

tests/test_audio_stream.py

@@ -0,0 +1,89 @@
+import os
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pytest
+
+from voiceMonitor.audio_stream import VoiceMonitor
+from voiceMonitor.config import Config
+
+
+def test_voice_monitor_init_creates_chunk_dir(tmp_path):
+    chunk_dir = tmp_path / "chunks_test"
+    vm = VoiceMonitor(chunk_dir=str(chunk_dir))
+    assert chunk_dir.exists()
+    assert vm.threshold == Config.DEFAULT_THRESHOLD
+
+
+def test_voice_monitor_init_custom_threshold(tmp_path):
+    vm = VoiceMonitor(threshold=55, chunk_dir=str(tmp_path / "chunks_test2"))
+    assert vm.threshold == 55
+
+
+def test_save_chunk_writes_wav_file(tmp_path):
+    vm = VoiceMonitor(chunk_dir=str(tmp_path / "chunks_test3"))
+    audio = np.zeros((Config.SAMPLE_RATE,), dtype="float32")
+    path = vm._save_chunk(audio, "20260101_000000")
+    assert os.path.exists(path)
+    assert path.endswith(".wav")
+
+
+@patch("voiceMonitor.audio_stream.extract_acoustic_features")
+@patch("voiceMonitor.audio_stream.score_audio")
+@patch("voiceMonitor.audio_stream.preprocess_audio")
+def test_process_chunk_calls_scoring_and_records_result(
+    mock_preprocess, mock_score, mock_features, tmp_path
+):
+    mock_preprocess.return_value = ["processed_chunk.wav"]
+    mock_score.return_value = 66.6
+    mock_features.return_value = {"jitter_local": 0.01}
+
+    vm = VoiceMonitor(chunk_dir=str(tmp_path / "chunks_test4"))
+    audio = np.zeros((Config.SAMPLE_RATE,), dtype="float32")
+    raw_file = vm._save_chunk(audio, "20260101_000000")
+
+    score = vm._process_chunk(raw_file, "20260101_000000", elapsed_seconds=0)
+
+    assert score == 66.6
+    assert len(vm.session.records) == 1
+    assert vm.session.records[0]["score"] == 66.6
+    assert vm.session.records[0]["features"] == {"jitter_local": 0.01}
+
+
+@patch("voiceMonitor.audio_stream.preprocess_audio")
+def test_process_chunk_returns_none_when_preprocessing_yields_nothing(
+    mock_preprocess, tmp_path
+):
+    mock_preprocess.return_value = []
+    vm = VoiceMonitor(chunk_dir=str(tmp_path / "chunks_test5"))
+    audio = np.zeros((Config.SAMPLE_RATE,), dtype="float32")
+    raw_file = vm._save_chunk(audio, "20260101_000000")
+
+    result = vm._process_chunk(raw_file, "20260101_000000", elapsed_seconds=0)
+    assert result is None
+    assert len(vm.session.records) == 0
+
+
+@patch("voiceMonitor.audio_stream.extract_acoustic_features", return_value={})
+@patch("voiceMonitor.audio_stream.score_audio", return_value=30.0)
+@patch("voiceMonitor.audio_stream.preprocess_audio", return_value=["processed.wav"])
+@patch("voiceMonitor.audio_stream.sd.InputStream")
+def test_start_runs_for_fixed_duration_and_returns_session(
+    mock_input_stream, mock_preprocess, mock_score, mock_features, tmp_path, monkeypatch
+):
+    # use small, fast values so the test does not depend on real audio timing
+    monkeypatch.setattr(Config, "SAMPLE_RATE", 10)
+    monkeypatch.setattr(Config, "WINDOW_SEC", 1)
+    monkeypatch.setattr(Config, "STEP_SEC", 1)
+
+    fake_stream = MagicMock()
+    fake_stream.read.return_value = (np.zeros((10, 1), dtype="float32"), False)
+    mock_input_stream.return_value.__enter__.return_value = fake_stream
+
+    vm = VoiceMonitor(chunk_dir=str(tmp_path / "chunks_test6"))
+    session = vm.start(duration_sec=2)
+
+    assert session is vm.session
+    # two one-second steps should have produced two processed windows
+    assert len(session.records) == 2
+    assert mock_score.call_count == 2

tests/test_cli.py

@@ -0,0 +1,35 @@
+from unittest.mock import MagicMock, patch
+
+from voiceMonitor import cli
+
+
+@patch("voiceMonitor.cli.VoiceMonitor")
+def test_main_parses_args_and_runs_session(mock_voice_monitor_cls, monkeypatch):
+    mock_instance = MagicMock()
+    mock_report = MagicMock()
+    mock_instance.start.return_value = mock_report
+    mock_voice_monitor_cls.return_value = mock_instance
+
+    monkeypatch.setattr(
+        "sys.argv", ["voicemonitor", "--duration", "30", "--threshold", "65"]
+    )
+
+    cli.main()
+
+    mock_voice_monitor_cls.assert_called_once_with(threshold=65.0)
+    mock_instance.start.assert_called_once_with(duration_sec=30)
+    mock_report.export_json.assert_called_once_with("session_report.json")
+
+
+@patch("voiceMonitor.cli.VoiceMonitor")
+def test_main_uses_defaults_when_no_args_given(mock_voice_monitor_cls, monkeypatch):
+    mock_instance = MagicMock()
+    mock_instance.start.return_value = MagicMock()
+    mock_voice_monitor_cls.return_value = mock_instance
+
+    monkeypatch.setattr("sys.argv", ["voicemonitor"])
+
+    cli.main()
+
+    mock_voice_monitor_cls.assert_called_once_with(threshold=None)
+    mock_instance.start.assert_called_once_with(duration_sec=None)

tests/test_session_report.py

@@ -0,0 +1,62 @@
+import json
+import os
+import tempfile
+
+from voiceMonitor.session import SessionReport
+
+
+def test_add_record_stores_fields():
+    report = SessionReport()
+    report.add_record("ts1", "chunks/ts1.wav", 42.5, elapsed_seconds=0)
+
+    assert len(report.records) == 1
+    record = report.records[0]
+    assert record["timestamp"] == "ts1"
+    assert record["chunk"] == "chunks/ts1.wav"
+    assert record["score"] == 42.5
+    assert record["features"] == {}
+
+
+def test_add_record_stores_features_when_given():
+    report = SessionReport()
+    features = {"jitter_local": 0.01, "shimmer_local": 0.02, "hnr": 15.0, "cpps": 8.0}
+    report.add_record("ts1", "chunks/ts1.wav", 42.5, elapsed_seconds=0, features=features)
+
+    assert report.records[0]["features"] == features
+
+
+def test_add_record_defaults_elapsed_seconds_when_omitted():
+    # backward compatible call, mirrors pre-existing callers
+    report = SessionReport()
+    report.add_record("ts1", "chunks/ts1.wav", 10)
+    report.add_record("ts2", "chunks/ts2.wav", 20)
+    assert report.analytics.average == 15
+    assert report.analytics.maximum == 20
+
+
+def test_add_record_updates_analytics():
+    report = SessionReport()
+    report.add_record("ts1", "chunks/ts1.wav", 10, elapsed_seconds=0)
+    report.add_record("ts2", "chunks/ts2.wav", 30, elapsed_seconds=5)
+
+    assert report.analytics.average == 20
+    assert report.analytics.maximum == 30
+    assert report.analytics.summary["readings"] == 2
+
+
+def test_export_json_writes_expected_structure():
+    report = SessionReport()
+    report.add_record("ts1", "chunks/ts1.wav", 50, elapsed_seconds=0)
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        out_path = os.path.join(tmp_dir, "session_report.json")
+        report.export_json(out_path)
+
+        assert os.path.exists(out_path)
+        with open(out_path) as fp:
+            data = json.load(fp)
+
+        assert "summary" in data
+        assert "records" in data
+        assert data["records"][0]["timestamp"] == "ts1"
+        assert data["summary"]["readings"] == 1

tests/test_utils.py

@@ -0,0 +1,35 @@
+import os
+import re
+import shutil
+import tempfile
+
+from voiceMonitor.utils import timestamp, ensure_dir
+
+
+def test_timestamp_format():
+    ts = timestamp()
+    # expected format: YYYYMMDD_HHMMSS
+    assert re.match(r"^\d{8}_\d{6}$", ts)
+
+
+def test_ensure_dir_creates_missing_directory():
+    tmp_root = tempfile.mkdtemp()
+    try:
+        target = os.path.join(tmp_root, "nested", "chunks")
+        assert not os.path.exists(target)
+        ensure_dir(target)
+        assert os.path.exists(target)
+        assert os.path.isdir(target)
+    finally:
+        shutil.rmtree(tmp_root)
+
+
+def test_ensure_dir_is_idempotent_on_existing_directory():
+    tmp_root = tempfile.mkdtemp()
+    try:
+        # should not raise when the directory already exists
+        ensure_dir(tmp_root)
+        ensure_dir(tmp_root)
+        assert os.path.exists(tmp_root)
+    finally:
+        shutil.rmtree(tmp_root)