diff --git a/README.md b/README.md index fb9c868..93bd136 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,10 @@ pymusiclooper -i tag --path "TRACK_NAME.mp3" --tag-names LOOP_START LOOP_END # 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 +# In interactive mode, the other export subcommands also offer to trim WAV, FLAC and OGG Vorbis files +# after the loop is chosen; with `tag`, a single tagged and trimmed copy is written +pymusiclooper -i tag --path "TRACK_NAME.ogg" --tag-names LOOPSTART LOOPLENGTH + # 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/core.py b/pymusiclooper/core.py index f85b94f..e2cb556 100644 --- a/pymusiclooper/core.py +++ b/pymusiclooper/core.py @@ -327,26 +327,40 @@ def trim( 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: + self._trim_to(output_file_path, loop_end, keep_after) + + return output_file_path + + def supports_lossless_trim(self) -> bool: + """Returns whether the source file's format can be losslessly trimmed with `trim`.""" + return self._supports_lossless_trim(soundfile.info(self.filepath)) + + @staticmethod + def _supports_lossless_trim(info) -> bool: + if info.format == "OGG": + return info.subtype == "VORBIS" + return info.format in ("WAV", "FLAC") and ( + info.subtype.startswith("PCM_") or info.subtype in ("FLOAT", "DOUBLE") + ) + + def _trim_to(self, output_file_path: str, loop_end: int, keep_after: int): + """Writes the source audio, losslessly cut `keep_after` samples after the loop end, to `output_file_path`.""" + info = soundfile.info(self.filepath) + if not self._supports_lossless_trim(info): + 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 info.format == "OGG": # 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 + return # Read the samples in their native representation so that writing them back is bit-exact dtype = {"FLOAT": "float32", "DOUBLE": "float64"}.get(info.subtype, "int32") @@ -364,8 +378,6 @@ def trim( 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: @@ -479,7 +491,9 @@ def export_tags( loop_start_tag: str, loop_end_tag: str, is_offset: Optional[bool] = None, - output_dir: Optional[str] = None + output_dir: Optional[str] = None, + trim: bool = False, + keep_after: int = 0, ) -> Tuple[str]: """Adds metadata tags of loop points to a copy of the source audio file. @@ -490,6 +504,11 @@ def export_tags( loop_end_tag (str): Name of the loop_end metadata tag. is_offset (bool, optional): Export second tag as relative length / absolute end. Defaults to auto-detecting based on tag name. output_dir (str, optional): Path to the output directory. Defaults to the same diretcory as the source audio file. + trim (bool, optional): Also losslessly cut the tagged copy `keep_after` samples after the loop end (see `trim`). Defaults to False. + keep_after (int, optional): Number of samples to keep after the loop end when trimming. Defaults to 0. + + Raises: + ValueError: if `trim` is set and the source file does not support lossless trimming. """ # Workaround for taglib import issues on Apple silicon devices # Import taglib only when needed to isolate ImportErrors @@ -503,7 +522,10 @@ def export_tags( exported_file_path = os.path.join( output_dir, f"{track_name}-tagged{file_extension}" ) - shutil.copyfile(self.mlaudio.filepath, exported_file_path) + if trim: + self._trim_to(exported_file_path, loop_end, keep_after) + else: + shutil.copyfile(self.mlaudio.filepath, exported_file_path) # Handle LOOPLENGTH tag if self._end_tag_is_offset(loop_end_tag, is_offset): diff --git a/pymusiclooper/handler.py b/pymusiclooper/handler.py index 64bf38e..2bcc48b 100644 --- a/pymusiclooper/handler.py +++ b/pymusiclooper/handler.py @@ -1,5 +1,6 @@ import logging import os +import signal import sys from contextlib import contextmanager from typing import List, Literal, Optional, Tuple @@ -13,6 +14,10 @@ from pymusiclooper.exceptions import AudioLoadError, LoopNotFoundError from pymusiclooper.utils import DEFAULT_OUTPUT_DIRECTORY_NAME +# Default number of samples to keep after the loop end when trimming interactively: enough for players +# that read slightly past the loop end (e.g. for resampling), while being inaudible (~2 ms at 44.1 kHz) +RECOMMENDED_KEEP_AFTER = 100 + class LoopHandler: def __init__( @@ -176,9 +181,8 @@ def get_user_input(): return get_user_input() return selected_index - except KeyboardInterrupt: - rich_console.print("\n[red]Operation terminated by user. Exiting.[/]") - sys.exit() + except (KeyboardInterrupt, EOFError): + _exit_by_user() class LoopExportHandler(LoopHandler): @@ -244,6 +248,10 @@ def run(self): loop_start = chosen_loop_pair.loop_start loop_end = chosen_loop_pair.loop_end + if self.interactive_mode and not self.trim and self.musiclooper.supports_lossless_trim(): + with _hideprogressbar(self._progressbar): + self.trim, self.keep_after = self.trim_prompt() + # Runners that do not need an output directory if self.to_stdout: self.stdout_export_runner(loop_start, loop_end) @@ -273,7 +281,8 @@ def run(self): if self.extended_length: self.extend_track_runner(loop_start, loop_end) - if self.trim: + # The tag runner trims its tagged copy itself, so that a single file is written + if self.trim and self.tag_names is None: self.trim_runner(loop_end) finally: if ( @@ -333,6 +342,28 @@ def extend_track_runner(self, loop_start: int, loop_end: int): except ValueError as e: logging.error(e) + def trim_prompt(self) -> Tuple[bool, int]: + """Asks whether to also losslessly trim the audio after the chosen loop end, and how many samples to keep after it.""" + try: + answer = rich_console.input("Trim the audio after the loop end? [cyan]\\[y/N][/]: ") + if answer.strip().lower() not in ("y", "yes"): + return False, 0 + + while True: + keep_after = rich_console.input( + f"Samples to keep after the loop end (press Enter for the recommended [cyan]{RECOMMENDED_KEEP_AFTER}[/]): " + ).strip() + if not keep_after: + return True, RECOMMENDED_KEEP_AFTER + try: + if int(keep_after) >= 0: + return True, int(keep_after) + except ValueError: + pass + rich_console.print(f"Please enter a whole number of samples (0 or more), or press Enter for {RECOMMENDED_KEEP_AFTER}.") + except (KeyboardInterrupt, EOFError): + _exit_by_user() + def trim_runner(self, loop_end: int): try: output_path = self.musiclooper.trim( @@ -404,8 +435,16 @@ def tag_runner(self, loop_start: int, loop_end: int): loop_end_tag, is_offset=self.tag_offset, output_dir=self.output_directory, + trim=self.trim, + keep_after=self.keep_after, ) - message = f"Exported {loop_start_tag}: {loop_start} and {loop_end_tag}: {loop_end} of \"{self.musiclooper.filename}\" to a copy in \"{self.output_directory}\"" + if not self.trim: + copy_description = "a copy" + elif self.keep_after: + copy_description = f"a copy trimmed {self.keep_after} samples after the loop end" + else: + copy_description = "a copy trimmed at the loop end" + message = f"Exported {loop_start_tag}: {loop_start} and {loop_end_tag}: {loop_end} of \"{self.musiclooper.filename}\" to {copy_description} in \"{self.output_directory}\"" if self.batch_mode: logging.info(message) else: @@ -541,6 +580,18 @@ def _cleanup_empty_created_dirs(self): os.removedirs(directory) +def _exit_by_user(): + """Exits after the user interrupted a prompt. + + Catch both KeyboardInterrupt and EOFError around prompts: on Windows, Ctrl+C at a prompt + can surface as an EOFError, with the actual interrupt delivered moments later. + """ + rich_console.print("\n[red]Operation terminated by user. Exiting.[/]") + # Ignore further Ctrl+C so that a late interrupt does not break the audio cleanup at exit + signal.signal(signal.SIGINT, signal.SIG_IGN) + sys.exit() + + @contextmanager def _hideprogressbar(progress: Progress): """ diff --git a/tests/test_core.py b/tests/test_core.py index ff3a717..e385d61 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -253,6 +253,51 @@ def test_trim_rejects_unsupported_formats(track, tmp_path, filename, format, sub MusicLooper(str(source_path)).trim(LOOP_END, output_dir=str(tmp_path)) +@pytest.mark.parametrize( + ("filename", "format", "subtype", "supported"), + [ + ("pcm16.wav", "WAV", "PCM_16", True), + ("float.wav", "WAV", "FLOAT", True), + ("track.flac", "FLAC", "PCM_16", True), + ("track.ogg", "OGG", "VORBIS", True), + ("track.opus", "OGG", "OPUS", False), + ("track.mp3", "MP3", "MPEG_LAYER_III", False), + ("ulaw.wav", "WAV", "ULAW", False), + ], +) +def test_supports_lossless_trim(track, tmp_path, filename, format, subtype, supported): + source_path = tmp_path / filename + sf.write(source_path, track, SR if subtype != "OPUS" else 48000, format=format, subtype=subtype) + + assert MusicLooper(str(source_path)).supports_lossless_trim() is supported + + +@pytest.mark.parametrize("filename", ["track.flac", "track.ogg"]) +def test_export_tags_with_trim_writes_single_tagged_trimmed_file(track, tmp_path, filename): + source_path = tmp_path / filename + sf.write(source_path, track, SR) + out_dir = tmp_path / "out" + out_dir.mkdir() + + MusicLooper(str(source_path)).export_tags( + LOOP_START, LOOP_END, "LOOPSTART", "LOOPLENGTH", output_dir=str(out_dir), trim=True, keep_after=100 + ) + + track_name, extension = os.path.splitext(filename) + assert os.listdir(out_dir) == [f"{track_name}-tagged{extension}"] + output_path = str(out_dir / f"{track_name}-tagged{extension}") + assert sf.info(output_path).frames == LOOP_END + 100 + assert MusicLooper(output_path).read_tags("LOOPSTART", "LOOPLENGTH") == (LOOP_START, LOOP_END) + + +def test_export_tags_with_trim_rejects_unsupported_formats(track, tmp_path): + source_path = tmp_path / "track.mp3" + sf.write(source_path, track, SR, format="MP3") + + with pytest.raises(ValueError): + MusicLooper(str(source_path)).export_tags(LOOP_START, LOOP_END, "LOOP_START", "LOOP_END", output_dir=str(tmp_path), trim=True) + + def test_ogg_vorbis_packet_durations_match_granule_positions(track, tmp_path): from pymusiclooper.ogg import _find_packet_cut, _read_pages diff --git a/tests/test_handler.py b/tests/test_handler.py index 61a0ad8..6010ef3 100644 --- a/tests/test_handler.py +++ b/tests/test_handler.py @@ -1,6 +1,8 @@ import os import pytest +import soundfile as sf +from conftest import SR from pymusiclooper import handler from pymusiclooper.analysis import LoopPair @@ -46,6 +48,99 @@ def test_interactive_more_then_select(fake_input): assert _bare_loop_handler().interactive_handler() == 27 +@pytest.mark.parametrize("interrupt", [KeyboardInterrupt, EOFError]) +def test_interactive_exits_cleanly_on_interrupt(monkeypatch, interrupt): + def raise_interrupt(*args, **kwargs): + raise interrupt + monkeypatch.setattr(handler.rich_console, "input", raise_interrupt) + monkeypatch.setattr(handler.signal, "signal", lambda *args: None) + with pytest.raises(SystemExit): + _bare_loop_handler().interactive_handler() + + +@pytest.mark.parametrize( + ("answers", "expected"), + [ + (("",), (False, 0)), + (("n",), (False, 0)), + (("y", ""), (True, handler.RECOMMENDED_KEEP_AFTER)), + (("Y", "0"), (True, 0)), + (("yes", "abc", "-5", "1.5", "10"), (True, 10)), + ], +) +def test_trim_prompt(fake_input, answers, expected): + fake_input(*answers) + assert LoopExportHandler.trim_prompt(None) == expected + + +def test_trim_prompt_exits_cleanly_on_interrupt(monkeypatch): + def raise_interrupt(*args, **kwargs): + raise EOFError + monkeypatch.setattr(handler.rich_console, "input", raise_interrupt) + monkeypatch.setattr(handler.signal, "signal", lambda *args: None) + with pytest.raises(SystemExit): + LoopExportHandler.trim_prompt(None) + + +def _interactive_export_handler(monkeypatch, source_path, output_dir, **kwargs): + monkeypatch.setenv("PML_INTERACTIVE_MODE", "1") + monkeypatch.setattr(handler.rich_console, "print", lambda *args, **kwargs: None) + return LoopExportHandler(path=str(source_path), min_duration_multiplier=0.35, output_dir=str(output_dir), **kwargs) + + +def test_interactive_tag_with_trim_prompt_writes_single_file(monkeypatch, fake_input, track, tmp_path): + source_path = tmp_path / "track.flac" + sf.write(source_path, track, SR) + out_dir = tmp_path / "out" + export_handler = _interactive_export_handler(monkeypatch, source_path, out_dir, tag_names=("LOOPSTART", "LOOPLENGTH")) + loop_pair = export_handler.loop_pair_list[0] + + fake_input("0", "y", "") + export_handler.run() + + assert os.listdir(out_dir) == ["track-tagged.flac"] + tagged = MusicLooper(str(out_dir / "track-tagged.flac")) + assert tagged.read_tags("LOOPSTART", "LOOPLENGTH") == (loop_pair.loop_start, loop_pair.loop_end) + assert sf.info(out_dir / "track-tagged.flac").frames == loop_pair.loop_end + handler.RECOMMENDED_KEEP_AFTER + + +def test_interactive_trim_prompt_declined(monkeypatch, fake_input, track, tmp_path): + source_path = tmp_path / "track.flac" + sf.write(source_path, track, SR) + out_dir = tmp_path / "out" + export_handler = _interactive_export_handler(monkeypatch, source_path, out_dir, tag_names=("LOOP_START", "LOOP_END")) + + fake_input("0", "") + export_handler.run() + + assert os.listdir(out_dir) == ["track-tagged.flac"] + assert sf.info(out_dir / "track-tagged.flac").frames == track.size + + +def test_interactive_trim_prompt_skipped_for_unsupported_formats(monkeypatch, fake_input, track, tmp_path): + source_path = tmp_path / "track.mp3" + sf.write(source_path, track, SR, format="MP3") + export_handler = _interactive_export_handler(monkeypatch, source_path, tmp_path / "out", to_stdout=True) + + # Only the loop selection is answered; a trim prompt would exhaust the answers + fake_input("0") + export_handler.run() + + +def test_interactive_trim_command_does_not_prompt_again(monkeypatch, fake_input, track, tmp_path): + source_path = tmp_path / "track.flac" + sf.write(source_path, track, SR) + out_dir = tmp_path / "out" + export_handler = _interactive_export_handler(monkeypatch, source_path, out_dir, trim=True, keep_after=7) + loop_pair = export_handler.loop_pair_list[0] + + fake_input("0") + export_handler.run() + + assert os.listdir(out_dir) == ["track-trimmed.flac"] + assert sf.info(out_dir / "track-trimmed.flac").frames == loop_pair.loop_end + 7 + + 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]