Interactive trim prompt and clean Ctrl+C exit on Windows

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/)

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):

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):
     """

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
 

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]