From 60f05cebd8e7968621a0ce015cf72987ec7577c7 Mon Sep 17 00:00:00 2001 From: "Somhairle H. Marisol" Date: Tue, 29 Sep 2026 10:50:12 +0800 Subject: feat(recorder): add offline voice clip pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [变更性质] - 本提交新增本机离线的语音片段录音能力，不涉及网络服务或语音识别。 [新增功能] - 通过 Pulse 默认输入持续采集音频，以本地 Silero VAD 触发片段。 - 以私有目录和权限写入 24 kbps Ogg/Opus 录音，并提供 systemd 用户服务安装器。 [实现方案] - 使用有限前置缓冲、静音封段和最长段滚动状态机，避免静音落盘及内存无限增长。 - 增加真实 Opus 编解码、服务 dry-run 和纯逻辑状态转换测试；记录模型归属和部署前置条件。 [影响范围] - 新增 Python CLI、运行时模块、中文运维文档和自动测试。 - 未安装、启用或启动任何用户 systemd 服务；不删除或修改录音数据。 --- tests/test_segmenter.py | 96 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 tests/test_segmenter.py (limited to 'tests/test_segmenter.py') 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,) -- cgit v1.2.3