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