1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
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)
|
