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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
"""User-local tools. No sudo and no changes to the application's interpreter."""
from __future__ import annotations
from collections import deque
from collections.abc import Callable
import json
import os
from pathlib import Path
import selectors
import shutil
import signal
import subprocess
import sys
import time
Progress = Callable[[str, float | None, float | None], None]
def data_dir() -> Path:
"""The VoiceForge data/prefix directory.
VOICEFORGE_HOME wins (the single-directory launcher exports it), then
XDG_DATA_HOME/voiceforge, then ~/.local/share/voiceforge. A relative
VOICEFORGE_HOME resolves against the current directory.
"""
home = os.environ.get("VOICEFORGE_HOME")
if home:
return Path(home).absolute()
root = Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share")))
if not root.is_absolute():
root = Path.home() / ".local/share"
return root / "voiceforge"
def tool_env() -> dict[str, str]:
"""Sanitized environment for bootstrap, setup, and worker subprocesses.
Host Python settings are removed so a virtualenv or sitecustomize on the
machine cannot leak into managed environments. Installation-target,
index, and constraint overrides are dropped and package-manager
configuration files disabled, so packages can only come from the
explicitly passed trusted indexes and land inside the data directory;
every cache (pip, uv, managed Python, XDG, temporary files) stays there
too.
"""
env = os.environ.copy()
for name in ("PYTHONPATH", "PYTHONHOME", "VIRTUAL_ENV",
"PIP_TARGET", "PIP_PREFIX", "PIP_USER", "PIP_FIND_LINKS",
"PIP_INDEX_URL", "PIP_EXTRA_INDEX_URL", "PIP_CONSTRAINT",
"UV_INDEX_URL", "UV_DEFAULT_INDEX", "UV_EXTRA_INDEX_URL",
"UV_INDEX", "UV_FIND_LINKS", "UV_CONSTRAINT"):
env.pop(name, None)
env["PYTHONNOUSERSITE"] = "1"
env["PIP_CONFIG_FILE"] = os.devnull
env["UV_NO_CONFIG"] = "1"
cache = data_dir() / "cache"
env["XDG_CACHE_HOME"] = str(cache)
env["PIP_CACHE_DIR"] = str(cache / "pip")
env["UV_CACHE_DIR"] = str(cache / "uv")
env["UV_PYTHON_INSTALL_DIR"] = str(data_dir() / "uv/python")
tmp = data_dir() / "tmp"
try:
tmp.mkdir(parents=True, exist_ok=True)
env["TMPDIR"] = str(tmp)
except OSError:
pass # Best effort: never break subprocesses over temporary-file containment.
return env
def terminate_group(process: subprocess.Popen) -> None:
"""SIGTERM the child's whole process group, escalate to SIGKILL, and reap.
The group is signalled even when the leader has already exited, because
descendants can outlive it while still holding the output pipe open.
A race between exit and signalling is tolerated.
"""
for sig in (signal.SIGTERM, signal.SIGKILL):
try:
os.killpg(process.pid, sig)
except ProcessLookupError:
pass
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
continue
try:
os.killpg(process.pid, 0)
except ProcessLookupError:
return
process.wait()
def run_process(args: list[str], stage: str, progress: Progress | None = None,
*, env: dict[str, str] | None = None) -> None:
"""Drain output without blocking; send heartbeat callbacks every 0.25s.
Worker JSON records carry frame/byte counts. Other processes report elapsed seconds
as completed with total=None. Only the last 64 KiB of diagnostics are retained.
Callback exceptions (including cancellation) terminate the process group.
"""
started = time.monotonic()
current_stage, completed, total = stage, None, None
diagnostics: deque[bytes] = deque(maxlen=16)
pending = b""
if progress:
progress(stage, None, None)
with subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
env=env, start_new_session=True) as process:
assert process.stdout is not None
os.set_blocking(process.stdout.fileno(), False)
try:
with selectors.DefaultSelector() as selector:
selector.register(process.stdout, selectors.EVENT_READ)
while selector.get_map() or process.poll() is None:
for key, _ in selector.select(timeout=0.25):
chunk = os.read(key.fd, 4096)
if not chunk:
selector.unregister(key.fileobj)
continue
diagnostics.append(chunk)
pending += chunk
while b"\n" in pending:
line, pending = pending.split(b"\n", 1)
if line.startswith(b"VOICEFORGE_PROGRESS "):
event = json.loads(line[len(b"VOICEFORGE_PROGRESS "):])
current_stage, completed, total = event
if len(pending) > 65536:
pending = pending[-4096:]
if progress:
progress(current_stage, completed if total is not None else
time.monotonic() - started, total)
code = process.wait()
if code:
detail = b"".join(diagnostics).decode("utf-8", errors="replace")
raise RuntimeError(f"{stage} failed (exit {code}):\n{detail}")
except BaseException:
terminate_group(process)
raise
def _uv_supports_relocatable(uv: str) -> bool:
"""Whether this uv understands the venv flags AI setup relies on.
A host uv can be older than the pinned one; capability is checked
directly instead of trusting the version string.
"""
try:
result = subprocess.run([uv, "venv", "--help"], capture_output=True,
text=True, check=True)
except (OSError, subprocess.SubprocessError):
return False
return "--relocatable" in result.stdout + result.stderr
def ensure_uv(progress: Progress | None = None) -> str:
"""Return a uv that supports the flags AI setup relies on.
The launcher's contained bootstrap uv is preferred; a host uv is trusted
only if it actually supports --relocatable. Otherwise uv==0.8.17 is
installed from PyPI (trust PyPI's distribution, not a downloaded shell
script).
"""
contained = data_dir() / "bootstrap/bin/uv"
if contained.is_file() and os.access(contained, os.X_OK):
return str(contained)
found = shutil.which("uv")
if found and _uv_supports_relocatable(found):
return found
root = contained.parent.parent
root.parent.mkdir(parents=True, exist_ok=True)
python = sys.executable or shutil.which("python3")
env = tool_env()
run_process([python, "-m", "venv", str(root)], "Creating uv bootstrap",
progress, env=env)
run_process([str(root / "bin/python"), "-m", "pip", "install",
"--index-url", "https://pypi.org/simple", "uv==0.8.17"],
"Installing uv from PyPI", progress, env=env)
if not (contained.is_file() and os.access(contained, os.X_OK)):
raise RuntimeError("Installing uv from PyPI did not produce a usable "
"uv; check for host pip settings (such as "
"PIP_TARGET) that redirect installations outside "
"the data directory.")
return str(contained)
def ensure_ffmpeg() -> str:
"""Return the bundled FFmpeg, or a system one where the bundle is
absent, not ffprobe.
The launcher installs imageio-ffmpeg via the bundled-ffmpeg extra; direct
package installs should declare it as a dependency. This helper never
runs pip in the current interpreter.
"""
try:
import imageio_ffmpeg
return imageio_ffmpeg.get_ffmpeg_exe()
except (ImportError, RuntimeError) as error:
found = shutil.which("ffmpeg")
if found:
return found
raise RuntimeError("FFmpeg is unavailable. Rebuild the contained "
"runtime with './producer.sh --rebuild', or "
"install a system FFmpeg.") from error
|
