summaryrefslogtreecommitdiff
path: root/src/mic_clipper/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 /src/mic_clipper/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 'src/mic_clipper/segmenter.py')
-rw-r--r--src/mic_clipper/segmenter.py120
1 files changed, 120 insertions, 0 deletions
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