summaryrefslogtreecommitdiff
path: root/src/mic_clipper/service.py
diff options
context:
space:
mode:
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)