From 303f003c15b590aa8b4d0803ca0e03a5d42572ac Mon Sep 17 00:00:00 2001 From: Thales Pereira <31625914+thcp@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:07:01 +0100 Subject: [PATCH 1/5] feat(settings): stem storage format setting, MP3 128 kbps by default Settings and the API gain stem_format (wav | mp3) and stem_mp3_kbps (128 | 192 | 256), whitelisted and validated together so one bad value changes nothing. STEMDECK_STEM_FORMAT and STEMDECK_STEM_MP3_KBPS seed the defaults for Docker and Unraid; a saved choice wins, as for separation_quality. The default is one constant in config.py. --- app/core/config.py | 13 ++++++++ app/core/settings.py | 72 ++++++++++++++++++++++++++++++++++++++++++++ app/main.py | 14 +++++++++ 3 files changed, 99 insertions(+) diff --git a/app/core/config.py b/app/core/config.py index b41ca515..5041cb79 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -403,6 +403,19 @@ def js_solver_available() -> bool: # A backing track without one stem (app/pipeline/backing.py): a sum of five # WAVs, seconds for a song; the ceiling only guards a stuck disk. TIMEOUT_BACKING = _env_int("STEMDECK_TIMEOUT_BACKING", 120) + +# How a finished song's stems are kept on disk (#783): "wav" (16-bit PCM, as +# separated) or "mp3" (CBR libmp3lame at one of STEM_MP3_KBPS_CHOICES). The +# default for an install with no saved choice and no env var. One line to +# change. 128 kbps is the maintainer's choice: stems summed again in the +# mixer carry the coding noise of each one. +DEFAULT_STEM_STORAGE: tuple[str, int] = ("mp3", 128) +STEM_FORMAT_CHOICES: tuple[str, ...] = ("wav", "mp3") +STEM_MP3_KBPS_CHOICES: tuple[int, ...] = (128, 192, 256) +# The MP3 step at the end of the pipeline (app/pipeline/stem_storage.py): one +# ffmpeg per stem, a few seconds each. Terminated when ffmpeg is silent this +# long; the song then keeps its WAVs. +TIMEOUT_STEM_TRANSCODE_STALL = _env_int("STEMDECK_TIMEOUT_STEM_TRANSCODE_STALL", 120) # Video render of a take (libx264 + AAC). Slower than any audio export: a # 5-minute 1080p30 encode runs about 1x to 3x realtime on a desktop CPU. TIMEOUT_VIDEO_RENDER = _env_int("STEMDECK_TIMEOUT_VIDEO_RENDER", 3600) diff --git a/app/core/settings.py b/app/core/settings.py index f18542c1..1014569d 100644 --- a/app/core/settings.py +++ b/app/core/settings.py @@ -18,6 +18,8 @@ - `discogs_token` - the user's Discogs token, for band profiles. - `transcribe_lyrics` - Whisper lyrics when none are found: auto | on | off. - `video_encoder` - H.264 encoder of the take render: auto | cpu | an encoder. +- `stem_format` - how new songs' stems are stored: wav | mp3 (#783). +- `stem_mp3_kbps` - the MP3 bitrate when stem_format is mp3: 128 | 192 | 256. Defaults fall back to the config.py constants (which honor their env vars), so nothing changes until the user overrides a value. @@ -36,8 +38,11 @@ from app.core.config import ( DATA_DIR, + DEFAULT_STEM_STORAGE, MAX_DURATION_SEC, PLAYLIST_MAX_ITEMS, + STEM_FORMAT_CHOICES, + STEM_MP3_KBPS_CHOICES, VIDEO_ENCODER_CHOICES, VIDEO_MAX_HEIGHT, available_torch_devices, @@ -635,6 +640,73 @@ def set_separation_quality(value: str) -> str: return choice +# ── stem_format / stem_mp3_kbps ── +# How a song's stems are stored once it finishes (#783): "wav" or "mp3" at a +# CBR bitrate. Read when a job finishes (app/pipeline/stem_storage.py), so a +# change applies to the next song without a restart; songs already in the +# library keep whatever they were stored as. STEMDECK_STEM_FORMAT and +# STEMDECK_STEM_MP3_KBPS seed the defaults for Docker and Unraid; a choice +# saved in Settings wins over them, as for separation_quality. +def _default_stem_format() -> str: + env = os.environ.get("STEMDECK_STEM_FORMAT", "").strip().lower() + return env if env in STEM_FORMAT_CHOICES else DEFAULT_STEM_STORAGE[0] + + +def _default_stem_mp3_kbps() -> int: + raw = os.environ.get("STEMDECK_STEM_MP3_KBPS", "").strip().lower().removesuffix("k") + try: + kbps = int(raw) + except ValueError: + return DEFAULT_STEM_STORAGE[1] + return kbps if kbps in STEM_MP3_KBPS_CHOICES else DEFAULT_STEM_STORAGE[1] + + +def get_stem_format() -> str: + with _LOCK: + v = _ensure().get("stem_format") + return v if isinstance(v, str) and v in STEM_FORMAT_CHOICES else _default_stem_format() + + +def get_stem_mp3_kbps() -> int: + with _LOCK: + v = _ensure().get("stem_mp3_kbps") + ok = isinstance(v, int) and not isinstance(v, bool) and v in STEM_MP3_KBPS_CHOICES + return v if ok else _default_stem_mp3_kbps() + + +def check_stem_format(value: object) -> str: + """The whitelisted format for ``value``, or ValueError (never echoing it).""" + choice = value.strip().lower() if isinstance(value, str) else "" + if choice not in STEM_FORMAT_CHOICES: + raise ValueError("stem_format must be one of: " + ", ".join(STEM_FORMAT_CHOICES)) + return STEM_FORMAT_CHOICES[STEM_FORMAT_CHOICES.index(choice)] + + +def check_stem_mp3_kbps(value: object) -> int: + """The whitelisted bitrate for ``value``, or ValueError (never echoing it).""" + kbps = value if isinstance(value, int) and not isinstance(value, bool) else None + if kbps not in STEM_MP3_KBPS_CHOICES: + raise ValueError( + "stem_mp3_kbps must be one of: " + ", ".join(map(str, STEM_MP3_KBPS_CHOICES)) + ) + return kbps + + +def set_stem_storage(fmt: object = None, kbps: object = None) -> tuple[str, int]: + """Persist either or both, after both are validated: a request with one + bad value changes nothing. Returns what now applies.""" + choice = check_stem_format(fmt) if fmt is not None else None + rate = check_stem_mp3_kbps(kbps) if kbps is not None else None + with _LOCK: + if choice is not None: + _ensure()["stem_format"] = choice + if rate is not None: + _ensure()["stem_mp3_kbps"] = rate + if choice is not None or rate is not None: + _save() + return get_stem_format(), get_stem_mp3_kbps() + + # ── video_encoder ── # The H.264 encoder of the take render (app/pipeline/video_encoder.py): "auto" # (default) uses a GPU encoder that passed a real test encode on this machine, diff --git a/app/main.py b/app/main.py index 12173c15..b582c953 100644 --- a/app/main.py +++ b/app/main.py @@ -72,6 +72,8 @@ get_playlist_max_items, get_port, get_separation_quality, + get_stem_format, + get_stem_mp3_kbps, get_transcribe_lyrics, get_video_encoder, get_video_max_height, @@ -90,6 +92,7 @@ set_playlist_max_items, set_port, set_separation_quality, + set_stem_storage, set_transcribe_lyrics, set_video_encoder, set_video_max_height, @@ -394,6 +397,10 @@ def _settings_payload() -> dict[str, object]: "video_max_height": get_video_max_height(), "export_sample_rate": get_export_sample_rate(), "separation_quality": get_separation_quality(), + # How new songs' stems are stored (#783). Read when a job finishes; + # songs already in the library keep their own format. + "stem_format": get_stem_format(), + "stem_mp3_kbps": get_stem_mp3_kbps(), # Whisper lyrics for a job no lookup found any for: "auto" (on a GPU # only) | "on" | "off". "transcribe_lyrics": get_transcribe_lyrics(), @@ -551,6 +558,13 @@ async def update_settings(request: Request) -> dict[str, object]: set_separation_quality(str(body["separation_quality"])) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) from None + if "stem_format" in body or "stem_mp3_kbps" in body: + try: + # Both checked before either is kept: one bad value changes nothing. + set_stem_storage(body.get("stem_format"), body.get("stem_mp3_kbps")) + except ValueError as e: + # The messages list the whitelist; they never echo the value. + raise HTTPException(status_code=422, detail=str(e)) from None if "transcribe_lyrics" in body: try: set_transcribe_lyrics(str(body["transcribe_lyrics"])) From 977c0d7858f939211d4a994f007b158a077fa073 Mon Sep 17 00:00:00 2001 From: Thales Pereira <31625914+thcp@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:08:12 +0100 Subject: [PATCH 2/5] feat(pipeline): store finished stems as MP3, one resolver for every reader A last pipeline stage (stem_storage.py) runs once nothing reads the stems any more. With MP3 selected it encodes each stem with libmp3lame CBR into a temporary file, checks that ffmpeg and libsndfile both decode it to the WAV's exact frame count and that its loudest block lines up at lag 0, and only when every stem passed moves the MP3s into place and deletes the WAVs. Any failure keeps the WAVs and records the song as WAV; a cancel cleans up and raises. The song is never failed. app/core/stem_files.py resolves a stem by the file on disk, WAV first (a WAV song's cached download MP3 never shadows its WAV), and every reader of a finished song goes through it: the stem, region, mixdown, zip, backing, vocal envelope, peaks, take length, loudness, fingerprint, lyrics timing, vocal split (handed a decoded WAV), recovery and the sweep, which would otherwise have deleted an unregistered MP3 song. Each job records stem_format and stem_mp3_kbps. The identify regression tests pin WAV: their stubbed stems are placeholder bytes. --- app/api/jobs.py | 23 +- app/api/stems.py | 37 +- app/api/takes.py | 5 +- app/core/models.py | 9 + app/core/registry.py | 16 +- app/core/stem_files.py | 134 ++++++ app/pipeline/backing.py | 12 +- app/pipeline/collect.py | 12 +- app/pipeline/lyrics_align.py | 3 +- app/pipeline/lyrics_retime.py | 22 +- app/pipeline/runner.py | 22 + app/pipeline/stem_storage.py | 489 ++++++++++++++++++++++ app/pipeline/transcribe.py | 7 +- app/pipeline/vocal_split.py | 17 +- docs/configuration.md | 2 + templates/stemdeck.xml | 4 + tests/test_identify_regressions.py | 10 + tests/test_stem_storage.py | 633 +++++++++++++++++++++++++++++ 18 files changed, 1414 insertions(+), 43 deletions(-) create mode 100644 app/core/stem_files.py create mode 100644 app/pipeline/stem_storage.py create mode 100644 tests/test_stem_storage.py diff --git a/app/api/jobs.py b/app/api/jobs.py index 1c52847d..62a8c0ee 100644 --- a/app/api/jobs.py +++ b/app/api/jobs.py @@ -63,6 +63,7 @@ get_discogs_token, get_max_duration_sec, ) +from app.core.stem_files import stem_path, stem_url from app.core.stems_location import is_relocating from app.pipeline import discogs, jobqueue from app.pipeline.analyze import dynamic_range_db, measure_loudness @@ -103,6 +104,7 @@ ) from app.pipeline.lyrics_retime import saved_view as saved_retime_view from app.pipeline.runner import _pipeline_lock +from app.pipeline.stem_storage import store_split_stems from app.pipeline.vocal_split import split_vocals from app.pipeline.work_lookup import find_work, might_have_work @@ -501,7 +503,7 @@ def _loudness_sources(job_id: str) -> list[Path]: stems_dir = (JOBS_DIR / job_id / "stems").resolve() if not stems_dir.is_relative_to(JOBS_DIR.resolve()): return [] - return [p for p in (stems_dir / f"{n}.wav" for n in STEM_NAMES) if p.is_file()] + return [p for p in (stem_path(stems_dir, n) for n in STEM_NAMES) if p.is_file()] def _upgrade_loudness(job: Job) -> None: @@ -805,6 +807,12 @@ async def start_vocal_split(job_id: str) -> Response: try: async with _pipeline_lock: new_names = await asyncio.to_thread(split_vocals, job, stems_dir) + # The peaks are read from the new WAVs first; then the two are kept + # the way the song keeps its stems (#783). Neither raises. + rms_values = await asyncio.to_thread( + merge_stem_peaks, stems_dir, ["vocals", *new_names] + ) + await asyncio.to_thread(store_split_stems, job, stems_dir, new_names) except Exception as e: cause = classify_failure("\n".join([*(getattr(e, "tail", None) or []), str(e)])) logger.warning("[%s] vocal split failed: %s", job_id, e, exc_info=True) @@ -817,11 +825,10 @@ async def start_vocal_split(job_id: str) -> Response: existing = {s["name"] for s in job.stems} for name in new_names: if name not in existing: - job.stems.append({"name": name, "url": f"/api/jobs/{job_id}/stems/{name}.wav"}) - # "vocals" rides along so presence_for_split can recover the scale the base - # stems were normalised against; without it the two new cards would have no - # percentage to show. - rms_values = merge_stem_peaks(stems_dir, ["vocals", *new_names]) + job.stems.append({"name": name, "url": stem_url(job_id, stems_dir, name)}) + # "vocals" rode along in merge_stem_peaks above so presence_for_split can + # recover the scale the base stems were normalised against; without it the + # two new cards would have no percentage to show. extra_presence = presence_for_split(rms_values, job.stem_presence) if extra_presence: job.stem_presence = {**(job.stem_presence or {}), **extra_presence} @@ -1195,7 +1202,7 @@ def _fingerprint_audio(job: Job) -> list[Path]: stems_root = (JOBS_DIR / job.id / "stems").resolve() stems = [] for name in STEM_NAMES: - path = (stems_root / f"{name}.wav").resolve() + path = stem_path(stems_root, name).resolve() if path.is_file() and path.is_relative_to(JOBS_DIR.resolve()): stems.append(path) return stems @@ -1560,7 +1567,7 @@ def _retime_ready(job_dir: Path, lyrics: dict | None) -> None: raise HTTPException(status_code=404, detail="no lyrics") if not lyric_pairs(lyrics): raise HTTPException(status_code=409, detail="lyrics have no lines") - vocals = (job_dir / "stems" / "vocals.wav").resolve() + vocals = stem_path(job_dir / "stems", "vocals").resolve() if not vocals.is_relative_to(JOBS_DIR.resolve()) or not vocals.is_file(): raise HTTPException(status_code=409, detail="no vocals stem") diff --git a/app/api/stems.py b/app/api/stems.py index 95f796e6..0c5113f4 100644 --- a/app/api/stems.py +++ b/app/api/stems.py @@ -35,6 +35,7 @@ ) from app.core.registry import get as registry_get from app.core.settings import get_export_sample_rate +from app.core.stem_files import stem_ext, stem_media_type, stem_path from app.core.takes import take_dir from app.pipeline.audio_stats import vocal_envelope from app.pipeline.click_render import cache_key as click_cache_key @@ -349,7 +350,8 @@ def _validate_stem_path(job_id: str, name: str): job = registry_get(job_id) if job is None or job.status != "done": raise HTTPException(status_code=404, detail="job not ready") - path = (JOBS_DIR / job_id / "stems" / f"{name}.wav").resolve() + # WAV or MP3, whichever this song keeps (#783). + path = stem_path(JOBS_DIR / job_id / "stems", name).resolve() if not path.is_file() or not path.is_relative_to(JOBS_DIR.resolve()): raise HTTPException(status_code=404, detail="stem not found") return path @@ -1044,7 +1046,7 @@ def _expand_render_lanes( # pointing at whichever component was tried last. paths: list[Path] = [] for comp in components: - comp_path = (JOBS_DIR / job_id / "stems" / f"{comp}.wav").resolve() + comp_path = stem_path(JOBS_DIR / job_id / "stems", comp).resolve() if not comp_path.is_file() or not comp_path.is_relative_to(JOBS_DIR.resolve()): paths = [] break @@ -1071,7 +1073,7 @@ def _job_original_set(job_id: str) -> frozenset[str] | None: """The stems the job's own original.wav holds (make_original_track: every base stem not selected), or None when it has no such file.""" job = registry_get(job_id) - path = (JOBS_DIR / job_id / "stems" / "original.wav").resolve() + path = stem_path(JOBS_DIR / job_id / "stems", "original").resolve() if job is None or not path.is_file() or not path.is_relative_to(JOBS_DIR.resolve()): return None selected = set(job.selected_stems) @@ -1128,7 +1130,7 @@ def _expand_mode_lanes( continue paths: list[Path] = [] for comp in components: - comp_path = (JOBS_DIR / job_id / "stems" / f"{comp}.wav").resolve() + comp_path = stem_path(JOBS_DIR / job_id / "stems", comp).resolve() if not comp_path.is_file() or not comp_path.is_relative_to(JOBS_DIR.resolve()): paths = [] break @@ -1498,7 +1500,7 @@ async def get_vocal_envelope(job_id: str) -> Response: if job is None or job.status != "done": raise HTTPException(status_code=404, detail="job not ready") stems_dir = (JOBS_DIR / job_id / "stems").resolve() - src = stems_dir / "vocals.wav" + src = stem_path(stems_dir, "vocals") dest = stems_dir / _VOCAL_ENVELOPE_FILE jobs_root = JOBS_DIR.resolve() if not src.is_file() or not src.is_relative_to(jobs_root) or not dest.is_relative_to(jobs_root): @@ -1557,12 +1559,16 @@ async def get_stem( click_accent: int = Query(default=-1, ge=-1, le=32, description="-1 auto, 0 off, N per bar"), click_gain: float = Query(default=0.6, ge=0, le=4, description="Click level"), ) -> FileResponse | StreamingResponse: - """Download a WAV stem. Optional ?start=&end= trims to a time region.""" + """A stem, as the song stores it: WAV, or MP3 for a song stored as MP3 + (#783), with its own type and file name whatever this URL says. Optional + ?start=&end= trims to a time region, always rendered as WAV.""" path = _validate_stem_path(job_id, name) if start is None and end is None: return FileResponse( - path, media_type="audio/wav", filename=_stem_download_name(job_id, name, "wav") + path, + media_type=stem_media_type(path), + filename=_stem_download_name(job_id, name, stem_ext(path)), ) if start is None or end is None or start >= end: @@ -2211,17 +2217,22 @@ def _arcname(prefix: str, name: str, ext: str) -> str: def _build_stems_zip( sources: list[tuple[str, Path]], fmt: str, dest: Path, prefix: str = "" ) -> None: - """Blocking: write the stems into a ZIP. WAV files are stored as-is; MP3, - FLAC, and OGG are transcoded per stem via ffmpeg. ZIP_STORED throughout - audio doesn't - meaningfully compress, and STORED keeps the build fast. Runs in a thread.""" - if fmt == "wav": + """Blocking: write the stems into a ZIP. A file already in `fmt` is stored + as-is (a WAV in a WAV zip, and an MP3-stored stem in an MP3 zip, which is + not encoded a second time); anything else is transcoded per stem via + ffmpeg. ZIP_STORED throughout - audio doesn't meaningfully compress, and + STORED keeps the build fast. Runs in a thread.""" + if all(p.suffix.lower() == f".{fmt}" for _name, p in sources): with zipfile.ZipFile(dest, "w", zipfile.ZIP_STORED) as zf: for name, p in sources: - zf.write(p, arcname=_arcname(prefix, name, "wav")) + zf.write(p, arcname=_arcname(prefix, name, fmt)) return encode = _ENCODE_ARGS[fmt] with tempfile.TemporaryDirectory() as td, zipfile.ZipFile(dest, "w", zipfile.ZIP_STORED) as zf: for name, p in sources: + if p.suffix.lower() == f".{fmt}": + zf.write(p, arcname=_arcname(prefix, name, fmt)) + continue out = os.path.join(td, f"{name}.{fmt}") cmd = [ ffmpeg_executable(), @@ -2292,7 +2303,7 @@ async def get_all_stems_zip( sources: list[tuple[str, Path]] = [] for name in wanted: - p = (stems_dir / f"{name}.wav").resolve() + p = stem_path(stems_dir, name).resolve() if p.is_file() and p.is_relative_to(jobs_root): sources.append((name, p)) labels = {t.get("id"): t.get("label") for t in job.takes} diff --git a/app/api/takes.py b/app/api/takes.py index b0e98dd2..277ec2a7 100644 --- a/app/api/takes.py +++ b/app/api/takes.py @@ -91,6 +91,7 @@ from app.core.registry import get as registry_get from app.core.registry import persist as registry_persist from app.core.settings import get_export_sample_rate +from app.core.stem_files import stem_path from app.core.takes import ( camera_file, make_track, @@ -268,7 +269,9 @@ def _reference_stem(job_id: str) -> Path | None: if not stems_dir.is_relative_to(JOBS_DIR.resolve()): return None for name in (*STEM_NAMES, *EXTRA_STEM_NAMES, "original"): - path = stems_dir / f"{name}.wav" + # WAV or MP3 (#783): an MP3 stem decodes to the WAV's exact frame + # count, which stem_storage verified before keeping it. + path = stem_path(stems_dir, name) if path.is_file(): return path return None diff --git a/app/core/models.py b/app/core/models.py index 7373ce7b..678ba6ea 100644 --- a/app/core/models.py +++ b/app/core/models.py @@ -338,6 +338,13 @@ class Job: # Wall-clock seconds per pipeline stage ({"download": 12.3, ...}); written # to metadata.json and the one-line completion summary in the log. stage_timings: dict[str, float] | None = None + # How this song's stems are kept on disk (#783): "wav" or "mp3", set when + # the job finishes from the Stem storage format setting of that moment, + # with the MP3 bitrate (kbps) or None. None for a song finished before the + # setting existed, whose stems are WAV. Readers do not branch on it: they + # resolve each file through app.core.stem_files, which looks at the disk. + stem_format: str | None = None + stem_mp3_kbps: int | None = None # On-demand lead/backing vocal split (#275) -- a post-hoc action on an # already-"done" job, not part of the main pipeline. "none" until the user # asks for it; "error" leaves the job's base stems untouched (see @@ -431,6 +438,8 @@ def to_state(self) -> dict[str, Any]: "compute_device": self.compute_device, "gpu_fallback": self.gpu_fallback, "stage_timings": self.stage_timings, + "stem_format": self.stem_format, + "stem_mp3_kbps": self.stem_mp3_kbps, "vocal_split": self.vocal_split, "trashed_at": self.trashed_at, "favorite": self.favorite, diff --git a/app/core/registry.py b/app/core/registry.py index 3bc7612e..c706a0cb 100644 --- a/app/core/registry.py +++ b/app/core/registry.py @@ -14,6 +14,7 @@ from app.core.config import DEMUCS_MODEL, EXTRA_STEM_NAMES, JOB_ID_RE, LYRICS_FILE, STEM_NAMES from app.core.models import Job, clean_artist, clean_audio_tags, clean_identity, clean_work from app.core.settings import quarantine_corrupt +from app.core.stem_files import stem_exists, stem_url from app.core.takes import reconcile_tracks, recover_takes logger = logging.getLogger("stemdeck.registry") @@ -566,22 +567,23 @@ def _recover_done_job(job_dir: Path) -> Job | None: stems_dir = job_dir / "stems" if not stems_dir.is_dir(): return None + # WAV or MP3, whichever the song was stored as (#783). stems = [ - {"name": name, "url": f"/api/jobs/{job_dir.name}/stems/{name}.wav"} + {"name": name, "url": stem_url(job_dir.name, stems_dir, name)} for name in ("original", *STEM_NAMES, *EXTRA_STEM_NAMES) - if (stems_dir / f"{name}.wav").is_file() + if stem_exists(stems_dir, name) ] if not stems: return None mix_url = None - if (stems_dir / "mix.wav").is_file(): - mix_url = f"/api/jobs/{job_dir.name}/stems/mix.wav" + if stem_exists(stems_dir, "mix"): + mix_url = stem_url(job_dir.name, stems_dir, "mix") selected = [stem["name"] for stem in stems if stem["name"] in STEM_NAMES] or list(STEM_NAMES) # A restart between the split finishing and its next registry persist # would otherwise report the job as never split -- derive from disk the # same way `stems` above does, so the recovered library entry doesn't # regress (#275). - has_split = all((stems_dir / f"{name}.wav").is_file() for name in EXTRA_STEM_NAMES) + has_split = all(stem_exists(stems_dir, name) for name in EXTRA_STEM_NAMES) vocal_split = "done" if has_split else "none" meta_path = job_dir / "metadata.json" meta: dict = {} @@ -633,6 +635,10 @@ def _recover_done_job(job_dir: Path) -> Job | None: # The file is the record: metadata.json predates it for most jobs. has_lyrics=(job_dir / LYRICS_FILE).is_file(), vocal_split=vocal_split, + stem_format=meta.get("stem_format") if meta.get("stem_format") in ("wav", "mp3") else None, + stem_mp3_kbps=meta.get("stem_mp3_kbps") + if isinstance(meta.get("stem_mp3_kbps"), int) + else None, # Takes are files first: their folders survive whatever the registry # lost, so the lanes come back with the job. takes=recover_takes(job_dir), diff --git a/app/core/stem_files.py b/app/core/stem_files.py new file mode 100644 index 00000000..0b0b72e9 --- /dev/null +++ b/app/core/stem_files.py @@ -0,0 +1,134 @@ +"""Where a stem's audio file is, whatever format the song keeps it in (#783). + +A library holds songs stored as WAV (every song before #783, and any song made +with the WAV setting) and songs stored as MP3. Every reader that opens a stem +of a finished song goes through `stem_path`, so none of them needs to know +which kind of song it has. + +The rule is the file on disk, WAV first: + +- A WAV song has `.wav`. It may also have `.mp3`, but that is the + download cache GET /stems/{name}.mp3 writes beside the WAV (a VBR transcode), + never the stem itself, so the WAV always wins. +- An MP3 song has only `.mp3`: its WAVs are deleted once the MP3s are + verified (app/pipeline/stem_storage.py). +- While that step runs, and if it fails, the WAVs are still there and win. + +So the answer is right in every state a song's folder can be in, without +reading the job record. `job.stem_format` records what the song was stored as, +for the page and for diagnostics. + +MP3 stems decode to exactly the WAV's samples in length and position: the LAME +header ffmpeg writes carries the encoder delay and padding, and ffmpeg, +libsndfile and Chromium's decodeAudioData all honour it (measured, see +tests/test_stem_storage.py). The storage step verifies the length with both +server-side decoders before it deletes anything. +""" + +from __future__ import annotations + +import contextlib +import shutil +import subprocess +import tempfile +from collections.abc import Iterator +from pathlib import Path + +from app.core.config import TIMEOUT_FFMPEG, ffmpeg_executable + +# In the order they are looked for. WAV first: see the module docstring. +STEM_EXTS: tuple[str, ...] = (".wav", ".mp3") + +_MEDIA_TYPES = {".wav": "audio/wav", ".mp3": "audio/mpeg"} + + +def stem_path(stems_dir: Path, name: str) -> Path: + """The file holding stem `name`: `.wav` when it exists, else + `.mp3` when that exists, else the `.wav` path (which does not + exist, so a caller's own is_file() check fails as it always did). + + `name` must already be whitelisted by the caller.""" + wav = stems_dir / f"{name}.wav" + if wav.is_file(): + return wav + mp3 = stems_dir / f"{name}.mp3" + if mp3.is_file(): + return mp3 + return wav + + +def stem_exists(stems_dir: Path, name: str) -> bool: + return stem_path(stems_dir, name).is_file() + + +def stem_ext(path: Path) -> str: + """ "wav" or "mp3" for a stem file.""" + return path.suffix.lower().removeprefix(".") + + +def stem_media_type(path: Path) -> str: + return _MEDIA_TYPES.get(path.suffix.lower(), "application/octet-stream") + + +def stem_url(job_id: str, stems_dir: Path, name: str) -> str: + """The URL the page loads stem `name` from, ending in the file's own + extension so the browser and a download both get the right type.""" + return f"/api/jobs/{job_id}/stems/{name}.{stem_ext(stem_path(stems_dir, name))}" + + +def stem_names_on_disk(stems_dir: Path) -> list[str]: + """Every name with a stem file in `stems_dir`, sorted, each once.""" + names: set[str] = set() + for ext in STEM_EXTS: + names.update( + p.stem for p in stems_dir.glob(f"*{ext}") if p.is_file() and not p.name.startswith(".") + ) + return sorted(names) + + +def is_stem_audio(path: Path) -> bool: + return path.suffix.lower() in STEM_EXTS + + +@contextlib.contextmanager +def decoded_wav(path: Path) -> Iterator[Path]: + """`path` itself when it is a WAV, else a float32 WAV decoded from it by + ffmpeg in a temporary folder that is removed afterwards. + + For a reader that hands the stem to a third-party library which may not + decode MP3 the way ffmpeg does (the vocal split's audio-separator). Raises + RuntimeError when the decode fails.""" + if path.suffix.lower() == ".wav": + yield path + return + folder = Path(tempfile.mkdtemp(prefix="stemdeck-decode-")) + try: + out = folder / f"{path.stem}.wav" + cmd = [ + ffmpeg_executable(), + "-nostdin", + "-loglevel", + "error", + "-i", + str(path), + "-map", + "0:a:0", + "-c:a", + "pcm_f32le", + "-f", + "wav", + "-y", + str(out), + ] + proc = subprocess.run( # noqa: S603 -- list args, no shell, trusted ffmpeg + cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + timeout=TIMEOUT_FFMPEG, + ) + if proc.returncode != 0 or not out.is_file(): + raise RuntimeError(f"could not decode {path.name}") + yield out + finally: + shutil.rmtree(folder, ignore_errors=True) diff --git a/app/pipeline/backing.py b/app/pipeline/backing.py index 2cbef475..426a789e 100644 --- a/app/pipeline/backing.py +++ b/app/pipeline/backing.py @@ -39,6 +39,7 @@ import soundfile as sf from app.core.config import EXTRA_STEM_NAMES, STEM_NAMES +from app.core.stem_files import stem_exists, stem_path logger = logging.getLogger("stemdeck.backing") @@ -64,12 +65,13 @@ def backing_inputs(stems_dir: Path, remove: str) -> list[Path]: if remove not in REMOVABLE: raise ValueError(f"unknown stem {remove!r}") names = [n for n in STEM_NAMES if n != remove] - paths = [stems_dir / f"{n}.wav" for n in names] + # WAV or MP3, whichever the song keeps (#783); libsndfile reads both. + paths = [stem_path(stems_dir, n) for n in names] present = [p for p in paths if p.is_file()] - vocals_gone = not (stems_dir / "vocals.wav").is_file() + vocals_gone = not stem_exists(stems_dir, "vocals") if remove != "vocals" and vocals_gone: # No vocals.wav but a finished split: its two halves stand in for it. - split = [stems_dir / f"{n}.wav" for n in EXTRA_STEM_NAMES] + split = [stem_path(stems_dir, n) for n in EXTRA_STEM_NAMES] if all(p.is_file() for p in split): present += split return present @@ -226,11 +228,11 @@ def rest_inputs(stems_dir: Path, names: tuple[str, ...]) -> list[Path]: vocals.wav already holds them.""" paths: list[Path] = [] for name in names: - p = stems_dir / f"{name}.wav" + p = stem_path(stems_dir, name) if p.is_file(): paths.append(p) continue - split = [stems_dir / f"{n}.wav" for n in EXTRA_STEM_NAMES] + split = [stem_path(stems_dir, n) for n in EXTRA_STEM_NAMES] if name == "vocals" and all(s.is_file() for s in split): paths += split continue diff --git a/app/pipeline/collect.py b/app/pipeline/collect.py index 5d7cc5dd..49fb5462 100644 --- a/app/pipeline/collect.py +++ b/app/pipeline/collect.py @@ -26,6 +26,7 @@ from app.core.registry import persist as registry_persist from app.core.registry import remove as registry_remove from app.core.registry import set_proc +from app.core.stem_files import is_stem_audio, stem_names_on_disk, stem_path from app.pipeline.audio_stats import scan_stem logger = logging.getLogger("stemdeck.collect") @@ -294,7 +295,7 @@ def merge_stem_peaks( ) for name in [*backfill, *new_names]: - wav = (sources or {}).get(name) or stems_dir / f"{name}.wav" + wav = (sources or {}).get(name) or stem_path(stems_dir, name) if not wav.is_file(): continue try: @@ -332,7 +333,8 @@ def merge_stem_peaks( def _stem_wav_names(stems_dir: Path) -> list[str]: - return sorted(p.stem for p in stems_dir.glob("*.wav") if p.is_file()) + # WAV or MP3 (#783): an MP3-stored song has no WAVs left. + return stem_names_on_disk(stems_dir) def fill_missing_peaks(stems_dir: Path, sources: dict[str, Path] | None = None) -> bool: @@ -353,7 +355,7 @@ def fill_missing_peaks(stems_dir: Path, sources: dict[str, Path] | None = None) if path.is_file(): return True wanted: dict[str, Path] = { - name: stems_dir / f"{name}.wav" for name in _stem_wav_names(stems_dir) + name: stem_path(stems_dir, name) for name in _stem_wav_names(stems_dir) } for name, wav in (sources or {}).items(): if wav.is_file(): @@ -442,7 +444,9 @@ def _has_takes_on_disk(job_dir: Path) -> bool: def _has_stems_on_disk(job_dir: Path) -> bool: stems = job_dir / "stems" try: - return stems.is_dir() and any(p.suffix.lower() == ".wav" for p in stems.iterdir()) + # An MP3-stored song (#783) holds stems too: missing it here would let + # the sweep delete an unregistered folder that still has a song in it. + return stems.is_dir() and any(is_stem_audio(p) for p in stems.iterdir()) except OSError: return True diff --git a/app/pipeline/lyrics_align.py b/app/pipeline/lyrics_align.py index fd63a35a..07ea24f7 100644 --- a/app/pipeline/lyrics_align.py +++ b/app/pipeline/lyrics_align.py @@ -44,6 +44,7 @@ LYRICS_ALIGN_TOLERANCE_SEC, ) from app.core.models import Job +from app.core.stem_files import stem_path from app.pipeline.audio_stats import vocal_envelope logger = logging.getLogger("stemdeck.lyrics") @@ -194,7 +195,7 @@ def estimate_offset(starts: list[float], strength: np.ndarray, hop: float) -> fl def _envelope(stems_dir: Path) -> tuple[float, list[int]] | None: """The vocals stem's envelope: the one kept beside it when it is not older than the stem, else read from the stem now. None without vocals.""" - vocals = stems_dir / "vocals.wav" + vocals = stem_path(stems_dir, "vocals") if not vocals.is_file(): return None kept = stems_dir / _ENVELOPE_FILE diff --git a/app/pipeline/lyrics_retime.py b/app/pipeline/lyrics_retime.py index c9a7c669..a4eefc87 100644 --- a/app/pipeline/lyrics_retime.py +++ b/app/pipeline/lyrics_retime.py @@ -112,6 +112,7 @@ LYRICS_TRANSCRIPT_FILE, LYRICS_USER_SYNCED_MAX_LINES, ) +from app.core.stem_files import stem_path from app.pipeline.lyrics_align import ( _HIT_RISE_DB, _LOUD_PERCENTILE, @@ -1458,12 +1459,29 @@ def _vocals_stamp(job_dir: Path) -> list[int] | None: was heard from. A stem made again (a re-separation writing it in place) no longer matches, and its transcript is heard afresh.""" try: - st = (job_dir / "stems" / "vocals.wav").stat() + st = stem_path(job_dir / "stems", "vocals").stat() except OSError: return None return [st.st_size, st.st_mtime_ns] +def restamp_transcript(job_dir: Path, old_stamp: list[int]) -> None: + """The vocals stem was stored again in another format, same audio (#783): + a transcript heard from it (stamped ``old_stamp``) is moved onto the new + file's stamp, so it is not heard afresh. Anything else is left as it is.""" + from app.pipeline.lyrics_lookup import _write_json_atomic + + path = transcript_path(job_dir) + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return + if not isinstance(data, dict) or data.get("vocals") != old_stamp: + return + data["vocals"] = _vocals_stamp(job_dir) + _write_json_atomic(path, data) + + def save_transcript( job_dir: Path, result: dict[str, Any] | None = None, @@ -1820,7 +1838,7 @@ def _hear( when the worker failed. Raises RetimeFailed without vocals, JobCancelled.""" from app.pipeline.transcribe import _device, _run_worker, _spawn_worker_cmd - vocals = job_dir / "stems" / "vocals.wav" + vocals = stem_path(job_dir / "stems", "vocals") if not vocals.is_file(): raise RetimeFailed("no vocals stem") cmd = _spawn_worker_cmd(vocals, _device(job)) diff --git a/app/pipeline/runner.py b/app/pipeline/runner.py index f6ed889d..e0b05bab 100644 --- a/app/pipeline/runner.py +++ b/app/pipeline/runner.py @@ -40,6 +40,7 @@ from app.pipeline.lyrics_retime import retime_after_separation from app.pipeline.sections import detect_sections from app.pipeline.separate import separate +from app.pipeline.stem_storage import store_stems from app.pipeline.transcribe import transcribe_lyrics logger = logging.getLogger("stemdeck.pipeline") @@ -324,12 +325,30 @@ def _run_with_band_lookup(job: Job, source: Path, job_dir: Path) -> None: _lap(job, "retime", mark) +def _store_stems(job: Job, job_dir: Path) -> None: + """WAV or MP3 storage of the stems (#783). Last, once no stage reads the + stems any more: separation and every analysis ran on the WAVs. Never + fails the song; store_stems keeps the WAVs on any failure, and the guard + is for a future change there, as for the beat grid.""" + _check_cancel(job) + mark = time.monotonic() + try: + store_stems(job, job_dir) + except JobCancelled: + raise + except Exception: + logger.exception("stem storage stage failed for job %s; stems kept as WAV", job.id) + _set(job, stem_format="wav", stem_mp3_kbps=None) + _lap(job, "store", mark) + + def _run_blocking(job: Job, url: str, job_dir: Path) -> None: _check_cancel(job) mark = time.monotonic() source = download(job, url, job_dir) _lap(job, "download", mark) _run_with_band_lookup(job, source, job_dir) + _store_stems(job, job_dir) def _run_local_blocking(job: Job, source_path: Path, job_dir: Path) -> None: @@ -338,6 +357,7 @@ def _run_local_blocking(job: Job, source_path: Path, job_dir: Path) -> None: source = _prepare_local_source(job, source_path, job_dir) _lap(job, "prepare", mark) _run_with_band_lookup(job, source, job_dir) + _store_stems(job, job_dir) def _write_metadata(job: Job, job_dir: Path) -> None: @@ -367,6 +387,8 @@ def _write_metadata(job: Job, job_dir: Path) -> None: "compute_device": job.compute_device, "gpu_fallback": job.gpu_fallback, "stage_timings": job.stage_timings, + "stem_format": job.stem_format, + "stem_mp3_kbps": job.stem_mp3_kbps, } try: (job_dir / "metadata.json").write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8") diff --git a/app/pipeline/stem_storage.py b/app/pipeline/stem_storage.py new file mode 100644 index 00000000..05946fce --- /dev/null +++ b/app/pipeline/stem_storage.py @@ -0,0 +1,489 @@ +"""The last pipeline stage: keep a finished song's stems as WAV or MP3 (#783). + +Separation and every analysis stage run on WAV exactly as they always did. +Only once the song has nothing left to read from its stems does this stage +look at the Stem storage format setting (read now, so a change applies to the +next song without a restart): + +- WAV: nothing is touched. The job records "wav". +- MP3: every stem file in stems/ (the six stems, original, mix) is encoded by + ffmpeg's libmp3lame at a constant bitrate into a temporary file, and each + one is verified: ffmpeg and libsndfile, the two decoders the server reads + stems with, must both decode it to exactly the WAV's frame count, and the + start of the decoded audio must line up with the WAV's. Only when every stem + passed are the MP3s moved into place (rename, so never half-written) and the + WAVs deleted. + +Any failure (ffmpeg without libmp3lame, a full disk, a stall, an output that +does not decode or does not line up) removes the temporary files, keeps every +WAV and records the song as "wav". The song finishes either way: this stage +never fails a job. A cancel raises JobCancelled, after the same cleanup. + +Sample accuracy: libmp3lame adds an encoder delay and padding. ffmpeg writes +both into the LAME/Xing header of the first frame (-write_xing 1, its +default, made explicit below), and ffmpeg, libsndfile (mpg123) and Chromium's +decodeAudioData all trim by it, so a decoded MP3 starts on the same sample as +the WAV and has the same length. tests/test_stem_storage.py measures it. + +`original.wav` here is the complement backing track (make_original_track), +a stem like the others, so it is stored the same way. The song's own source is +not in stems/ and is never touched: an upload's kept source. stays as it +was uploaded. +""" + +from __future__ import annotations + +import logging +import os +import subprocess +import threading +import time +import uuid +from pathlib import Path + +from app.core.config import ( + EXTRA_STEM_NAMES, + STEM_NAMES, + TIMEOUT_FFMPEG, + TIMEOUT_STEM_TRANSCODE_STALL, + ffmpeg_executable, +) +from app.core.models import Job, JobCancelled, _set +from app.core.registry import set_proc +from app.core.settings import get_stem_format, get_stem_mp3_kbps +from app.core.stem_files import stem_url + +logger = logging.getLogger("stemdeck.pipeline") + +# Every stem file a finished song can hold. +STORED_NAMES: tuple[str, ...] = (*STEM_NAMES, *EXTRA_STEM_NAMES, "original", "mix") +# The alignment check compares the loudest block of this many seconds of each +# stem, found in one sequential pass over both files (no seeking, constant +# memory). After the exact length check it catches an output that decodes to +# the right length but not to the right place. +_ALIGN_BLOCK_SEC = 10 +# A block quieter than this RMS is near silence: a stem that never plays has +# nothing to line up, so only its length is checked. +_ALIGN_MIN_RMS = 0.01 +# The lags searched either side of 0, in samples: past the largest offset a +# decoder that ignored the LAME header would add (1105 = 576 encoder delay + +# 529 decoder delay at 44.1 kHz). +_ALIGN_MAX_LAG = 4096 +# A correct encode correlates at about 0.99 at lag 0 even at 128 kbps. +# Refused: a block whose lag-0 correlation is under the floor (not this audio), +# or clearly below the best one within the searched lags (shifted). The ratio +# leaves room for steady tones, whose correlation repeats every period. +_ALIGN_MIN_CORR = 0.5 +_ALIGN_MIN_RATIO = 0.98 +# The encoder's output is shown to nobody. Kept for the log line on failure. +_TAIL_LINES = 15 + + +class StemStorageError(RuntimeError): + """The MP3s could not be made or did not verify. The WAVs stay.""" + + +def _encode_cmd(src: Path, dest: Path, kbps: int) -> list[str]: + """One stem to CBR MP3. Module-level seam: tests swap in a stub encoder. + + `-progress pipe:1` prints a status block about once a second, which is + what the stall watchdog listens to.""" + return [ + ffmpeg_executable(), + "-nostdin", + "-hide_banner", + "-loglevel", + "error", + "-nostats", + "-progress", + "pipe:1", + "-i", + str(src), + "-map", + "0:a:0", + "-c:a", + "libmp3lame", + "-b:a", + f"{kbps}k", + # The LAME header carries the encoder delay and padding, which is + # what makes the decoded MP3 sample-aligned with the WAV. + "-write_xing", + "1", + "-f", + "mp3", + "-y", + str(dest), + ] + + +def _encode(job: Job, src: Path, dest: Path, kbps: int) -> None: + """Run the encoder for one stem, registered for cancel, under a stall + watchdog. Raises JobCancelled or StemStorageError.""" + try: + proc = subprocess.Popen( # noqa: S603 -- list args, no shell, trusted ffmpeg + _encode_cmd(src, dest, kbps), + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + ) + except OSError as exc: + raise StemStorageError("the MP3 encoder could not be started") from exc + set_proc(job.id, proc) + last_output = [time.monotonic()] + stalled = threading.Event() + done = threading.Event() + err_tail: list[str] = [] + + def _watchdog() -> None: + while not done.wait(timeout=0.2): + if job.cancel_requested: + proc.terminate() + return + if time.monotonic() - last_output[0] > TIMEOUT_STEM_TRANSCODE_STALL: + logger.warning( + "[%s] stem encode silent for %ss; stopping it", + job.id, + TIMEOUT_STEM_TRANSCODE_STALL, + ) + stalled.set() + proc.terminate() + return + + def _drain_stderr() -> None: + assert proc.stderr is not None + for line in proc.stderr: + err_tail.append(line.rstrip()) + del err_tail[:-_TAIL_LINES] + + watch = threading.Thread(target=_watchdog, daemon=True) + drain = threading.Thread(target=_drain_stderr, daemon=True) + watch.start() + drain.start() + try: + assert proc.stdout is not None + for _line in proc.stdout: + last_output[0] = time.monotonic() + try: + proc.wait(timeout=TIMEOUT_FFMPEG) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + raise StemStorageError("the MP3 encoder did not exit") from None + finally: + done.set() + set_proc(job.id, None) + watch.join(timeout=2) + drain.join(timeout=2) + if proc.poll() is None: + proc.terminate() + try: + proc.communicate(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + if job.cancel_requested: + raise JobCancelled() + if stalled.is_set(): + raise StemStorageError("the MP3 encoder stalled") + if proc.returncode != 0: + detail = " | ".join(err_tail[-3:]) or f"exit {proc.returncode}" + raise StemStorageError(f"the MP3 encoder failed: {detail}") + if not dest.is_file() or dest.stat().st_size == 0: + raise StemStorageError("the MP3 encoder wrote nothing") + + +def _ffmpeg_frames(job: Job, path: Path, channels: int) -> int: + """Frames ffmpeg decodes `path` to: the whole file, counted as it streams.""" + cmd = [ + ffmpeg_executable(), + "-nostdin", + "-loglevel", + "error", + "-i", + str(path), + "-map", + "0:a:0", + "-ac", + str(channels), + "-c:a", + "pcm_s16le", + "-f", + "s16le", + "-", + ] + proc = subprocess.Popen( # noqa: S603 -- list args, no shell, trusted ffmpeg + cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL + ) + set_proc(job.id, proc) + total = 0 + try: + assert proc.stdout is not None + while chunk := proc.stdout.read(1 << 20): + total += len(chunk) + proc.wait(timeout=TIMEOUT_FFMPEG) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + raise StemStorageError("the MP3 check did not finish") from None + finally: + set_proc(job.id, None) + if job.cancel_requested: + raise JobCancelled() + if proc.returncode != 0: + raise StemStorageError("the MP3 does not decode") + return total // (2 * channels) + + +def verify_mp3(job: Job, wav: Path, mp3: Path) -> None: + """Raise StemStorageError unless `mp3` decodes to `wav`'s exact length in + both ffmpeg and libsndfile, at its rate and channel count, and its + loudest block lines up with the WAV's (cross-correlation peak at lag 0).""" + import numpy as np + import soundfile as sf + + try: + ref = sf.info(str(wav)) + got = sf.info(str(mp3)) + except Exception as exc: # libsndfile raises its own types + raise StemStorageError("the MP3 cannot be read by libsndfile") from exc + if (got.samplerate, got.channels) != (ref.samplerate, ref.channels): + raise StemStorageError("the MP3 has another rate or channel count") + if got.frames != ref.frames: + raise StemStorageError(f"libsndfile decodes {got.frames} frames, the WAV has {ref.frames}") + ff = _ffmpeg_frames(job, mp3, ref.channels) + if ff != ref.frames: + raise StemStorageError(f"ffmpeg decodes {ff} frames, the WAV has {ref.frames}") + try: + a, b = _loudest_block(wav, mp3, ref.samplerate * _ALIGN_BLOCK_SEC) + except Exception as exc: + raise StemStorageError("the MP3 cannot be read back") from exc + if a is None or float(np.sqrt(np.mean(a * a))) < _ALIGN_MIN_RMS: + return + zero, best, lag = _correlation(a, b, _ALIGN_MAX_LAG) + if zero < _ALIGN_MIN_CORR or zero < _ALIGN_MIN_RATIO * best: + raise StemStorageError( + f"the MP3 does not line up with the WAV (lag {lag}, {zero:.3f} at 0 against {best:.3f})" + ) + + +def _loudest_block(wav: Path, mp3: Path, size: int): + """Channel 0 of the loudest `size`-frame block of `wav`, and the same + block of `mp3`, read side by side from the start. (None, None) for an + empty file.""" + import numpy as np + import soundfile as sf + + best_rms, best = -1.0, (None, None) + pairs = zip( + sf.blocks(str(wav), blocksize=size, dtype="float32", always_2d=True), + sf.blocks(str(mp3), blocksize=size, dtype="float32", always_2d=True), + strict=True, + ) + for a, b in pairs: + if len(a) != len(b): + raise StemStorageError("the MP3 reads back at another length") + a0 = a[:, 0].astype(np.float64) + rms = float(np.sqrt(np.mean(a0 * a0))) if a0.size else 0.0 + if rms > best_rms: + best_rms, best = rms, (a0, b[:, 0].astype(np.float64)) + return best + + +def _correlation(a, b, max_lag: int) -> tuple[float, float, int]: + """Normalised cross-correlation of b against a: its value at lag 0, its + best value within +-max_lag, and that lag.""" + import numpy as np + + n = len(a) + nfft = 1 << (2 * n).bit_length() + c = np.fft.irfft(np.fft.rfft(b, nfft) * np.conj(np.fft.rfft(a, nfft)), nfft) + denom = float(np.sqrt(np.sum(a * a) * np.sum(b * b))) + if denom <= 0: + return 0.0, 0.0, 0 + k = min(max_lag, n - 1) + window = np.concatenate([c[-k:], c[: k + 1]]) / denom if k else c[:1] / denom + i = int(np.argmax(window)) + return float(c[0] / denom), float(window[i]), i - k + + +def _stamp(path: Path) -> list[int] | None: + try: + st = path.stat() + except OSError: + return None + return [st.st_size, st.st_mtime_ns] + + +def _keep_derived_files(job_dir: Path, vocals_wav_stamp: list[int] | None) -> None: + """Files kept beside the stems that are checked against the vocals stem's + size and time: the same audio is now in another file, so they stay valid + and are not recomputed. Best effort.""" + stems_dir = job_dir / "stems" + vocals_mp3 = stems_dir / "vocals.mp3" + if vocals_wav_stamp is None or not vocals_mp3.is_file(): + return + try: + from app.pipeline.lyrics_retime import restamp_transcript + + restamp_transcript(job_dir, vocals_wav_stamp) + except Exception: + logger.info("[%s] could not keep the transcript's stamp", job_dir.name, exc_info=True) + envelope = stems_dir / "vocal_envelope.json" + try: + # Kept while it is not older than the stem (lyrics_align._envelope, + # GET /vocal-envelope): it was made from the WAV, and still holds. + if envelope.is_file() and envelope.stat().st_mtime_ns >= vocals_wav_stamp[1]: + os.utime(envelope) + except OSError: + logger.info("[%s] could not keep the vocal envelope", job_dir.name, exc_info=True) + + +def _record(job: Job, fmt: str, kbps: int | None) -> None: + _set(job, stem_format=fmt, stem_mp3_kbps=kbps) + + +def _refresh_urls(job: Job, stems_dir: Path) -> None: + """Point the job's stem URLs (and the mix link, when it is a stem file) at + the files now on disk.""" + stems = [] + for entry in job.stems: + name = entry.get("name", "") + stems.append({**entry, "url": stem_url(job.id, stems_dir, name)} if name else entry) + prefix = f"/api/jobs/{job.id}/stems/" + mix_url = job.mix_url + if mix_url and mix_url.startswith(prefix): + name = Path(mix_url[len(prefix) :]).stem + if name in STORED_NAMES: + mix_url = stem_url(job.id, stems_dir, name) + _set(job, stems=stems, mix_url=mix_url) + + +def store_stems(job: Job, job_dir: Path) -> None: + """Keep the song's stems in the format the setting asks for now. Never + raises but for a cancel; see the module docstring.""" + if job.cancel_requested: + raise JobCancelled() + fmt = get_stem_format() + if fmt != "mp3": + _record(job, "wav", None) + return + kbps = get_stem_mp3_kbps() + stems_dir = job_dir / "stems" + wavs = [p for p in (stems_dir / f"{n}.wav" for n in STORED_NAMES) if p.is_file()] + if not wavs: + _record(job, "wav", None) + return + + _set(job, stage="Compressing stems...") + started = time.monotonic() + made: list[tuple[Path, Path]] = [] + try: + for wav in wavs: + if job.cancel_requested: + raise JobCancelled() + # Dot-prefixed and .tmp-marked: no reader takes it for a stem. + tmp = stems_dir / f".{wav.stem}.{uuid.uuid4().hex}.tmp.mp3" + made.append((wav, tmp)) + _encode(job, wav, tmp, kbps) + verify_mp3(job, wav, tmp) + except JobCancelled: + _discard(made) + raise + except Exception as exc: + # StemStorageError, or anything unexpected (OSError on a full disk, + # libsndfile): the WAVs are all still here, so the song is a WAV song. + _discard(made) + logger.warning("[%s] stems kept as WAV: MP3 storage failed: %s", job.id, exc) + _record(job, "wav", None) + return + + vocals_stamp = _stamp(stems_dir / "vocals.wav") + moved: list[Path] = [] + try: + for wav, tmp in made: + final = wav.with_suffix(".mp3") + os.replace(tmp, final) + moved.append(final) + except OSError as exc: + # Every WAV is still here and wins over an MP3 of the same name + # (stem_files.stem_path); the moved ones are removed so nothing + # half-done is left behind. + _discard(made) + for final in moved: + final.unlink(missing_ok=True) + logger.warning("[%s] stems kept as WAV: could not move the MP3s: %s", job.id, exc) + _record(job, "wav", None) + return + + before = sum(_size(w) for w in wavs) + after = sum(_size(w.with_suffix(".mp3")) for w in wavs) + for wav in wavs: + try: + wav.unlink() + except OSError: + # Harmless: a WAV left behind is still a valid stem and wins over + # its MP3. Logged, so the disk it costs is not a mystery. + logger.warning("[%s] could not delete %s after storing it as MP3", job.id, wav.name) + _keep_derived_files(job_dir, vocals_stamp) + _refresh_urls(job, stems_dir) + _record(job, "mp3", kbps) + logger.info( + "[%s] stems stored as MP3 %dk: %d files, %.1f MB -> %.1f MB in %.1fs", + job.id, + kbps, + len(wavs), + before / 1e6, + after / 1e6, + time.monotonic() - started, + ) + + +def _size(path: Path) -> int: + try: + return path.stat().st_size + except OSError: + return 0 + + +def _discard(made: list[tuple[Path, Path]]) -> None: + for _wav, tmp in made: + try: + tmp.unlink(missing_ok=True) + except OSError: + logger.debug("could not remove %s", tmp, exc_info=True) + + +def store_split_stems(job: Job, stems_dir: Path, names: list[str]) -> None: + """The on-demand lead/backing split writes WAVs into a song that may be + stored as MP3. They are stored the way the song is, with the same + verification, and stay WAV on any failure. Never raises.""" + if job.stem_format != "mp3" or not job.stem_mp3_kbps: + return + made: list[tuple[Path, Path]] = [] + try: + for name in names: + wav = stems_dir / f"{name}.wav" + if not wav.is_file(): + continue + tmp = stems_dir / f".{name}.{uuid.uuid4().hex}.tmp.mp3" + made.append((wav, tmp)) + _encode(job, wav, tmp, int(job.stem_mp3_kbps)) + verify_mp3(job, wav, tmp) + for wav, tmp in made: + os.replace(tmp, wav.with_suffix(".mp3")) + except Exception as exc: + _discard(made) + for wav, _tmp in made: + if wav.is_file(): + # Never both for long: the WAV wins anyway, the MP3 is clutter. + wav.with_suffix(".mp3").unlink(missing_ok=True) + logger.warning("[%s] split stems kept as WAV: %s", job.id, exc) + return + for wav, _tmp in made: + try: + wav.unlink() + except OSError: + logger.warning("[%s] could not delete %s after storing it as MP3", job.id, wav.name) diff --git a/app/pipeline/transcribe.py b/app/pipeline/transcribe.py index 054cecfd..bbb8fa59 100644 --- a/app/pipeline/transcribe.py +++ b/app/pipeline/transcribe.py @@ -62,6 +62,7 @@ from app.core.models import Job, JobCancelled, _set from app.core.registry import set_proc from app.core.settings import lyrics_mending_enabled, transcribe_lyrics_enabled +from app.core.stem_files import stem_exists, stem_path from app.pipeline.lyrics_lookup import lyrics_path, read_lyrics, write_lyrics from app.pipeline.lyrics_repair import looks_stripped, mend_from_transcript, words_of from app.pipeline.lyrics_retime import save_transcript @@ -144,7 +145,7 @@ def skip_reason(job: Job, job_dir: Path) -> str | None: def _vocals_reason(job: Job, job_dir: Path) -> str | None: """Why this job's vocals are not worth transcribing, or None.""" - if not (job_dir / "stems" / "vocals.wav").is_file(): + if not stem_exists(job_dir / "stems", "vocals"): return "no vocals stem" presence = (job.stem_presence or {}).get("vocals") if isinstance(presence, int) and presence < TRANSCRIBE_MIN_VOCAL_PRESENCE: @@ -483,7 +484,7 @@ def transcribe_lyrics(job: Job, job_dir: Path) -> bool: logger.info("[%s] lyrics transcription skipped: %s", job.id, reason) return False _set(job, stage=_STAGE) - vocals = job_dir / "stems" / "vocals.wav" + vocals = stem_path(job_dir / "stems", "vocals") result = _run_worker(job, _spawn_worker_cmd(vocals, _device(job))) if result is None: return False @@ -571,7 +572,7 @@ def mend_lyrics(job: Job, job_dir: Path) -> bool: return False started = time.monotonic() _set(job, stage=_STAGE) - vocals = job_dir / "stems" / "vocals.wav" + vocals = stem_path(job_dir / "stems", "vocals") result = _run_worker(job, _spawn_worker_cmd(vocals, _device(job)) + _language_gate()) if result is None: return False diff --git a/app/pipeline/vocal_split.py b/app/pipeline/vocal_split.py index a3d1ebf8..84310a37 100644 --- a/app/pipeline/vocal_split.py +++ b/app/pipeline/vocal_split.py @@ -25,6 +25,7 @@ from app.core.models import Job from app.core.registry import set_proc from app.core.settings import get_demucs_device +from app.core.stem_files import decoded_wav, stem_path from app.pipeline.errors import SeparationError logger = logging.getLogger("stemdeck.pipeline") @@ -50,9 +51,23 @@ def split_vocals(job: Job, stems_dir: Path) -> list[str]: responsible for catching failures and leaving the job's base stems untouched -- this function never mutates anything but stems_dir's contents, and never touches vocals.wav itself.""" - vocals_path = stems_dir / "vocals.wav" + vocals_path = stem_path(stems_dir, "vocals") if not vocals_path.is_file(): raise SeparationError("vocals.wav not found -- job has no vocals stem to split") + # audio-separator gets a WAV whatever the song keeps (#783): an MP3 vocals + # stem is decoded by ffmpeg first, so the model reads exactly the samples + # every other reader does. + try: + with decoded_wav(vocals_path) as wav_in: + return _split(job, wav_in, stems_dir) + except SeparationError: + raise + except RuntimeError as exc: + raise SeparationError("could not read the vocals stem") from exc + + +def _split(job: Job, vocals_path: Path, stems_dir: Path) -> list[str]: + """split_vocals on a WAV vocals file.""" device = get_demucs_device() env = os.environ.copy() diff --git a/docs/configuration.md b/docs/configuration.md index 914b0f86..4075cff2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -19,6 +19,8 @@ | `STEMDECK_MAX_DURATION_SEC` | `1200` | Reject audio longer than this (seconds). | | `STEMDECK_JOB_TTL_SECONDS` | `86400` | How long to keep job dirs on disk. | | `STEMDECK_MAX_PENDING_JOBS` | `3` | Max queued jobs before returning 503. | +| `STEMDECK_STEM_FORMAT` | `mp3` | How new songs' stems are stored: `mp3` or `wav`. Songs already in the library keep their format. A choice saved in Settings > Library wins over this. | +| `STEMDECK_STEM_MP3_KBPS` | `128` | MP3 stem bitrate: `128`, `192` or `256` (constant bitrate). Used only with `mp3`. | | `STEMDECK_TIMEOUT_FFMPEG` | `300` | ffmpeg subprocess timeout (seconds). | | `STEMDECK_TIMEOUT_ANALYZE` | `120` | Audio analysis timeout (seconds). | | `STEMDECK_TIMEOUT_DEMUCS_STALL` | `1800` | Kill Demucs if no output for this many seconds. | diff --git a/templates/stemdeck.xml b/templates/stemdeck.xml index b092e999..88137e8e 100644 --- a/templates/stemdeck.xml +++ b/templates/stemdeck.xml @@ -51,6 +51,10 @@ 3 + + + + 99 100 diff --git a/tests/test_identify_regressions.py b/tests/test_identify_regressions.py index 9a8f0bc3..7d0e7e48 100644 --- a/tests/test_identify_regressions.py +++ b/tests/test_identify_regressions.py @@ -86,6 +86,16 @@ } +@pytest.fixture(autouse=True) +def _wav_storage(monkeypatch): + """The stubbed stages write placeholder bytes, not audio, so the MP3 + storage step (#783, the default) could only ever take its failure path + here. These tests hold the pipeline's contract from before identification; + WAV storage is that contract. MP3 storage has its own tests + (test_stem_storage.py).""" + monkeypatch.setenv("STEMDECK_STEM_FORMAT", "wav") + + @pytest.fixture(autouse=True) def _isolate_registry(): _jobs.clear() diff --git a/tests/test_stem_storage.py b/tests/test_stem_storage.py new file mode 100644 index 00000000..d5e8f014 --- /dev/null +++ b/tests/test_stem_storage.py @@ -0,0 +1,633 @@ +"""Stem storage format (#783): WAV or MP3 stems, one resolver, sample accuracy. + +- The setting: MP3 128 by default, env seeds it, a saved choice wins, a + whitelist guards both values, a non-object body changes nothing. +- The storage stage: MP3 stems replace the WAVs only after every one decoded + to the WAV's exact length and lined up; every failure keeps the WAVs and + records the song as WAV; a cancel cleans up and raises; WAV touches nothing. +- Sample accuracy: an MP3 stem decodes to the WAV's samples in length and + position (lag 0 by cross-correlation) in ffmpeg and libsndfile, at every + offered bitrate, and an export from MP3 stems lines up with one from WAVs. +- Readers: the stem URL serves the right file and type for either format, the + zip, the region, the mixdown, the backing, the take frame count, recovery + and the sweep all find MP3 stems, and a WAV song with a cached download MP3 + still reads its WAV. +""" + +from __future__ import annotations + +import io +import json +import os +import sys +import threading +import time +import zipfile +from pathlib import Path + +import httpx +import numpy as np +import pytest +import soundfile as sf + +from app.api import stems as stems_mod +from app.core import settings +from app.core.models import Job, JobCancelled +from app.core.registry import _jobs +from app.core.stem_files import stem_names_on_disk, stem_path, stem_url +from app.pipeline import stem_storage +from tests.ffmpeg_probe import skip_without_ffmpeg + +SR = 44100 +# The signal every alignment test uses: broadband noise bursts on a tone, with +# hard clicks once a second, so the cross-correlation has one sharp peak. +SECONDS = 6.0 + + +def _signal(seed: int = 7, seconds: float = SECONDS) -> np.ndarray: + rng = np.random.default_rng(seed) + n = int(SR * seconds) + t = np.arange(n) / SR + env = np.where((t % 0.5) < 0.05, 1.0, 0.25)[:, None] + x = rng.standard_normal((n, 2)) * 0.08 * env + 0.2 * np.sin(2 * np.pi * 220 * t)[:, None] + for k in range(1, int(seconds)): + x[k * SR, :] = 0.9 + return x.clip(-1, 1) + + +def _lag(a: np.ndarray, b: np.ndarray) -> int: + """Lag of b against a in samples (positive: b is late), by FFT xcorr.""" + m = min(len(a), len(b)) + a, b = a[:m], b[:m] + nfft = 1 << (2 * m).bit_length() + c = np.fft.irfft(np.fft.rfft(b, nfft) * np.conj(np.fft.rfft(a, nfft)), nfft) + c = np.concatenate([c[-(m - 1) :], c[:m]]) + return int(np.argmax(c)) - (m - 1) + + +def _ffmpeg_decode(path: Path) -> np.ndarray: + import subprocess + + from app.core.config import ffmpeg_executable + + out = subprocess.run( + [ffmpeg_executable(), "-v", "error", "-i", str(path), "-f", "f32le", "-ac", "2", "-"], + capture_output=True, + check=True, + ).stdout + return np.frombuffer(out, dtype=np.float32).reshape(-1, 2) + + +def _job(jobs_dir: Path, job_id: str, names=("vocals", "drums", "bass"), seconds=SECONDS) -> Job: + job = Job(id=job_id, status="done", title="Storage", duration_sec=seconds) + stems = jobs_dir / job_id / "stems" + stems.mkdir(parents=True, exist_ok=True) + for i, name in enumerate(names): + sf.write(stems / f"{name}.wav", _signal(seed=i + 1, seconds=seconds), SR, subtype="PCM_16") + job.stems.append({"name": name, "url": f"/api/jobs/{job_id}/stems/{name}.wav"}) + _jobs[job.id] = job + return job + + +@pytest.fixture(autouse=True) +def _isolate_registry(): + _jobs.clear() + yield + _jobs.clear() + + +@pytest.fixture +def jobs_dir(tmp_path, monkeypatch): + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_DIR", tmp_path / "cache" / "mixdown") + monkeypatch.setattr(stems_mod, "_CLICK_CACHE_DIR", tmp_path / "cache" / "click") + monkeypatch.setattr(stems_mod, "_MASTER_PLAN_DIR", tmp_path / "cache" / "master") + monkeypatch.delenv("STEMDECK_STEM_FORMAT", raising=False) + monkeypatch.delenv("STEMDECK_STEM_MP3_KBPS", raising=False) + from app.core import config + + return config.JOBS_DIR + + +@pytest.fixture +def client(jobs_dir): + from app.main import app + + return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") + + +def _store(job: Job, jobs_dir: Path, fmt: str = "mp3", kbps: int = 128) -> Path: + settings.set_stem_storage(fmt, kbps) + stem_storage.store_stems(job, jobs_dir / job.id) + return jobs_dir / job.id / "stems" + + +# ── The setting ── + + +def test_default_is_mp3_128_with_nothing_saved(jobs_dir): + assert settings.get_stem_format() == "mp3" + assert settings.get_stem_mp3_kbps() == 128 + + +def test_env_seeds_the_default_and_a_saved_choice_wins(jobs_dir, monkeypatch): + monkeypatch.setenv("STEMDECK_STEM_FORMAT", "wav") + monkeypatch.setenv("STEMDECK_STEM_MP3_KBPS", "256") + assert (settings.get_stem_format(), settings.get_stem_mp3_kbps()) == ("wav", 256) + settings.set_stem_storage("mp3", 192) + assert (settings.get_stem_format(), settings.get_stem_mp3_kbps()) == ("mp3", 192) + + +@pytest.mark.parametrize("fmt", ["flac", "MP4", "", "../wav"]) +def test_env_outside_the_whitelist_falls_back_to_the_default(jobs_dir, monkeypatch, fmt): + monkeypatch.setenv("STEMDECK_STEM_FORMAT", fmt) + monkeypatch.setenv("STEMDECK_STEM_MP3_KBPS", "320") + assert (settings.get_stem_format(), settings.get_stem_mp3_kbps()) == ("mp3", 128) + + +async def test_settings_endpoint_reads_and_saves_both(client): + async with client as c: + r = await c.get("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/api/settings") + assert (r.json()["stem_format"], r.json()["stem_mp3_kbps"]) == ("mp3", 128) + r = await c.post("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/api/settings", json={"stem_format": "wav"}) + assert r.status_code == 200 + assert r.json()["stem_format"] == "wav" + r = await c.post("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/api/settings", json={"stem_format": "mp3", "stem_mp3_kbps": 256}) + assert (r.json()["stem_format"], r.json()["stem_mp3_kbps"]) == ("mp3", 256) + + +@pytest.mark.parametrize( + "body", + [ + {"stem_format": "flac"}, + {"stem_format": 1}, + {"stem_mp3_kbps": 320}, + {"stem_mp3_kbps": "128"}, + {"stem_mp3_kbps": True}, + {"stem_format": "wav", "stem_mp3_kbps": 64}, + ], +) +async def test_settings_endpoint_rejects_values_outside_the_whitelist(client, body): + async with client as c: + r = await c.post("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/api/settings", json=body) + assert r.status_code == 422 + detail = r.json()["detail"] + assert "must be one of" in detail + # Never echoes what was sent. + assert "flac" not in detail and "320" not in detail and "64" not in detail + # One bad value changes nothing, not even the good one beside it. + r = await c.get("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/api/settings") + assert (r.json()["stem_format"], r.json()["stem_mp3_kbps"]) == ("mp3", 128) + + +@pytest.mark.parametrize("raw", ["[1, 2]", '"mp3"', "null", "not json"]) +async def test_settings_endpoint_ignores_a_body_that_is_not_an_object(client, raw): + async with client as c: + r = await c.post("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/api/settings", content=raw, headers={"Content-Type": "application/json"}) + assert r.status_code == 200 + assert (r.json()["stem_format"], r.json()["stem_mp3_kbps"]) == ("mp3", 128) + + +def test_the_stage_reads_the_setting_when_the_job_finishes(jobs_dir): + """No restart: a change applies to the next song that finishes.""" + first = _job(jobs_dir, "a1b2c3d4e501", names=("vocals",)) + _store(first, jobs_dir, "wav") + second = _job(jobs_dir, "a1b2c3d4e502", names=("vocals",)) + settings.set_stem_storage("mp3", 192) + stem_storage.store_stems(second, jobs_dir / second.id) + assert (first.stem_format, first.stem_mp3_kbps) == ("wav", None) + assert (second.stem_format, second.stem_mp3_kbps) == ("mp3", 192) + + +# ── The resolver ── + + +def test_resolver_prefers_wav_then_mp3_then_the_missing_wav_path(tmp_path): + (tmp_path / "vocals.wav").write_bytes(b"w") + (tmp_path / "vocals.mp3").write_bytes(b"m") # a WAV song's download cache + (tmp_path / "drums.mp3").write_bytes(b"m") + (tmp_path / ".bass.abc.tmp.mp3").write_bytes(b"t") # a storage temp file + assert stem_path(tmp_path, "vocals").name == "vocals.wav" + assert stem_path(tmp_path, "drums").name == "drums.mp3" + assert stem_path(tmp_path, "bass") == tmp_path / "bass.wav" + assert not stem_path(tmp_path, "bass").is_file() + assert stem_names_on_disk(tmp_path) == ["drums", "vocals"] + assert stem_url("abcdef123456", tmp_path, "drums") == "/api/jobs/abcdef123456/stems/drums.mp3" + assert stem_url("abcdef123456", tmp_path, "vocals") == "/api/jobs/abcdef123456/stems/vocals.wav" + + +# ── The stage: success ── + + +@pytest.mark.parametrize("kbps", [128, 192, 256]) +def test_mp3_storage_is_sample_aligned_in_ffmpeg_and_libsndfile(jobs_dir, kbps): + """The alignment guard. Fails if encoder delay or padding ever leaks into + a decoded stem: length must match exactly, and the cross-correlation + peak must sit at lag 0, in both server-side decoders.""" + skip_without_ffmpeg() + job = _job(jobs_dir, "a1b2c3d4e5f6", names=("vocals",)) + wav = jobs_dir / job.id / "stems" / "vocals.wav" + ref, _ = sf.read(wav) + stems = _store(job, jobs_dir, "mp3", kbps) + mp3 = stems / "vocals.mp3" + assert mp3.is_file() and not wav.exists() + via_ffmpeg = _ffmpeg_decode(mp3) + via_sndfile, rate = sf.read(mp3) + assert rate == SR + for decoded in (via_ffmpeg, via_sndfile): + assert len(decoded) == len(ref) + assert _lag(ref[:, 0], decoded[:, 0]) == 0 + assert _lag(ref[:, 1], decoded[:, 1]) == 0 + + +def test_mp3_storage_replaces_every_stem_and_records_the_job(jobs_dir): + skip_without_ffmpeg() + job = _job(jobs_dir, "a1b2c3d4e5f7", names=("vocals", "drums", "bass", "original", "mix")) + job.mix_url = f"/api/jobs/{job.id}/stems/mix.wav" + before = sum(p.stat().st_size for p in (jobs_dir / job.id / "stems").glob("*.wav")) + stems = _store(job, jobs_dir, "mp3", 128) + assert sorted(p.name for p in stems.iterdir()) == sorted( + f"{n}.mp3" for n in ("vocals", "drums", "bass", "original", "mix") + ) + after = sum(p.stat().st_size for p in stems.glob("*.mp3")) + assert after < before / 5 + assert (job.stem_format, job.stem_mp3_kbps) == ("mp3", 128) + assert all(s["url"].endswith(".mp3") for s in job.stems) + assert job.mix_url == f"/api/jobs/{job.id}/stems/mix.mp3" + + +def test_wav_storage_touches_nothing(jobs_dir): + job = _job(jobs_dir, "a1b2c3d4e5f8", names=("vocals", "drums")) + stems = jobs_dir / job.id / "stems" + snapshot = {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in stems.iterdir()} + urls = [dict(s) for s in job.stems] + _store(job, jobs_dir, "wav") + assert {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in stems.iterdir()} == snapshot + assert job.stems == urls + assert (job.stem_format, job.stem_mp3_kbps) == ("wav", None) + + +def test_nothing_final_exists_until_every_stem_verified(jobs_dir, monkeypatch): + """Atomic: no .mp3 appears, and no WAV goes, before the last + stem passed its check.""" + skip_without_ffmpeg() + job = _job(jobs_dir, "a1b2c3d4e5f9", names=("vocals", "drums", "bass")) + stems = jobs_dir / job.id / "stems" + real = stem_storage.verify_mp3 + seen: list[tuple[list[str], list[str]]] = [] + + def watching(job_, wav, mp3): + seen.append( + ( + sorted(p.name for p in stems.glob("*.mp3") if not p.name.startswith(".")), + sorted(p.name for p in stems.glob("*.wav")), + ) + ) + real(job_, wav, mp3) + + monkeypatch.setattr(stem_storage, "verify_mp3", watching) + _store(job, jobs_dir) + assert len(seen) == 3 + for finals, wavs in seen: + assert finals == [] + assert wavs == ["bass.wav", "drums.wav", "vocals.wav"] + + +def test_transcript_and_envelope_kept_from_the_wav_stay_valid(jobs_dir): + skip_without_ffmpeg() + from app.pipeline.lyrics_retime import read_transcript, save_transcript + + job = _job(jobs_dir, "a1b2c3d4e5fa", names=("vocals",)) + job_dir = jobs_dir / job.id + save_transcript(job_dir, words=[{"start": 0.5, "end": 1.0, "text": "la"}], language="en") + envelope = job_dir / "stems" / "vocal_envelope.json" + envelope.write_text(json.dumps({"hop": 0.04, "db": [-20]}), encoding="utf-8") + time.sleep(0.01) + _store(job, jobs_dir) + assert read_transcript(job_dir) is not None + vocals = job_dir / "stems" / "vocals.mp3" + assert envelope.stat().st_mtime_ns >= vocals.stat().st_mtime_ns + + +# ── The stage: every failure keeps the WAVs ── + + +def _stub(monkeypatch, code: str) -> None: + """Replace the encoder with a Python one-liner; argv[1] is the output.""" + monkeypatch.setattr( + stem_storage, + "_encode_cmd", + lambda src, dest, kbps: [sys.executable, "-c", code, str(dest), str(src)], + ) + + +def _assert_kept_as_wav(job: Job, jobs_dir: Path, names) -> None: + stems = jobs_dir / job.id / "stems" + assert sorted(p.name for p in stems.iterdir()) == sorted(f"{n}.wav" for n in names) + assert (job.stem_format, job.stem_mp3_kbps) == ("wav", None) + assert all(s["url"].endswith(".wav") for s in job.stems) + + +def test_ffmpeg_without_libmp3lame_keeps_the_wavs(jobs_dir, monkeypatch): + skip_without_ffmpeg() + real = stem_storage._encode_cmd + + def no_lame(src, dest, kbps): + cmd = real(src, dest, kbps) + cmd[cmd.index("libmp3lame")] = "libmp3lame_not_built_in" + return cmd + + monkeypatch.setattr(stem_storage, "_encode_cmd", no_lame) + job = _job(jobs_dir, "b1b2c3d4e501", names=("vocals", "drums")) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals", "drums")) + + +def test_a_full_disk_keeps_the_wavs(jobs_dir, monkeypatch): + _stub( + monkeypatch, + "import sys; open(sys.argv[1], 'wb').write(b'\\xff\\xfb' * 100); " + "sys.stderr.write('Error writing trailer: No space left on device\\n'); sys.exit(1)", + ) + job = _job(jobs_dir, "b1b2c3d4e502", names=("vocals", "drums")) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals", "drums")) + + +def test_a_corrupt_output_keeps_the_wavs(jobs_dir, monkeypatch): + skip_without_ffmpeg() + _stub(monkeypatch, "import sys; open(sys.argv[1], 'wb').write(b'not an mp3 at all' * 999)") + job = _job(jobs_dir, "b1b2c3d4e503", names=("vocals",)) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals",)) + + +def test_a_short_output_keeps_the_wavs(jobs_dir, monkeypatch): + """Decodes fine, one second too short: the length check refuses it.""" + skip_without_ffmpeg() + real = stem_storage._encode_cmd + monkeypatch.setattr( + stem_storage, + "_encode_cmd", + lambda src, dest, kbps: [*real(src, dest, kbps)[:-1], "-t", "5", str(dest)], + ) + job = _job(jobs_dir, "b1b2c3d4e504", names=("vocals",)) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals",)) + + +@pytest.mark.parametrize("shift", [576, 1105, 2000]) +def test_a_shifted_output_keeps_the_wavs(jobs_dir, monkeypatch, shift): + """Right length, audio late by an encoder/decoder delay (what a decoder + that ignored the LAME header would produce): the alignment check + refuses it.""" + skip_without_ffmpeg() + job = _job(jobs_dir, "b1b2c3d4e505", names=("vocals",)) + frames = int(SR * SECONDS) + real = stem_storage._encode_cmd + + def shifted(src, dest, kbps): + cmd = real(src, dest, kbps) + at = cmd.index("-c:a") + return [ + *cmd[:at], + "-af", + f"adelay=delays={shift}S:all=1,atrim=end_sample={frames}", + *cmd[at:], + ] + + monkeypatch.setattr(stem_storage, "_encode_cmd", shifted) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals",)) + + +def test_a_failed_move_keeps_the_wavs_and_leaves_no_mp3(jobs_dir, monkeypatch): + skip_without_ffmpeg() + job = _job(jobs_dir, "b1b2c3d4e506", names=("vocals", "drums", "bass")) + real = os.replace + calls = {"n": 0} + + def flaky(src, dst): + calls["n"] += 1 + if calls["n"] == 2: + raise OSError(28, "No space left on device") + real(src, dst) + + monkeypatch.setattr(stem_storage.os, "replace", flaky) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals", "drums", "bass")) + + +def test_a_stalled_encoder_is_stopped_and_the_wavs_kept(jobs_dir, monkeypatch): + monkeypatch.setattr(stem_storage, "TIMEOUT_STEM_TRANSCODE_STALL", 1) + _stub(monkeypatch, "import time; time.sleep(60)") + job = _job(jobs_dir, "b1b2c3d4e507", names=("vocals",)) + started = time.monotonic() + _store(job, jobs_dir) + assert time.monotonic() - started < 20 + _assert_kept_as_wav(job, jobs_dir, ("vocals",)) + + +def test_a_cancel_mid_encode_stops_it_cleans_up_and_raises(jobs_dir, monkeypatch): + _stub( + monkeypatch, + "import sys, time\n" + "open(sys.argv[1], 'wb').write(b'partial')\n" + "for _ in range(600):\n" + " print('progress=continue', flush=True); time.sleep(0.1)\n", + ) + job = _job(jobs_dir, "b1b2c3d4e508", names=("vocals", "drums")) + settings.set_stem_storage("mp3", 128) + threading.Timer(0.5, lambda: setattr(job, "cancel_requested", True)).start() + started = time.monotonic() + with pytest.raises(JobCancelled): + stem_storage.store_stems(job, jobs_dir / job.id) + assert time.monotonic() - started < 10 + stems = jobs_dir / job.id / "stems" + assert sorted(p.name for p in stems.iterdir()) == ["drums.wav", "vocals.wav"] + + +def test_an_unexpected_error_still_keeps_the_wavs(jobs_dir, monkeypatch): + def boom(job_, wav, mp3): + raise MemoryError("libsndfile ran out") + + monkeypatch.setattr(stem_storage, "verify_mp3", boom) + _stub(monkeypatch, "import sys; open(sys.argv[1], 'wb').write(b'x')") + job = _job(jobs_dir, "b1b2c3d4e509", names=("vocals",)) + _store(job, jobs_dir) + _assert_kept_as_wav(job, jobs_dir, ("vocals",)) + + +def test_the_runner_never_fails_a_song_over_storage(jobs_dir, monkeypatch): + from app.pipeline import runner + + def explode(job_, job_dir): + raise RuntimeError("storage bug") + + monkeypatch.setattr(runner, "store_stems", explode) + job = _job(jobs_dir, "b1b2c3d4e50a", names=("vocals",)) + runner._store_stems(job, jobs_dir / job.id) + assert job.stem_format == "wav" + assert "store" in job.stage_timings + + +# ── Readers on an MP3 song ── + + +@pytest.fixture +def mp3_job(jobs_dir): + skip_without_ffmpeg() + job = _job(jobs_dir, "c1b2c3d4e5f0", names=("vocals", "drums", "bass")) + _store(job, jobs_dir, "mp3", 128) + assert job.stem_format == "mp3" + return job + + +@pytest.fixture +def wav_job(jobs_dir): + job = _job(jobs_dir, "c1b2c3d4e5f1", names=("vocals", "drums", "bass")) + _store(job, jobs_dir, "wav") + return job + + +async def test_stem_url_serves_the_mp3_with_its_own_type(client, mp3_job, jobs_dir): + stored = (jobs_dir / mp3_job.id / "stems" / "vocals.mp3").read_bytes() + async with client as c: + for url in (f"/api/jobs/{mp3_job.id}/stems/vocals.wav", mp3_job.stems[0]["url"]): + r = await c.get(url) + assert r.status_code == 200 + assert r.headers["content-type"] == "audio/mpeg" + assert ".mp3" in r.headers["content-disposition"] + # The stored file itself, never encoded a second time. + assert r.content == stored + + +async def test_stem_url_of_a_wav_song_is_unchanged(client, wav_job, jobs_dir): + async with client as c: + r = await c.get(f"/api/jobs/{wav_job.id}/stems/vocals.wav") + assert r.status_code == 200 + assert r.headers["content-type"] == "audio/wav" + assert r.content == (jobs_dir / wav_job.id / "stems" / "vocals.wav").read_bytes() + + +async def test_a_wav_song_with_a_cached_download_mp3_still_plays_its_wav(client, wav_job, jobs_dir): + skip_without_ffmpeg() + async with client as c: + r = await c.get(f"/api/jobs/{wav_job.id}/stems/vocals.mp3") + assert r.status_code == 200 + assert (jobs_dir / wav_job.id / "stems" / "vocals.mp3").is_file() + r = await c.get(f"/api/jobs/{wav_job.id}/stems/vocals.wav") + assert r.headers["content-type"] == "audio/wav" + + +async def test_region_of_an_mp3_stem_is_a_wav_of_the_right_samples(client, mp3_job, jobs_dir): + ref = _signal(seed=1) + async with client as c: + r = await c.get(f"/api/jobs/{mp3_job.id}/stems/vocals.wav?start=2&end=4") + assert r.status_code == 200 + data, rate = sf.read(io.BytesIO(r.content)) + assert rate == SR and len(data) == 2 * SR + assert _lag(ref[2 * SR : 4 * SR, 0], data[:, 0]) == 0 + + +async def test_mixdown_from_mp3_stems_lines_up_with_one_from_wavs(client, mp3_job, wav_job): + """The export path: same lanes, same gains; the MP3 song's mix has the + WAV song's length and its peak at lag 0.""" + q = "stems=vocals,drums,bass&gains=1,0.5,0.8" + async with client as c: + a = await c.get(f"/api/jobs/{wav_job.id}/mixdown.wav?{q}") + b = await c.get(f"/api/jobs/{mp3_job.id}/mixdown.wav?{q}") + assert a.status_code == b.status_code == 200 + x, _ = sf.read(io.BytesIO(a.content)) + y, _ = sf.read(io.BytesIO(b.content)) + assert len(x) == len(y) + assert _lag(x[:, 0], y[:, 0]) == 0 + + +async def test_zip_of_an_mp3_song(client, mp3_job, jobs_dir): + async with client as c: + r = await c.get(f"/api/jobs/{mp3_job.id}/stems/all.zip?format=mp3") + assert r.status_code == 200 + with zipfile.ZipFile(io.BytesIO(r.content)) as zf: + names = sorted(zf.namelist()) + assert names == ["Storage_bass.mp3", "Storage_drums.mp3", "Storage_vocals.mp3"] + stored = (jobs_dir / mp3_job.id / "stems" / "vocals.mp3").read_bytes() + assert zf.read("Storage_vocals.mp3") == stored + r = await c.get(f"/api/jobs/{mp3_job.id}/stems/all.zip?format=wav") + assert r.status_code == 200 + with zipfile.ZipFile(io.BytesIO(r.content)) as zf: + data, rate = sf.read(io.BytesIO(zf.read("Storage_vocals.wav"))) + assert rate == SR and len(data) == int(SR * SECONDS) + + +def test_backing_and_rest_from_mp3_stems(mp3_job, jobs_dir): + from app.pipeline.backing import ensure_backing, ensure_rest + + job_dir = jobs_dir / mp3_job.id + out, built = ensure_backing(mp3_job.id, job_dir, "vocals") + assert built and sf.info(str(out)).frames == int(SR * SECONDS) + out, _ = ensure_rest(mp3_job.id, job_dir, ("bass", "drums")) + assert sf.info(str(out)).frames == int(SR * SECONDS) + + +def test_take_length_comes_from_an_mp3_stem(mp3_job): + from app.api.takes import _song_frames + + assert _song_frames(mp3_job.id) == int(SR * SECONDS) + + +async def test_vocal_envelope_from_an_mp3_stem(client, mp3_job): + async with client as c: + r = await c.get(f"/api/jobs/{mp3_job.id}/vocal-envelope") + assert r.status_code == 200 + assert len(r.json()["db"]) > 0 + + +def test_recovery_and_sweep_see_an_mp3_song(mp3_job, jobs_dir): + from app.core.registry import _recover_done_job + from app.pipeline.collect import _has_stems_on_disk + + job_dir = jobs_dir / mp3_job.id + (job_dir / "metadata.json").write_text( + json.dumps({"title": "Storage", "stem_format": "mp3", "stem_mp3_kbps": 128}), + encoding="utf-8", + ) + assert _has_stems_on_disk(job_dir) + recovered = _recover_done_job(job_dir) + assert recovered is not None + assert {s["name"] for s in recovered.stems} == {"vocals", "drums", "bass"} + assert all(s["url"].endswith(".mp3") for s in recovered.stems) + assert (recovered.stem_format, recovered.stem_mp3_kbps) == ("mp3", 128) + + +def test_missing_peaks_are_rebuilt_from_mp3_stems(mp3_job, jobs_dir): + from app.pipeline.collect import fill_missing_peaks + + stems = jobs_dir / mp3_job.id / "stems" + assert fill_missing_peaks(stems) + assert set(json.loads((stems / "peaks.json").read_text())) == {"vocals", "drums", "bass"} + + +def test_split_stems_are_stored_like_the_song(mp3_job, jobs_dir): + stems = jobs_dir / mp3_job.id / "stems" + for i, name in enumerate(("lead_vocals", "backing_vocals")): + sf.write(stems / f"{name}.wav", _signal(seed=10 + i), SR, subtype="PCM_16") + stem_storage.store_split_stems(mp3_job, stems, ["lead_vocals", "backing_vocals"]) + assert (stems / "lead_vocals.mp3").is_file() and not (stems / "lead_vocals.wav").exists() + assert (stems / "backing_vocals.mp3").is_file() + + +def test_vocal_split_hands_the_model_a_decoded_wav(mp3_job, jobs_dir, monkeypatch): + from app.pipeline import vocal_split + + seen = {} + + def fake(job, vocals_path, stems_dir): + seen["path"] = vocals_path + seen["frames"] = sf.info(str(vocals_path)).frames + return ["lead_vocals", "backing_vocals"] + + monkeypatch.setattr(vocal_split, "_split", fake) + vocal_split.split_vocals(mp3_job, jobs_dir / mp3_job.id / "stems") + assert seen["path"].suffix == ".wav" + assert seen["frames"] == int(SR * SECONDS) + assert not seen["path"].exists() # the temporary decode is gone From dcbc0c1d5b063edfcf1c31a5cbbb2d5b4178d2d9 Mon Sep 17 00:00:00 2001 From: Thales Pereira <31625914+thcp@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:35:43 +0100 Subject: [PATCH 3/5] feat(player): open MP3-stored songs on the full-decode engine The chunked engine streams PCM out of WAVs by Range request and cannot read an MP3. A song whose stem URLs end in .mp3 opens on the full-decode engine instead (desktop and phone), whose decodeAudioData honours the LAME header, so the stems stay sample-aligned with the click and takes. Over the full-decode RAM ceiling it plays on the streaming path, as a fulldecode-mode song always did. Lane downloads are named after the stored format, which the desktop app saves the bytes under. --- docs/api.md | 4 ++-- static/js/mixer.js | 3 ++- static/js/playbackStems.js | 26 +++++++++++++++++++++++ static/js/player.js | 10 ++++++--- static/mobile/app.js | 8 ++++++- tests/js/stem-format.test.mjs | 39 +++++++++++++++++++++++++++++++++++ 6 files changed, 83 insertions(+), 7 deletions(-) create mode 100644 tests/js/stem-format.test.mjs diff --git a/docs/api.md b/docs/api.md index d78010b6..0e8ee30e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -13,7 +13,7 @@ The endpoints the web app itself uses. They are not a stable public interface an | GET | `/api/jobs/{id}/events` | SSE stream of job state | | POST | `/api/jobs/{id}/cancel` | Terminate active subprocess and cancel job | | PATCH | `/api/jobs/{id}/sections` | Save waveform section markers for a job | -| GET | `/api/jobs/{id}/stems/{name}.wav` | Stream a single stem WAV file | -| GET | `/api/jobs/{id}/stems/{name}.mp3` | Transcode and stream a stem as MP3 | +| GET | `/api/jobs/{id}/stems/{name}.wav` | Stream a single stem as the song stores it: WAV, or MP3 (`audio/mpeg`) for a song stored as MP3 (#783) | +| GET | `/api/jobs/{id}/stems/{name}.mp3` | A stem as MP3: the stored file for a song stored as MP3, else a cached transcode | | GET | `/api/jobs/{id}/video.mp4` | Mux the current mix with the source video (MP4 upload or YouTube) into an MP4 | | DELETE | `/api/jobs/{id}` | Remove job dir from disk (terminal jobs only) | diff --git a/static/js/mixer.js b/static/js/mixer.js index 8a65b01f..13fcbef6 100644 --- a/static/js/mixer.js +++ b/static/js/mixer.js @@ -20,6 +20,7 @@ import { import { storeGet, storeSetDebounced, storeFlushPending } from "./utils.js"; import { getMode } from "./mode.js"; import { t } from "./i18n.js"; +import { stemFileExt } from "./playbackStems.js"; import { PITCH_MAX, PITCH_MIN, clampPitch, pitchBlockedKey } from "./pitchBus.js"; function defaultMixerEntry() { @@ -733,7 +734,7 @@ export function renderMixerRow(stem) { const dl = document.createElement("a"); dl.className = "lane-dl mx-btn"; dl.href = stem.url; - dl.download = `${stem.name}.wav`; + dl.download = `${stem.name}.${stemFileExt(stem.url)}`; dl.title = t("aria.download", { name: display }); dl.appendChild(downloadIcon()); diff --git a/static/js/playbackStems.js b/static/js/playbackStems.js index 05ea162c..87c65a64 100644 --- a/static/js/playbackStems.js +++ b/static/js/playbackStems.js @@ -101,3 +101,29 @@ export function addVisualOnlyStems(playbackStems, visibleStems) { .map((stem) => ({ ...stem, controlName: stem.name, pitched: false, visualOnly: true })), ]; } + +/** + * Whether any of these stems is stored compressed (MP3, #783), by the + * extension its URL ends in. The chunked engine streams PCM straight out of + * a WAV by Range request and cannot read an MP3, so a song stored as MP3 + * opens on the full-decode engine instead, whose decodeAudioData honours the + * MP3's encoder delay and padding and lines the stems up to the sample (the + * server checks the same before it keeps an MP3). + * + * @param {{url?:string}[]} stems + */ +export function hasCompressedStems(stems) { + return (stems || []).some((stem) => stemFileExt(stem?.url) === "mp3"); +} + +/** + * The file extension a stem downloads with: "mp3" for a stem stored as MP3, + * else "wav". The desktop app saves a lane download under the name the page + * gives it, so the name has to say what the bytes are. + * + * @param {string|undefined} url + */ +export function stemFileExt(url) { + const path = String(url || "").split(/[?#]/)[0]; + return /\.mp3$/i.test(path) ? "mp3" : "wav"; +} diff --git a/static/js/player.js b/static/js/player.js index eca70315..08cc7ce0 100644 --- a/static/js/player.js +++ b/static/js/player.js @@ -41,7 +41,7 @@ import { createAudioEngine, estimateDecodedBytes } from "./audioEngine.js"; import { createChunkedAudioEngine } from "./chunkedAudioEngine.js"; import { createPlaybackContext } from "./audioContext.js"; import { effectivePitch } from "./pitchBus.js"; -import { addVisualOnlyStems, buildPlaybackStems, sumPeaks } from "./playbackStems.js"; +import { addVisualOnlyStems, buildPlaybackStems, hasCompressedStems, stemFileExt, sumPeaks } from "./playbackStems.js"; import { layoutLanes, originalSource, jobOriginalSet, layoutSignature, baseLanes } from "./partChoice.js"; import { applyAllLaneEq, exportFxParam } from "./eqEngine.js"; import { exportMasterParam, setMixSource, resetForSong as resetMasterForSong } from "./masterEngine.js"; @@ -1520,7 +1520,7 @@ export function wireUpAudio(jobId, stems, duration, thumbnail, mixUrl = null, ti // A recording track with no take yet has nothing to download. if (dl && stem.url) { dl.href = stem.url; - dl.download = _stemFilename(stem.name); + dl.download = _stemFilename(stem.name, stemFileExt(stem.url)); } } @@ -1556,7 +1556,11 @@ export function wireUpAudio(jobId, stems, duration, thumbnail, mixUrl = null, ti // the 6-connection HTTP/1.1 limit and `canplay` stalls (the WebView2/WKWebView // freeze). Null URLs make the multitrack's