From 53dd94b20a9f91620e002695b099cc7a990e2f79 Mon Sep 17 00:00:00 2001 From: Xehanort88 Date: Wed, 23 Sep 2026 13:50:24 +0100 Subject: [PATCH 1/3] Add test suite Tests generate their own audio: an intro followed by a repeated 8-note pattern, so the correct loop points are known in advance and loop detection can be checked for correctness, not just for running. Covers audio loading, loop detection and scoring, zero-crossing snapping, split/extend/tag/txt exports, the interactive picker, batch file discovery and every CLI command. Known bugs are marked as strict xfail, so each fix is flagged until its marker is removed. Also adds pytest as a dev dependency and a CI workflow running the tests on Ubuntu and Windows with Python 3.10 and 3.13. --- .github/workflows/tests.yml | 25 +++++ pyproject.toml | 8 ++ tests/conftest.py | 84 +++++++++++++++++ tests/test_analysis.py | 176 ++++++++++++++++++++++++++++++++++++ tests/test_audio.py | 72 +++++++++++++++ tests/test_cli.py | 93 +++++++++++++++++++ tests/test_core.py | 116 ++++++++++++++++++++++++ tests/test_handler.py | 82 +++++++++++++++++ uv.lock | 110 ++++++++++++++++++++++ 9 files changed, 766 insertions(+) create mode 100644 .github/workflows/tests.yml create mode 100644 tests/conftest.py create mode 100644 tests/test_analysis.py create mode 100644 tests/test_audio.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_core.py create mode 100644 tests/test_handler.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..bfadfed --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,25 @@ +name: Tests + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + python-version: ["3.10", "3.13"] + steps: + - uses: actions/checkout@v4 + - name: Set up uv + uses: astral-sh/setup-uv@v6 + with: + python-version: ${{ matrix.python-version }} + - name: Run tests + run: uv run pytest diff --git a/pyproject.toml b/pyproject.toml index 9b37b98..1fc9414 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,3 +64,11 @@ select = [ # isort "I", ] + +[dependency-groups] +dev = [ + "pytest>=9.1.1", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ef930f1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,84 @@ +"""Shared fixtures: synthetic tracks whose correct loop points are known in advance. + +The test track is an intro followed by the same 8-note pattern repeated several times, +so any loop whose length is a whole number of patterns is seamless. +""" + +import numpy as np +import pytest +import soundfile as sf + +SR = 22050 +BEAT_SAMPLES = SR // 2 # 120 bpm +INTRO_NOTES = [50, 53, 57, 59] +PATTERN_NOTES = [60, 64, 67, 72, 69, 65, 62, 67] +PATTERN_REPEATS = 4 + +INTRO_SAMPLES = len(INTRO_NOTES) * BEAT_SAMPLES +PATTERN_SAMPLES = len(PATTERN_NOTES) * BEAT_SAMPLES + +# Loop points are located on ~23ms STFT frames, then nudged to a zero crossing (+/-5ms) +SAMPLE_TOLERANCE = int(0.03 * SR) + + +def _note(midi: int) -> np.ndarray: + t = np.arange(BEAT_SAMPLES) / SR + freq = 440.0 * 2 ** ((midi - 69) / 12) + tone = sum(np.sin(2 * np.pi * freq * h * t) / h for h in range(1, 5)) + attack = np.minimum(1.0, t / 0.005) + return tone * attack * np.exp(-4 * t) + + +def make_track() -> np.ndarray: + notes = INTRO_NOTES + PATTERN_NOTES * PATTERN_REPEATS + PATTERN_NOTES[:1] + y = np.concatenate([_note(n) for n in notes]) + return (0.5 * y / np.max(np.abs(y))).astype(np.float32) + + +@pytest.fixture(scope="session") +def track() -> np.ndarray: + return make_track() + + +@pytest.fixture(scope="session") +def audio_dir(tmp_path_factory): + return tmp_path_factory.mktemp("audio") + + +@pytest.fixture(scope="session") +def track_path(audio_dir, track): + path = audio_dir / "track.wav" + sf.write(path, track, SR, subtype="FLOAT") + return str(path) + + +@pytest.fixture(scope="session") +def stereo_track_path(audio_dir, track): + path = audio_dir / "stereo.wav" + sf.write(path, np.stack([track, 0.8 * track], axis=1), SR, subtype="FLOAT") + return str(path) + + +@pytest.fixture(scope="session") +def silent_path(audio_dir): + path = audio_dir / "silent.wav" + sf.write(path, np.zeros(SR, dtype=np.float32), SR) + return str(path) + + +@pytest.fixture +def flac_track_path(tmp_path, track): + """A fresh FLAC copy per test, since tagging tests write next to it.""" + path = tmp_path / "track.flac" + sf.write(path, track, SR, format="FLAC") + return str(path) + + +def assert_whole_patterns(loop_start: int, loop_end: int): + """Asserts that a loop spans a whole number of repeated patterns.""" + length = loop_end - loop_start + n_patterns = round(length / PATTERN_SAMPLES) + assert n_patterns >= 1, f"loop of {length} samples is shorter than one pattern" + assert abs(length - n_patterns * PATTERN_SAMPLES) <= SAMPLE_TOLERANCE, ( + f"loop of {length} samples is not a multiple of the {PATTERN_SAMPLES}-sample pattern" + ) diff --git a/tests/test_analysis.py b/tests/test_analysis.py new file mode 100644 index 0000000..1682e5c --- /dev/null +++ b/tests/test_analysis.py @@ -0,0 +1,176 @@ +import numpy as np +import pytest +from conftest import INTRO_SAMPLES, SAMPLE_TOLERANCE, SR, assert_whole_patterns + +from pymusiclooper import analysis +from pymusiclooper.analysis import ( + LoopPair, + _calculate_loop_score, + _calculate_subseq_beat_similarity, + _prioritize_duration, + _weights, + nearest_zero_crossing, +) +from pymusiclooper.core import MusicLooper +from pymusiclooper.exceptions import LoopNotFoundError + + +@pytest.fixture(scope="module") +def looper(track_path): + return MusicLooper(track_path) + + +@pytest.fixture(scope="module") +def pairs(looper): + return looper.find_loop_pairs() + + +# --- End-to-end loop detection --- + + +def test_best_loop_spans_whole_patterns(pairs): + best = pairs[0] + assert_whole_patterns(best.loop_start, best.loop_end) + assert best.loop_start >= INTRO_SAMPLES - SAMPLE_TOLERANCE, "loop should not start inside the intro" + + +def test_best_loop_scores_highly(pairs): + assert pairs[0].score > 0.95 + + +def test_pairs_are_sorted_ordered_and_in_bounds(looper, pairs): + scores = [pair.score for pair in pairs[1:]] + assert scores == sorted(scores, reverse=True) + for pair in pairs: + assert 0 <= pair.loop_start < pair.loop_end <= looper.mlaudio.length + + +def test_default_min_duration_is_35_percent_of_track(looper, pairs): + min_samples = 0.35 * looper.mlaudio.length + assert all(p.loop_end - p.loop_start >= min_samples - SAMPLE_TOLERANCE - SR for p in pairs) + + +def test_min_and_max_loop_duration_respected(looper): + pairs = looper.find_loop_pairs(min_loop_duration=5, max_loop_duration=9) + for pair in pairs: + seconds = (pair.loop_end - pair.loop_start) / SR + assert 5 - 0.05 <= seconds <= 9 + 0.05 + assert_whole_patterns(pairs[0].loop_start, pairs[0].loop_end) + + +def test_approx_loop_position_respected(looper): + pairs = looper.find_loop_pairs(approx_loop_start=6.0, approx_loop_end=14.0) + best = pairs[0] + assert abs(best.loop_start / SR - 6.0) <= 2.05 + assert abs(best.loop_end / SR - 14.0) <= 2.05 + assert_whole_patterns(best.loop_start, best.loop_end) + + +def test_disable_pruning_keeps_at_least_as_many_pairs(looper, pairs): + unpruned = looper.find_loop_pairs(disable_pruning=True) + assert len(unpruned) >= len(pairs) + + +def test_impossible_constraints_raise(looper): + with pytest.raises(LoopNotFoundError): + looper.find_loop_pairs(min_loop_duration=60) + + +@pytest.mark.xfail(strict=True, reason="bug: _prioritize_duration runs before loop_start/loop_end are set, so it never reorders anything") +def test_prioritize_duration_sees_real_loop_positions(monkeypatch, looper): + durations_seen = [] + original = analysis._prioritize_duration + + def spy(pair_list): + durations_seen.extend(p.loop_end - p.loop_start for p in pair_list) + return original(pair_list) + + monkeypatch.setattr(analysis, "_prioritize_duration", spy) + looper.find_loop_pairs() + + assert durations_seen, "_prioritize_duration was not called" + assert max(durations_seen) > 0 + + +# --- Scoring --- + + +def _pair(start, end, score, loudness=0.1): + return LoopPair(0, 0, note_distance=0.0, loudness_difference=loudness, score=score, loop_start=start, loop_end=end) + + +def test_prioritize_duration_prefers_longest_among_tied_scores(): + pairs = [_pair(0, 100, 1.0), _pair(0, 300, 1.0), _pair(0, 900, 0.5)] + _prioritize_duration(pairs) + assert pairs[0].loop_end == 300 + + +def test_prioritize_duration_keeps_clearly_better_score_first(): + pairs = [_pair(0, 100, 0.99), _pair(0, 300, 0.90)] + _prioritize_duration(pairs) + assert pairs[0].loop_end == 100 + + +def _random_chroma(n_frames=200, seed=0): + return np.random.default_rng(seed).random((12, n_frames)) + + +def test_identical_sequences_score_one(): + chroma = _random_chroma() + chroma[:, 100:150] = chroma[:, 20:70] + score = _calculate_loop_score(30, 110, chroma, test_duration=20, weights=_weights(20, start=5)) + assert score == pytest.approx(1.0) + + +def test_unrelated_sequences_score_lower(): + chroma = _random_chroma() + chroma[:, 100:150] = chroma[:, 20:70] + matching = _calculate_loop_score(30, 110, chroma, test_duration=20, weights=_weights(20, start=5)) + unrelated = _calculate_loop_score(30, 160, chroma, test_duration=20, weights=_weights(20, start=5)) + assert unrelated < matching + + +@pytest.mark.xfail(strict=True, reason="bug: truncated look-behind windows are zero-padded on the side nearest the loop point") +def test_truncated_lookbehind_weights_frames_nearest_the_loop_point(): + chroma = _random_chroma() + # Only 3 frames exist before b1=3; they match the 3 frames before b2=100 + chroma[:, 0:3] = chroma[:, 97:100] + weights = _weights(20, start=10)[::-1] # heaviest weight on the frame right before the loop point + + score = _calculate_subseq_beat_similarity(3, 100, chroma, -20, weights=weights) + + assert score == pytest.approx(weights[-3:].sum() / weights.sum()) + + +# --- Zero crossings --- + + +def _sine(freq=100.0, seconds=1.0, channels=1): + t = np.arange(int(seconds * SR)) / SR + y = np.sin(2 * np.pi * freq * t) + return np.tile(y[:, np.newaxis], (1, channels)) + + +def _assert_rising_zero_crossing(audio, idx): + assert abs(audio[idx, 0]) < 0.05 + assert audio[idx + 1, 0] > audio[idx, 0] + + +@pytest.mark.parametrize("channels", [1, 2]) +def test_zero_crossing_snaps_to_nearby_rising_crossing(channels): + audio = _sine(channels=channels) + idx = nearest_zero_crossing(audio, SR, 1000) + assert abs(idx - 1000) <= SR // 200 + _assert_rising_zero_crossing(audio, idx) + + +def test_zero_crossing_near_start_of_audio(): + audio = _sine() + idx = nearest_zero_crossing(audio, SR, 30) + assert 0 <= idx <= 30 + SR // 200 + _assert_rising_zero_crossing(audio, idx) + + +def test_zero_crossing_keeps_index_without_a_crossing(): + audio = np.full((SR, 1), 0.8) + assert nearest_zero_crossing(audio, SR, 1000) == 1000 diff --git a/tests/test_audio.py b/tests/test_audio.py new file mode 100644 index 0000000..63c0816 --- /dev/null +++ b/tests/test_audio.py @@ -0,0 +1,72 @@ +import numpy as np +import pytest +import soundfile as sf +from conftest import SR, make_track + +from pymusiclooper.audio import MLAudio +from pymusiclooper.exceptions import AudioLoadError + + +def test_loads_mono(track_path, track): + audio = MLAudio(track_path) + assert audio.rate == SR + assert audio.n_channels == 1 + assert audio.length == track.size + assert audio.playback_audio.shape == (track.size, 1) + assert audio.total_duration == pytest.approx(track.size / SR) + assert audio.filename == "track.wav" + + +def test_loads_stereo(stereo_track_path, track): + audio = MLAudio(stereo_track_path) + assert audio.n_channels == 2 + assert audio.playback_audio.shape == (track.size, 2) + + +def test_analysis_signal_is_normalized_mono(stereo_track_path): + audio = MLAudio(stereo_track_path) + assert audio.audio.ndim == 1 + assert np.max(np.abs(audio.audio)) == pytest.approx(1.0) + + +@pytest.mark.xfail(strict=True, reason="bug: for mono input, to_mono returns the same array, so normalizing it in place also rescales the playback/export audio") +def test_playback_audio_is_untouched(track_path, track): + audio = MLAudio(track_path) + np.testing.assert_array_equal(audio.playback_audio[:, 0], track) + + +def test_silent_file_raises(silent_path): + with pytest.raises(AudioLoadError): + MLAudio(silent_path) + + +def test_non_audio_file_raises(tmp_path): + path = tmp_path / "notes.txt" + path.write_text("not audio") + with pytest.raises(AudioLoadError): + MLAudio(str(path)) + + +def test_leading_silence_is_trimmed_on_frame_boundary(tmp_path): + silence = np.zeros(SR + 1234, dtype=np.float32) + path = tmp_path / "padded.wav" + sf.write(path, np.concatenate([silence, make_track()]), SR, subtype="FLOAT") + + audio = MLAudio(str(path)) + + assert 0 < audio.trim_offset <= silence.size + # apply_trim_offset round-trips through frames, which is only exact + # because librosa trims on hop-length (512 sample) boundaries + assert audio.trim_offset % 512 == 0 + assert audio.frames_to_samples(audio.apply_trim_offset(10)) == ( + audio.frames_to_samples(10) + audio.trim_offset + ) + + +def test_time_conversions(track_path): + audio = MLAudio(track_path) + assert audio.seconds_to_samples(1) == SR + assert audio.samples_to_seconds(SR) == pytest.approx(1.0) + assert audio.frames_to_samples(1) == 512 + assert audio.samples_to_frames(1024) == 2 + assert audio.samples_to_ftime(int(61.5 * SR)) == "01:01.500" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a4765e1 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,93 @@ +import re +import shutil + +import pytest +from click.testing import CliRunner +from conftest import SR, assert_whole_patterns + +from pymusiclooper import __version__ +from pymusiclooper.cli import cli_main + + +@pytest.fixture +def run(): + def invoke(*args): + result = CliRunner().invoke(cli_main, [str(arg) for arg in args]) + assert result.exception is None or isinstance(result.exception, SystemExit), result.output + return result + return invoke + + +def test_version(run): + result = run("--version") + assert result.exit_code == 0 + assert __version__ in result.output + + +def test_path_is_required(run): + result = run("export-points") + assert result.exit_code != 0 + + +def test_export_points_to_stdout(run, track_path): + result = run("export-points", "--path", track_path) + + assert result.exit_code == 0 + loop_start = int(re.search(r"LOOP_START: (\d+)", result.output).group(1)) + loop_end = int(re.search(r"LOOP_END: (\d+)", result.output).group(1)) + assert_whole_patterns(loop_start, loop_end) + + +def test_export_points_in_seconds(run, track_path): + result = run("export-points", "--path", track_path, "--fmt", "SECONDS") + + loop_start = float(re.search(r"LOOP_START: ([\d.]+)", result.output).group(1)) + loop_end = float(re.search(r"LOOP_END: ([\d.]+)", result.output).group(1)) + assert_whole_patterns(round(loop_start * SR), round(loop_end * SR)) + + +def test_export_points_alt_export_top(run, track_path): + result = run("export-points", "--path", track_path, "--alt-export-top", "3") + + lines = result.output.strip().splitlines() + assert len(lines) == 3 + assert all(len(line.split()) == 5 for line in lines) + + +def test_export_points_to_txt(run, track_path, tmp_path): + run("export-points", "--path", track_path, "--export-to", "TXT", "--output-dir", tmp_path) + + lines = (tmp_path / "loops.txt").read_text().splitlines() + assert len(lines) == 1 + assert lines[0].endswith(" track.wav") + + +def test_split_audio(run, track_path, tmp_path): + run("split-audio", "--path", track_path, "--output-dir", tmp_path) + + assert sorted(p.name for p in tmp_path.iterdir()) == [ + "track.wav-intro.wav", + "track.wav-loop.wav", + "track.wav-outro.wav", + ] + + +def test_extend(run, track_path, tmp_path): + run("extend", "--path", track_path, "--output-dir", tmp_path, "--extended-length", 45, "--format", "WAV") + + assert [p.name for p in tmp_path.iterdir()] == ["track.wav-extended-0m45s.wav"] + + +def test_batch_export_skips_non_audio_files(run, track_path, stereo_track_path, tmp_path): + in_dir = tmp_path / "in" + out_dir = tmp_path / "out" + in_dir.mkdir() + shutil.copy(track_path, in_dir) + shutil.copy(stereo_track_path, in_dir) + (in_dir / "notes.txt").write_text("not audio") + + result = run("export-points", "--path", in_dir, "--export-to", "TXT", "--output-dir", out_dir) + + assert result.exit_code == 0 + lines = (out_dir / "loops.txt").read_text().splitlines() + assert sorted(line.split()[-1] for line in lines) == ["stereo.wav", "track.wav"] diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..7c833eb --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,116 @@ +import os + +import numpy as np +import pytest +import soundfile as sf +import taglib +from conftest import INTRO_SAMPLES, PATTERN_SAMPLES, SR + +from pymusiclooper.core import MusicLooper + +LOOP_START = INTRO_SAMPLES +LOOP_END = INTRO_SAMPLES + 2 * PATTERN_SAMPLES + + +@pytest.fixture(scope="module") +def looper(track_path): + return MusicLooper(track_path) + + +# --- Split / extend --- + + +def test_export_splits_into_intro_loop_outro(looper, track, tmp_path): + looper.export(LOOP_START, LOOP_END, format="WAV", output_dir=str(tmp_path)) + + sections = [ + sf.read(tmp_path / f"track.wav-{name}.wav")[0] for name in ("intro", "loop", "outro") + ] + + assert [s.shape[0] for s in sections] == [LOOP_START, LOOP_END - LOOP_START, track.size - LOOP_END] + np.testing.assert_allclose(np.concatenate(sections), looper.mlaudio.playback_audio[:, 0], atol=1e-4) + + +def test_extend_with_fade_out_reaches_requested_length(looper, tmp_path): + output_path = looper.extend(LOOP_START, LOOP_END, extended_length=60, fade_length=3, format="WAV", output_dir=str(tmp_path)) + + extended, rate = sf.read(output_path) + assert rate == SR + assert extended.shape[0] / SR == pytest.approx(60, abs=0.05) + assert np.max(np.abs(extended[-100:])) < 0.01, "track should fade out to silence" + assert os.path.basename(output_path) == "track.wav-extended-1m00s.wav" + + +def test_extend_without_fade_out_keeps_outro(looper, tmp_path): + output_path = looper.extend(LOOP_START, LOOP_END, extended_length=60, disable_fade_out=True, format="WAV", output_dir=str(tmp_path)) + + extended, _ = sf.read(output_path) + outro = looper.mlaudio.playback_audio[LOOP_END:, 0] + assert extended.shape[0] / SR >= 60 + np.testing.assert_allclose(extended[-outro.size:], outro, atol=1e-4) + + +def test_extend_shorter_than_track_raises(looper, tmp_path): + with pytest.raises(ValueError): + looper.extend(LOOP_START, LOOP_END, extended_length=5, output_dir=str(tmp_path)) + + +@pytest.mark.xfail(strict=True, reason="bug: fade_length=0 slices the whole final loop section (x[-0:]) and fails to broadcast") +def test_extend_with_zero_fade_length(looper, tmp_path): + looper.extend(LOOP_START, LOOP_END, extended_length=60, fade_length=0, format="WAV", output_dir=str(tmp_path)) + + +# --- Loop point text export --- + + +def test_export_txt_appends_lines(looper, tmp_path): + looper.export_txt(LOOP_START, LOOP_END, output_dir=str(tmp_path)) + looper.export_txt(1, 2, output_dir=str(tmp_path)) + + lines = (tmp_path / "loops.txt").read_text().splitlines() + assert lines == [f"{LOOP_START} {LOOP_END} track.wav", "1 2 track.wav"] + + +# --- Metadata tags --- + + +@pytest.mark.parametrize( + ("start_tag", "end_tag", "stored_end"), + [ + ("LOOP_START", "LOOP_END", LOOP_END), + ("LOOPSTART", "LOOPLENGTH", LOOP_END - LOOP_START), + ], +) +def test_tags_roundtrip(flac_track_path, tmp_path, start_tag, end_tag, stored_end): + out_dir = tmp_path / "out" + out_dir.mkdir() + + written = MusicLooper(flac_track_path).export_tags(LOOP_START, LOOP_END, start_tag, end_tag, output_dir=str(out_dir)) + assert written == (str(LOOP_START), str(stored_end)) + + tagged = MusicLooper(str(out_dir / "track-tagged.flac")) + assert tagged.read_tags(start_tag, end_tag) == (LOOP_START, LOOP_END) + assert tagged.read_tags(None, None) == (LOOP_START, LOOP_END), "tags should be auto-detected" + + +def test_read_tags_without_loop_tags_raises(flac_track_path): + with pytest.raises(ValueError): + MusicLooper(flac_track_path).read_tags(None, None) + + +@pytest.mark.xfail(strict=True, reason="bug: export_tags defaults output_dir to the source file path instead of its directory") +def test_export_tags_defaults_to_source_directory(flac_track_path, tmp_path): + MusicLooper(flac_track_path).export_tags(LOOP_START, LOOP_END, "LOOP_START", "LOOP_END") + assert (tmp_path / "track-tagged.flac").exists() + + +def test_extend_copies_source_tags(flac_track_path, tmp_path): + with taglib.File(flac_track_path, save_on_exit=True) as source: + source.tags["TITLE"] = ["Test Track"] + + output_path = MusicLooper(flac_track_path).extend( + LOOP_START, LOOP_END, extended_length=60, format="FLAC", output_dir=str(tmp_path) + ) + + with taglib.File(output_path) as extended: + assert extended.tags["TITLE"] == ["Test Track"] diff --git a/tests/test_handler.py b/tests/test_handler.py new file mode 100644 index 0000000..5f33702 --- /dev/null +++ b/tests/test_handler.py @@ -0,0 +1,82 @@ +import os + +import pytest + +from pymusiclooper import handler +from pymusiclooper.analysis import LoopPair +from pymusiclooper.core import MusicLooper +from pymusiclooper.handler import BatchHandler, LoopExportHandler, LoopHandler + + +def _bare_loop_handler(n_pairs=30): + """A LoopHandler with preset loop pairs, skipping audio loading and analysis.""" + loop_handler = LoopHandler.__new__(LoopHandler) + loop_handler.loop_pair_list = [ + LoopPair(0, 0, note_distance=0.0, loudness_difference=0.0, score=1 - i / 100, loop_start=i, loop_end=i + 1000) + for i in range(n_pairs) + ] + loop_handler.filepath = "track.wav" + loop_handler.in_samples = True + loop_handler._musiclooper = None + loop_handler._progressbar = None + return loop_handler + + +@pytest.fixture +def fake_input(monkeypatch): + """Feeds the given answers to the interactive prompt, in order.""" + def feed(*answers): + remaining = iter(answers) + monkeypatch.setattr(handler.rich_console, "input", lambda *args, **kwargs: next(remaining)) + return feed + + +def test_interactive_selects_entered_index(fake_input): + fake_input("3") + assert _bare_loop_handler().interactive_handler() == 3 + + +def test_interactive_reprompts_on_invalid_input(fake_input): + fake_input("abc", "", "99", "4") + assert _bare_loop_handler().interactive_handler() == 4 + + +@pytest.mark.xfail(strict=True, reason="bug: the choice made after 'more'/'all'/'reset' is discarded and the user is prompted again") +def test_interactive_more_then_select(fake_input): + fake_input("more", "27", "1") + assert _bare_loop_handler().interactive_handler() == 27 + + +def test_choose_loop_pair_defaults_to_best(): + loop_handler = _bare_loop_handler() + assert loop_handler.choose_loop_pair(interactive_mode=False) is loop_handler.loop_pair_list[0] + + +@pytest.mark.xfail(strict=True, reason="bug: success message names loop.txt but the file written is loops.txt") +def test_txt_export_message_names_written_file(monkeypatch, track_path, tmp_path): + export_handler = LoopExportHandler.__new__(LoopExportHandler) + export_handler._musiclooper = MusicLooper(track_path) + export_handler.output_directory = str(tmp_path) + export_handler.alt_export_top = 0 + export_handler.fmt = "samples" + export_handler.batch_mode = False + + messages = [] + monkeypatch.setattr(handler.rich_console, "print", lambda msg, *args, **kwargs: messages.append(msg)) + export_handler.txt_export_runner(100, 200) + + written = os.listdir(tmp_path) + assert written == ["loops.txt"] + assert str(tmp_path / "loops.txt") in messages[0] + + +def test_get_files_in_directory(tmp_path): + (tmp_path / "a.wav").touch() + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.wav").touch() + + flat = BatchHandler.get_files_in_directory(str(tmp_path)) + recursive = BatchHandler.get_files_in_directory(str(tmp_path), recursive=True) + + assert flat == [str(tmp_path / "a.wav")] + assert sorted(recursive) == sorted([str(tmp_path / "a.wav"), str(tmp_path / "sub" / "b.wav")]) diff --git a/uv.lock b/uv.lock index 784e524..6f288ca 100644 --- a/uv.lock +++ b/uv.lock @@ -324,6 +324,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/84/d0/205d54408c08b13550c733c4b85429e7ead111c7f0014309637425520a9a/deprecated-1.3.1-py2.py3-none-any.whl", hash = "sha256:597bfef186b6f60181535a29fbe44865ce137a5079f295b479886c82729d5f3f", size = 11298, upload-time = "2025-10-30T08:19:00.758Z" }, ] +[[package]] +name = "exceptiongroup" +version = "1.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8a/0e/97c33bf5009bdbac74fd2beace167cab3f978feb69cc36f1ef79360d6c4e/exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598", size = 16740, upload-time = "2025-11-21T23:01:53.443Z" }, +] + [[package]] name = "idna" version = "3.11" @@ -333,6 +345,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/61/66938bbb5fc52dbdf84594873d5b51fb1f7c7794e9c0f5bd885f30bc507b/idna-3.11-py3-none-any.whl", hash = "sha256:771a87f49d9defaf64091e6e6fe9c18d4833f140bd19464795bc32d966ca37ea", size = 71008, upload-time = "2025-10-12T14:55:18.883Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "joblib" version = "1.5.2" @@ -693,6 +714,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/73/cb/ac7874b3e5d58441674fb70742e6c374b28b0c7cb988d37d991cde47166c/platformdirs-4.5.0-py3-none-any.whl", hash = "sha256:e578a81bb873cbb89a41fcc904c7ef523cc18284b7e3b3ccf06aca1403b7ebd3", size = 18651, upload-time = "2025-10-08T17:44:47.223Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "pooch" version = "1.8.2" @@ -745,6 +775,11 @@ dependencies = [ { name = "yt-dlp" }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, +] + [package.metadata] requires-dist = [ { name = "click-option-group", specifier = ">=0.5.6,<0.6" }, @@ -761,6 +796,9 @@ requires-dist = [ { name = "yt-dlp", specifier = ">=2025.1.26" }, ] +[package.metadata.requires-dev] +dev = [{ name = "pytest", specifier = ">=9.1.1" }] + [[package]] name = "pytaglib" version = "3.0.1" @@ -789,6 +827,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/87/d9/a81226deee5f687edc5e914b7c0a4f1f2cfd1302fb265be37025cf9b352b/pytaglib-3.0.1-cp313-cp313-win_amd64.whl", hash = "sha256:a65f310df7cd78e4218a7bb581da7775192fc81dfcb2612da1ed93091c8671b9", size = 256377, upload-time = "2025-03-21T22:32:37.669Z" }, ] +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "exceptiongroup", marker = "python_full_version < '3.11'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + [[package]] name = "requests" version = "2.32.5" @@ -1122,6 +1178,60 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/32/d5/f9a850d79b0851d1d4ef6456097579a9005b31fea68726a4ae5f2d82ddd9/threadpoolctl-3.6.0-py3-none-any.whl", hash = "sha256:43a0b8fd5a2928500110039e43a5eed8480b918967083ea48dc3ab9f13c4a7fb", size = 18638, upload-time = "2025-03-13T13:49:21.846Z" }, ] +[[package]] +name = "tomli" +version = "2.4.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/22/de/48c59722572767841493b26183a0d1cc411d54fd759c5607c4590b6563a6/tomli-2.4.1.tar.gz", hash = "sha256:7c7e1a961a0b2f2472c1ac5b69affa0ae1132c39adcb67aba98568702b9cc23f", size = 17543, upload-time = "2026-03-25T20:22:03.828Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/11/db3d5885d8528263d8adc260bb2d28ebf1270b96e98f0e0268d32b8d9900/tomli-2.4.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:f8f0fc26ec2cc2b965b7a3b87cd19c5c6b8c5e5f436b984e85f486d652285c30", size = 154704, upload-time = "2026-03-25T20:21:10.473Z" }, + { url = "https://files.pythonhosted.org/packages/6d/f7/675db52c7e46064a9aa928885a9b20f4124ecb9bc2e1ce74c9106648d202/tomli-2.4.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4ab97e64ccda8756376892c53a72bd1f964e519c77236368527f758fbc36a53a", size = 149454, upload-time = "2026-03-25T20:21:12.036Z" }, + { url = "https://files.pythonhosted.org/packages/61/71/81c50943cf953efa35bce7646caab3cf457a7d8c030b27cfb40d7235f9ee/tomli-2.4.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:96481a5786729fd470164b47cdb3e0e58062a496f455ee41b4403be77cb5a076", size = 237561, upload-time = "2026-03-25T20:21:13.098Z" }, + { url = "https://files.pythonhosted.org/packages/48/c1/f41d9cb618acccca7df82aaf682f9b49013c9397212cb9f53219e3abac37/tomli-2.4.1-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5a881ab208c0baf688221f8cecc5401bd291d67e38a1ac884d6736cbcd8247e9", size = 243824, upload-time = "2026-03-25T20:21:14.569Z" }, + { url = "https://files.pythonhosted.org/packages/22/e4/5a816ecdd1f8ca51fb756ef684b90f2780afc52fc67f987e3c61d800a46d/tomli-2.4.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:47149d5bd38761ac8be13a84864bf0b7b70bc051806bc3669ab1cbc56216b23c", size = 242227, upload-time = "2026-03-25T20:21:15.712Z" }, + { url = "https://files.pythonhosted.org/packages/6b/49/2b2a0ef529aa6eec245d25f0c703e020a73955ad7edf73e7f54ddc608aa5/tomli-2.4.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:ec9bfaf3ad2df51ace80688143a6a4ebc09a248f6ff781a9945e51937008fcbc", size = 247859, upload-time = "2026-03-25T20:21:17.001Z" }, + { url = "https://files.pythonhosted.org/packages/83/bd/6c1a630eaca337e1e78c5903104f831bda934c426f9231429396ce3c3467/tomli-2.4.1-cp311-cp311-win32.whl", hash = "sha256:ff2983983d34813c1aeb0fa89091e76c3a22889ee83ab27c5eeb45100560c049", size = 97204, upload-time = "2026-03-25T20:21:18.079Z" }, + { url = "https://files.pythonhosted.org/packages/42/59/71461df1a885647e10b6bb7802d0b8e66480c61f3f43079e0dcd315b3954/tomli-2.4.1-cp311-cp311-win_amd64.whl", hash = "sha256:5ee18d9ebdb417e384b58fe414e8d6af9f4e7a0ae761519fb50f721de398dd4e", size = 108084, upload-time = "2026-03-25T20:21:18.978Z" }, + { url = "https://files.pythonhosted.org/packages/b8/83/dceca96142499c069475b790e7913b1044c1a4337e700751f48ed723f883/tomli-2.4.1-cp311-cp311-win_arm64.whl", hash = "sha256:c2541745709bad0264b7d4705ad453b76ccd191e64aa6f0fc66b69a293a45ece", size = 95285, upload-time = "2026-03-25T20:21:20.309Z" }, + { url = "https://files.pythonhosted.org/packages/c1/ba/42f134a3fe2b370f555f44b1d72feebb94debcab01676bf918d0cb70e9aa/tomli-2.4.1-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:c742f741d58a28940ce01d58f0ab2ea3ced8b12402f162f4d534dfe18ba1cd6a", size = 155924, upload-time = "2026-03-25T20:21:21.626Z" }, + { url = "https://files.pythonhosted.org/packages/dc/c7/62d7a17c26487ade21c5422b646110f2162f1fcc95980ef7f63e73c68f14/tomli-2.4.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7f86fd587c4ed9dd76f318225e7d9b29cfc5a9d43de44e5754db8d1128487085", size = 150018, upload-time = "2026-03-25T20:21:23.002Z" }, + { url = "https://files.pythonhosted.org/packages/5c/05/79d13d7c15f13bdef410bdd49a6485b1c37d28968314eabee452c22a7fda/tomli-2.4.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ff18e6a727ee0ab0388507b89d1bc6a22b138d1e2fa56d1ad494586d61d2eae9", size = 244948, upload-time = "2026-03-25T20:21:24.04Z" }, + { url = "https://files.pythonhosted.org/packages/10/90/d62ce007a1c80d0b2c93e02cab211224756240884751b94ca72df8a875ca/tomli-2.4.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:136443dbd7e1dee43c68ac2694fde36b2849865fa258d39bf822c10e8068eac5", size = 253341, upload-time = "2026-03-25T20:21:25.177Z" }, + { url = "https://files.pythonhosted.org/packages/1a/7e/caf6496d60152ad4ed09282c1885cca4eea150bfd007da84aea07bcc0a3e/tomli-2.4.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:5e262d41726bc187e69af7825504c933b6794dc3fbd5945e41a79bb14c31f585", size = 248159, upload-time = "2026-03-25T20:21:26.364Z" }, + { url = "https://files.pythonhosted.org/packages/99/e7/c6f69c3120de34bbd882c6fba7975f3d7a746e9218e56ab46a1bc4b42552/tomli-2.4.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:5cb41aa38891e073ee49d55fbc7839cfdb2bc0e600add13874d048c94aadddd1", size = 253290, upload-time = "2026-03-25T20:21:27.46Z" }, + { url = "https://files.pythonhosted.org/packages/d6/2f/4a3c322f22c5c66c4b836ec58211641a4067364f5dcdd7b974b4c5da300c/tomli-2.4.1-cp312-cp312-win32.whl", hash = "sha256:da25dc3563bff5965356133435b757a795a17b17d01dbc0f42fb32447ddfd917", size = 98141, upload-time = "2026-03-25T20:21:28.492Z" }, + { url = "https://files.pythonhosted.org/packages/24/22/4daacd05391b92c55759d55eaee21e1dfaea86ce5c571f10083360adf534/tomli-2.4.1-cp312-cp312-win_amd64.whl", hash = "sha256:52c8ef851d9a240f11a88c003eacb03c31fc1c9c4ec64a99a0f922b93874fda9", size = 108847, upload-time = "2026-03-25T20:21:29.386Z" }, + { url = "https://files.pythonhosted.org/packages/68/fd/70e768887666ddd9e9f5d85129e84910f2db2796f9096aa02b721a53098d/tomli-2.4.1-cp312-cp312-win_arm64.whl", hash = "sha256:f758f1b9299d059cc3f6546ae2af89670cb1c4d48ea29c3cacc4fe7de3058257", size = 95088, upload-time = "2026-03-25T20:21:30.677Z" }, + { url = "https://files.pythonhosted.org/packages/07/06/b823a7e818c756d9a7123ba2cda7d07bc2dd32835648d1a7b7b7a05d848d/tomli-2.4.1-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:36d2bd2ad5fb9eaddba5226aa02c8ec3fa4f192631e347b3ed28186d43be6b54", size = 155866, upload-time = "2026-03-25T20:21:31.65Z" }, + { url = "https://files.pythonhosted.org/packages/14/6f/12645cf7f08e1a20c7eb8c297c6f11d31c1b50f316a7e7e1e1de6e2e7b7e/tomli-2.4.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:eb0dc4e38e6a1fd579e5d50369aa2e10acfc9cace504579b2faabb478e76941a", size = 149887, upload-time = "2026-03-25T20:21:33.028Z" }, + { url = "https://files.pythonhosted.org/packages/5c/e0/90637574e5e7212c09099c67ad349b04ec4d6020324539297b634a0192b0/tomli-2.4.1-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c7f2c7f2b9ca6bdeef8f0fa897f8e05085923eb091721675170254cbc5b02897", size = 243704, upload-time = "2026-03-25T20:21:34.51Z" }, + { url = "https://files.pythonhosted.org/packages/10/8f/d3ddb16c5a4befdf31a23307f72828686ab2096f068eaf56631e136c1fdd/tomli-2.4.1-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f3c6818a1a86dd6dca7ddcaaf76947d5ba31aecc28cb1b67009a5877c9a64f3f", size = 251628, upload-time = "2026-03-25T20:21:36.012Z" }, + { url = "https://files.pythonhosted.org/packages/e3/f1/dbeeb9116715abee2485bf0a12d07a8f31af94d71608c171c45f64c0469d/tomli-2.4.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:d312ef37c91508b0ab2cee7da26ec0b3ed2f03ce12bd87a588d771ae15dcf82d", size = 247180, upload-time = "2026-03-25T20:21:37.136Z" }, + { url = "https://files.pythonhosted.org/packages/d3/74/16336ffd19ed4da28a70959f92f506233bd7cfc2332b20bdb01591e8b1d1/tomli-2.4.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:51529d40e3ca50046d7606fa99ce3956a617f9b36380da3b7f0dd3dd28e68cb5", size = 251674, upload-time = "2026-03-25T20:21:38.298Z" }, + { url = "https://files.pythonhosted.org/packages/16/f9/229fa3434c590ddf6c0aa9af64d3af4b752540686cace29e6281e3458469/tomli-2.4.1-cp313-cp313-win32.whl", hash = "sha256:2190f2e9dd7508d2a90ded5ed369255980a1bcdd58e52f7fe24b8162bf9fedbd", size = 97976, upload-time = "2026-03-25T20:21:39.316Z" }, + { url = "https://files.pythonhosted.org/packages/6a/1e/71dfd96bcc1c775420cb8befe7a9d35f2e5b1309798f009dca17b7708c1e/tomli-2.4.1-cp313-cp313-win_amd64.whl", hash = "sha256:8d65a2fbf9d2f8352685bc1364177ee3923d6baf5e7f43ea4959d7d8bc326a36", size = 108755, upload-time = "2026-03-25T20:21:40.248Z" }, + { url = "https://files.pythonhosted.org/packages/83/7a/d34f422a021d62420b78f5c538e5b102f62bea616d1d75a13f0a88acb04a/tomli-2.4.1-cp313-cp313-win_arm64.whl", hash = "sha256:4b605484e43cdc43f0954ddae319fb75f04cc10dd80d830540060ee7cd0243cd", size = 95265, upload-time = "2026-03-25T20:21:41.219Z" }, + { url = "https://files.pythonhosted.org/packages/3c/fb/9a5c8d27dbab540869f7c1f8eb0abb3244189ce780ba9cd73f3770662072/tomli-2.4.1-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:fd0409a3653af6c147209d267a0e4243f0ae46b011aa978b1080359fddc9b6cf", size = 155726, upload-time = "2026-03-25T20:21:42.23Z" }, + { url = "https://files.pythonhosted.org/packages/62/05/d2f816630cc771ad836af54f5001f47a6f611d2d39535364f148b6a92d6b/tomli-2.4.1-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:a120733b01c45e9a0c34aeef92bf0cf1d56cfe81ed9d47d562f9ed591a9828ac", size = 149859, upload-time = "2026-03-25T20:21:43.386Z" }, + { url = "https://files.pythonhosted.org/packages/ce/48/66341bdb858ad9bd0ceab5a86f90eddab127cf8b046418009f2125630ecb/tomli-2.4.1-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:559db847dc486944896521f68d8190be1c9e719fced785720d2216fe7022b662", size = 244713, upload-time = "2026-03-25T20:21:44.474Z" }, + { url = "https://files.pythonhosted.org/packages/df/6d/c5fad00d82b3c7a3ab6189bd4b10e60466f22cfe8a08a9394185c8a8111c/tomli-2.4.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:01f520d4f53ef97964a240a035ec2a869fe1a37dde002b57ebc4417a27ccd853", size = 252084, upload-time = "2026-03-25T20:21:45.62Z" }, + { url = "https://files.pythonhosted.org/packages/00/71/3a69e86f3eafe8c7a59d008d245888051005bd657760e96d5fbfb0b740c2/tomli-2.4.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7f94b27a62cfad8496c8d2513e1a222dd446f095fca8987fceef261225538a15", size = 247973, upload-time = "2026-03-25T20:21:46.937Z" }, + { url = "https://files.pythonhosted.org/packages/67/50/361e986652847fec4bd5e4a0208752fbe64689c603c7ae5ea7cb16b1c0ca/tomli-2.4.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ede3e6487c5ef5d28634ba3f31f989030ad6af71edfb0055cbbd14189ff240ba", size = 256223, upload-time = "2026-03-25T20:21:48.467Z" }, + { url = "https://files.pythonhosted.org/packages/8c/9a/b4173689a9203472e5467217e0154b00e260621caa227b6fa01feab16998/tomli-2.4.1-cp314-cp314-win32.whl", hash = "sha256:3d48a93ee1c9b79c04bb38772ee1b64dcf18ff43085896ea460ca8dec96f35f6", size = 98973, upload-time = "2026-03-25T20:21:49.526Z" }, + { url = "https://files.pythonhosted.org/packages/14/58/640ac93bf230cd27d002462c9af0d837779f8773bc03dee06b5835208214/tomli-2.4.1-cp314-cp314-win_amd64.whl", hash = "sha256:88dceee75c2c63af144e456745e10101eb67361050196b0b6af5d717254dddf7", size = 109082, upload-time = "2026-03-25T20:21:50.506Z" }, + { url = "https://files.pythonhosted.org/packages/d5/2f/702d5e05b227401c1068f0d386d79a589bb12bf64c3d2c72ce0631e3bc49/tomli-2.4.1-cp314-cp314-win_arm64.whl", hash = "sha256:b8c198f8c1805dc42708689ed6864951fd2494f924149d3e4bce7710f8eb5232", size = 96490, upload-time = "2026-03-25T20:21:51.474Z" }, + { url = "https://files.pythonhosted.org/packages/45/4b/b877b05c8ba62927d9865dd980e34a755de541eb65fffba52b4cc495d4d2/tomli-2.4.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:d4d8fe59808a54658fcc0160ecfb1b30f9089906c50b23bcb4c69eddc19ec2b4", size = 164263, upload-time = "2026-03-25T20:21:52.543Z" }, + { url = "https://files.pythonhosted.org/packages/24/79/6ab420d37a270b89f7195dec5448f79400d9e9c1826df982f3f8e97b24fd/tomli-2.4.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:7008df2e7655c495dd12d2a4ad038ff878d4ca4b81fccaf82b714e07eae4402c", size = 160736, upload-time = "2026-03-25T20:21:53.674Z" }, + { url = "https://files.pythonhosted.org/packages/02/e0/3630057d8eb170310785723ed5adcdfb7d50cb7e6455f85ba8a3deed642b/tomli-2.4.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1d8591993e228b0c930c4bb0db464bdad97b3289fb981255d6c9a41aedc84b2d", size = 270717, upload-time = "2026-03-25T20:21:55.129Z" }, + { url = "https://files.pythonhosted.org/packages/7a/b4/1613716072e544d1a7891f548d8f9ec6ce2faf42ca65acae01d76ea06bb0/tomli-2.4.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:734e20b57ba95624ecf1841e72b53f6e186355e216e5412de414e3c51e5e3c41", size = 278461, upload-time = "2026-03-25T20:21:56.228Z" }, + { url = "https://files.pythonhosted.org/packages/05/38/30f541baf6a3f6df77b3df16b01ba319221389e2da59427e221ef417ac0c/tomli-2.4.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:8a650c2dbafa08d42e51ba0b62740dae4ecb9338eefa093aa5c78ceb546fcd5c", size = 274855, upload-time = "2026-03-25T20:21:57.653Z" }, + { url = "https://files.pythonhosted.org/packages/77/a3/ec9dd4fd2c38e98de34223b995a3b34813e6bdadf86c75314c928350ed14/tomli-2.4.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:504aa796fe0569bb43171066009ead363de03675276d2d121ac1a4572397870f", size = 283144, upload-time = "2026-03-25T20:21:59.089Z" }, + { url = "https://files.pythonhosted.org/packages/ef/be/605a6261cac79fba2ec0c9827e986e00323a1945700969b8ee0b30d85453/tomli-2.4.1-cp314-cp314t-win32.whl", hash = "sha256:b1d22e6e9387bf4739fbe23bfa80e93f6b0373a7f1b96c6227c32bef95a4d7a8", size = 108683, upload-time = "2026-03-25T20:22:00.214Z" }, + { url = "https://files.pythonhosted.org/packages/12/64/da524626d3b9cc40c168a13da8335fe1c51be12c0a63685cc6db7308daae/tomli-2.4.1-cp314-cp314t-win_amd64.whl", hash = "sha256:2c1c351919aca02858f740c6d33adea0c5deea37f9ecca1cc1ef9e884a619d26", size = 121196, upload-time = "2026-03-25T20:22:01.169Z" }, + { url = "https://files.pythonhosted.org/packages/5a/cd/e80b62269fc78fc36c9af5a6b89c835baa8af28ff5ad28c7028d60860320/tomli-2.4.1-cp314-cp314t-win_arm64.whl", hash = "sha256:eab21f45c7f66c13f2a9e0e1535309cee140182a9cdae1e041d02e47291e8396", size = 100393, upload-time = "2026-03-25T20:22:02.137Z" }, + { url = "https://files.pythonhosted.org/packages/7b/61/cceae43728b7de99d9b847560c262873a1f6c98202171fd5ed62640b494b/tomli-2.4.1-py3-none-any.whl", hash = "sha256:0d85819802132122da43cb86656f8d1f8c6587d54ae7dcaf30e90533028b49fe", size = 14583, upload-time = "2026-03-25T20:22:03.012Z" }, +] + [[package]] name = "typing-extensions" version = "4.15.0" From a715a60d849d30b44eb33a5ac8b7b3277015c20b Mon Sep 17 00:00:00 2001 From: Xehanort88 Date: Wed, 23 Sep 2026 13:50:34 +0100 Subject: [PATCH 2/3] Fix bugs found by the test suite - Mono tracks were played and exported louder than the original: the analysis-only normalization was applied in place to the audio array that is also used for playback, since to_mono returns its input as-is for mono audio. - Longer loops were never preferred among near-identical scores: _prioritize_duration ran before loop_start/loop_end were set, so every duration it compared was 0. - Loops starting in the first seconds of a track were underscored: the truncated look-behind window was zero-padded on the side nearest the loop point, where the weights are heaviest. - Interactive mode discarded the choice made after 'more', 'all' or 'reset' and prompted again. - extend with fade_length=0 crashed (x[-0:] selects the whole array). - export_tags() without output_dir used the file path as the directory. - The txt export message named loop.txt instead of loops.txt. --- pymusiclooper/analysis.py | 15 ++++++++++----- pymusiclooper/audio.py | 3 ++- pymusiclooper/core.py | 19 ++++++++++++------- pymusiclooper/handler.py | 9 ++++----- tests/test_analysis.py | 2 -- tests/test_audio.py | 1 - tests/test_core.py | 2 -- tests/test_handler.py | 2 -- 8 files changed, 28 insertions(+), 25 deletions(-) diff --git a/pymusiclooper/analysis.py b/pymusiclooper/analysis.py index 6940db2..4745ae9 100644 --- a/pymusiclooper/analysis.py +++ b/pymusiclooper/analysis.py @@ -172,10 +172,6 @@ def find_best_loop_points( mlaudio, chroma, bpm, candidate_pairs, disable_pruning ) - # prefer longer loops for highly similar sequences - if len(filtered_candidate_pairs) > 1: - _prioritize_duration(filtered_candidate_pairs) - # Set the exact loop start and end in samples and adjust them # to the nearest zero crossing. Avoids audio popping/clicking while looping # as much as possible. @@ -203,6 +199,11 @@ def find_best_loop_points( f"No loop points found for {mlaudio.filename} with current parameters." ) + # prefer longer loops for highly similar sequences + # (must run after the loop positions in samples are set, since it compares loop durations) + if len(filtered_candidate_pairs) > 1: + _prioritize_duration(filtered_candidate_pairs) + logging.info( f"Filtered to {len(filtered_candidate_pairs)} best candidate loop points" ) @@ -522,8 +523,12 @@ def _calculate_subseq_beat_similarity( cosine_sim = dot_prod / (np.maximum(b1_norm * b2_norm, 1e-10)) if max_offset < test_length: + # Pad the missing frames on the side farthest from the loop point: + # after the tested frames when looking ahead, before them when looking behind + missing_frames = test_length - max_offset + pad_width = (missing_frames, 0) if test_end_offset < 0 else (0, missing_frames) return np.average( - np.pad(cosine_sim, pad_width=(0, test_length - max_offset), mode="constant", constant_values=0), + np.pad(cosine_sim, pad_width=pad_width, mode="constant", constant_values=0), weights=weights, ) else: diff --git a/pymusiclooper/audio.py b/pymusiclooper/audio.py index 138613b..1742e08 100644 --- a/pymusiclooper/audio.py +++ b/pymusiclooper/audio.py @@ -48,7 +48,8 @@ def __init__(self, filepath: str) -> None: raise AudioLoadError(f"\"{filepath}\" only contains silence and cannot be analyzed.") # Normalize audio channels to between -1.0 and +1.0 before analysis - mono_signal /= np.max(np.abs(mono_signal)) + # (not in-place: for mono input, to_mono returns raw_audio itself, which is also used for playback/export) + mono_signal = mono_signal / np.max(np.abs(mono_signal)) self.audio, self.trim_offset = librosa.effects.trim(mono_signal, top_db=40) self.trim_offset = self.trim_offset[0] diff --git a/pymusiclooper/core.py b/pymusiclooper/core.py index 82819d2..2ca953e 100644 --- a/pymusiclooper/core.py +++ b/pymusiclooper/core.py @@ -206,10 +206,12 @@ def extend( samples_to_fade = min( self.mlaudio.seconds_to_samples(fade_length), final_loop.shape[0] ) - final_loop[-samples_to_fade:] = ( - final_loop[-samples_to_fade:] - * np.linspace(1, 0, samples_to_fade)[:, np.newaxis] - ) + # Guard against x[-0:], which would select the whole section + if samples_to_fade > 0: + final_loop[-samples_to_fade:] = ( + final_loop[-samples_to_fade:] + * np.linspace(1, 0, samples_to_fade)[:, np.newaxis] + ) # Format extended file name with its duration suffixed extended_loop_length = final_loop.shape[0] + ( @@ -272,9 +274,10 @@ def export_txt( loop_end: Union[str, int, float, str], txt_name: str = "loops", output_dir: Optional[str] = None - ): - """Exports the given loop points to a text file named `loop.txt` in append mode with the format: + ) -> str: + """Exports the given loop points to a text file named `loops.txt` in append mode with the format: `{loop_start} {loop_end} {filename}` + Returns the path to the text file. Args: loop_start (Union[int, float, str]): Loop start in samples, seconds or ftime. @@ -290,6 +293,8 @@ def export_txt( with open(out_path, "a") as file: file.write(f"{loop_start} {loop_end} {self.mlaudio.filename}\n") + return out_path + def _find_start_tag( self, @@ -377,7 +382,7 @@ def export_tags( import taglib if output_dir is None: - output_dir = os.path.abspath(self.mlaudio.filepath) + output_dir = os.path.dirname(os.path.abspath(self.mlaudio.filepath)) track_name, file_extension = os.path.splitext(self.mlaudio.filename) diff --git a/pymusiclooper/handler.py b/pymusiclooper/handler.py index 2d4c148..e514d06 100644 --- a/pymusiclooper/handler.py +++ b/pymusiclooper/handler.py @@ -132,11 +132,11 @@ def get_user_input(): preview = False if num_input == "more": - self.interactive_handler(show_top=show_top * 2) + return self.interactive_handler(show_top=show_top * 2) if num_input == "all": - self.interactive_handler(show_top=total_candidates) + return self.interactive_handler(show_top=total_candidates) if num_input == "reset": - self.interactive_handler() + return self.interactive_handler() if num_input[-1] == "p": idx = int(num_input[:-1]) @@ -322,12 +322,11 @@ def txt_export_runner(self, loop_start: int, loop_end: int): if self.alt_export_top != 0: self.alt_export_runner(mode="TXT") else: - self.musiclooper.export_txt( + out_path = self.musiclooper.export_txt( self._fmt(loop_start), self._fmt(loop_end), output_dir=self.output_directory, ) - out_path = os.path.join(self.output_directory, "loop.txt") message = f'Successfully added "{self.musiclooper.filename}" loop points to "{out_path}"' if self.batch_mode: logging.info(message) diff --git a/tests/test_analysis.py b/tests/test_analysis.py index 1682e5c..37c34c2 100644 --- a/tests/test_analysis.py +++ b/tests/test_analysis.py @@ -76,7 +76,6 @@ def test_impossible_constraints_raise(looper): looper.find_loop_pairs(min_loop_duration=60) -@pytest.mark.xfail(strict=True, reason="bug: _prioritize_duration runs before loop_start/loop_end are set, so it never reorders anything") def test_prioritize_duration_sees_real_loop_positions(monkeypatch, looper): durations_seen = [] original = analysis._prioritize_duration @@ -130,7 +129,6 @@ def test_unrelated_sequences_score_lower(): assert unrelated < matching -@pytest.mark.xfail(strict=True, reason="bug: truncated look-behind windows are zero-padded on the side nearest the loop point") def test_truncated_lookbehind_weights_frames_nearest_the_loop_point(): chroma = _random_chroma() # Only 3 frames exist before b1=3; they match the 3 frames before b2=100 diff --git a/tests/test_audio.py b/tests/test_audio.py index 63c0816..d8b714f 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -29,7 +29,6 @@ def test_analysis_signal_is_normalized_mono(stereo_track_path): assert np.max(np.abs(audio.audio)) == pytest.approx(1.0) -@pytest.mark.xfail(strict=True, reason="bug: for mono input, to_mono returns the same array, so normalizing it in place also rescales the playback/export audio") def test_playback_audio_is_untouched(track_path, track): audio = MLAudio(track_path) np.testing.assert_array_equal(audio.playback_audio[:, 0], track) diff --git a/tests/test_core.py b/tests/test_core.py index 7c833eb..931fa60 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -55,7 +55,6 @@ def test_extend_shorter_than_track_raises(looper, tmp_path): looper.extend(LOOP_START, LOOP_END, extended_length=5, output_dir=str(tmp_path)) -@pytest.mark.xfail(strict=True, reason="bug: fade_length=0 slices the whole final loop section (x[-0:]) and fails to broadcast") def test_extend_with_zero_fade_length(looper, tmp_path): looper.extend(LOOP_START, LOOP_END, extended_length=60, fade_length=0, format="WAV", output_dir=str(tmp_path)) @@ -98,7 +97,6 @@ def test_read_tags_without_loop_tags_raises(flac_track_path): MusicLooper(flac_track_path).read_tags(None, None) -@pytest.mark.xfail(strict=True, reason="bug: export_tags defaults output_dir to the source file path instead of its directory") def test_export_tags_defaults_to_source_directory(flac_track_path, tmp_path): MusicLooper(flac_track_path).export_tags(LOOP_START, LOOP_END, "LOOP_START", "LOOP_END") assert (tmp_path / "track-tagged.flac").exists() diff --git a/tests/test_handler.py b/tests/test_handler.py index 5f33702..61a0ad8 100644 --- a/tests/test_handler.py +++ b/tests/test_handler.py @@ -41,7 +41,6 @@ def test_interactive_reprompts_on_invalid_input(fake_input): assert _bare_loop_handler().interactive_handler() == 4 -@pytest.mark.xfail(strict=True, reason="bug: the choice made after 'more'/'all'/'reset' is discarded and the user is prompted again") def test_interactive_more_then_select(fake_input): fake_input("more", "27", "1") assert _bare_loop_handler().interactive_handler() == 27 @@ -52,7 +51,6 @@ def test_choose_loop_pair_defaults_to_best(): assert loop_handler.choose_loop_pair(interactive_mode=False) is loop_handler.loop_pair_list[0] -@pytest.mark.xfail(strict=True, reason="bug: success message names loop.txt but the file written is loops.txt") def test_txt_export_message_names_written_file(monkeypatch, track_path, tmp_path): export_handler = LoopExportHandler.__new__(LoopExportHandler) export_handler._musiclooper = MusicLooper(track_path) From 23ce1b4062fc421ff98f823be0a347039b6f51c5 Mon Sep 17 00:00:00 2001 From: Xehanort88 Date: Wed, 23 Sep 2026 13:50:40 +0100 Subject: [PATCH 3/3] Use embedded loop tags as the first choice and add a lossless trim command Embedded loop tags: - Loop points already stored in a file's metadata (LOOP_START/LOOP_END, LOOPSTART/LOOPLENGTH and the other known tag names) are now returned first by find_loop_pairs, ahead of the detected candidates. They are still returned if detection finds no loop, and are skipped when an approximate loop position is given or when they are out of range. - LoopPair gains a from_metadata flag; interactive mode shows these points as "from tags" and play reports them as such. - New --ignore-tags option to skip them. trim command: - Cuts the audio a given number of samples (--keep-after) after the loop end, keeping the original format, bit depth and tags. - WAV and FLAC are rewritten bit-exact from their native sample format. - Ogg Vorbis is cut at the container level without re-encoding: the audio packets after the one containing the end sample are dropped and the granule position of the final page is lowered to the exact end, so decoders that only trim the last packet (e.g. FFmpeg) also end on the exact sample. Falls back to a page-level cut if the packet durations cannot be determined. - Other formats are rejected, since trimming them would require lossy re-encoding. Tests cover both features; an optional test checks trimmed Ogg files with FFmpeg as an independent decoder when it is available. --- README.md | 6 + pymusiclooper/analysis.py | 2 + pymusiclooper/cli.py | 23 ++- pymusiclooper/console.py | 3 + pymusiclooper/core.py | 142 +++++++++++++++++-- pymusiclooper/handler.py | 43 +++++- pymusiclooper/ogg.py | 290 ++++++++++++++++++++++++++++++++++++++ tests/test_core.py | 170 ++++++++++++++++++++++ tests/test_ogg_ffmpeg.py | 70 +++++++++ 9 files changed, 724 insertions(+), 25 deletions(-) create mode 100644 pymusiclooper/ogg.py create mode 100644 tests/test_ogg_ffmpeg.py diff --git a/README.md b/README.md index d5e361a..fb9c868 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,8 @@ pymusiclooper -i play --url "https://www.youtube.com/watch?v=dQw4w9WgXcQ" pymusiclooper play-tagged --path "TRACK_NAME.mp3" --tag-names LOOP_START LOOP_END ``` +*Note: if a file already has loop metadata tags (e.g. LOOP_START/LOOP_END or LOOPSTART/LOOPLENGTH), they are used as the first loop choice by all commands. Use `--ignore-tags` to skip them.* + ### Export *Note: batch processing is available for all export subcommands. Simply specify a directory instead of a file as the path to be used.* @@ -145,6 +147,10 @@ pymusiclooper export-points --path "/path/to/track.wav" --alt-export-top -1 # (or all audio files in a directory, if a directory path is used instead) pymusiclooper -i tag --path "TRACK_NAME.mp3" --tag-names LOOP_START LOOP_END +# Losslessly cut everything after the loop end (WAV, FLAC and OGG Vorbis only), keeping 1000 samples past it +# The original format, bit depth and tags are preserved; OGG files are cut without re-encoding +pymusiclooper -i trim --path "TRACK_NAME.wav" --keep-after 1000 + # Export the loop points (in samples) of all tracks in a particular directory to a loops.txt file # (compatible with https://github.com/libertyernie/LoopingAudioConverter/) diff --git a/pymusiclooper/analysis.py b/pymusiclooper/analysis.py index 4745ae9..78475eb 100644 --- a/pymusiclooper/analysis.py +++ b/pymusiclooper/analysis.py @@ -20,6 +20,7 @@ class LoopPair: note_distance: float loudness_difference: float score: float. Defaults to 0. + from_metadata: bool (loop points were read from the file's metadata tags instead of being detected). Defaults to False. """ _loop_start_frame_idx: int @@ -29,6 +30,7 @@ class LoopPair: score: float = 0 loop_start: int = 0 loop_end: int = 0 + from_metadata: bool = False def find_best_loop_points( diff --git a/pymusiclooper/cli.py b/pymusiclooper/cli.py index b5284b9..5c6b703 100644 --- a/pymusiclooper/cli.py +++ b/pymusiclooper/cli.py @@ -77,6 +77,7 @@ def common_loop_options(f): @click.option('--approx-loop-position', type=click.FloatRange(min=0), nargs=2, default=None, help='The approximate desired loop start and loop end in seconds. [dim]([cyan]+/-2[/] second search window for each point)[/]') @click.option("--brute-force", is_flag=True, default=False, help=r"Check the entire audio track instead of just the detected beats. [dim yellow](Warning: may take several minutes to complete.)[/]") @click.option("--disable-pruning", is_flag=True, default=False, help="Disables filtering of the detected loop points from the initial pass.") + @click.option("--ignore-tags", is_flag=True, default=False, help="Ignore loop points already stored in the file's metadata tags (e.g. LOOP_START/LOOP_END), which are otherwise used as the first choice.") @functools.wraps(f) def wrapper_common_options(*args, **kwargs): @@ -123,12 +124,13 @@ def play(**kwargs): start_time = handler.format_time(chosen_loop_pair.loop_start, in_samples=in_samples) end_time = handler.format_time(chosen_loop_pair.loop_end, in_samples=in_samples) + source = ( + "loop points from metadata tags" + if chosen_loop_pair.from_metadata + else f"similarity: {chosen_loop_pair.score:.2%}" + ) rich_console.print( - "\nPlaying with looping active from [green]{}[/] back to [green]{}[/]; similarity: {:.2%}".format( - end_time, - start_time, - chosen_loop_pair.score, - ) + f"\nPlaying with looping active from [green]{end_time}[/] back to [green]{start_time}[/]; {source}" ) rich_console.print("(Press [red]Ctrl+C[/] to stop looping.)") @@ -226,6 +228,17 @@ def tag(**kwargs): run_handler(**kwargs) +@cli_main.command() +@common_path_options +@common_loop_options +@common_export_options +@click.option('--keep-after', type=click.IntRange(min=0), default=0, show_default=True, help="Number of samples to keep after the loop end.") +def trim(**kwargs): + """Losslessly cut the audio after the loop end, keeping its original format, bit depth and tags. [dim](WAV, FLAC and OGG Vorbis only)[/]""" + kwargs["trim"] = True + run_handler(**kwargs) + + def run_handler(**kwargs): try: if kwargs.get("url", None) is not None: diff --git a/pymusiclooper/console.py b/pymusiclooper/console.py index 20e1642..377ad2a 100644 --- a/pymusiclooper/console.py +++ b/pymusiclooper/console.py @@ -12,6 +12,7 @@ "--approx-loop-position", "--brute-force", "--disable-pruning", + "--ignore-tags", ] _export_options = ["--output-dir", "--format"] _batch_options = ["--recursive", "--flatten"] @@ -49,6 +50,7 @@ def _option_groups(additional_basic_options=None): "pymusiclooper tag": _option_groups(["--tag-names", "--tag-offset"]), "pymusiclooper export-points": _option_groups(["--export-to", "--alt-export-top", "--fmt"]), "pymusiclooper extend": _option_groups(["--extended-length", "--fade-length", "--disable-fade-out"]), + "pymusiclooper trim": _option_groups(["--keep-after"]), } _COMMAND_GROUPS = { "pymusiclooper": [ @@ -66,6 +68,7 @@ def _option_groups(additional_basic_options=None): "split-audio", "tag", "extend", + "trim", ], } ] diff --git a/pymusiclooper/core.py b/pymusiclooper/core.py index 2ca953e..f85b94f 100644 --- a/pymusiclooper/core.py +++ b/pymusiclooper/core.py @@ -1,6 +1,7 @@ """Contains the core MusicLooper class that can be used for programmatic access to the CLI's main features.""" +import logging import os import shutil from math import ceil @@ -11,6 +12,8 @@ from pymusiclooper.analysis import LoopPair, find_best_loop_points from pymusiclooper.audio import MLAudio +from pymusiclooper.exceptions import LoopNotFoundError +from pymusiclooper.ogg import trim_vorbis from pymusiclooper.playback import PlaybackHandler # Lazy-load external libraries when they're needed @@ -35,8 +38,10 @@ def find_loop_pairs( approx_loop_end: Optional[float] = None, brute_force: bool = False, disable_pruning: bool = False, + use_embedded_tags: bool = True, ) -> List[LoopPair]: """Finds the best loop points for the track, according to the parameters specified. + If the file already has loop points stored in its metadata tags, they are returned as the first choice. Args: min_duration_multiplier (float, optional): The minimum duration of a loop as a multiplier of track duration. Defaults to 0.35. @@ -46,22 +51,70 @@ def find_loop_pairs( approx_loop_end (float, optional): The approximate location of the desired loop end (in seconds). If specified, must specify approx_loop_start as well. Defaults to None. brute_force (bool, optional): Checks the entire track instead of the detected beats (disclaimer: runtime may be significantly longer). Defaults to False. disable_pruning (bool, optional): Returns all the candidate loop points without filtering. Defaults to False. - + use_embedded_tags (bool, optional): Places the loop points found in the file's metadata tags (if any) first in the returned list. Ignored if an approximate loop position is specified. Defaults to True. + Raises: LoopNotFoundError: raised in case no loops were found Returns: List[LoopPair]: A list of `LoopPair` objects containing the loop points related data. See the `LoopPair` class for more info. """ - return find_best_loop_points( - mlaudio=self.mlaudio, - min_duration_multiplier=min_duration_multiplier, - min_loop_duration=min_loop_duration, - max_loop_duration=max_loop_duration, - approx_loop_start=approx_loop_start, - approx_loop_end=approx_loop_end, - brute_force=brute_force, - disable_pruning=disable_pruning + embedded_pair = None + if use_embedded_tags and approx_loop_start is None and approx_loop_end is None: + embedded_pair = self.read_embedded_loop_pair() + + try: + loop_pairs = find_best_loop_points( + mlaudio=self.mlaudio, + min_duration_multiplier=min_duration_multiplier, + min_loop_duration=min_loop_duration, + max_loop_duration=max_loop_duration, + approx_loop_start=approx_loop_start, + approx_loop_end=approx_loop_end, + brute_force=brute_force, + disable_pruning=disable_pruning + ) + except LoopNotFoundError: + if embedded_pair is None: + raise + loop_pairs = [] + + if embedded_pair is not None: + loop_pairs.insert(0, embedded_pair) + + return loop_pairs + + def read_embedded_loop_pair(self) -> Optional[LoopPair]: + """Reads the loop points stored in the file's metadata tags (e.g. LOOP_START/LOOP_END), auto-detecting the tag names. + + Returns: + Optional[LoopPair]: A `LoopPair` with `from_metadata=True`, or None if the file has no valid loop tags. + """ + try: + loop_start, loop_end = self.read_tags(None, None) + except Exception as e: + logging.debug(f"No embedded loop points read from \"{self.filename}\": {e}") + return None + + if not 0 <= loop_start < loop_end <= self.mlaudio.length: + logging.warning( + f"Ignoring embedded loop points of \"{self.filename}\" ({loop_start}, {loop_end}):" + f" outside the audio's range of {self.mlaudio.length} samples." + ) + return None + + logging.info(f"Found embedded loop points in the metadata tags: {loop_start} -> {loop_end}") + + # Metadata loop points are not analyzed, so the similarity metrics are nominal + return LoopPair( + _loop_start_frame_idx=self.samples_to_frames(loop_start), + _loop_end_frame_idx=self.samples_to_frames(loop_end), + note_distance=0.0, + loudness_difference=0.0, + score=1.0, + loop_start=loop_start, + loop_end=loop_end, + from_metadata=True, ) @property @@ -251,14 +304,77 @@ def extend( if disable_fade_out: sf.buffer_write(outro.tobytes(order="C"), dtype) - # attempt to copy over the tags + self._copy_tags(output_file_path) + + return output_file_path + + def trim( + self, + loop_end: int, + keep_after: int = 0, + output_dir: Optional[str] = None, + ) -> str: + """Losslessly cuts the audio `keep_after` samples after the loop end, removing the rest of the track. + The original container, bit depth and tags are kept; only WAV, FLAC and Ogg Vorbis files are supported. + Ogg Vorbis files are trimmed at the container level, without re-encoding the audio. + Returns the path to the trimmed audio file. + + Args: + loop_end (int): Loop end in samples. + keep_after (int, optional): Number of samples to keep after the loop end. Defaults to 0. + output_dir (str, optional): Path to the output directory. Defaults to the same directory as the source audio file. + + Raises: + ValueError: if the source file is not a PCM/float WAV, FLAC or single-stream Ogg Vorbis file. + """ + info = soundfile.info(self.filepath) + is_ogg_vorbis = info.format == "OGG" and info.subtype == "VORBIS" + if not is_ogg_vorbis and ( + info.format not in ("WAV", "FLAC") + or not (info.subtype.startswith("PCM_") or info.subtype in ("FLOAT", "DOUBLE")) + ): + raise ValueError( + f"Lossless trimming is only supported for PCM/float WAV, FLAC and Ogg Vorbis files; \"{self.filename}\" is {info.format} ({info.subtype})." + ) + + if output_dir is None: + output_dir = os.path.dirname(os.path.abspath(self.mlaudio.filepath)) + + track_name, file_extension = os.path.splitext(self.mlaudio.filename) + output_file_path = os.path.join(output_dir, f"{track_name}-trimmed{file_extension}") + + if is_ogg_vorbis: + # The pages holding the tags are copied verbatim, so the tags are kept as-is + trim_vorbis(self.filepath, output_file_path, loop_end + keep_after) + return output_file_path + + # Read the samples in their native representation so that writing them back is bit-exact + dtype = {"FLOAT": "float32", "DOUBLE": "float64"}.get(info.subtype, "int32") + n_frames = min(info.frames, loop_end + keep_after) + audio, rate = soundfile.read(self.filepath, frames=n_frames, dtype=dtype, always_2d=True) + + soundfile.write( + output_file_path, + audio, + rate, + format=info.format, + subtype=info.subtype, + endian=info.endian, + ) + + self._copy_tags(output_file_path) + + return output_file_path + + def _copy_tags(self, dest_filepath: str): + """Attempts to copy the metadata tags of the source audio file to `dest_filepath`.""" try: import taglib original_tags = None with taglib.File(self.filepath, save_on_exit=False) as src_file: original_tags = src_file.tags - with taglib.File(output_file_path, save_on_exit=True) as dest_file: + with taglib.File(dest_filepath, save_on_exit=True) as dest_file: for tag in original_tags: dest_file.tags[tag] = original_tags[tag] except Exception: @@ -266,8 +382,6 @@ def extend( # TODO: implement logging for debugging pass - return output_file_path - def export_txt( self, loop_start: Union[int, float, str], diff --git a/pymusiclooper/handler.py b/pymusiclooper/handler.py index e514d06..64bf38e 100644 --- a/pymusiclooper/handler.py +++ b/pymusiclooper/handler.py @@ -25,6 +25,7 @@ def __init__( approx_loop_position: Optional[tuple] = None, brute_force: bool = False, disable_pruning: bool = False, + ignore_tags: bool = False, _progressbar: Progress = None, **kwargs, ): @@ -48,6 +49,7 @@ def __init__( approx_loop_end=self.approx_loop_end, brute_force=brute_force, disable_pruning=disable_pruning, + use_embedded_tags=not ignore_tags, ) self.interactive_mode = "PML_INTERACTIVE_MODE" in os.environ self.in_samples = "PML_DISPLAY_SAMPLES" in os.environ @@ -109,17 +111,20 @@ def interactive_handler(self, show_top=25): if self.in_samples else preview_looper.samples_to_ftime(pair.loop_end - pair.loop_start) ) - score = pair.score - loudness_difference = pair.loudness_difference - note_distance = pair.note_distance + if pair.from_metadata: + metrics = ("-", "-", "[bold]from tags[/]") + else: + metrics = ( + f"{pair.note_distance:.4f}", + f"{pair.loudness_difference:.4f}", + f"{pair.score:.2%}", + ) table.add_row( str(idx), str(start_time), str(end_time), str(length), - f"{note_distance:.4f}", - f"{loudness_difference:.4f}", - f"{score:.2%}", + *metrics, ) rich_console.print(table) @@ -188,6 +193,7 @@ def __init__( approx_loop_position: Optional[tuple] = None, brute_force: bool = False, disable_pruning: bool = False, + ignore_tags: bool = False, split_audio: bool = False, format: Literal["WAV", "FLAC", "OGG", "MP3"] = "WAV", to_txt: bool = False, @@ -200,6 +206,8 @@ def __init__( extended_length: float = 0, fade_length: float = 0, disable_fade_out: bool = False, + trim: bool = False, + keep_after: int = 0, **kwargs, ): super().__init__( @@ -210,6 +218,7 @@ def __init__( approx_loop_position=approx_loop_position, brute_force=brute_force, disable_pruning=disable_pruning, + ignore_tags=ignore_tags, **kwargs, ) self.output_directory = output_dir @@ -225,6 +234,8 @@ def __init__( self.extended_length = extended_length self.disable_fade_out = disable_fade_out self.fade_length = fade_length + self.trim = trim + self.keep_after = keep_after self._is_autocreated_outdir = False def run(self): @@ -245,6 +256,7 @@ def run(self): or self.to_txt or self.split_audio or self.extended_length + or self.trim ) and not os.path.exists(self.output_directory): os.mkdir(self.output_directory) self._is_autocreated_outdir = True @@ -260,6 +272,9 @@ def run(self): if self.extended_length: self.extend_track_runner(loop_start, loop_end) + + if self.trim: + self.trim_runner(loop_end) finally: if ( self._is_autocreated_outdir @@ -318,6 +333,22 @@ def extend_track_runner(self, loop_start: int, loop_end: int): except ValueError as e: logging.error(e) + def trim_runner(self, loop_end: int): + try: + output_path = self.musiclooper.trim( + loop_end, + keep_after=self.keep_after, + output_dir=self.output_directory, + ) + message = f'Successfully exported a trimmed version of "{self.musiclooper.filename}" to "{output_path}"' + if self.batch_mode: + logging.info(message) + else: + rich_console.print(message) + # Unsupported (lossy) source format + except ValueError as e: + logging.error(e) + def txt_export_runner(self, loop_start: int, loop_end: int): if self.alt_export_top != 0: self.alt_export_runner(mode="TXT") diff --git a/pymusiclooper/ogg.py b/pymusiclooper/ogg.py new file mode 100644 index 0000000..1933dc6 --- /dev/null +++ b/pymusiclooper/ogg.py @@ -0,0 +1,290 @@ +"""Lossless trimming of Ogg Vorbis files at the container level, without re-encoding the audio. + +The Ogg pages up to the requested length are copied verbatim. On the last kept page, the audio packets after the +one containing the requested end sample are dropped, then its granule position (the number of samples decoded by +the end of a page) is lowered to the requested length and the page is marked as the end of the stream. +The Vorbis specification defines this as the way to end a stream on a sample that is not on a block boundary: +decoders must discard the samples past it. Keeping the trimmed samples within the last packet matches what encoders +produce, which is required by decoders that only trim the last packet (e.g. FFmpeg). +""" + +import logging +import struct +from typing import Iterator, List, NamedTuple, Optional, Tuple + +_PAGE_HEADER = struct.Struct("<4sBBqIIIB") +_CAPTURE_PATTERN = b"OggS" +_END_OF_STREAM_FLAG = 0x04 + + +def _make_crc_table() -> List[int]: + table = [] + for i in range(256): + crc = i << 24 + for _ in range(8): + crc = ((crc << 1) ^ 0x04C11DB7) if crc & 0x80000000 else (crc << 1) + table.append(crc & 0xFFFFFFFF) + return table + + +_CRC_TABLE = _make_crc_table() + + +def _crc(data: bytes) -> int: + """Ogg page checksum: CRC-32 with polynomial 0x04C11DB7, no reflection, zero initial value and no final XOR.""" + crc = 0 + for byte in data: + crc = ((crc << 8) & 0xFFFFFFFF) ^ _CRC_TABLE[(crc >> 24) ^ byte] + return crc + + +class _Page(NamedTuple): + offset: int + header_type: int + granule_position: int + serial_number: int + sequence_number: int + segment_table: bytes + body: bytes + + @property + def length(self) -> int: + return _PAGE_HEADER.size + len(self.segment_table) + len(self.body) + + +def _read_pages(data: bytes) -> Iterator[_Page]: + offset = 0 + while offset < len(data): + if data[offset:offset + 4] != _CAPTURE_PATTERN or offset + _PAGE_HEADER.size > len(data): + raise ValueError(f"Invalid Ogg page at byte offset {offset}.") + _, version, header_type, granule, serial, sequence, _, n_segments = _PAGE_HEADER.unpack_from(data, offset) + if version != 0: + raise ValueError(f"Unsupported Ogg page version {version}.") + segment_table_start = offset + _PAGE_HEADER.size + segment_table = data[segment_table_start:segment_table_start + n_segments] + body_start = segment_table_start + n_segments + body = data[body_start:body_start + sum(segment_table)] + page = _Page(offset, header_type, granule, serial, sequence, segment_table, body) + yield page + offset += page.length + + +class _ReverseBitReader: + """Reads the bits of a Vorbis packet backwards, starting from its last bit.""" + + def __init__(self, data: bytes): + self.data = data[::-1] + self.pos = 0 + + def bits_left(self) -> int: + return len(self.data) * 8 - self.pos + + def read(self, n_bits: int) -> int: + value = 0 + for _ in range(n_bits): + bit = (self.data[self.pos >> 3] >> (7 - (self.pos & 7))) & 1 + value = (value << 1) | bit + self.pos += 1 + return value + + +class _VorbisPacketDuration: + """Computes the number of samples each Vorbis audio packet decodes to. + + Only the block size flags of the modes are needed from the setup header. Since they sit at its very end, + they are found by scanning the header backwards instead of parsing all of it (same approach as FFmpeg's + vorbis_parser.c and liboggz). + """ + + def __init__(self, identification_header: bytes, setup_header: bytes): + self.blocksizes = (1 << (identification_header[28] & 0x0F), 1 << (identification_header[28] >> 4)) + self.mode_blockflags = self._parse_mode_blockflags(setup_header) + mode_bits = max(1, (len(self.mode_blockflags) - 1).bit_length()) + self.mode_mask = ((1 << mode_bits) - 1) << 1 + # The previous window flag is the bit after the mode number + self.prev_mask = (self.mode_mask | 0x1) + 1 + self.previous_blocksize = None + + @staticmethod + def _parse_mode_blockflags(setup_header: bytes) -> List[int]: + reader = _ReverseBitReader(setup_header) + + framing_bit_pos = None + while reader.bits_left() > 97: + if reader.read(1): + framing_bit_pos = reader.pos + break + if framing_bit_pos is None: + raise ValueError("Invalid Vorbis setup header: no framing bit.") + + # Each mode is 41 bits: blockflag (1), windowtype (16, always 0), transformtype (16, always 0), mapping (8). + # Walk backwards over them while they look valid; the mode count stored just before them confirms a match. + mode_count = 0 + confirmed_mode_count = 0 + while reader.bits_left() >= 97: + if reader.read(8) > 63 or reader.read(16) or reader.read(16): + break + reader.read(1) + mode_count += 1 + if mode_count > 64: + break + mode_header_pos = reader.pos + if reader.read(6) + 1 == mode_count: + confirmed_mode_count = mode_count + reader.pos = mode_header_pos + if not confirmed_mode_count: + raise ValueError("Invalid Vorbis setup header: no modes found.") + + reader.pos = framing_bit_pos + blockflags = [0] * confirmed_mode_count + for i in reversed(range(confirmed_mode_count)): + reader.read(40) + blockflags[i] = reader.read(1) + return blockflags + + def __call__(self, first_byte: int) -> int: + mode = 0 if len(self.mode_blockflags) == 1 else (first_byte & self.mode_mask) >> 1 + if mode >= len(self.mode_blockflags): + raise ValueError("Invalid Vorbis audio packet.") + blockflag = self.mode_blockflags[mode] + current_blocksize = self.blocksizes[blockflag] + if blockflag: + previous_blocksize = self.blocksizes[1 if first_byte & self.prev_mask else 0] + else: + previous_blocksize = self.previous_blocksize + is_first_packet = self.previous_blocksize is None + self.previous_blocksize = current_blocksize + # The first audio packet only primes the decoder and returns no samples + return 0 if is_first_packet else (previous_blocksize + current_blocksize) // 4 + + +def _find_packet_cut(pages: List[_Page], n_samples: int) -> Optional[Tuple[int, int]]: + """Finds where to cut the stream so that the last kept packet contains sample `n_samples`. + + Returns: + Optional[Tuple[int, int]]: (index of the last page to keep, number of its segments to keep), + or None if the requested length is at or past the end of the stream. + """ + header_packets = [] + packet_duration = None + packet_data = b"" + previous_granule = 0 + + for page_idx, page in enumerate(pages): + # (duration, number of segments up to the packet's end) of each packet completed on this page + completed_packets = [] + body_pos = 0 + for segment_idx, lacing_value in enumerate(page.segment_table): + packet_data += page.body[body_pos:body_pos + lacing_value] + body_pos += lacing_value + if lacing_value == 255: + continue + if len(header_packets) < 3: + header_packets.append(packet_data) + if len(header_packets) == 3: + packet_duration = _VorbisPacketDuration(header_packets[0], header_packets[2]) + elif packet_data: + completed_packets.append((packet_duration(packet_data[0]), segment_idx + 1)) + packet_data = b"" + # Only the first byte of audio packets is needed when a packet continues on the next page + if len(header_packets) == 3 and len(packet_data) > 1: + packet_data = packet_data[:1] + + if not completed_packets or page.granule_position == -1: + continue + + if page.granule_position >= n_samples: + page_end = previous_granule + sum(duration for duration, _ in completed_packets) + # The final page of a stream may already end it before the end of its last packet + is_final_page = page.header_type & _END_OF_STREAM_FLAG or page_idx == len(pages) - 1 + if page_end != page.granule_position and not (is_final_page and page_end > page.granule_position): + raise ValueError("Vorbis packet durations do not match the page granule positions.") + packet_end = previous_granule + for duration, n_segments in completed_packets: + packet_end += duration + if packet_end >= n_samples: + return page_idx, n_segments + + previous_granule = page.granule_position + + return None + + +def _find_page_cut(pages: List[_Page], n_samples: int) -> Optional[Tuple[int, int]]: + """Same as `_find_packet_cut`, but keeps all the packets completed on the last page.""" + # The first page whose completed packets reach the requested length + # (header pages have a granule position of 0, and -1 means no packet ends on the page) + for page_idx, page in enumerate(pages): + if page.granule_position >= n_samples: + # Drop the segments of a trailing packet that continues on the next page (a run of 255 lacing values), + # since that packet would be incomplete. At least one packet ends on this page (granule position != -1), + # so the run belongs entirely to a packet that starts on this page. + return page_idx, len(page.segment_table.rstrip(b"\xff")) + return None + + +def _build_page(page: _Page, header_type: int, granule_position: int, segment_table: bytes, body: bytes) -> bytes: + def pack(crc: int) -> bytes: + return _PAGE_HEADER.pack( + _CAPTURE_PATTERN, 0, header_type, granule_position, page.serial_number, + page.sequence_number, crc, len(segment_table), + ) + segment_table + body + + return pack(_crc(pack(0))) + + +def trim_vorbis(src_filepath: str, dest_filepath: str, n_samples: int) -> int: + """Writes a copy of an Ogg Vorbis file that ends after `n_samples` samples, without re-encoding the audio. + + Args: + src_filepath (str): Path to the source Ogg Vorbis file. + dest_filepath (str): Path of the trimmed file to write. + n_samples (int): Number of samples (per channel) to keep from the start of the stream. + + Raises: + ValueError: if the file is not a single-stream Ogg Vorbis file. + + Returns: + int: The length of the trimmed file in samples. + """ + with open(src_filepath, "rb") as f: + data = f.read() + + pages = list(_read_pages(data)) + if not pages or not pages[0].body.startswith(b"\x01vorbis"): + raise ValueError("Not an Ogg Vorbis file.") + if any(page.serial_number != pages[0].serial_number for page in pages): + raise ValueError("Chained or multiplexed Ogg files are not supported.") + + try: + cut = _find_packet_cut(pages, max(1, n_samples)) + except (ValueError, IndexError) as e: + # Fall back to keeping all the packets of the last page, which is valid but + # relies on decoders trimming more than the last packet + logging.warning(f"Could not determine the Vorbis packet durations ({e}); trimming at the page level.") + cut = _find_page_cut(pages, max(1, n_samples)) + + if cut is None: + # Requested length is at or past the end of the stream: keep everything + with open(dest_filepath, "wb") as f: + f.write(data) + return max(0, pages[-1].granule_position) + + page_idx, n_segments = cut + page = pages[page_idx] + segment_table = page.segment_table[:n_segments] + body = page.body[:sum(segment_table)] + + last_page = _build_page( + page, + header_type=page.header_type | _END_OF_STREAM_FLAG, + granule_position=n_samples, + segment_table=segment_table, + body=body, + ) + + with open(dest_filepath, "wb") as f: + f.write(data[:page.offset]) + f.write(last_page) + + return n_samples diff --git a/tests/test_core.py b/tests/test_core.py index 931fa60..ff3a717 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -112,3 +112,173 @@ def test_extend_copies_source_tags(flac_track_path, tmp_path): with taglib.File(output_path) as extended: assert extended.tags["TITLE"] == ["Test Track"] + + +# --- Embedded loop points --- + + +def _write_loop_tags(path: str, loop_start: int, loop_end: int): + with taglib.File(path, save_on_exit=True) as audio_file: + audio_file.tags["LOOP_START"] = [str(loop_start)] + audio_file.tags["LOOP_END"] = [str(loop_end)] + + +def test_find_loop_pairs_puts_embedded_tags_first(flac_track_path): + _write_loop_tags(flac_track_path, LOOP_START + 123, LOOP_END + 123) + + pairs = MusicLooper(flac_track_path).find_loop_pairs() + + assert pairs[0].from_metadata + assert (pairs[0].loop_start, pairs[0].loop_end) == (LOOP_START + 123, LOOP_END + 123) + assert len(pairs) > 1, "detected loop points should still follow the embedded ones" + assert not any(pair.from_metadata for pair in pairs[1:]) + + +def test_find_loop_pairs_can_ignore_embedded_tags(flac_track_path): + _write_loop_tags(flac_track_path, LOOP_START + 123, LOOP_END + 123) + + pairs = MusicLooper(flac_track_path).find_loop_pairs(use_embedded_tags=False) + + assert not any(pair.from_metadata for pair in pairs) + + +def test_find_loop_pairs_ignores_embedded_tags_with_approx_position(flac_track_path): + _write_loop_tags(flac_track_path, LOOP_START + 123, LOOP_END + 123) + + pairs = MusicLooper(flac_track_path).find_loop_pairs( + approx_loop_start=LOOP_START / SR, approx_loop_end=LOOP_END / SR + ) + + assert not any(pair.from_metadata for pair in pairs) + + +def test_embedded_tags_out_of_range_are_ignored(flac_track_path, track): + _write_loop_tags(flac_track_path, LOOP_START, track.size + 1) + + assert MusicLooper(flac_track_path).read_embedded_loop_pair() is None + + +def test_no_embedded_tags(track_path): + assert MusicLooper(track_path).read_embedded_loop_pair() is None + + +# --- Lossless trim --- + + +@pytest.mark.parametrize( + ("filename", "subtype", "dtype"), + [ + ("pcm16.wav", "PCM_16", "int16"), + ("pcm24.flac", "PCM_24", "int32"), + ("float.wav", "FLOAT", "float32"), + ], +) +def test_trim_is_bit_exact(track, tmp_path, filename, subtype, dtype): + source_path = tmp_path / filename + sf.write(source_path, np.stack([track, 0.8 * track], axis=1), SR, subtype=subtype) + out_dir = tmp_path / "out" + out_dir.mkdir() + + output_path = MusicLooper(str(source_path)).trim(LOOP_END, keep_after=100, output_dir=str(out_dir)) + + source, _ = sf.read(source_path, dtype=dtype) + trimmed, rate = sf.read(output_path, dtype=dtype) + assert rate == SR + assert sf.info(output_path).subtype == subtype + assert os.path.basename(output_path) == f"{os.path.splitext(filename)[0]}-trimmed{os.path.splitext(filename)[1]}" + np.testing.assert_array_equal(trimmed, source[: LOOP_END + 100]) + + +def test_trim_keep_after_past_track_end_keeps_whole_track(flac_track_path, track, tmp_path): + output_path = MusicLooper(flac_track_path).trim(LOOP_END, keep_after=10 * track.size, output_dir=str(tmp_path)) + + assert sf.info(output_path).frames == track.size + + +def test_trim_copies_source_tags(flac_track_path, tmp_path): + out_dir = tmp_path / "out" + out_dir.mkdir() + with taglib.File(flac_track_path, save_on_exit=True) as source: + source.tags["LOOP_START"] = [str(LOOP_START)] + source.tags["LOOP_END"] = [str(LOOP_END)] + + output_path = MusicLooper(flac_track_path).trim(LOOP_END, output_dir=str(out_dir)) + + assert MusicLooper(output_path).read_tags(None, None) == (LOOP_START, LOOP_END) + + +@pytest.mark.parametrize("keep_after", [0, 777]) +def test_trim_ogg_vorbis_without_reencoding(track, tmp_path, keep_after): + source_path = tmp_path / "track.ogg" + sf.write(source_path, np.stack([track, 0.8 * track], axis=1), SR, format="OGG", subtype="VORBIS") + with taglib.File(str(source_path), save_on_exit=True) as source: + source.tags["LOOPSTART"] = [str(LOOP_START)] + source.tags["LOOPLENGTH"] = [str(LOOP_END - LOOP_START)] + out_dir = tmp_path / "out" + out_dir.mkdir() + + output_path = MusicLooper(str(source_path)).trim(LOOP_END, keep_after=keep_after, output_dir=str(out_dir)) + + source, _ = sf.read(source_path, dtype="float32") + trimmed, _ = sf.read(output_path, dtype="float32") + assert os.path.basename(output_path) == "track-trimmed.ogg" + np.testing.assert_array_equal(trimmed, source[: LOOP_END + keep_after]) + assert MusicLooper(output_path).read_tags(None, None) == (LOOP_START, LOOP_END) + + +def test_trim_ogg_vorbis_past_track_end_keeps_whole_track(track, tmp_path): + source_path = tmp_path / "track.ogg" + sf.write(source_path, track, SR, format="OGG", subtype="VORBIS") + out_dir = tmp_path / "out" + out_dir.mkdir() + + output_path = MusicLooper(str(source_path)).trim(LOOP_END, keep_after=10 * track.size, output_dir=str(out_dir)) + + assert open(output_path, "rb").read() == open(source_path, "rb").read() + + +@pytest.mark.parametrize( + ("filename", "format", "subtype"), + [ + ("track.opus", "OGG", "OPUS"), + ("track.mp3", "MP3", "MPEG_LAYER_III"), + ("ulaw.wav", "WAV", "ULAW"), + ], +) +def test_trim_rejects_unsupported_formats(track, tmp_path, filename, format, subtype): + source_path = tmp_path / filename + sf.write(source_path, track, SR if subtype != "OPUS" else 48000, format=format, subtype=subtype) + + with pytest.raises(ValueError): + MusicLooper(str(source_path)).trim(LOOP_END, output_dir=str(tmp_path)) + + +def test_ogg_vorbis_packet_durations_match_granule_positions(track, tmp_path): + from pymusiclooper.ogg import _find_packet_cut, _read_pages + + source_path = tmp_path / "track.ogg" + sf.write(source_path, np.stack([track, 0.8 * track], axis=1), SR, format="OGG", subtype="VORBIS") + pages = list(_read_pages(source_path.read_bytes())) + + # Raises if the computed packet durations do not add up to the page granule positions + for n_samples in range(1, track.size, 997): + assert _find_packet_cut(pages, n_samples) is not None + assert _find_packet_cut(pages, track.size + 1) is None + + +def test_trim_ogg_vorbis_falls_back_to_page_level_cut(track, tmp_path, monkeypatch): + from pymusiclooper import ogg + + def fail(*args, **kwargs): + raise ValueError("unparseable") + + monkeypatch.setattr(ogg, "_find_packet_cut", fail) + source_path = tmp_path / "track.ogg" + sf.write(source_path, track, SR, format="OGG", subtype="VORBIS") + output_path = tmp_path / "trimmed.ogg" + + assert ogg.trim_vorbis(str(source_path), str(output_path), LOOP_END) == LOOP_END + + source, _ = sf.read(source_path, dtype="float32") + trimmed, _ = sf.read(output_path, dtype="float32") + np.testing.assert_array_equal(trimmed, source[:LOOP_END]) diff --git a/tests/test_ogg_ffmpeg.py b/tests/test_ogg_ffmpeg.py new file mode 100644 index 0000000..994e329 --- /dev/null +++ b/tests/test_ogg_ffmpeg.py @@ -0,0 +1,70 @@ +"""Checks that trimmed Ogg Vorbis files also end on the exact sample with FFmpeg, an independent Vorbis decoder. + +FFmpeg is looked up in the PML_FFMPEG environment variable, then in tools/ffmpeg/, then on the PATH; +the tests are skipped if it is not found, or if it does not honor Vorbis end trimming at all (older builds). +""" + +import os +import shutil +import subprocess +from pathlib import Path + +import numpy as np +import pytest +import soundfile as sf +from conftest import SR + +from pymusiclooper.ogg import trim_vorbis + + +def _find_ffmpeg(): + local_ffmpeg = Path(__file__).parent.parent / "tools" / "ffmpeg" / ("ffmpeg.exe" if os.name == "nt" else "ffmpeg") + for candidate in (os.environ.get("PML_FFMPEG"), local_ffmpeg, shutil.which("ffmpeg")): + if candidate and os.path.isfile(candidate): + return str(candidate) + return None + + +def _decode(ffmpeg: str, path, decoder: str, n_channels: int) -> np.ndarray: + raw = subprocess.run( + [ffmpeg, "-v", "error", "-c:a", decoder, "-i", str(path), "-f", "f32le", "-"], + capture_output=True, + check=True, + ).stdout + return np.frombuffer(raw, dtype=np.float32).reshape(-1, n_channels) + + +@pytest.fixture(scope="module") +def ffmpeg(): + path = _find_ffmpeg() + if path is None: + pytest.skip("ffmpeg not found (set PML_FFMPEG or place it in tools/ffmpeg/)") + return path + + +@pytest.fixture(scope="module") +def stereo_ogg_path(track, tmp_path_factory): + path = tmp_path_factory.mktemp("ogg") / "track.ogg" + sf.write(path, np.stack([track, 0.8 * track], axis=1), SR, format="OGG", subtype="VORBIS") + return path + + +@pytest.mark.parametrize("decoder", ["vorbis", "libvorbis"]) +def test_ffmpeg_decodes_trimmed_ogg_to_exact_length(ffmpeg, stereo_ogg_path, tmp_path, decoder): + source_length = sf.info(stereo_ogg_path).frames + try: + source = _decode(ffmpeg, stereo_ogg_path, decoder, n_channels=2) + except subprocess.CalledProcessError: + pytest.skip(f"this ffmpeg build has no '{decoder}' decoder") + if source.shape[0] != source_length: + pytest.skip("this ffmpeg build does not honor Vorbis end trimming, even on untouched files") + + # Cut points past the first audio page (a stream whose first audio page is also its last + # is a special case in FFmpeg, and loops cannot realistically end that early) + for n_samples in range(source_length // 4, source_length, 4999): + output_path = tmp_path / "trimmed.ogg" + trim_vorbis(str(stereo_ogg_path), str(output_path), n_samples) + + trimmed = _decode(ffmpeg, output_path, decoder, n_channels=2) + assert trimmed.shape[0] == n_samples + np.testing.assert_array_equal(trimmed, source[:n_samples])