summaryrefslogtreecommitdiff
path: root/tests/test_segmenter.py
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/test_segmenter.py
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/test_segmenter.py')
-rw-r--r--tests/test_segmenter.py96
1 files changed, 96 insertions, 0 deletions
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,)