From 60f05cebd8e7968621a0ce015cf72987ec7577c7 Mon Sep 17 00:00:00 2001
From: "Somhairle H. Marisol" <muqiu-han@outlook.com>
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 服务；不删除或修改录音数据。
---
 src/mic_clipper/segmenter.py | 120 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 120 insertions(+)
 create mode 100644 src/mic_clipper/segmenter.py

(limited to 'src/mic_clipper/segmenter.py')

diff --git a/src/mic_clipper/segmenter.py b/src/mic_clipper/segmenter.py
new file mode 100644
index 0000000..aa0e3b2
--- /dev/null
+++ b/src/mic_clipper/segmenter.py
@@ -0,0 +1,120 @@
+"""Pure streaming voice-clip state machine."""
+
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+import numpy as np
+
+
+SAMPLE_RATE = 16_000
+
+
+@dataclass(frozen=True)
+class Start:
+    started_at: datetime
+    samples: np.ndarray
+
+
+@dataclass(frozen=True)
+class Audio:
+    samples: np.ndarray
+
+
+@dataclass(frozen=True)
+class End:
+    pass
+
+
+class Segmenter:
+    """Emit stream events while retaining only the configured pre-roll in memory."""
+
+    def __init__(
+        self,
+        *,
+        pre_roll_seconds: float,
+        silence_seconds: float,
+        max_segment_seconds: float = 600,
+        sample_rate: int = SAMPLE_RATE,
+    ) -> None:
+        self.sample_rate = sample_rate
+        self.pre_roll_samples = round(pre_roll_seconds * sample_rate)
+        self.silence_samples_limit = round(silence_seconds * sample_rate)
+        self.max_segment_samples = round(max_segment_seconds * sample_rate)
+        self._pre_roll: deque[np.ndarray] = deque()
+        self._pre_roll_size = 0
+        self._active = False
+        self._segment_samples = 0
+        self._silence_samples = 0
+
+    @property
+    def active(self) -> bool:
+        return self._active
+
+    def push(
+        self, samples: np.ndarray, is_speech: bool, captured_at: datetime
+    ) -> tuple[Start | Audio | End, ...]:
+        samples = np.asarray(samples, dtype=np.float32)
+        if samples.ndim != 1:
+            raise ValueError("audio must be a mono, one-dimensional array")
+        if not len(samples):
+            return ()
+
+        if not self._active:
+            if not is_speech:
+                self._append_pre_roll(samples)
+                return ()
+            pre_roll = self._take_pre_roll()
+            started_at = captured_at - timedelta(
+                seconds=len(pre_roll) / self.sample_rate
+            )
+            self._active = True
+            self._segment_samples = len(pre_roll) + len(samples)
+            self._silence_samples = 0
+            self._append_pre_roll(samples)
+            return (Start(started_at, np.concatenate((pre_roll, samples))),)
+
+        if self._segment_samples + len(samples) > self.max_segment_samples:
+            self._append_pre_roll(samples)
+            self._active = True
+            self._segment_samples = len(samples)
+            self._silence_samples = 0
+            return (End(), Start(captured_at, samples))
+
+        self._segment_samples += len(samples)
+        if is_speech:
+            self._silence_samples = 0
+        else:
+            self._silence_samples += len(samples)
+        self._append_pre_roll(samples)
+
+        events: tuple[Start | Audio | End, ...] = (Audio(samples),)
+        if self._silence_samples >= self.silence_samples_limit:
+            self._active = False
+            self._segment_samples = 0
+            self._silence_samples = 0
+            events += (End(),)
+        return events
+
+    def _append_pre_roll(self, samples: np.ndarray) -> None:
+        self._pre_roll.append(samples)
+        self._pre_roll_size += len(samples)
+        while self._pre_roll_size > self.pre_roll_samples:
+            excess = self._pre_roll_size - self.pre_roll_samples
+            oldest = self._pre_roll[0]
+            if len(oldest) <= excess:
+                self._pre_roll.popleft()
+                self._pre_roll_size -= len(oldest)
+            else:
+                self._pre_roll[0] = oldest[excess:]
+                self._pre_roll_size -= excess
+
+    def _take_pre_roll(self) -> np.ndarray:
+        if not self._pre_roll:
+            return np.empty(0, dtype=np.float32)
+        samples = np.concatenate(tuple(self._pre_roll))
+        self._pre_roll.clear()
+        self._pre_roll_size = 0
+        return samples
-- 
cgit v1.2.3