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 服务;不删除或修改录音数据。 --- src/mic_clipper/service.py | 101 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 src/mic_clipper/service.py (limited to 'src/mic_clipper/service.py') diff --git a/src/mic_clipper/service.py b/src/mic_clipper/service.py new file mode 100644 index 0000000..ecf5673 --- /dev/null +++ b/src/mic_clipper/service.py @@ -0,0 +1,101 @@ +"""Install and remove the systemd user service without touching recordings.""" + +from __future__ import annotations + +import os +import subprocess +import tempfile +from collections.abc import Callable +from pathlib import Path + + +UNIT_NAME = "mic-clipper.service" + + +def render_unit(python: Path, project_root: Path) -> str: + return f"""[Unit] +Description=Offline voice-activated microphone clip recorder +Wants=pipewire.service wireplumber.service pipewire-pulse.service +After=pipewire.service wireplumber.service pipewire-pulse.service + +[Service] +Type=simple +WorkingDirectory={project_root} +Environment=PYTHONPATH={project_root / 'src'} +ExecStart={python} -m mic_clipper run +Restart=on-failure +RestartSec=5s +RestartSteps=5 +RestartMaxDelaySec=60s +StartLimitIntervalSec=300 +StartLimitBurst=10 +NoNewPrivileges=yes +PrivateTmp=yes +UMask=0077 + +[Install] +WantedBy=default.target +""" + + +def install( + *, + unit_path: Path, + python: Path, + project_root: Path, + dry_run: bool = False, + run_command: Callable[[list[str]], object] | None = None, +) -> list[list[str]]: + commands = [ + ["systemctl", "--user", "daemon-reload"], + ["systemctl", "--user", "enable", "--now", UNIT_NAME], + ] + if dry_run: + return commands + unit_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + _write_private(unit_path, render_unit(python, project_root)) + runner = run_command or _run_command + for command in commands: + runner(command) + return commands + + +def uninstall( + *, + unit_path: Path, + run_command: Callable[[list[str]], object] | None = None, +) -> list[list[str]]: + if not unit_path.exists(): + return [] + commands = [ + ["systemctl", "--user", "disable", "--now", UNIT_NAME], + ["systemctl", "--user", "daemon-reload"], + ] + runner = run_command or _run_command + for command in commands[:1]: + runner(command) + unit_path.unlink() + runner(commands[1]) + return commands + + +def status() -> int: + return subprocess.run( + ["systemctl", "--user", "status", UNIT_NAME], check=False + ).returncode + + +def _write_private(path: Path, content: str) -> None: + with tempfile.NamedTemporaryFile( + "w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False + ) as temporary: + temporary.write(content) + temporary.flush() + os.fchmod(temporary.fileno(), 0o600) + temporary_path = Path(temporary.name) + os.replace(temporary_path, path) + os.chmod(path, 0o600) + + +def _run_command(command: list[str]) -> None: + subprocess.run(command, check=True) -- cgit v1.2.3