summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
authorSomhairle H. Marisol <[email protected]>2026-09-29 10:50:12 +0800
committerSomhairle H. Marisol <[email protected]>2026-09-29 10:50:12 +0800
commit60f05cebd8e7968621a0ce015cf72987ec7577c7 (patch)
tree7f3c572ae82015b17ceaff6eb901ec88ba809856 /tests
downloadmic-clipper-60f05cebd8e7968621a0ce015cf72987ec7577c7.tar.gz
feat(recorder): add offline voice clip pipeline
[变更性质] - 本提交新增本机离线的语音片段录音能力,不涉及网络服务或语音识别。 [新增功能] - 通过 Pulse 默认输入持续采集音频,以本地 Silero VAD 触发片段。 - 以私有目录和权限写入 24 kbps Ogg/Opus 录音,并提供 systemd 用户服务安装器。 [实现方案] - 使用有限前置缓冲、静音封段和最长段滚动状态机,避免静音落盘及内存无限增长。 - 增加真实 Opus 编解码、服务 dry-run 和纯逻辑状态转换测试;记录模型归属和部署前置条件。 [影响范围] - 新增 Python CLI、运行时模块、中文运维文档和自动测试。 - 未安装、启用或启动任何用户 systemd 服务;不删除或修改录音数据。
Diffstat (limited to 'tests')
-rw-r--r--tests/test_cli.py25
-rw-r--r--tests/test_runtime.py82
-rw-r--r--tests/test_segmenter.py96
-rw-r--r--tests/test_service.py60
-rw-r--r--tests/test_storage.py33
-rw-r--r--tests/test_vad.py19
6 files changed, 315 insertions, 0 deletions
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 0000000..09de22c
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,25 @@
+from pathlib import Path
+
+from mic_clipper.cli import main
+
+
+def test_service_install_dry_run_prints_commands_without_creating_unit(tmp_path, capsys):
+ unit_path = tmp_path / "mic-clipper.service"
+
+ result = main(
+ [
+ "service",
+ "install",
+ "--dry-run",
+ "--unit-path",
+ str(unit_path),
+ "--project-root",
+ str(tmp_path),
+ "--python",
+ "/runtime/python",
+ ]
+ )
+
+ assert result == 0
+ assert not unit_path.exists()
+ assert "systemctl --user enable --now mic-clipper.service" in capsys.readouterr().out
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
new file mode 100644
index 0000000..d0d45fb
--- /dev/null
+++ b/tests/test_runtime.py
@@ -0,0 +1,82 @@
+from datetime import datetime, timezone
+
+import numpy as np
+
+from mic_clipper.audio_input import PulseAudioInput
+from mic_clipper.runner import record_frames
+from mic_clipper.segmenter import Segmenter
+
+
+class FakeVad:
+ def __init__(self, probabilities: list[float]) -> None:
+ self.probabilities = iter(probabilities)
+
+ def probability(self, _samples: np.ndarray) -> float:
+ return next(self.probabilities)
+
+
+class FakeClip:
+ def __init__(self) -> None:
+ self.writes: list[np.ndarray] = []
+ self.closed = False
+
+ def write(self, samples: np.ndarray) -> None:
+ self.writes.append(samples)
+
+ def close(self) -> None:
+ self.closed = True
+
+ def abort(self) -> None:
+ self.closed = True
+
+
+class FakeWriter:
+ def __init__(self) -> None:
+ self.opened: list[FakeClip] = []
+
+ def open(self, _started_at: datetime) -> FakeClip:
+ clip = FakeClip()
+ self.opened.append(clip)
+ return clip
+
+
+def samples() -> np.ndarray:
+ return np.zeros(512, dtype=np.float32)
+
+
+def test_silence_never_opens_an_output_clip():
+ writer = FakeWriter()
+
+ record_frames(
+ [samples(), samples()],
+ vad=FakeVad([0.1, 0.1]),
+ writer=writer,
+ segmenter=Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5),
+ now=lambda: datetime(2026, 9, 29, tzinfo=timezone.utc),
+ )
+
+ assert writer.opened == []
+
+
+def test_end_event_closes_the_active_encoder():
+ writer = FakeWriter()
+
+ record_frames(
+ [samples()] + [samples()] * 47,
+ vad=FakeVad([0.9] + [0.1] * 47),
+ writer=writer,
+ segmenter=Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5),
+ now=lambda: datetime(2026, 9, 29, tzinfo=timezone.utc),
+ )
+
+ assert len(writer.opened) == 1
+ assert writer.opened[0].closed
+
+
+def test_pulse_input_uses_dynamic_default_device_at_16khz_mono():
+ command = PulseAudioInput.command()
+
+ assert command[0] == "ffmpeg"
+ assert command[command.index("-f") + 1] == "pulse"
+ assert command[command.index("-i") + 1] == "default"
+ assert command[-2:] == ["f32le", "pipe:1"]
diff --git a/tests/test_segmenter.py b/tests/test_segmenter.py
new file mode 100644
index 0000000..d1f7ee5
--- /dev/null
+++ b/tests/test_segmenter.py
@@ -0,0 +1,96 @@
+from datetime import datetime, timedelta, timezone
+
+import numpy as np
+
+from mic_clipper.segmenter import Audio, End, Segmenter, Start
+
+
+SAMPLE_RATE = 16_000
+FRAME_SAMPLES = 512
+FRAME_DURATION = timedelta(seconds=FRAME_SAMPLES / SAMPLE_RATE)
+START = datetime(2026, 9, 29, 23, 59, 59, tzinfo=timezone.utc)
+
+
+def frame(value: float = 0.0) -> np.ndarray:
+ return np.full(FRAME_SAMPLES, value, dtype=np.float32)
+
+
+def push(segmenter: Segmenter, index: int, speech: bool, value: float = 0.0):
+ return segmenter.push(frame(value), speech, START + index * FRAME_DURATION)
+
+
+def test_speech_starts_segment_with_half_second_preroll():
+ segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5)
+
+ for index in range(16):
+ assert push(segmenter, index, False) == ()
+
+ events = push(segmenter, 16, True, 1.0)
+
+ assert len(events) == 1
+ assert isinstance(events[0], Start)
+ assert events[0].started_at == START + 16 * FRAME_DURATION - timedelta(seconds=0.5)
+ assert events[0].samples.shape == (8_512,)
+ assert np.all(events[0].samples[:8_000] == 0.0)
+ assert np.all(events[0].samples[8_000:] == 1.0)
+
+
+def test_short_silence_is_merged_into_active_segment():
+ segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5)
+
+ assert isinstance(push(segmenter, 0, True, 1.0)[0], Start)
+ assert isinstance(push(segmenter, 1, False)[0], Audio)
+ assert isinstance(push(segmenter, 2, True, 1.0)[0], Audio)
+ assert segmenter.active
+
+
+def test_one_point_five_seconds_of_silence_closes_segment():
+ segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5)
+
+ push(segmenter, 0, True, 1.0)
+ events = ()
+ for index in range(1, 48):
+ events = push(segmenter, index, False)
+
+ assert isinstance(events[0], Audio)
+ assert isinstance(events[-1], End)
+ assert not segmenter.active
+
+
+def test_maximum_segment_rolls_without_losing_audio():
+ segmenter = Segmenter(
+ pre_roll_seconds=0.5,
+ silence_seconds=1.5,
+ max_segment_seconds=FRAME_SAMPLES * 2 / SAMPLE_RATE,
+ )
+
+ first = push(segmenter, 0, True, 1.0)
+ second = push(segmenter, 1, True, 2.0)
+ third = push(segmenter, 2, True, 3.0)
+
+ assert isinstance(first[0], Start)
+ assert isinstance(second[0], Audio)
+ assert isinstance(third[0], End)
+ assert isinstance(third[1], Start)
+ assert np.all(third[1].samples == 3.0)
+
+
+def test_segment_start_date_is_retained_across_midnight():
+ segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5)
+
+ events = push(segmenter, 16, True, 1.0)
+
+ assert isinstance(events[0], Start)
+ assert events[0].started_at.date().isoformat() == "2026-09-29"
+
+
+def test_speech_after_a_closed_segment_keeps_terminal_silence_as_preroll():
+ segmenter = Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5)
+
+ push(segmenter, 0, True, 1.0)
+ for index in range(1, 48):
+ push(segmenter, index, False)
+ events = push(segmenter, 48, True, 1.0)
+
+ assert isinstance(events[0], Start)
+ assert events[0].samples.shape == (8_512,)
diff --git a/tests/test_service.py b/tests/test_service.py
new file mode 100644
index 0000000..2bda2ab
--- /dev/null
+++ b/tests/test_service.py
@@ -0,0 +1,60 @@
+from pathlib import Path
+
+from mic_clipper.service import install, render_unit, uninstall
+
+
+def test_rendered_service_waits_for_audio_services_and_restarts_only_on_failure():
+ unit = render_unit(Path("/runtime/python"), Path("/project"))
+
+ assert "After=pipewire.service wireplumber.service pipewire-pulse.service" in unit
+ assert "Restart=on-failure" in unit
+ assert "RestartMaxDelaySec=60s" in unit
+ assert "ExecStart=/runtime/python -m mic_clipper run" in unit
+ assert "UMask=0077" in unit
+
+
+def test_install_and_uninstall_only_manage_the_tool_unit(tmp_path):
+ unit_path = tmp_path / "systemd" / "mic-clipper.service"
+ clips = tmp_path / "Mic Clips" / "2026-09-29" / "keep.opus"
+ clips.parent.mkdir(parents=True)
+ clips.write_bytes(b"recording")
+ commands: list[list[str]] = []
+
+ install(
+ unit_path=unit_path,
+ python=Path("/runtime/python"),
+ project_root=Path("/project"),
+ run_command=commands.append,
+ )
+ install(
+ unit_path=unit_path,
+ python=Path("/runtime/python"),
+ project_root=Path("/project"),
+ run_command=commands.append,
+ )
+ uninstall(unit_path=unit_path, run_command=commands.append)
+
+ assert not unit_path.exists()
+ assert clips.read_bytes() == b"recording"
+ assert commands == [
+ ["systemctl", "--user", "daemon-reload"],
+ ["systemctl", "--user", "enable", "--now", "mic-clipper.service"],
+ ["systemctl", "--user", "daemon-reload"],
+ ["systemctl", "--user", "enable", "--now", "mic-clipper.service"],
+ ["systemctl", "--user", "disable", "--now", "mic-clipper.service"],
+ ["systemctl", "--user", "daemon-reload"],
+ ]
+
+
+def test_dry_run_does_not_write_a_unit(tmp_path):
+ unit_path = tmp_path / "mic-clipper.service"
+
+ commands = install(
+ unit_path=unit_path,
+ python=Path("/runtime/python"),
+ project_root=Path("/project"),
+ dry_run=True,
+ )
+
+ assert not unit_path.exists()
+ assert commands[-1] == ["systemctl", "--user", "enable", "--now", "mic-clipper.service"]
diff --git a/tests/test_storage.py b/tests/test_storage.py
new file mode 100644
index 0000000..389f199
--- /dev/null
+++ b/tests/test_storage.py
@@ -0,0 +1,33 @@
+import os
+import subprocess
+from datetime import datetime, timezone
+
+import numpy as np
+
+from mic_clipper.storage import ClipWriter
+
+
+def test_writer_creates_private_opus_clip_in_segment_start_date(tmp_path):
+ started_at = datetime(2026, 9, 29, 23, 59, 59, tzinfo=timezone.utc)
+ samples = np.sin(np.linspace(0, 100, 16_000, dtype=np.float32))
+ writer = ClipWriter(tmp_path)
+
+ clip = writer.open(started_at)
+ clip.write(samples)
+ clip.close()
+
+ destination = tmp_path / "2026-09-29"
+ files = list(destination.glob("*.opus"))
+ assert len(files) == 1
+ assert files[0].name.startswith("23-59-59")
+ assert os.stat(destination).st_mode & 0o777 == 0o700
+ assert os.stat(files[0]).st_mode & 0o777 == 0o600
+ assert files[0].read_bytes()[:4] == b"OggS"
+ assert b"OpusHead" in files[0].read_bytes()[:128]
+
+ decoded = subprocess.run(
+ ["ffmpeg", "-v", "error", "-i", str(files[0]), "-f", "f32le", "pipe:1"],
+ check=True,
+ capture_output=True,
+ )
+ assert len(decoded.stdout) > 0
diff --git a/tests/test_vad.py b/tests/test_vad.py
new file mode 100644
index 0000000..98826ae
--- /dev/null
+++ b/tests/test_vad.py
@@ -0,0 +1,19 @@
+import numpy as np
+import pytest
+
+from mic_clipper.vad import FRAME_SAMPLES, SileroVad
+
+
+def test_bundled_silero_vad_accepts_512_samples_and_returns_probability():
+ vad = SileroVad()
+
+ probability = vad.probability(np.zeros(FRAME_SAMPLES, dtype=np.float32))
+
+ assert 0.0 <= probability <= 1.0
+
+
+def test_bundled_silero_vad_rejects_non_512_sample_frames():
+ vad = SileroVad()
+
+ with pytest.raises(ValueError, match="512"):
+ vad.probability(np.zeros(FRAME_SAMPLES - 1, dtype=np.float32))