got-feedBack/feedBack
1
0
Fork 0
mirror of https://github.com/got-feedBack/feedBack.git synced 2026-09-10 19:44:30 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix(sloppak): the full mix is a stem — drop the invented original_audio key (#946)
Nightly / build-docker (push) Has been cancelled
Details
ship-ci / ci (push) Waiting to run
Details

Browse Source 
* fix(sloppak): the full mix is a stem — drop the invented `original_audio` key (#933)

Core read, served, and depended on `original_audio:` — a top-level manifest key
this repo invented in #583 that the feedpak spec never defined. The format
already had a home for the pre-separation mixdown: it is a stem. feedpak 1.15.0
(feedpak-spec#53) RESERVES the id `full` for it, so read it from there.

The key existed to work around a bug in our own reader. The packer's comment
said so plainly: "we must NOT list the full mix as a playable stem — the player
sums every entry in `stems` and does not gate playback on `default`, so a listed
full mix plays on top of the stems". Faced with a reader that would double the
song, the packer put the mixdown outside `stems` and invented a key to point at
it. The fix belongs in the reader, and that is what this is.

load_song() now partitions the stem list: `full` comes out as
LoadedSloppak.full_mix, the instruments stay in .stems. Nothing that sums stems
or draws one fader per stem can see the mixdown, so retaining it is safe — which
is what lets the packer put it where the format says it goes.

- ws_highway: `song_info` gains full_mix_url / has_full_mix. The old
  original_audio_url / has_original_audio remain as deprecated aliases for one
  release so an older stems plugin keeps working (#945).
- `stems` on the wire, and stem_ids / stem_count in the library index, are now
  INSTRUMENT stems only — a separated pack that retains its mixdown no longer
  advertises a bogus "full" chip or an inflated stem count.
- enrichment: fingerprint against the mixdown wherever it lives. This widens
  coverage — _song_audio_file() previously returned None for any pack without
  the invented key, so fingerprinting silently did nothing for nearly every pack.
- sloppak: `original_audio:` is still READ as a deprecated fallback, because
  every pack in the wild carries it and would otherwise lose its pristine mix.
  tools/migrate_full_mix_stem.py rewrites those packs into the spec shape
  (original/full.ogg -> stems/full.ogg, add the `full` stem at default:off, drop
  the key); the fallback and the aliases die with #945.

The spec gate keeps the debt honest: the grandfather entry now tracks #945, and
the gate fails if it goes stale.

Verified: spec gate OK (4/4, incl. ingesting the spec's new example pack that
retains `full`); 2493 python tests, 995 js tests; migrator round-tripped over
real packs from the library and the results pass the spec's reference validator.

* fix(migrate): discover directory-form packs instead of silently skipping them

iter_packs() searched only files, so a directory-form pack (`song.sloppak/`, the
authoring shape) was walked INTO and never yielded — silently missed by a run
that's meant to be exhaustive. Discover suffix-named directories too (yielded
whole, not descended into), and route packs through migrate_pack/verify_pack.

Directory packs are REPORTED as `dir-form-unsupported`, not rewritten in place:
a single-file pack is replaced atomically (a fully-built temp archive swapped in
with one os.replace), but a populated directory can't be swapped that way, so an
interrupted in-place rewrite could leave an authoring pack half-migrated. The
status is a problem status, so it counts against the run's exit code and shows
in the summary — the operator re-packs or migrates it as a `.feedpak` instead of
it vanishing from the report. Addresses a CodeRabbit review finding.

Signed-off-by: Kris Anderson <topkoa@gmail.com>

* fix(migrate): verify requires an explicit `off` on a retained full mix

verify_zip accepted any non-truthy `default` on a multi-stem `full` (missing,
empty, boolean, `false`/`no`/`0`, malformed) as "ok". But core defaults an ABSENT
`default` to True — ON (lib/sloppak.py: `s.get("default", True)`) — and treats an
empty/unrecognized string as ON too, so a migrated-shape pack whose `full` stem
has a missing or blank default beside instrument stems would actually play the
mixdown on open and double the song. verify was certifying that as safe.

Require an explicit normalized `off` beside instrument stems: `on`-ish values are
reported `full-stem-default-on` (actively plays), everything that is not a
normalized `off` is reported `full-stem-default-not-off`. The migrator already
writes the literal `off`, so its own output is unaffected; this also certifies
the pack is in the tool's canonical, most-portable shape. The len>1 gate is kept,
so a sole `full` stem (which IS the audio) is not policed.

Adds parametrized coverage for missing / empty / boolean / off-ish / malformed
defaults, and a sole-full-stem case. Addresses a CodeRabbit review finding.

Signed-off-by: Kris Anderson <topkoa@gmail.com>

---------

Signed-off-by: Kris Anderson <topkoa@gmail.com>
Co-authored-by: Kris Anderson <topkoa@gmail.com>
This commit is contained in:
Byron Gamatos
2026-07-13 12:22:42 -04:00
committed by GitHub
co-authored by Kris Anderson
parent d876ded00f
commit 329cc86315
10 changed files with 1430 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+34
View File
@@ -7,6 +7,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Changed
- **The full mix is a stem** (#933) — core no longer depends on `original_audio:`, a
top-level manifest key this repo invented (#583) that the feedpak spec never had.
The format already carried the pre-separation mixdown as a stem; feedpak 1.15.0
(feedpak-spec#53) reserves the id **`full`** for it, so that is where core reads it
from now.
`full` is a mixdown, not a layer — it already contains every instrument — so
`load_song()` lifts it OUT of `LoadedSloppak.stems` onto `LoadedSloppak.full_mix`.
Nothing that sums stems or renders one fader per stem can see it, which is what
makes retaining it safe; leaving it in the list would double the whole song and
leave "guitar" audible with the guitar fader muted. That trap is exactly why the
packer invented the key instead of putting the mixdown where the format says it
goes — the bug was in the reader, and this fixes the reader.
Consequences worth knowing:
- The highway WS `song_info` frame gains `full_mix_url` / `has_full_mix`.
`original_audio_url` / `has_original_audio` remain as **deprecated aliases**
(same values) for one release so a client built against the old frame keeps
working; they go with the fallback below (#945).
- `stems` on `song_info`, and `stem_ids` / `stem_count` in the library index, now
describe *instrument* stems only — a separated pack that retains its mixdown no
longer advertises a bogus "full" stem chip or an inflated stem count.
- Audio fingerprinting (`lib/enrichment.py`) now resolves the mixdown the same
way, which **widens** its coverage: it previously returned `None` for any pack
without the invented key, so fingerprinting silently did nothing for the
overwhelming majority of packs.
- Core still **reads** `original_audio:` as a deprecated fallback, because every
pack written before the spec caught up carries it and would otherwise lose its
pristine mix. `tools/migrate_full_mix_stem.py` rewrites those packs into the
spec shape (moves `original/full.ogg` → `stems/full.ogg`, adds the `full` stem
at `default: off`, drops the key); the fallback and the aliases are removed once
they are migrated (#945).
### Added
- **Genres fall back to MusicBrainz enrichment** — the effective genre now
resolves override → pack genre → the enrichment match's primary genre
feedpak-spec-exceptions.yml
+15 -9
View File
@@ -34,17 +34,23 @@
exceptions:
- key: original_audio
issue: https://github.com/got-feedback/feedback/issues/933
issue: https://github.com/got-feedback/feedback/issues/945
reason: >-
Added by #583 (the full mix played while every stem fader sits at unity,
since demucs recombination is lossy). Core, lib/enrichment.py, and the
stems plugin all depend on it, but it never went through a FEP and the
since demucs recombination is lossy). It never went through a FEP and the
spec does not define it — the drift this gate exists to prevent.
The resolution is REMOVAL, not a FEP: the spec already carries the mixdown
as a stem ({id: full, file: stems/full.ogg}), so this key added a second,
redundant location for audio to a format that already had one. See #933.
#933 fixed the drift: feedpak 1.15.0 RESERVES the stem id `full` for the
complete mixdown (feedpak-spec#53), and core now reads the full mix from
that stem. Nothing depends on this key any more — not the loader, not
lib/enrichment.py, not the stems plugin, and the packer no longer writes it.
Grandfathered so the gate can land green and start blocking the *next*
instance immediately, rather than blocking on #933. This entry goes away
when core no longer reads or writes the key.
What remains is a READ-ONLY deprecated fallback in lib/sloppak.py
(_legacy_full_mix), kept for one release because every pack produced before
the spec caught up carries `original_audio: original/full.ogg` and would
otherwise silently lose its pristine mix. tools/migrate_full_mix_stem.py
rewrites those packs into the spec shape.
This entry disappears with that fallback — tracked by #945, which cannot be
forgotten: the gate fails if the entry goes stale, and deleting the read is
what makes it stale.
lib/enrichment.py
+20 -5
View File
@@ -368,10 +368,12 @@ def _acoustid_gate() -> "JSONResponse | None":
def _song_audio_file(filename: str) -> "str | None":
"""Resolve a LIBRARY song (by filename/id) to a local master-audio file for
fingerprinting: the full-mix `original_audio` extracted from a sloppak, or a
loose folder's audio. None when the song can't be found or ships no full-mix
audio (some packs carry only stems). Mirrors serve_sloppak_file's containment
guards so a crafted filename can't read outside DLC_DIR / the pack."""
fingerprinting: a sloppak's complete mixdown, or a loose folder's audio. None
when the song can't be found or carries no mixdown (a pack that kept only its
separated stems — an acoustic fingerprint of one re-summed from them would not
match the recording, so we decline rather than submit a lossy reconstruction).
Mirrors serve_sloppak_file's containment guards so a crafted filename can't
read outside DLC_DIR / the pack."""
dlc = _get_dlc_dir()
if not dlc:
return None
@@ -383,7 +385,20 @@ def _song_audio_file(filename: str) -> "str | None":
canon = resolved.relative_to(dlc.resolve()).as_posix()
except ValueError:
return None
rel = (sloppak_mod.load_manifest(resolved) or {}).get("original_audio")
manifest = sloppak_mod.load_manifest(resolved) or {}
# The mixdown is the RESERVED `full` stem (spec §5.3). Unlike playback,
# fingerprinting wants it even when it is the pack's ONLY stem — a
# single-mix pack is exactly the master audio we want to fingerprint —
# so this asks find_full_mix() rather than partition_stems().
stems = manifest.get("stems") or []
full = sloppak_mod.find_full_mix(
[s for s in stems if isinstance(s, dict)]
)
rel = full.get("file") if full else None
# DEPRECATED fallback: packs written before the spec reserved `full` put
# the mixdown behind a top-level `original_audio:` key instead (#933).
if not isinstance(rel, str) or not rel.strip():
rel = manifest.get("original_audio")
if not isinstance(rel, str) or not rel.strip():
return None
src = sloppak_mod.get_cached_source_dir(canon)
lib/routers/ws_highway.py
+43 -19
View File
@@ -321,11 +321,16 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
audio_url = None
audio_error: str | None = None # Surfaced in song_info when audio_url is None
stems_payload: list[dict] = []
# URL of the single full-mix audio (sloppak `original_audio:`), when the
# pack ships one. The stems plugin uses this to play the untouched mix
# while every stem slider is at unity; None otherwise (separate stems
# only, loose folder, or archive).
original_audio_url: str | None = None
# URL of the pack's complete mixdown — the RESERVED `full` stem (spec
# §5.3), which sloppak.load_song() lifts out of `stems` because it is a
# mixdown, not a layer. The stems plugin plays it while every stem slider
# is at unity (separation is lossy, so it beats re-summing the stems) and
# crosses to the separated stems as soon as one is attenuated.
#
# None when the pack has no mixdown to offer separately from its stems:
# a single-mix pack (its one stem IS the mixdown), a loose folder, or an
# archive.
full_mix_url: str | None = None
if is_loose:
# Loose folder filenames are relative paths (artist/album/song).
# Hash the *canonical* dlc-relative path (so two URL spellings
@@ -365,21 +370,25 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
url = f"/api/sloppak/{q_fn}/file/{quote(s['file'])}"
stems_payload.append({"id": s["id"], "url": url, "default": s["default"]})
# Full-mix URL (served by the same /api/sloppak/.../file/ endpoint).
if loaded_slop is not None and loaded_slop.original_audio:
original_audio_url = (
f"/api/sloppak/{q_fn}/file/{quote(loaded_slop.original_audio)}"
if loaded_slop is not None and loaded_slop.full_mix:
full_mix_url = (
f"/api/sloppak/{q_fn}/file/{quote(loaded_slop.full_mix)}"
)
if stems_payload:
# Stems present: keep the core <audio> pointed at stem[0]. This
# URL is only ever heard in the degraded path (stems plugin
# refuses takeover / decode fails); the full-mix↔stems switch is
# driven client-side by `original_audio_url`, not `audio_url`.
# driven client-side by `full_mix_url`, not `audio_url`.
audio_url = stems_payload[0]["url"]
elif original_audio_url:
elif full_mix_url:
# Stem-less full-mix pack: nothing to separate, so play the full
# mix natively through the core <audio>. The stems plugin's
# onSongReady returns early on an empty stems list (no graph).
audio_url = original_audio_url
# Reachable only via the deprecated `original_audio:` key, whose
# packs put the mixdown outside `stems` — a pack that carries its
# mixdown as the `full` stem has it IN `stems`, so it lands in the
# branch above with stems_payload == [full].
audio_url = full_mix_url
else:
audio_error = "This sloppak has no playable stems."
else:
@@ -521,16 +530,31 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
# for the credits overlay, so minigames / synthetic highway uses
# (no manifest) never trigger it.
"authors": _sanitize_authors(loaded_slop.manifest) if (is_slop and loaded_slop is not None) else [],
# Instrument stems ONLY. The pack's complete mixdown (the RESERVED
# `full` stem, spec §5.3) is deliberately NOT in this list: consumers
# sum `stems` into one mix and render one fader per entry, and the
# mixdown is neither a layer nor an instrument — summing it would
# double the whole song. It is surfaced separately, below.
"stems": stems_payload,
# Full-mix audio (sloppak `original_audio:`) served alongside the
# separate `stems`. The stems plugin plays this single file while
# every stem slider is at unity and switches to the separate stems
# the moment one drops below 100%. None when the pack ships stems
# only. `has_*` flags mirror the has_drum_tab/has_keys convention so
# a client can branch without re-deriving from the URLs.
"original_audio_url": original_audio_url,
"has_original_audio": bool(original_audio_url),
# The complete mixdown, served by the same /api/sloppak/.../file/
# endpoint as the stems. The stems plugin plays this single file
# while every stem slider is at unity and crosses to the separated
# stems the moment one drops below 100% — separation is lossy, so the
# mixdown is strictly better audio when nothing is muted. None when
# the pack has no mixdown apart from its stems. The `has_*` flags
# mirror the has_drum_tab/has_keys convention so a client can branch
# without re-deriving from the URLs.
"full_mix_url": full_mix_url,
"has_full_mix": bool(full_mix_url),
"has_stems": bool(stems_payload),
# DEPRECATED aliases of the two keys above, kept so a client built
# against the old frame keeps working across one release. They were
# named after `original_audio:` — a manifest key this repo invented
# and the feedpak spec never had (#933). The key is gone; the mixdown
# is a stem. Remove these once the shipped stems plugin reads
# `full_mix_url` (#945).
"original_audio_url": full_mix_url,
"has_original_audio": bool(full_mix_url),
# Surface a drum_tab presence flag so the visualization picker
# can auto-activate the drums plugin even when the chosen
# arrangement isn't named "Drums" (drum_tab.json lives next
lib/sloppak.py
+153 -33
View File
@@ -35,6 +35,21 @@ FEEDPAK_EXT = ".feedpak"
SLOPPAK_EXT = ".sloppak"
SONG_EXTS = (FEEDPAK_EXT, SLOPPAK_EXT) # accepted on read/discovery
# ── The full mix ──────────────────────────────────────────────────────────────
#
# Spec §5.3 RESERVES the stem id `full` for the song's complete mixdown: the
# whole song in one file, as heard before source separation. It is a stem — it
# lives in `stems` like every other audio file in a pack — but it is a *mixdown,
# not a layer*. A reader that sums stems must never include it in the sum: it
# already contains every instrument, so summing it doubles the whole song and
# muting `guitar` still leaves guitar audible inside it.
#
# Keeping it matters because separation is lossy: re-summing guitar+bass+drums+
# vocals does NOT reproduce the file they came from. The mixdown is the only
# faithful rendering of the song a pack can carry, so we play it whenever every
# stem sits at unity and nothing is muted.
FULL_MIX_STEM_ID = "full"
import yaml
from jsonc import load_json
@@ -52,6 +67,97 @@ import drums as drums_mod
import notation as notation_mod
def find_full_mix(stems: list[dict]) -> dict | None:
"""The RESERVED `full` stem (spec §5.3) — the pack's complete mixdown — or None.
Answers "what is this pack's master audio", which is what fingerprinting
wants. For playback use partition_stems() instead: a pack whose *only* stem
is `full` has no mixdown to play *separately from* its stems, and this
function still returns it.
"""
return next(
(s for s in stems if str(s.get("id", "")) == FULL_MIX_STEM_ID), None
)
def partition_stems(stems: list[dict]) -> tuple[dict | None, list[dict]]:
"""Split stem descriptors into (mixdown, instrument_stems) for PLAYBACK.
The mixdown is lifted OUT of the stem list because every consumer of `stems`
treats that list as layers to sum or to show as mixer channels, and `full` is
neither (spec §5.3). Leaving it in is precisely the bug that made the packer
invent `original_audio` in the first place: a listed full mix plays on top of
the stems.
A pack whose only stem is `full` is a single-mix pack, not a separated one:
there are no instruments to be pristine *against*, so `full` stays the sole
playable stem and no mixdown is surfaced. That keeps the freshly-converted
single-stem pack — much the most common shape — behaving exactly as before.
EVERY entry with the reserved id is removed, not just the one we surface. A
malformed pack that lists `full` twice would otherwise leave a copy of the
whole song behind in the stem list, to be summed with the instruments — the
precise failure this function exists to prevent, reintroduced by a duplicate.
"""
if len(stems) < 2:
return None, stems
full = find_full_mix(stems)
if full is None:
return None, stems
return full, [s for s in stems if str(s.get("id", "")) != FULL_MIX_STEM_ID]
def _legacy_full_mix(manifest: dict, source_dir: Path) -> str | None:
"""Full mix from the DEPRECATED `original_audio:` manifest key, or None.
Before feedpak 1.15.0 reserved `full`, §5.3 said the mixdown was "commonly
replaced" by the per-instrument stems on splitting — so it had nowhere to
live, and this repo invented a top-level key pointing at a parallel
`original/` directory (#583) to hold it. That key was never in the spec, and
#933 removed our dependence on it: the mixdown is a stem.
We still READ it, because every pack written before the spec caught up
carries `original_audio: original/full.ogg` and would otherwise lose its full
mix. We never write it. Delete this once those packs are migrated (#945);
`tools/migrate_full_mix_stem.py` is the migration.
NOTE the string literal below. tools/check_spec_conformance.py AST-scans for
`manifest.get("<literal>")` to prove every manifest key core reads is one the
spec declares. Hoisting "original_audio" into a named constant would hide
this read from that scan — the gate would conclude core no longer touches the
key, and the grandfather entry that documents this debt would go stale. The
literal is what keeps the deprecation honest and visible to CI. Leave it.
Same permissive, path-traversal-guarded posture as the optional side-files: a
missing / escaping / unreadable file leaves the pack without a full mix (the
player falls back to the separated stems) rather than aborting the load.
Returns the manifest-relative string, so callers build its URL exactly as
they build a stem's.
"""
rel_raw = manifest.get("original_audio")
if not isinstance(rel_raw, str) or not rel_raw.strip():
return None
rel = rel_raw.strip()
try:
target = (source_dir / rel).resolve()
target.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: original_audio path %r escapes source_dir — skipped", rel)
return None
except OSError as e:
log.warning("sloppak: original_audio path resolution failed (%s) — skipped", e)
return None
if not target.is_file():
return None
log.info(
"sloppak: pack uses the deprecated `original_audio:` key (%r) — the full mix "
"is a stem (id `full`, feedpak spec §5.3). Re-pack with "
"tools/migrate_full_mix_stem.py; support for this key will be removed.",
rel,
)
return rel
# ── Format detection ──────────────────────────────────────────────────────────
def is_sloppak(path: Path) -> bool:
@@ -595,14 +701,21 @@ class LoadedSloppak:
# song.arrangements (not to manifest["arrangements"]) — skipped entries are
# absent so indexing by song.arrangements index is safe.
arrangement_ids: list[str | None] = field(default_factory=list)
# Manifest-relative path to the single full-mix audio file, taken from the
# manifest `original_audio:` key (e.g. "original/full.ogg"). This is the
# pre-separation mixdown that exists alongside the per-instrument `stems`.
# None when the key is absent, points outside source_dir, or the file is
# missing on disk. Served to the front-end via the highway WS as
# `original_audio_url`; the stems plugin uses it to play the untouched mix
# when every stem slider is at unity (and the separate stems otherwise).
original_audio: str | None = None
# Manifest-relative path to the pack's complete mixdown — the whole song in
# one file, as heard before source separation. This is the RESERVED `full`
# stem (spec §5.3), lifted out of `stems` above precisely because it is NOT
# an instrument layer: summing it with the per-instrument stems it was split
# into would double the entire song. See partition_stems().
#
# None when the pack has no mixdown to offer *separately* from its stems —
# which includes the common single-mix pack, whose only stem IS the mixdown
# (there is nothing to be pristine against, so it stays in `stems`).
#
# Served to the front-end via the highway WS as `full_mix_url`; the stems
# plugin plays it while every stem slider sits at unity and crosses to the
# separated stems the moment one drops below 100% — demucs recombination is
# lossy, so the mixdown is strictly the better audio when nothing is muted.
full_mix: str | None = None
def load_song(
@@ -994,6 +1107,13 @@ def load_song(
default_on = bool(default_val)
stems.append({"id": sid, "file": sfile, "default": default_on})
# The complete mixdown is a stem (spec §5.3), but it is not a *layer*: lift
# it out so that no consumer of `stems` — the mixer, the library's stem
# chips, the WS payload — sums it with, or lists it beside, the instruments
# it was separated into. `full_mix_stem` is None for a single-mix pack,
# whose only stem IS the mixdown and stays in the list.
full_mix_stem, stems = partition_stems(stems)
# Optional keys.json — song-level, instrument-independent key/scale track
# (manifest `keys:` key, spec §7.7). Permissive like the other side-files:
# missing / unreadable / malformed -> None, never fatal. Stored as a
@@ -1056,28 +1176,22 @@ def load_song(
}
_fpv = manifest.get("feedpak_version")
# Optional full-mix audio — manifest `original_audio:` key. The single
# pre-separation mixdown that ships alongside the per-instrument stems.
# Same permissive, path-traversal-guarded posture as drum_tab above: a
# missing/escaping/absent file simply leaves the full mix unavailable (the
# player falls back to the separate stems) rather than aborting the load.
# We store the manifest-relative string so server.py can build its URL the
# same way it builds stem URLs (via the /api/sloppak/.../file/ endpoint).
original_audio_data: str | None = None
original_audio_rel = manifest.get("original_audio")
if isinstance(original_audio_rel, str) and original_audio_rel.strip():
rel = original_audio_rel.strip()
try:
oa_path = (source_dir / rel).resolve()
oa_path.relative_to(source_dir.resolve())
except ValueError:
log.warning("sloppak: original_audio path %r escapes source_dir — skipped", rel)
oa_path = None
except OSError as e:
log.warning("sloppak: original_audio path resolution failed (%s) — skipped", e)
oa_path = None
if oa_path is not None and oa_path.is_file():
original_audio_data = rel
# The pack's full mix. Normally the RESERVED `full` stem partitioned out
# above (spec §5.3) — no path work needed, it was validated with the other
# stems and its URL is built the same way. Only when the pack has no `full`
# stem do we fall back to the DEPRECATED `original_audio:` key, which is the
# shape every pack written before feedpak 1.15.0 uses.
if full_mix_stem is not None:
full_mix_data: str | None = full_mix_stem["file"]
elif find_full_mix(stems) is not None:
# Single-mix pack: its ONE stem is the mixdown, so there is no mixdown to
# offer *apart from* the stems. Never fall through to the legacy key here
# — a pack that both carries a `full` stem and names the old key would
# otherwise surface the mixdown twice (once as the stem the player is
# already playing, once as a "pristine" track to cross to).
full_mix_data = None
else:
full_mix_data = _legacy_full_mix(manifest, source_dir)
return LoadedSloppak(
song=song,
@@ -1092,7 +1206,7 @@ def load_song(
keys=keys_data,
notation_by_id=notation_by_id_data,
arrangement_ids=arrangement_ids_acc,
original_audio=original_audio_data,
full_mix=full_mix_data,
)
@@ -1137,7 +1251,7 @@ def extract_meta(path: Path) -> dict:
tuning_offsets = _tuning_for_meta(arr_list)
stems_list = manifest.get("stems", []) or []
stem_ids: list[str] = []
valid_stems: list[dict] = []
for s in stems_list:
if not isinstance(s, dict):
continue
@@ -1151,7 +1265,13 @@ def extract_meta(path: Path) -> dict:
isinstance(sid, str) and sid
and isinstance(sfile, str) and sfile
):
stem_ids.append(sid)
valid_stems.append({"id": sid, "file": sfile})
# Partition exactly as load_song() does, for the same reason the library
# filter must not lie: `full` is the mixdown, not an instrument (spec §5.3).
# A separated pack that retains it would otherwise offer the user a "full"
# stem chip alongside guitar/bass/drums and count it as a seventh stem.
_full, instrument_stems = partition_stems(valid_stems)
stem_ids = [s["id"] for s in instrument_stems]
stem_count = len(stem_ids)
return {
static/highway.js
+14 -10
View File
@@ -1908,17 +1908,21 @@ function createHighway() {
// never routable.
const isAudioUrl = msg.audio_url.startsWith('/audio/');
// "Full mix" covers BOTH single-mix pack shapes:
// - stem-less packs (original_audio: in the manifest,
// audio_url == original_audio_url), and
// - single-stem packs (stems: [full.ogg] only) — the server
// puts the full mix in the stems list, has_original_audio
// is false, and audio_url points at the one stem. With one
// stem there is no per-stem mix to preserve, so routing it
// natively loses nothing. Real multi-stem (>1) stays out
// until Phase 2.
// - single-stem packs (stems: [full.ogg] only) — the pack's
// one stem IS its mixdown, so the server leaves it in the
// stems list, has_full_mix is false, and audio_url points
// at that one stem; and
// - legacy stem-less packs, whose mixdown sits outside stems
// behind the deprecated original_audio: key, so has_stems
// is false and audio_url == full_mix_url.
// Either way there is one audible source and no per-stem mix
// to preserve, so routing it natively loses nothing. A pack
// that retains its `full` stem ALONGSIDE separated stems is
// multi-stem (has_full_mix && has_stems) and stays out until
// Phase 2 — routing it natively would drop the mixer.
const isFeedpakFullMix = !isAudioUrl
&& msg.audio_url.startsWith('/api/sloppak/')
&& ((!!msg.has_original_audio && !msg.has_stems)
&& ((!!msg.has_full_mix && !msg.has_stems)
|| (msg.stems || []).length === 1);
// Record the loaded song's audio so app.js can re-route it
// between the HTML5 and JUCE paths if the audio engine is
@@ -1943,7 +1947,7 @@ function createHighway() {
'isFeedpakFullMix=', isFeedpakFullMix,
'has_stems=', !!msg.has_stems,
'stems=', (msg.stems || []).length,
'has_original_audio=', !!msg.has_original_audio,
'has_full_mix=', !!msg.has_full_mix,
'format=', msg.format,
'alreadyLoaded=', alreadyLoaded,
'juceApi=', !!window.feedBackDesktop?.audio);
tests/test_migrate_full_mix_stem.py
+429
View File
@@ -0,0 +1,429 @@
"""tools/migrate_full_mix_stem.py — packs off the deprecated `original_audio:` key.
The migration moves real audio inside tens of thousands of archives, so the
interesting cases are the ones where it must NOT act: a pack it would corrupt, a
pack it has already done, a pack whose mixdown isn't where the key claims.
"""
from __future__ import annotations
import importlib.util
import zipfile
from pathlib import Path
import pytest
import yaml
_SPEC = importlib.util.spec_from_file_location(
"migrate_full_mix_stem",
Path(__file__).resolve().parent.parent / "tools" / "migrate_full_mix_stem.py",
)
mig = importlib.util.module_from_spec(_SPEC)
_SPEC.loader.exec_module(mig)
def _manifest(**extra) -> dict:
m = {
"feedpak_version": "1.13.0",
"title": "T",
"artist": "A",
"duration": 1.0,
"arrangements": [{"id": "lead", "file": "arrangements/lead.json"}],
"stems": [
{"id": "guitar", "file": "stems/guitar.ogg", "default": "on"},
{"id": "drums", "file": "stems/drums.ogg", "default": "on"},
],
"original_audio": "original/full.ogg",
}
m.update(extra)
return m
def _write_pack(path: Path, manifest: dict, files: dict[str, bytes] | None = None) -> Path:
files = files or {
"original/full.ogg": b"MIXDOWN",
"stems/guitar.ogg": b"g",
"stems/drums.ogg": b"d",
"arrangements/lead.json": b"{}",
}
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("manifest.yaml", yaml.safe_dump(manifest, sort_keys=False))
for name, data in files.items():
zf.writestr(name, data)
return path
def _read(path: Path) -> tuple[dict, set[str]]:
with zipfile.ZipFile(path) as zf:
return yaml.safe_load(zf.read("manifest.yaml")), set(zf.namelist())
# ── plan_manifest: the decisions, without the archives ──────────────────────
def test_plan_adds_the_full_stem_and_drops_the_key():
new, move = mig.plan_manifest(_manifest())
assert move == "original/full.ogg"
assert "original_audio" not in new
assert new["stems"][0] == {
"id": "full",
"file": "stems/full.ogg",
"default": "off",
}
# The separated stems survive, in order, untouched.
assert [s["id"] for s in new["stems"]] == ["full", "guitar", "drums"]
assert new["feedpak_version"] == "1.15.0"
def test_plan_marks_the_retained_mixdown_default_off():
"""The one line that keeps a pre-1.15.0 reader from doubling the song: a
reader that sums every stem still won't play `full` on open if it honours
`default`, which has been normative since 1.0.0."""
new, _ = mig.plan_manifest(_manifest())
assert new["stems"][0]["default"] == "off"
def test_plan_marks_a_sole_mixdown_default_on():
"""With no separated stems the mixdown IS the audio — off would mute the pack."""
new, _ = mig.plan_manifest(_manifest(stems=[]))
assert new["stems"] == [{"id": "full", "file": "stems/full.ogg", "default": "on"}]
def test_plan_preserves_unknown_keys_verbatim():
"""Spec §3: a writer that re-emits a pack SHOULD preserve unknown keys."""
new, _ = mig.plan_manifest(_manifest(source_tool="ExampleTool v1.2.3", rigs="rigs.json"))
assert new["source_tool"] == "ExampleTool v1.2.3"
assert new["rigs"] == "rigs.json"
def test_plan_skips_an_already_migrated_pack():
m = _manifest(
stems=[{"id": "full", "file": "stems/full.ogg", "default": "off"}],
)
del m["original_audio"]
with pytest.raises(mig.Skip):
mig.plan_manifest(m)
def test_plan_skips_a_pack_that_never_had_the_key():
m = _manifest()
del m["original_audio"]
with pytest.raises(mig.Skip):
mig.plan_manifest(m)
def test_plan_drops_a_stale_key_without_moving_anything():
"""Mixdown already a stem, dead key lingering beside it."""
new, move = mig.plan_manifest(
_manifest(stems=[{"id": "full", "file": "stems/full.ogg", "default": "off"}])
)
assert move == ""
assert "original_audio" not in new
assert [s["id"] for s in new["stems"]] == ["full"]
def test_plan_forces_an_existing_full_stem_off_beside_instrument_stems():
"""Dropping the stale key is not enough if the mixdown it duplicated is left
ENABLED: a reader that honours `default` would then play the whole song on top
of the stems on open. The migration must not hand back a pack in the exact
state it exists to remove."""
new, move = mig.plan_manifest(
_manifest(
stems=[
{"id": "full", "file": "stems/full.ogg", "default": "on"},
{"id": "guitar", "file": "stems/guitar.ogg", "default": "on"},
]
)
)
assert move == ""
assert new["stems"][0] == {"id": "full", "file": "stems/full.ogg", "default": "off"}
assert new["stems"][1]["default"] == "on" # instruments untouched
def test_plan_leaves_a_sole_full_stem_enabled_when_dropping_a_stale_key():
"""No instruments beside it — the mixdown IS the audio. Forcing it off here
would mute the pack."""
new, _ = mig.plan_manifest(
_manifest(stems=[{"id": "full", "file": "stems/full.ogg", "default": "on"}])
)
assert new["stems"] == [{"id": "full", "file": "stems/full.ogg", "default": "on"}]
def test_plan_needs_no_move_when_the_key_already_points_at_the_canonical_path():
new, move = mig.plan_manifest(_manifest(original_audio="stems/full.ogg"))
assert move == ""
assert new["stems"][0]["file"] == "stems/full.ogg"
# ── migrate_zip: the archive rewrite ────────────────────────────────────────
def test_migrate_moves_the_audio_and_rewrites_the_manifest(tmp_path: Path):
pak = _write_pack(tmp_path / "song.feedpak", _manifest())
assert mig.migrate_zip(pak, dry_run=False) == "migrated"
manifest, names = _read(pak)
assert "original/full.ogg" not in names # the invented directory is gone
assert "stems/full.ogg" in names # audio lives where the format says
assert "original_audio" not in manifest
assert manifest["stems"][0]["id"] == "full"
assert mig.verify_zip(pak) == "ok"
def test_migrate_preserves_the_mixdown_bytes(tmp_path: Path):
"""It is a rename, not a re-encode. Losing a byte here loses the master audio."""
pak = _write_pack(tmp_path / "song.feedpak", _manifest())
mig.migrate_zip(pak, dry_run=False)
with zipfile.ZipFile(pak) as zf:
assert zf.read("stems/full.ogg") == b"MIXDOWN"
assert zf.read("stems/guitar.ogg") == b"g"
def test_migrate_is_idempotent(tmp_path: Path):
pak = _write_pack(tmp_path / "song.feedpak", _manifest())
assert mig.migrate_zip(pak, dry_run=False) == "migrated"
before = pak.read_bytes()
assert mig.migrate_zip(pak, dry_run=False) == "skip"
assert pak.read_bytes() == before # a re-run touches nothing
def test_dry_run_changes_nothing(tmp_path: Path):
pak = _write_pack(tmp_path / "song.feedpak", _manifest())
before = pak.read_bytes()
assert mig.migrate_zip(pak, dry_run=True) == "would-migrate"
assert pak.read_bytes() == before
def test_migrate_refuses_when_the_mixdown_is_absent(tmp_path: Path):
"""The key points at audio the archive doesn't contain. Fabricating a stem
entry for a missing file would break every reader — refuse, don't guess."""
pak = _write_pack(
tmp_path / "song.feedpak",
_manifest(),
files={"stems/guitar.ogg": b"g", "arrangements/lead.json": b"{}"},
)
before = pak.read_bytes()
assert mig.migrate_zip(pak, dry_run=False) == "missing-audio"
assert pak.read_bytes() == before
def test_migrate_refuses_when_the_target_path_is_taken(tmp_path: Path):
"""A `stems/full.ogg` that is NOT the mixdown already occupies the target.
Overwriting it would destroy a stem."""
pak = _write_pack(
tmp_path / "song.feedpak",
_manifest(),
files={
"original/full.ogg": b"MIXDOWN",
"stems/full.ogg": b"SOMETHING-ELSE",
"arrangements/lead.json": b"{}",
},
)
assert mig.migrate_zip(pak, dry_run=False) == "target-occupied"
with zipfile.ZipFile(pak) as zf:
assert zf.read("stems/full.ogg") == b"SOMETHING-ELSE"
def test_migrate_drops_a_stale_key_beside_a_non_canonical_full_stem(tmp_path: Path):
"""The mixdown is already a stem, but at a path of the pack's own choosing —
which is legal (§2.2: readers resolve through the manifest, never by
filename). Only the dead key needs removing. Demanding `stems/full.ogg` here
would reject a perfectly valid pack as `missing-audio`."""
m = _manifest(stems=[{"id": "full", "file": "audio/mixdown.ogg", "default": "off"}])
pak = _write_pack(
tmp_path / "song.feedpak",
m,
files={"audio/mixdown.ogg": b"MIXDOWN", "arrangements/lead.json": b"{}"},
)
assert mig.migrate_zip(pak, dry_run=False) == "migrated"
manifest, names = _read(pak)
assert "original_audio" not in manifest
assert manifest["stems"] == [
{"id": "full", "file": "audio/mixdown.ogg", "default": "off"}
]
assert "audio/mixdown.ogg" in names # the audio never moved
assert mig.verify_zip(pak) == "ok"
# ── verify_zip ──────────────────────────────────────────────────────────────
def test_verify_rejects_a_retained_mixdown_that_plays_on_open(tmp_path: Path):
"""The hazard the migration must never create: `full` alongside instrument
stems AND default-on means a summing reader plays the whole song twice."""
m = _manifest(
stems=[
{"id": "full", "file": "stems/full.ogg", "default": "on"},
{"id": "guitar", "file": "stems/guitar.ogg", "default": "on"},
]
)
del m["original_audio"]
pak = _write_pack(
tmp_path / "song.feedpak",
m,
files={"stems/full.ogg": b"M", "stems/guitar.ogg": b"g"},
)
assert mig.verify_zip(pak) == "full-stem-default-on"
@pytest.mark.parametrize(
"default, expected",
[
({"default": "off"}, "ok"), # the one safe, canonical shape
({"default": "OFF"}, "ok"), # case-insensitive
({"default": " off "}, "ok"), # surrounding whitespace tolerated
({}, "full-stem-default-not-off"), # MISSING — core defaults to True (ON)
({"default": ""}, "full-stem-default-not-off"), # empty → ON in core
({"default": False}, "full-stem-default-not-off"), # boolean, not the string
({"default": True}, "full-stem-default-on"), # boolean truthy → plays
({"default": "false"}, "full-stem-default-not-off"), # off-ish but non-canonical
({"default": "0"}, "full-stem-default-not-off"),
({"default": "no"}, "full-stem-default-not-off"),
({"default": "maybe"}, "full-stem-default-not-off"), # malformed
({"default": "on"}, "full-stem-default-on"),
({"default": "yes"}, "full-stem-default-on"),
({"default": "1"}, "full-stem-default-on"),
],
)
def test_verify_requires_an_explicit_off_on_a_retained_mixdown(tmp_path, default, expected):
"""Beside instrument stems, `full` is safe only with an explicit normalized
`off`. Core defaults an ABSENT `default` to ON and treats empty/unknown as
ON, so a missing or blank default is the double-audio hazard itself, not a
lesser one — `verify` must not certify it."""
m = _manifest(
stems=[
{"id": "full", "file": "stems/full.ogg", **default},
{"id": "guitar", "file": "stems/guitar.ogg", "default": "on"},
]
)
del m["original_audio"]
pak = _write_pack(
tmp_path / f"{tmp_path.name}.feedpak",
m,
files={"stems/full.ogg": b"M", "stems/guitar.ogg": b"g"},
)
assert mig.verify_zip(pak) == expected
def test_verify_ignores_default_on_a_sole_full_stem(tmp_path: Path):
"""A single `full` stem IS the audio — the len>1 gate means its default is
not policed, so an on/absent default is fine (off would mute the pack)."""
for default in ({"default": "on"}, {}, {"default": ""}):
m = _manifest(stems=[{"id": "full", "file": "stems/full.ogg", **default}])
del m["original_audio"]
pak = _write_pack(
tmp_path / f"{tmp_path.name}-{len(default)}.feedpak",
m,
files={"stems/full.ogg": b"M"},
)
assert mig.verify_zip(pak) == "ok"
def test_verify_rejects_an_unmigrated_pack(tmp_path: Path):
pak = _write_pack(tmp_path / "song.feedpak", _manifest())
assert mig.verify_zip(pak) == "still-has-key"
# ── Unsafe manifest paths must not be laundered into playable audio ─────────
@pytest.mark.parametrize(
"rel", ["../outside.ogg", "/etc/passwd", "a/../../x.ogg", "C:/x.ogg", "a\\b.ogg"]
)
def test_migrate_refuses_an_unsafe_full_mix_path(tmp_path: Path, rel: str):
"""Core's loader REFUSES a full-mix path that escapes the pack — such a pack
simply has no full mix, and the audio is inert. Migrating it into
`stems/full.ogg` would take content the reader deliberately rejected and hand
it back as a valid, playable stem. Report it; never promote it."""
pak = _write_pack(
tmp_path / "song.feedpak",
_manifest(original_audio=rel),
files={rel: b"EVIL", "stems/guitar.ogg": b"g"},
)
before = pak.read_bytes()
assert mig.migrate_zip(pak, dry_run=False) == "unsafe-path"
assert pak.read_bytes() == before
def test_safe_relpath_accepts_ordinary_pack_paths():
assert mig.is_safe_relpath("stems/full.ogg")
assert mig.is_safe_relpath("original/full.ogg")
assert not mig.is_safe_relpath("")
assert not mig.is_safe_relpath("a//b.ogg")
# ── Damaged packs must not abort the run ────────────────────────────────────
def test_a_corrupt_archive_is_reported_not_fatal(tmp_path: Path, capsys):
"""A real library has damage in it — a truncated download, an archive left
half-written by an interrupted converter. One of those must not kill a
50,000-pack run and throw away the summary: the pack is reported, skipped,
and everything else still migrates."""
good = _write_pack(tmp_path / "good.feedpak", _manifest())
bad = tmp_path / "bad.feedpak"
bad.write_bytes(b"this is not a zip file at all")
rc = mig.main([str(tmp_path)])
out = capsys.readouterr().out
assert rc == 1 # a problem pack fails the run's exit code
assert "corrupt-zip" in out
assert "migrated" in out
assert mig.verify_zip(good) == "ok" # the healthy pack still got migrated
assert bad.read_bytes() == b"this is not a zip file at all" # untouched
# ── Directory-form (authoring) packs are discovered, not silently skipped ────
def _write_dir_pack(path: Path, manifest: dict, files: dict[str, bytes] | None = None) -> Path:
"""Build a directory-form pack (`song.sloppak/`), the authoring shape."""
files = files or {
"original/full.ogg": b"MIXDOWN",
"stems/guitar.ogg": b"g",
"stems/drums.ogg": b"d",
"arrangements/lead.json": b"{}",
}
path.mkdir()
(path / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False))
for name, data in files.items():
p = path / name
p.parent.mkdir(parents=True, exist_ok=True)
p.write_bytes(data)
return path
def test_iter_packs_discovers_directory_form_packs(tmp_path: Path):
"""A `song.sloppak/` directory is a pack; os.walk must yield it whole and
NOT descend into it (its stems/ are contents, not packs)."""
d = _write_dir_pack(tmp_path / "song.sloppak", _manifest())
z = _write_pack(tmp_path / "other.feedpak", _manifest())
found = set(mig.iter_packs(tmp_path))
assert d in found and z in found
# Nothing inside the directory pack was yielded as its own pack.
assert not any(d in p.parents for p in found)
def test_iter_packs_yields_a_directly_passed_dir_pack(tmp_path: Path):
d = _write_dir_pack(tmp_path / "song.sloppak", _manifest())
assert list(mig.iter_packs(d)) == [d]
def test_directory_form_pack_is_reported_not_silently_skipped(tmp_path: Path, capsys):
"""The migrator rewrites single-file packs atomically; a directory can't be
swapped that way, so it is surfaced as a problem rather than vanishing from
the run (the silent-skip this guards against) or being rewritten unsafely."""
d = _write_dir_pack(tmp_path / "song.sloppak", _manifest())
good = _write_pack(tmp_path / "good.feedpak", _manifest())
rc = mig.main([str(tmp_path)])
out = capsys.readouterr().out
assert rc == 1 # a reported problem fails the exit code
assert "dir-form-unsupported" in out
assert mig.verify_zip(good) == "ok" # the zip pack still migrated
# The directory pack is untouched: legacy key intact, mixdown not moved.
manifest = yaml.safe_load((d / "manifest.yaml").read_text())
assert manifest.get("original_audio") == "original/full.ogg"
assert (d / "original" / "full.ogg").read_bytes() == b"MIXDOWN"
def test_verify_reports_directory_form_packs(tmp_path: Path):
d = _write_dir_pack(tmp_path / "song.sloppak", _manifest())
assert mig.verify_pack(d) == "dir-form-unsupported"
tests/test_sloppak_full_mix_load.py
+277
View File
@@ -0,0 +1,277 @@
"""The sloppak loader's handling of a pack's complete mixdown (#933).
The mixdown is a stem: feedpak spec §5.3 RESERVES the id `full` for it. It is a
mixdown, not a layer — it already contains every instrument, so a reader that
sums `stems` must never include it in that sum, and `load_song()` therefore
lifts it OUT of `LoadedSloppak.stems` and onto `LoadedSloppak.full_mix`.
Also covers the DEPRECATED `original_audio:` manifest key — a key this repo
invented (#583) before the spec reserved `full`, which every pack in the wild
still carries. We read it as a fallback so those packs keep their full mix; we
never write it. Those tests are the deprecation contract: they go when the key
does (#945).
"""
from __future__ import annotations
import json
from pathlib import Path
import yaml
import sloppak as sloppak_mod
def _write_dir_sloppak(
root: Path,
manifest_extras: dict,
*,
write_legacy_full_mix: bool = False,
stems: list[dict] | None = None,
) -> Path:
"""Build a minimal directory-form sloppak that load_song will accept.
Uses the tmp_path leaf name to make the sloppak filename unique per test,
avoiding the module-level ``resolve_source_dir`` cache being poisoned by a
previous test that happened to share the same "song.sloppak" filename.
"""
pak = root / f"{root.name}.sloppak"
pak.mkdir()
arr_dir = pak / "arrangements"
arr_dir.mkdir()
arr = {
"name": "Lead",
"tuning": [0, 0, 0, 0, 0, 0],
"capo": 0,
"notes": [],
"chords": [],
"anchors": [],
"handshapes": [],
"templates": [],
"beats": [],
"sections": [],
}
(arr_dir / "lead.json").write_text(json.dumps(arr))
manifest = {
"title": "Test",
"artist": "Tester",
"album": "",
"year": 2026,
"duration": 10.0,
"arrangements": [{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"}],
"stems": (
stems
if stems is not None
else [{"id": "guitar", "file": "stems/guitar.ogg", "default": True}]
),
}
manifest.update(manifest_extras)
(pak / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False))
if write_legacy_full_mix:
orig_dir = pak / "original"
orig_dir.mkdir()
# The loader only checks presence (is_file); contents are irrelevant.
(orig_dir / "full.ogg").write_bytes(b"OggS-not-real")
return pak
def _load(pak_path: Path, tmp_path: Path):
dlc_root = pak_path.parent
cache = tmp_path / "cache"
cache.mkdir()
return sloppak_mod.load_song(pak_path.name, dlc_root, cache)
def _separated(**extra) -> list[dict]:
"""A separated pack that RETAINS its mixdown, as spec §5.3 asks writers to."""
return [
{"id": "full", "file": "stems/full.ogg", "default": False, **extra},
{"id": "guitar", "file": "stems/guitar.ogg", "default": True},
{"id": "drums", "file": "stems/drums.ogg", "default": True},
]
# ── The `full` stem is the mixdown (spec §5.3) ───────────────────────────────
def test_full_stem_is_surfaced_as_the_mixdown(tmp_path: Path):
pak = _write_dir_sloppak(tmp_path, {}, stems=_separated())
loaded = _load(pak, tmp_path)
# Manifest-relative, so the WS builds its URL exactly as it builds a stem's.
assert loaded.full_mix == "stems/full.ogg"
def test_full_stem_is_removed_from_the_stem_list(tmp_path: Path):
"""The regression this whole change exists to prevent.
Every consumer sums `stems` into one mix and renders one fader per entry. The
mixdown already contains every instrument, so leaving it in the list doubles
the entire song — and muting `guitar` would still leave guitar audible inside
it. That exact trap is why the packer invented `original_audio` rather than
putting the mixdown where the format says it goes.
"""
pak = _write_dir_sloppak(tmp_path, {}, stems=_separated())
loaded = _load(pak, tmp_path)
assert [s["id"] for s in loaded.stems] == ["guitar", "drums"]
def test_single_mix_pack_keeps_full_as_its_only_stem(tmp_path: Path):
"""A pack whose ONLY stem is `full` is a single-mix pack, not a separated one.
There are no instruments to be pristine against, so the mixdown stays the sole
playable stem and nothing is surfaced separately. Anything else would strip the
stem list of the most common pack shape in the library and leave it silent.
"""
pak = _write_dir_sloppak(
tmp_path, {}, stems=[{"id": "full", "file": "stems/full.ogg", "default": True}]
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
assert [s["id"] for s in loaded.stems] == ["full"]
def test_every_full_entry_is_removed_not_just_the_first(tmp_path: Path):
"""A malformed pack listing `full` twice must not leave one behind.
Removing the mixdown by object identity would drop only the entry we surface
and leave its duplicate in the stem list — a whole copy of the song, summed
with the instruments. That is the exact bug this partition prevents, so a
duplicate must not smuggle it back in.
"""
pak = _write_dir_sloppak(
tmp_path,
{},
stems=[
{"id": "full", "file": "stems/full.ogg", "default": False},
{"id": "guitar", "file": "stems/guitar.ogg", "default": True},
{"id": "full", "file": "original/full.ogg", "default": True},
],
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix == "stems/full.ogg"
assert [s["id"] for s in loaded.stems] == ["guitar"]
def test_separated_pack_without_a_full_stem_has_no_mixdown(tmp_path: Path):
"""Stems only, mixdown discarded — the pre-1.15.0 shape. Nothing to surface."""
pak = _write_dir_sloppak(
tmp_path,
{},
stems=[
{"id": "guitar", "file": "stems/guitar.ogg", "default": True},
{"id": "drums", "file": "stems/drums.ogg", "default": True},
],
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
assert [s["id"] for s in loaded.stems] == ["guitar", "drums"]
def test_single_mix_pack_ignores_a_lingering_deprecated_key(tmp_path: Path):
"""`full` is the pack's only stem AND the old key is still there.
The stem wins, and it stays the sole playable stem — falling back to the key
would surface the mixdown twice: once as the stem the player is already
playing, and once as a "pristine" track for it to cross over to.
"""
pak = _write_dir_sloppak(
tmp_path,
{"original_audio": "original/full.ogg"},
stems=[{"id": "full", "file": "stems/full.ogg", "default": True}],
write_legacy_full_mix=True,
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
assert [s["id"] for s in loaded.stems] == ["full"]
def test_full_stem_wins_over_the_deprecated_key(tmp_path: Path):
"""A migrated pack that still carries the old key must use the stem."""
pak = _write_dir_sloppak(
tmp_path,
{"original_audio": "original/full.ogg"},
stems=_separated(),
write_legacy_full_mix=True,
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix == "stems/full.ogg"
# ── The library index must not advertise the mixdown as an instrument ────────
def test_extract_meta_excludes_the_mixdown_from_stem_ids(tmp_path: Path):
"""The library's stem chips / stem_count come from here, and must agree with
load_song() — otherwise the filter offers a "full" chip beside guitar+drums
and counts a third stem that no mixer will ever show."""
pak = _write_dir_sloppak(tmp_path, {}, stems=_separated())
meta = sloppak_mod.extract_meta(pak)
assert meta["stem_ids"] == ["guitar", "drums"]
assert meta["stem_count"] == 2
def test_extract_meta_keeps_full_for_a_single_mix_pack(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {}, stems=[{"id": "full", "file": "stems/full.ogg", "default": True}]
)
meta = sloppak_mod.extract_meta(pak)
assert meta["stem_ids"] == ["full"]
assert meta["stem_count"] == 1
# ── DEPRECATED `original_audio:` fallback — delete with the key (#945) ───────
def test_legacy_key_still_provides_the_full_mix(tmp_path: Path):
"""Every pack written before the spec reserved `full` looks like this. Dropping
the read would silently take the pristine mix away from all of them."""
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "original/full.ogg"}, write_legacy_full_mix=True
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix == "original/full.ogg"
# The legacy mixdown lives OUTSIDE `stems`, so the stem list is untouched.
assert [s["id"] for s in loaded.stems] == ["guitar"]
def test_legacy_key_absent_means_no_full_mix(tmp_path: Path):
pak = _write_dir_sloppak(tmp_path, {}, write_legacy_full_mix=True)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
def test_legacy_key_none_when_file_missing(tmp_path: Path):
# Manifest points at a full mix that isn't on disk — disabled silently.
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "original/full.ogg"}, write_legacy_full_mix=False
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
def test_legacy_key_none_when_value_blank(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {"original_audio": " "}, write_legacy_full_mix=True
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
# ── Security / path-traversal branches (legacy key only — a stem `file` is
# resolved through the same /api/sloppak/.../file/ guard as every other stem)
def test_legacy_key_none_when_path_escapes_sloppak(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "../outside.ogg"}, write_legacy_full_mix=True
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
def test_legacy_key_none_when_path_is_absolute(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "/etc/passwd"}, write_legacy_full_mix=True
)
loaded = _load(pak, tmp_path)
assert loaded.full_mix is None
tests/test_sloppak_original_audio_load.py
-119
View File
@@ -1,119 +0,0 @@
"""End-to-end test for the sloppak loader recognising an `original_audio:`
manifest key (the single full-mix file shipped alongside the separate stems)
and surfacing the manifest-relative path on the LoadedSloppak."""
from __future__ import annotations
import json
from pathlib import Path
import yaml
import sloppak as sloppak_mod
def _write_dir_sloppak(root: Path, manifest_extras: dict, *, write_full_mix: bool) -> Path:
"""Build a minimal directory-form sloppak that load_song will accept.
Uses the tmp_path leaf name to make the sloppak filename unique per test,
avoiding the module-level ``resolve_source_dir`` cache being poisoned by a
previous test that happened to share the same "song.sloppak" filename.
"""
pak = root / f"{root.name}.sloppak"
pak.mkdir()
arr_dir = pak / "arrangements"
arr_dir.mkdir()
arr = {
"name": "Lead",
"tuning": [0, 0, 0, 0, 0, 0],
"capo": 0,
"notes": [],
"chords": [],
"anchors": [],
"handshapes": [],
"templates": [],
"beats": [],
"sections": [],
}
(arr_dir / "lead.json").write_text(json.dumps(arr))
manifest = {
"title": "Test",
"artist": "Tester",
"album": "",
"year": 2026,
"duration": 10.0,
"arrangements": [{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"}],
"stems": [{"id": "guitar", "file": "stems/guitar.ogg", "default": True}],
}
manifest.update(manifest_extras)
(pak / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False))
if write_full_mix:
orig_dir = pak / "original"
orig_dir.mkdir()
# The loader only checks presence (is_file); contents are irrelevant.
(orig_dir / "full.ogg").write_bytes(b"OggS-not-real")
return pak
def _load(pak_path: Path, tmp_path: Path):
dlc_root = pak_path.parent
cache = tmp_path / "cache"
cache.mkdir()
return sloppak_mod.load_song(pak_path.name, dlc_root, cache)
# ── Happy path ───────────────────────────────────────────────────────────────
def test_load_song_attaches_original_audio_when_manifest_opts_in(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "original/full.ogg"}, write_full_mix=True
)
loaded = _load(pak, tmp_path)
# Stored as the manifest-relative string so server.py can build the URL the
# same way it builds stem URLs.
assert loaded.original_audio == "original/full.ogg"
# ── Absent / degraded branches ───────────────────────────────────────────────
def test_load_song_original_audio_none_when_manifest_silent(tmp_path: Path):
pak = _write_dir_sloppak(tmp_path, {}, write_full_mix=True)
loaded = _load(pak, tmp_path)
assert loaded.original_audio is None
def test_load_song_original_audio_none_when_file_missing(tmp_path: Path):
# Manifest points at a full mix that isn't on disk — disabled silently.
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "original/full.ogg"}, write_full_mix=False
)
loaded = _load(pak, tmp_path)
assert loaded.original_audio is None
def test_load_song_original_audio_none_when_value_blank(tmp_path: Path):
pak = _write_dir_sloppak(tmp_path, {"original_audio": " "}, write_full_mix=True)
loaded = _load(pak, tmp_path)
assert loaded.original_audio is None
# ── Security / path-traversal branches ──────────────────────────────────────
def test_load_song_original_audio_none_when_path_escapes_sloppak(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "../outside.ogg"}, write_full_mix=True
)
loaded = _load(pak, tmp_path)
assert loaded.original_audio is None
def test_load_song_original_audio_none_when_path_is_absolute(tmp_path: Path):
pak = _write_dir_sloppak(
tmp_path, {"original_audio": "/etc/passwd"}, write_full_mix=True
)
loaded = _load(pak, tmp_path)
assert loaded.original_audio is None
tools/migrate_full_mix_stem.py
+445
View File
@@ -0,0 +1,445 @@
#!/usr/bin/env python3
"""Migrate packs off the deprecated `original_audio:` key — the full mix is a stem.
Before feedpak 1.15.0 reserved the stem id `full`, spec §5.3 said the mixdown was
"commonly replaced" by the per-instrument stems when a pack was split — so after
separation it had nowhere to live. This repo worked around that by inventing a
top-level `original_audio:` manifest key pointing at a parallel `original/`
directory (#583). The key was never in the spec, and #933 removed core's
dependence on it: the mixdown is a stem, and its id is `full`.
This rewrites a pack into the shape the spec now defines:
original/full.ogg -> stems/full.ogg (entry moved)
original_audio: original/full.ogg -> stems: [{id: full, file: stems/full.ogg,
default: 'off'}, ...]
`default: 'off'` is what makes the retained mixdown safe: a reader that honours
`default` (normative since feedpak 1.0.0) will not play it on open, so it never
doubles the mix even in a reader that predates the reserved id.
Nothing else in the pack is touched — every other key, file and stem is preserved
verbatim, and `feedpak_version` is stamped to the version the result conforms to.
The rewrite is atomic per pack: a new archive is built beside the original and
renamed over it only on success, so an interrupted run leaves every pack either
fully migrated or untouched — never truncated.
Idempotent: a pack that already carries a `full` stem and no `original_audio:` is
reported as `skip` and left alone, so a partial run can simply be re-run.
Usage:
python tools/migrate_full_mix_stem.py --dry-run /path/to/packs # report only
python tools/migrate_full_mix_stem.py /path/to/packs # migrate
python tools/migrate_full_mix_stem.py --verify /path/to/packs # check results
Exit status is 0 only when every pack ended up in the migrated shape (or was
already there).
"""
from __future__ import annotations
import argparse
import os
import shutil
import sys
import tempfile
import threading
import time
import zipfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
import yaml
# The version this migration brings a pack up to: the one that reserved `full`.
TARGET_FEEDPAK_VERSION = "1.15.0"
FULL_MIX_STEM_ID = "full"
LEGACY_KEY = "original_audio"
# Where the mixdown lands. §2.1's conventional layout — readers resolve through
# the manifest and never care about the path, but a pack that says `stems/` and
# means it is the one a human can read.
CANONICAL_FULL_MIX_PATH = "stems/full.ogg"
PACK_EXTS = (".feedpak", ".sloppak")
class Skip(Exception):
"""Pack needs no migration."""
def is_safe_relpath(rel: str) -> bool:
"""True when `rel` is a manifest path the spec allows (§2.2 rule 2).
POSIX-style relative: forward slashes, no leading `/`, no `..` segments, no
empty segments, no colon (which excludes drive letters and NTFS alternate
data streams), no backslashes.
This is a TRUST BOUNDARY, not a tidiness check. Core's loader refuses a
full-mix path that escapes the pack and reports the pack as having no full
mix — the audio is inert. A migration that moved such an entry into
`stems/full.ogg` would take content the reader deliberately rejected and
hand it back as a valid, playable stem. So a pack like this is reported, not
migrated.
"""
if not rel or rel.startswith("/") or "\\" in rel or ":" in rel:
return False
parts = rel.split("/")
return all(p and p != ".." for p in parts)
def plan_manifest(manifest: dict) -> tuple[dict, str]:
"""Return (new_manifest, relpath_of_audio_to_move); "" = no file needs moving.
Raises Skip when the pack needs no migration. Pure — no I/O — so the part
with the decisions in it is testable without building archives.
"""
raw_stems = manifest.get("stems")
stems: list = raw_stems if isinstance(raw_stems, list) else []
has_full_stem = any(
isinstance(s, dict) and str(s.get("id", "")) == FULL_MIX_STEM_ID for s in stems
)
legacy_rel = manifest.get(LEGACY_KEY)
legacy_rel = legacy_rel.strip() if isinstance(legacy_rel, str) else ""
if not legacy_rel:
# Nothing invented to undo: either the pack already keeps its mixdown as
# the `full` stem, or it never carried one.
raise Skip("already migrated" if has_full_stem else "no original_audio key")
if has_full_stem:
# The mixdown is already a stem and the dead key merely lingers beside it.
# Drop the key; move nothing. But do NOT trust its `default`: a mixdown
# left enabled beside instrument stems is the double-audio hazard this
# migration exists to remove, and a reader that honours `default` would
# play the whole song on top of the stems on open. Force it off — unless
# `full` is the only stem, in which case it IS the audio.
others = [
s
for s in stems
if isinstance(s, dict) and str(s.get("id", "")) != FULL_MIX_STEM_ID
]
new_stems = []
for s in stems:
if isinstance(s, dict) and str(s.get("id", "")) == FULL_MIX_STEM_ID and others:
s = {**s, "default": "off"}
new_stems.append(s)
to_move = ""
else:
# `default` decides whether a reader plays this on open, and that is the
# whole safety margin: alongside per-instrument stems the mixdown must be
# OFF (a reader that sums the list would otherwise double the song), but
# when it is the pack's only stem it IS the audio and must be ON.
entry = {
"id": FULL_MIX_STEM_ID,
"file": CANONICAL_FULL_MIX_PATH,
"default": "off" if stems else "on",
}
# First in the list, matching the spec's §5.3 example.
new_stems = [entry, *stems]
# If the key already pointed at the canonical path, only the manifest is wrong.
to_move = "" if legacy_rel == CANONICAL_FULL_MIX_PATH else legacy_rel
out: dict = {}
for k, v in manifest.items():
if k == LEGACY_KEY:
continue # the invented key disappears
out[k] = new_stems if k == "stems" else v
out.setdefault("stems", new_stems) # a pack that had no stems list gets one
out["feedpak_version"] = TARGET_FEEDPAK_VERSION
return out, to_move
def migrate_zip(path: Path, dry_run: bool) -> str:
"""Rewrite one zipped pack in place. Returns a one-word status."""
with zipfile.ZipFile(path) as zf:
try:
raw = zf.read("manifest.yaml")
except KeyError:
return "no-manifest"
manifest = yaml.safe_load(raw) or {}
try:
new_manifest, old_rel = plan_manifest(manifest)
except Skip:
return "skip"
names = set(zf.namelist())
if old_rel:
if not is_safe_relpath(old_rel):
# Core refuses this path and plays no full mix for the pack. Do
# not launder it into a valid stem — see is_safe_relpath().
return "unsafe-path"
if old_rel not in names:
# The key points at audio that isn't in the archive. Core already
# treats that as "no full mix"; migrating would fabricate a stem
# entry for a file that does not exist and break every reader.
return "missing-audio"
if CANONICAL_FULL_MIX_PATH in names:
return "target-occupied"
else:
# Manifest-only rewrite (a stale key beside a mixdown that is already
# a stem, or a key that already pointed at the canonical path). Check
# the file the resulting `full` stem will actually NAME — not the
# canonical path, which an already-migrated pack is free not to use:
# §2.2 says readers resolve through the manifest, so a valid pack may
# keep its mixdown anywhere.
full_file = next(
(
s.get("file")
for s in new_manifest.get("stems", [])
if isinstance(s, dict) and str(s.get("id", "")) == FULL_MIX_STEM_ID
),
None,
)
if full_file not in names:
return "missing-audio"
if dry_run:
return "would-migrate"
# Build the replacement beside the original, on the same filesystem, so
# the final rename is atomic and an interrupted run can't truncate a pack.
tmp_fd, tmp_name = tempfile.mkstemp(
dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp"
)
os.close(tmp_fd)
tmp_path = Path(tmp_name)
try:
with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as out:
for item in zf.infolist():
if item.filename == "manifest.yaml":
out.writestr(
item,
yaml.safe_dump(
new_manifest, sort_keys=False, allow_unicode=True
),
)
continue
data = zf.read(item.filename)
if old_rel and item.filename == old_rel:
# Same bytes, same compression, new name: the mixdown moves
# from original/ into stems/ where the format says audio goes.
moved = zipfile.ZipInfo(
CANONICAL_FULL_MIX_PATH, date_time=item.date_time
)
moved.compress_type = item.compress_type
moved.external_attr = item.external_attr
out.writestr(moved, data)
continue
out.writestr(item, data)
except BaseException:
tmp_path.unlink(missing_ok=True)
raise
shutil.copystat(path, tmp_path)
os.replace(tmp_path, path) # atomic
return "migrated"
def verify_zip(path: Path) -> str:
"""Confirm a pack is in the migrated shape and its mixdown is really there."""
with zipfile.ZipFile(path) as zf:
try:
manifest = yaml.safe_load(zf.read("manifest.yaml")) or {}
except KeyError:
return "no-manifest"
if LEGACY_KEY in manifest:
return "still-has-key"
stems = manifest.get("stems") or []
full = next(
(
s
for s in stems
if isinstance(s, dict) and str(s.get("id", "")) == FULL_MIX_STEM_ID
),
None,
)
if full is None:
return "no-full-stem"
if full.get("file") not in set(zf.namelist()):
return "full-stem-missing-file"
# A retained mixdown that plays on open would double the mix in any reader
# that sums the stem list — the whole hazard this migration must not create.
# Beside instrument stems, `full` MUST carry an explicit, normalized "off":
# - core (lib/sloppak.py) defaults an ABSENT `default` to True (ON) and
# treats an empty / unrecognized string as ON, so a missing or blank
# default is not merely non-canonical — core would play the mixdown on
# open, doubling the song. It is the exact hazard, not a lesser one.
# - the migrator always writes the literal "off", so requiring it also
# certifies the pack is in the shape this tool produces — the most
# portable spelling, understood even by a reader that only knows
# "on"/"off" and would choke on a boolean or `false`/`0`/`no`.
# So: `on`-ish values are reported as actively-playing; everything that is
# not a normalized "off" (missing, empty, boolean, `false`/`no`/`0`,
# malformed) is reported as an unsafe/non-canonical default.
if len(stems) > 1:
default = str(full.get("default", "")).strip().lower()
if default in ("true", "on", "yes", "1"):
return "full-stem-default-on"
if default != "off":
return "full-stem-default-not-off"
return "ok"
def migrate_pack(path: Path, dry_run: bool) -> str:
"""Dispatch by pack form. ZIP-file packs are rewritten in place; directory
(authoring) packs are REPORTED, not rewritten.
A single-file pack is replaced atomically — a fully-built temp archive
swapped in with one os.replace(), so an interrupted run leaves it either
fully migrated or untouched. A directory can't be swapped that way (no
atomic replace of a populated directory), so an in-place rewrite could leave
an authoring pack half-migrated. Rather than risk that, directory packs are
surfaced as `dir-form-unsupported` (a problem status, so the run's exit code
and summary flag them) for the operator to re-pack or migrate as a `.feedpak`.
"""
if path.is_dir():
return "dir-form-unsupported"
return migrate_zip(path, dry_run)
def verify_pack(path: Path) -> str:
"""Verify a pack; directory (authoring) packs are reported, see migrate_pack."""
if path.is_dir():
return "dir-form-unsupported"
return verify_zip(path)
def iter_packs(root: Path):
"""Yield every pack under `root`. A pack is a suffix-named ZIP FILE or a
suffix-named DIRECTORY (the authoring form) — both are discovered so a
directory-form pack is never silently walked past. A directory pack is
yielded whole, not descended into: its `stems/` and `arrangements/` are pack
contents, not packs. (migrate/verify then report directory packs rather than
rewriting them in place — see migrate_pack.)"""
if root.is_file():
yield root
return
# A directory whose OWN name is a pack suffix is a single directory-form
# pack passed directly, not a tree of packs to search.
if root.name.endswith(PACK_EXTS):
yield root
return
for dirpath, dirnames, filenames in os.walk(root):
for fn in sorted(filenames):
if fn.endswith(PACK_EXTS):
yield Path(dirpath) / fn
for dn in sorted(dn for dn in dirnames if dn.endswith(PACK_EXTS)):
yield Path(dirpath) / dn
# Don't descend INTO a pack directory — its contents aren't packs.
dirnames[:] = [dn for dn in dirnames if not dn.endswith(PACK_EXTS)]
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("root", type=Path, help="pack, or directory of packs")
ap.add_argument("--dry-run", action="store_true", help="report, change nothing")
ap.add_argument("--verify", action="store_true", help="check the migrated shape")
ap.add_argument("--jobs", type=int, default=8, help="parallel packs (default 8)")
args = ap.parse_args(argv)
if not args.root.exists():
print(f"error: {args.root} does not exist", file=sys.stderr)
return 2
packs = list(iter_packs(args.root))
if not packs:
print(f"no packs found under {args.root}", file=sys.stderr)
return 2
action = verify_pack if args.verify else (lambda p: migrate_pack(p, args.dry_run))
def work(pack: Path) -> str:
"""Never raise. One unreadable pack must not kill a 50,000-pack run.
A library this size has damage in it — a truncated download, an archive
left half-written by an interrupted converter. Letting that propagate
aborts the whole job partway through and throws away the summary, which
is exactly when you most need to know what happened. Report it as a
problem status instead: the pack is untouched, the run continues, and the
final report names it.
"""
try:
return action(pack)
except zipfile.BadZipFile:
return "corrupt-zip"
except OSError as e:
return f"io-error ({e.__class__.__name__})"
except Exception as e: # malformed YAML, unexpected manifest shape, …
return f"error ({e.__class__.__name__})"
counts: dict[str, int] = {}
problems: list[tuple[str, Path]] = []
# A real run rewrites every archive under `root` — tens of thousands of packs
# and hundreds of gigabytes. Printing only a final summary means hours of
# silence, in which a stall and steady progress look identical. Emit a
# heartbeat instead: rate and ETA come from the packs actually finished, so
# it stays honest when the disk slows down. stderr, so `> report.txt` keeps
# the summary clean.
total = len(packs)
started = time.monotonic()
# The heartbeat runs on its OWN CLOCK, in its own thread.
#
# Two weaker designs were tried and both go quiet exactly when you need them
# to speak. Ticking every N packs ties the cadence to how slow a pack is: 500
# packs is a blink in a --dry-run and many minutes in a real migration, so the
# run that most needs watching says the least. Ticking on time but only when a
# pack *finishes* is no better: if every worker is grinding on a huge archive,
# nothing completes, so nothing prints — and a stall becomes indistinguishable
# from progress, which is the one thing a progress meter must never allow.
#
# A daemon thread on a fixed interval reports regardless. If the count stops
# advancing between beats, you are looking at a stall, and you can see it.
HEARTBEAT_SECONDS = 10.0
done = 0 # only the main loop writes it; the beat thread only reads
stop_beat = threading.Event()
def heartbeat() -> None:
while not stop_beat.wait(HEARTBEAT_SECONDS):
elapsed = time.monotonic() - started
rate = done / elapsed if elapsed > 0 else 0.0
eta = (total - done) / rate if rate > 0 else 0.0
print(
f" {done}/{total} ({100 * done / total:.1f}%) "
f"{rate:.1f} packs/s eta {eta / 60:.0f}m "
f"[{len(problems)} problem(s)]",
file=sys.stderr,
flush=True,
)
print(
f"{total} pack(s) under {args.root} — "
f"{'verifying' if args.verify else 'dry run' if args.dry_run else 'migrating'} "
f"with {max(1, args.jobs)} job(s)",
file=sys.stderr,
flush=True,
)
beat = threading.Thread(target=heartbeat, daemon=True)
beat.start()
# as_completed, not pool.map: map yields in SUBMISSION order, so the counter
# would stall behind one slow pack while later ones were already done — a
# progress meter that lies about progress. Count each pack as it finishes.
try:
with ThreadPoolExecutor(max_workers=max(1, args.jobs)) as pool:
futures = {pool.submit(work, p): p for p in packs}
for fut in as_completed(futures):
pack = futures[fut]
status = fut.result()
counts[status] = counts.get(status, 0) + 1
if status not in ("migrated", "skip", "would-migrate", "ok"):
problems.append((status, pack))
done += 1
finally:
stop_beat.set()
beat.join(timeout=1)
print(f"\n{len(packs)} pack(s) under {args.root}")
for status, n in sorted(counts.items(), key=lambda kv: -kv[1]):
print(f" {n:>7} {status}")
if problems:
print(f"\n{len(problems)} pack(s) need a look:", file=sys.stderr)
for status, pack in problems[:20]:
print(f" {status:<22} {pack}", file=sys.stderr)
if len(problems) > 20:
print(f" … and {len(problems) - 20} more", file=sys.stderr)
return 1 if problems else 0
if __name__ == "__main__":
raise SystemExit(main())