From 687b9344986879bc0eb545b67c3fa5badde34390 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 26 Sep 2026 07:42:41 +0000 Subject: [PATCH 1/2] Preserve all metadata when trimming and handle Ogg streams with a start offset - Trim WAV files at the chunk level: every chunk is copied verbatim (tags, sampler loops, cue points, bext, the exact format chunk), only the data chunk is shortened and the fact sample count updated. - Restore the source's FLAC metadata blocks (tags, cover art, application blocks, padding, surrounding ID3 tags) onto the re-encoded audio, dropping only the seek table and cue sheet, which point into the old audio. - Log a warning instead of silently ignoring failures to copy tags. - Trim Ogg Vorbis streams whose granule positions start past 0 at the right sample, as decoders number samples from the stream's start granule. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QTLaHUDYw58dcfrK65spTz --- README.md | 4 +- pymusiclooper/cli.py | 2 +- pymusiclooper/core.py | 33 +++++-- pymusiclooper/flac.py | 96 ++++++++++++++++++ pymusiclooper/ogg.py | 66 +++++++++---- pymusiclooper/wav.py | 83 ++++++++++++++++ tests/test_core.py | 222 ++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 477 insertions(+), 29 deletions(-) create mode 100644 pymusiclooper/flac.py create mode 100644 pymusiclooper/wav.py diff --git a/README.md b/README.md index 93bd136..aee2cac 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,9 @@ pymusiclooper export-points --path "/path/to/track.wav" --alt-export-top -1 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 +# The original format, bit depth and metadata are preserved; WAV and OGG files are cut without re-encoding. +# WAV and OGG files keep all their metadata (e.g. tags, WAV sampler loops and cue points); +# FLAC files keep all theirs (e.g. tags and cover art) except the seek table and cue sheet, which no longer apply 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 diff --git a/pymusiclooper/cli.py b/pymusiclooper/cli.py index 5c6b703..bcb2a9a 100644 --- a/pymusiclooper/cli.py +++ b/pymusiclooper/cli.py @@ -234,7 +234,7 @@ def tag(**kwargs): @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)[/]""" + """Losslessly cut the audio after the loop end, keeping its original format, bit depth and metadata. [dim](WAV, FLAC and OGG Vorbis only)[/]""" kwargs["trim"] = True run_handler(**kwargs) diff --git a/pymusiclooper/core.py b/pymusiclooper/core.py index e2cb556..e0065bc 100644 --- a/pymusiclooper/core.py +++ b/pymusiclooper/core.py @@ -13,8 +13,10 @@ from pymusiclooper.analysis import LoopPair, find_best_loop_points from pymusiclooper.audio import MLAudio from pymusiclooper.exceptions import LoopNotFoundError +from pymusiclooper.flac import copy_flac_metadata from pymusiclooper.ogg import trim_vorbis from pymusiclooper.playback import PlaybackHandler +from pymusiclooper.wav import trim_wav # Lazy-load external libraries when they're needed soundfile = lazy.load("soundfile") @@ -315,8 +317,10 @@ def trim( 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. + The original container, bit depth and metadata are kept; only WAV, FLAC and Ogg Vorbis files are supported. + WAV and Ogg Vorbis files are trimmed at the container level, without re-encoding the audio, and all their + metadata is copied (e.g. WAV sampler loops and cue points). FLAC audio is re-encoded losslessly, and all its + metadata except the seek table and cue sheet is copied (e.g. tags and cover art). Returns the path to the trimmed audio file. Args: @@ -357,11 +361,15 @@ def _trim_to(self, output_file_path: str, loop_end: int, keep_after: int): f"Lossless trimming is only supported for PCM/float WAV, FLAC and Ogg Vorbis files; \"{self.filename}\" is {info.format} ({info.subtype})." ) + # The pages/chunks holding the metadata are copied verbatim, so the metadata is kept as-is 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 + if info.format == "WAV": + trim_wav(self.filepath, output_file_path, loop_end + keep_after) + return + # FLAC audio frames cannot be cut, so the kept samples are re-encoded (losslessly). # 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) @@ -376,10 +384,16 @@ def _trim_to(self, output_file_path: str, loop_end: int, keep_after: int): endian=info.endian, ) - self._copy_tags(output_file_path) + try: + copy_flac_metadata(self.filepath, output_file_path) + except ValueError as e: + logging.warning( + f"Could not copy the FLAC metadata blocks of \"{self.filename}\" ({e}); copying its tags only." + ) + self._copy_tags(output_file_path) def _copy_tags(self, dest_filepath: str): - """Attempts to copy the metadata tags of the source audio file to `dest_filepath`.""" + """Attempts to copy the metadata tags of the source audio file to `dest_filepath`, logging a warning on failure.""" try: import taglib original_tags = None @@ -389,10 +403,11 @@ def _copy_tags(self, dest_filepath: str): 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: - # silently ignore errors for now; - # TODO: implement logging for debugging - pass + except Exception as e: + # Tag copying is best-effort: the exported audio is still valid without them + logging.warning( + f"Could not copy the metadata tags of \"{self.filename}\" to \"{os.path.basename(dest_filepath)}\": {e}" + ) def export_txt( self, diff --git a/pymusiclooper/flac.py b/pymusiclooper/flac.py new file mode 100644 index 0000000..fdf10a7 --- /dev/null +++ b/pymusiclooper/flac.py @@ -0,0 +1,96 @@ +"""Restores the metadata of a FLAC file onto a re-encoded copy of its audio. + +A FLAC file is a list of metadata blocks followed by the audio frames. The metadata blocks of the source file +are copied verbatim (tags, pictures such as cover art, application blocks, padding), except for the ones that +describe the audio itself: the stream info (length, MD5 checksum) is taken from the re-encoded copy, and the seek table +and cue sheet are dropped, since they point to positions in the source audio that no longer exist in the copy. +""" + +from typing import List, Tuple + +_STREAMINFO = 0 +_SEEKTABLE = 3 +_CUESHEET = 5 +_LAST_BLOCK_FLAG = 0x80 +_ID3V1_LENGTH = 128 + + +def _id3v2_length(data: bytes) -> int: + """Returns the length of the ID3v2 tag at the start of the file (some taggers add one to FLAC files), or 0.""" + if len(data) < 10 or data[:3] != b"ID3": + return 0 + # The tag size is stored as a 28-bit "synchsafe" integer (7 bits per byte) + size = 0 + for byte in data[6:10]: + size = (size << 7) | (byte & 0x7F) + has_footer = data[5] & 0x10 + return 10 + size + (10 if has_footer else 0) + + +def _read_metadata(data: bytes) -> Tuple[int, List[Tuple[int, bytes]], int]: + """Parses the metadata blocks of a FLAC file. + + Returns: + Tuple[int, List[Tuple[int, bytes]], int]: (offset of the "fLaC" marker, [(block type, block data)], + offset of the first audio frame) + """ + marker_offset = _id3v2_length(data) + if data[marker_offset:marker_offset + 4] != b"fLaC": + raise ValueError("Not a FLAC file.") + + blocks = [] + offset = marker_offset + 4 + while True: + if offset + 4 > len(data): + raise ValueError("Invalid FLAC file: truncated metadata.") + header = data[offset] + length = int.from_bytes(data[offset + 1:offset + 4], "big") + block_data = data[offset + 4:offset + 4 + length] + if len(block_data) != length: + raise ValueError("Invalid FLAC file: truncated metadata.") + blocks.append((header & 0x7F, block_data)) + offset += 4 + length + if header & _LAST_BLOCK_FLAG: + return marker_offset, blocks, offset + + +def copy_flac_metadata(src_filepath: str, dest_filepath: str): + """Replaces the metadata of the FLAC file `dest_filepath` with the metadata of `src_filepath`, + except for the blocks describing the audio itself (see the module docstring). + ID3 tags that taggers sometimes add before or after the FLAC data are copied too. + + Args: + src_filepath (str): Path to the FLAC file to copy the metadata from. + dest_filepath (str): Path to the FLAC file to write the metadata to, in place. + + Raises: + ValueError: if either file is not a valid FLAC file. + """ + with open(src_filepath, "rb") as f: + src = f.read() + with open(dest_filepath, "rb") as f: + dest = f.read() + + src_marker_offset, src_blocks, _ = _read_metadata(src) + _, dest_blocks, dest_audio_offset = _read_metadata(dest) + + dest_streaminfo = [block_data for block_type, block_data in dest_blocks if block_type == _STREAMINFO] + if not dest_streaminfo: + raise ValueError("Invalid FLAC file: missing stream info.") + + blocks = [(_STREAMINFO, dest_streaminfo[0])] + [ + (block_type, block_data) + for block_type, block_data in src_blocks + if block_type not in (_STREAMINFO, _SEEKTABLE, _CUESHEET) + ] + + output = [src[:src_marker_offset], b"fLaC"] + for idx, (block_type, block_data) in enumerate(blocks): + last_block_flag = _LAST_BLOCK_FLAG if idx == len(blocks) - 1 else 0 + output += [bytes([last_block_flag | block_type]), len(block_data).to_bytes(3, "big"), block_data] + output.append(dest[dest_audio_offset:]) + if src[-_ID3V1_LENGTH:-_ID3V1_LENGTH + 3] == b"TAG": + output.append(src[-_ID3V1_LENGTH:]) + + with open(dest_filepath, "wb") as f: + f.write(b"".join(output)) diff --git a/pymusiclooper/ogg.py b/pymusiclooper/ogg.py index 1933dc6..199d170 100644 --- a/pymusiclooper/ogg.py +++ b/pymusiclooper/ogg.py @@ -2,7 +2,8 @@ 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 end of a page, counted from the stream's start granule) 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). @@ -158,17 +159,13 @@ def __call__(self, first_byte: int) -> int: 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. +def _completed_audio_packets(pages: List[_Page]) -> Iterator[Tuple[int, List[Tuple[int, int]]]]: + """Yields, for each page on which audio packets end, its index and the + (duration, number of segments up to the packet's end) of each audio packet completed on it. """ 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 @@ -190,10 +187,40 @@ def _find_packet_cut(pages: List[_Page], n_samples: int) -> Optional[Tuple[int, 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 completed_packets and page.granule_position != -1: + yield page_idx, completed_packets + + +def _start_granule(pages: List[_Page]) -> int: + """Returns the granule position of the stream's first sample. + + It is usually 0, but a stream cut out of a longer one (e.g. recorded from a live stream) can start later: + decoders then take the granule position of the first audio page minus the samples decoded from it as the start, + and number the decoded samples from there. A lower granule position on the first audio page instead means that + the stream also ends on it, before the end of its last packet. + """ + for page_idx, completed_packets in _completed_audio_packets(pages): + return max(0, pages[page_idx].granule_position - sum(duration for duration, _ in completed_packets)) + return 0 + + +def _find_packet_cut(pages: List[_Page], end_granule: int, start_granule: int = 0) -> Optional[Tuple[int, int]]: + """Finds where to cut the stream so that the last kept packet contains the sample at granule position `end_granule`. + + Args: + pages (List[_Page]): The pages of the stream. + end_granule (int): Granule position of the requested end. + start_granule (int, optional): Granule position of the stream's first sample (see `_start_granule`). Defaults to 0. + + 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. + """ + previous_granule = start_granule - if page.granule_position >= n_samples: + for page_idx, completed_packets in _completed_audio_packets(pages): + page = pages[page_idx] + if page.granule_position >= end_granule: 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 @@ -202,7 +229,7 @@ def _find_packet_cut(pages: List[_Page], n_samples: int) -> Optional[Tuple[int, packet_end = previous_granule for duration, n_segments in completed_packets: packet_end += duration - if packet_end >= n_samples: + if packet_end >= end_granule: return page_idx, n_segments previous_granule = page.granule_position @@ -210,12 +237,12 @@ def _find_packet_cut(pages: List[_Page], n_samples: int) -> Optional[Tuple[int, return None -def _find_page_cut(pages: List[_Page], n_samples: int) -> Optional[Tuple[int, int]]: +def _find_page_cut(pages: List[_Page], end_granule: 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: + if page.granule_position >= end_granule: # 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. @@ -256,19 +283,22 @@ def trim_vorbis(src_filepath: str, dest_filepath: str, n_samples: int) -> int: if any(page.serial_number != pages[0].serial_number for page in pages): raise ValueError("Chained or multiplexed Ogg files are not supported.") + # Sample positions are counted from the stream's first sample, granule positions from its start granule + start_granule = 0 try: - cut = _find_packet_cut(pages, max(1, n_samples)) + start_granule = _start_granule(pages) + cut = _find_packet_cut(pages, start_granule + max(1, n_samples), start_granule) 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)) + cut = _find_page_cut(pages, start_granule + 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) + return max(0, pages[-1].granule_position - start_granule) page_idx, n_segments = cut page = pages[page_idx] @@ -278,7 +308,7 @@ def trim_vorbis(src_filepath: str, dest_filepath: str, n_samples: int) -> int: last_page = _build_page( page, header_type=page.header_type | _END_OF_STREAM_FLAG, - granule_position=n_samples, + granule_position=start_granule + n_samples, segment_table=segment_table, body=body, ) diff --git a/pymusiclooper/wav.py b/pymusiclooper/wav.py new file mode 100644 index 0000000..b98146e --- /dev/null +++ b/pymusiclooper/wav.py @@ -0,0 +1,83 @@ +"""Lossless trimming of WAV files at the chunk level, without rewriting the audio. + +Every chunk of the source file is copied verbatim, so everything stored besides the audio is kept as-is: +the format chunk, tags (LIST/INFO and ID3 chunks), sampler loops (smpl), cue points, broadcast extension (bext), etc. +Only the audio data chunk is shortened, and the sample count of the fact chunk (if any) and the RIFF size are updated. +""" + +import struct +from typing import List, NamedTuple + + +class _Chunk(NamedTuple): + chunk_id: bytes + data: bytes + + +def _read_chunks(data: bytes, endian: str) -> List[_Chunk]: + chunks = [] + offset = 12 + while offset + 8 <= len(data): + chunk_id = data[offset:offset + 4] + (size,) = struct.unpack_from(f"{endian}I", data, offset + 4) + # Files written by interrupted or streaming encoders can leave the data size as a placeholder: + # 0 (then the data runs to the end of the file), or a size past the end of the file + if chunk_id == b"data" and size == 0: + size = len(data) - offset - 8 + chunk_data = data[offset + 8:offset + 8 + size] + chunks.append(_Chunk(chunk_id, chunk_data)) + # Chunks are padded to an even size + offset += 8 + size + (size & 1) + return chunks + + +def trim_wav(src_filepath: str, dest_filepath: str, n_frames: int) -> int: + """Writes a copy of a PCM/float WAV file that ends after `n_frames` frames, keeping all its other chunks. + + Args: + src_filepath (str): Path to the source WAV file. + dest_filepath (str): Path of the trimmed file to write. + n_frames (int): Number of frames (samples per channel) to keep from the start of the audio. + + Raises: + ValueError: if the file is not a RIFF/RIFX WAVE file with a format and a data chunk. + + Returns: + int: The length of the trimmed file in frames. + """ + with open(src_filepath, "rb") as f: + data = f.read() + + if data[:4] not in (b"RIFF", b"RIFX") or data[8:12] != b"WAVE": + raise ValueError("Not a RIFF WAVE file.") + endian = "<" if data[:4] == b"RIFF" else ">" + + chunks = _read_chunks(data, endian) + fmt_chunk = next((chunk for chunk in chunks if chunk.chunk_id == b"fmt "), None) + if fmt_chunk is None or len(fmt_chunk.data) < 14 or not any(chunk.chunk_id == b"data" for chunk in chunks): + raise ValueError("Invalid WAVE file: missing format or data chunk.") + + (block_align,) = struct.unpack_from(f"{endian}H", fmt_chunk.data, 12) + if block_align == 0: + raise ValueError("Invalid WAVE file: zero block alignment.") + + audio_chunk = next(chunk for chunk in chunks if chunk.chunk_id == b"data") + kept_frames = min(max(0, n_frames), len(audio_chunk.data) // block_align) + + body = [] + for chunk in chunks: + chunk_data = chunk.data + if chunk is audio_chunk: + chunk_data = chunk_data[:kept_frames * block_align] + elif chunk.chunk_id == b"fact" and len(chunk_data) >= 4: + # The fact chunk starts with the number of frames (required for non-PCM formats, e.g. float) + chunk_data = struct.pack(f"{endian}I", kept_frames) + chunk_data[4:] + body += [chunk.chunk_id, struct.pack(f"{endian}I", len(chunk_data)), chunk_data] + if len(chunk_data) & 1: + body.append(b"\x00") + body = b"".join(body) + + with open(dest_filepath, "wb") as f: + f.write(data[:4] + struct.pack(f"{endian}I", 4 + len(body)) + b"WAVE" + body) + + return kept_frames diff --git a/tests/test_core.py b/tests/test_core.py index e385d61..d55eb31 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,4 +1,6 @@ import os +import struct +from pathlib import Path import numpy as np import pytest @@ -207,6 +209,163 @@ def test_trim_copies_source_tags(flac_track_path, tmp_path): assert MusicLooper(output_path).read_tags(None, None) == (LOOP_START, LOOP_END) +def _append_wav_chunk(path, chunk_id: bytes, chunk_data: bytes): + data = path.read_bytes() + chunk_id + struct.pack(" dict: + from pymusiclooper.wav import _read_chunks + + return {chunk.chunk_id: chunk.data for chunk in _read_chunks(path.read_bytes(), "<")} + + +@pytest.mark.parametrize( + ("subtype", "keep_after"), + [ + ("PCM_16", 100), + # 8-bit mono with an odd number of frames: the data chunk needs a pad byte + ("PCM_U8", 1), + # Float WAV files have a fact chunk holding the number of frames + ("FLOAT", 0), + ], +) +def test_trim_wav_keeps_all_chunks(track, tmp_path, subtype, keep_after): + source_path = tmp_path / "track.wav" + sf.write(source_path, track, SR, subtype=subtype) + with taglib.File(str(source_path), save_on_exit=True) as source: + source.tags["TITLE"] = ["Loop"] + sampler_chunk = bytes(range(60)) + odd_chunk = b"odd" + _append_wav_chunk(source_path, b"smpl", sampler_chunk) + _append_wav_chunk(source_path, b"xodd", odd_chunk) + 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)) + + n_frames = LOOP_END + keep_after + source_chunks = _wav_chunks(source_path) + trimmed_chunks = _wav_chunks(tmp_path / "out" / "track-trimmed.wav") + assert list(trimmed_chunks) == list(source_chunks) + for chunk_id in source_chunks: + if chunk_id not in (b"data", b"fact"): + assert trimmed_chunks[chunk_id] == source_chunks[chunk_id] + assert trimmed_chunks[b"smpl"] == sampler_chunk + assert trimmed_chunks[b"xodd"] == odd_chunk + if b"fact" in source_chunks: + assert struct.unpack_from("II{len(mime_type)}sI{len(description)}sIIIII{len(picture)}s", + 3, len(mime_type), mime_type, len(description), description, 16, 16, 24, 0, len(picture), picture, + )) + application_block = (2, b"TEST" + bytes(range(40))) + # A seek table with a single placeholder point, which does not apply to the trimmed audio + seektable_block = (3, b"\xff" * 8 + b"\x00" * 10) + id3v2 = b"ID3\x03\x00\x00\x00\x00\x00\x14" + b"\x00" * 20 + id3v1 = b"TAG" + b"\x00" * 125 + _flac_with_metadata_blocks(source_path, [seektable_block, picture_block, application_block], id3v2, id3v1) + 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)) + + trimmed_data = open(output_path, "rb").read() + _, source_blocks, _ = _read_metadata(source_path.read_bytes()) + _, trimmed_blocks, _ = _read_metadata(trimmed_data) + assert trimmed_blocks[0][0] == 0 + assert trimmed_blocks[1:] == [block for block in source_blocks if block[0] not in (0, 3)] + assert picture_block in trimmed_blocks and application_block in trimmed_blocks + assert trimmed_data.startswith(id3v2) and trimmed_data.endswith(id3v1) + + source, _ = sf.read(source_path, dtype="int32") + trimmed, _ = sf.read(output_path, dtype="int32") + assert sf.info(output_path).frames == LOOP_END + 100 + np.testing.assert_array_equal(trimmed, source[: LOOP_END + 100]) + with taglib.File(output_path) as trimmed_file: + assert trimmed_file.tags["TITLE"] == ["Loop"] + + +def test_trim_flac_falls_back_to_copying_tags(flac_track_path, tmp_path, monkeypatch, caplog): + from pymusiclooper import core + + def fail(*args, **kwargs): + raise ValueError("unparseable") + + monkeypatch.setattr(core, "copy_flac_metadata", fail) + with taglib.File(flac_track_path, save_on_exit=True) as source: + source.tags["TITLE"] = ["Loop"] + out_dir = tmp_path / "out" + out_dir.mkdir() + + output_path = MusicLooper(flac_track_path).trim(LOOP_END, output_dir=str(out_dir)) + + assert "copying its tags only" in caplog.text + with taglib.File(output_path) as trimmed_file: + assert trimmed_file.tags["TITLE"] == ["Loop"] + + +def test_copy_tags_failure_is_logged(looper, tmp_path, caplog): + looper._copy_tags(str(tmp_path / "missing.wav")) + + assert "Could not copy the metadata tags" in caplog.text + + @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" @@ -327,3 +486,66 @@ def fail(*args, **kwargs): source, _ = sf.read(source_path, dtype="float32") trimmed, _ = sf.read(output_path, dtype="float32") np.testing.assert_array_equal(trimmed, source[:LOOP_END]) + + +def _ogg_with_start_granule(source_path, output_path, start_granule: int): + """Writes a copy of an Ogg Vorbis file whose samples are numbered from `start_granule` instead of 0, + like a stream recorded from the middle of a longer one.""" + from pymusiclooper.ogg import _build_page, _read_pages + + output_path.write_bytes(b"".join( + _build_page( + page, + page.header_type, + page.granule_position + start_granule if page.granule_position > 0 else page.granule_position, + page.segment_table, + page.body, + ) + for page in _read_pages(source_path.read_bytes()) + )) + + +@pytest.mark.parametrize("keep_after", [0, 777]) +def test_trim_ogg_vorbis_with_start_granule(track, tmp_path, keep_after): + from pymusiclooper.ogg import _find_packet_cut, _read_pages, _start_granule + + start_granule = 44100 + plain_path = tmp_path / "plain.ogg" + sf.write(plain_path, np.stack([track, 0.8 * track], axis=1), SR, format="OGG", subtype="VORBIS") + source_path = tmp_path / "track.ogg" + _ogg_with_start_granule(plain_path, source_path, start_granule) + out_dir = tmp_path / "out" + out_dir.mkdir() + + pages = list(_read_pages(source_path.read_bytes())) + assert _start_granule(pages) == start_granule + # Raises if the computed packet durations do not add up to the page granule positions + assert _find_packet_cut(pages, start_granule + LOOP_END, start_granule) is not None + + 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 source.shape[0] == track.size + np.testing.assert_array_equal(trimmed, source[: LOOP_END + keep_after]) + assert list(_read_pages(open(output_path, "rb").read()))[-1].granule_position == start_granule + LOOP_END + keep_after + + +def test_trim_ogg_vorbis_with_start_granule_past_track_end(track, tmp_path): + from pymusiclooper.ogg import trim_vorbis + + plain_path = tmp_path / "plain.ogg" + sf.write(plain_path, track, SR, format="OGG", subtype="VORBIS") + source_path = tmp_path / "track.ogg" + _ogg_with_start_granule(plain_path, source_path, 44100) + + assert trim_vorbis(str(source_path), str(tmp_path / "trimmed.ogg"), 10 * track.size) == track.size + + +def test_ogg_vorbis_start_granule_is_zero_for_regular_files(track, tmp_path): + from pymusiclooper.ogg import _read_pages, _start_granule + + source_path = tmp_path / "track.ogg" + sf.write(source_path, track, SR, format="OGG", subtype="VORBIS") + + assert _start_granule(list(_read_pages(source_path.read_bytes()))) == 0 From 6b401678b6654b67cafae1315fed5fabd6d28573 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 26 Sep 2026 08:01:32 +0000 Subject: [PATCH 2/2] Support WAV smpl loops, skip the trim prompt for console export, dedupe embedded loops - Read loop points from a WAV file's sampler (smpl) chunk when it has no loop tags (tag names auto-detected), so they are used as the first loop choice and by play-tagged; write them to the smpl chunk when tagging WAV files, keeping any existing sampler data and other loops. - Parse WAV chunks from their headers with seeks, so looking for the smpl chunk does not read the whole file. - Do not offer to trim when export-points only prints to the console. - Leave out detected loops within 5 ms of the embedded loop points, which would otherwise be listed twice. - Default `trim --keep-after` to the same 100 samples as the interactive prompt. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QTLaHUDYw58dcfrK65spTz --- README.md | 6 +- pymusiclooper/cli.py | 4 +- pymusiclooper/core.py | 44 +++++++-- pymusiclooper/handler.py | 10 ++- pymusiclooper/wav.py | 188 +++++++++++++++++++++++++++++++-------- tests/test_cli.py | 13 +++ tests/test_core.py | 130 ++++++++++++++++++++++++++- tests/test_handler.py | 15 +++- 8 files changed, 359 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index aee2cac..30fd3b7 100644 --- a/README.md +++ b/README.md @@ -119,7 +119,7 @@ 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.* +*Note: if a file already has loop metadata tags (e.g. LOOP_START/LOOP_END or LOOPSTART/LOOPLENGTH), or is a WAV file with a loop in its sampler (`smpl`) chunk, those loop points are used as the first loop choice by all commands. Use `--ignore-tags` to skip them.* ### Export @@ -145,16 +145,18 @@ pymusiclooper export-points --path "/path/to/track.wav" --alt-export-top -1 # Add metadata tags of the best discovered loop points to a copy of the input audio file # (or all audio files in a directory, if a directory path is used instead) +# WAV files also get the loop points in their sampler (`smpl`) chunk, which many game engines and samplers read 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 +# (100 by default, which avoids clicks in players that read slightly past the loop end; use 0 to cut exactly at it) # The original format, bit depth and metadata are preserved; WAV and OGG files are cut without re-encoding. # WAV and OGG files keep all their metadata (e.g. tags, WAV sampler loops and cue points); # FLAC files keep all theirs (e.g. tags and cover art) except the seek table and cue sheet, which no longer apply 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 +# after the loop is chosen (except when printing loop points to the terminal); with `tag`, a single tagged and trimmed copy is written pymusiclooper -i tag --path "TRACK_NAME.ogg" --tag-names LOOPSTART LOOPLENGTH diff --git a/pymusiclooper/cli.py b/pymusiclooper/cli.py index bcb2a9a..73563ad 100644 --- a/pymusiclooper/cli.py +++ b/pymusiclooper/cli.py @@ -19,7 +19,7 @@ from pymusiclooper.console import _COMMAND_GROUPS, _OPTION_GROUPS, rich_console from pymusiclooper.core import MusicLooper from pymusiclooper.exceptions import AudioLoadError, LoopNotFoundError -from pymusiclooper.handler import BatchHandler, LoopExportHandler, LoopHandler +from pymusiclooper.handler import RECOMMENDED_KEEP_AFTER, BatchHandler, LoopExportHandler, LoopHandler from pymusiclooper.utils import download_audio, get_outputdir, mk_outputdir # CLI --help styling @@ -232,7 +232,7 @@ def tag(**kwargs): @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.") +@click.option('--keep-after', type=click.IntRange(min=0), default=RECOMMENDED_KEEP_AFTER, show_default=True, help="Number of samples to keep after the loop end. [dim](A few samples avoid clicks in players that read slightly past the loop end, e.g. for resampling; use [cyan]0[/] to cut exactly at the loop end.)[/]") def trim(**kwargs): """Losslessly cut the audio after the loop end, keeping its original format, bit depth and metadata. [dim](WAV, FLAC and OGG Vorbis only)[/]""" kwargs["trim"] = True diff --git a/pymusiclooper/core.py b/pymusiclooper/core.py index e0065bc..71d6c62 100644 --- a/pymusiclooper/core.py +++ b/pymusiclooper/core.py @@ -16,7 +16,10 @@ from pymusiclooper.flac import copy_flac_metadata from pymusiclooper.ogg import trim_vorbis from pymusiclooper.playback import PlaybackHandler -from pymusiclooper.wav import trim_wav +from pymusiclooper.wav import read_smpl_loop, trim_wav, write_smpl_loop + +# Detected loop points this close to the embedded ones (in seconds) are the same loop, so they are not listed twice +_DUPLICATE_LOOP_TOLERANCE = 0.005 # Lazy-load external libraries when they're needed soundfile = lazy.load("soundfile") @@ -43,7 +46,8 @@ def find_loop_pairs( 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. + If the file already has loop points stored in its metadata tags, they are returned as the first choice, + and detected loop points matching them (within 5ms) are left out. Args: min_duration_multiplier (float, optional): The minimum duration of a loop as a multiplier of track duration. Defaults to 0.35. @@ -82,12 +86,20 @@ def find_loop_pairs( loop_pairs = [] if embedded_pair is not None: + tolerance = self.mlaudio.seconds_to_samples(_DUPLICATE_LOOP_TOLERANCE) + loop_pairs = [ + pair + for pair in loop_pairs + if abs(pair.loop_start - embedded_pair.loop_start) > tolerance + or abs(pair.loop_end - embedded_pair.loop_end) > tolerance + ] 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. + """Reads the loop points stored in the file's metadata tags (e.g. LOOP_START/LOOP_END), auto-detecting the tag names, + or else in a WAV file's sampler (smpl) chunk. Returns: Optional[LoopPair]: A `LoopPair` with `from_metadata=True`, or None if the file has no valid loop tags. @@ -511,6 +523,7 @@ def export_tags( keep_after: int = 0, ) -> Tuple[str]: """Adds metadata tags of loop points to a copy of the source audio file. + For WAV files, the loop points are also written to the sampler (smpl) chunk. Args: loop_start (int): Loop start in samples. @@ -543,18 +556,24 @@ def export_tags( shutil.copyfile(self.mlaudio.filepath, exported_file_path) # Handle LOOPLENGTH tag - if self._end_tag_is_offset(loop_end_tag, is_offset): - loop_end = loop_end - loop_start + loop_end_tag_value = loop_end - loop_start if self._end_tag_is_offset(loop_end_tag, is_offset) else loop_end with taglib.File(exported_file_path, save_on_exit=True) as audio_file: audio_file.tags[loop_start_tag] = [str(loop_start)] - audio_file.tags[loop_end_tag] = [str(loop_end)] + audio_file.tags[loop_end_tag] = [str(loop_end_tag_value)] + + # Game engines and samplers read the loop points of WAV files from their sampler chunk + # (written after the tags, so that saving the tags cannot affect it) + if soundfile.info(exported_file_path).format in ("WAV", "WAVEX"): + write_smpl_loop(exported_file_path, loop_start, loop_end, self.mlaudio.rate) - return str(loop_start), str(loop_end) + return str(loop_start), str(loop_end_tag_value) def read_tags(self, loop_start_tag: str, loop_end_tag: str, is_offset: Optional[bool] = None) -> Tuple[int, int]: - """Reads the tags provided from the file and returns the read loop points + """Reads the tags provided from the file and returns the read loop points. + If both tag names are None, they are auto-detected, and a WAV file without loop tags + has its loop points read from its sampler (smpl) chunk instead. Args: loop_start_tag (str): The name of the metadata tag containing the loop_start value @@ -564,6 +583,15 @@ def read_tags(self, loop_start_tag: str, loop_end_tag: str, is_offset: Optional[ Returns: Tuple[int, int]: A tuple containing (loop_start, loop_end) """ + try: + return self._read_loop_tags(loop_start_tag, loop_end_tag, is_offset) + except ValueError: + smpl_loop = read_smpl_loop(self.filepath) if loop_start_tag is None and loop_end_tag is None else None + if smpl_loop is None: + raise + return smpl_loop + + def _read_loop_tags(self, loop_start_tag: str, loop_end_tag: str, is_offset: Optional[bool] = None) -> Tuple[int, int]: # Workaround for taglib import issues on Apple silicon devices # Import taglib only when needed to isolate ImportErrors import taglib diff --git a/pymusiclooper/handler.py b/pymusiclooper/handler.py index 2bcc48b..2f24a36 100644 --- a/pymusiclooper/handler.py +++ b/pymusiclooper/handler.py @@ -14,7 +14,7 @@ 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 +# Default number of samples to keep after the loop end when trimming (interactively or with the trim command): 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 @@ -248,7 +248,13 @@ 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(): + # Exporting the loop points to the console only prints them, so it does not offer to write a trimmed file + if ( + self.interactive_mode + and not self.trim + and not self.to_stdout + and self.musiclooper.supports_lossless_trim() + ): with _hideprogressbar(self._progressbar): self.trim, self.keep_after = self.trim_prompt() diff --git a/pymusiclooper/wav.py b/pymusiclooper/wav.py index b98146e..b00a310 100644 --- a/pymusiclooper/wav.py +++ b/pymusiclooper/wav.py @@ -1,36 +1,83 @@ -"""Lossless trimming of WAV files at the chunk level, without rewriting the audio. +"""Chunk-level editing of WAV files: lossless trimming, and reading/writing loop points in the sampler (smpl) chunk. Every chunk of the source file is copied verbatim, so everything stored besides the audio is kept as-is: the format chunk, tags (LIST/INFO and ID3 chunks), sampler loops (smpl), cue points, broadcast extension (bext), etc. -Only the audio data chunk is shortened, and the sample count of the fact chunk (if any) and the RIFF size are updated. +When trimming, only the audio data chunk is shortened, and the sample count of the fact chunk (if any) and the RIFF size +are updated. + +The sampler chunk is where many game engines, samplers and audio tools read the loop points of WAV files from. """ import struct -from typing import List, NamedTuple +from typing import BinaryIO, List, NamedTuple, Optional, Tuple + +# Sampler chunk header: manufacturer, product, sample period (ns), MIDI unity note, MIDI pitch fraction, +# SMPTE format, SMPTE offset, number of loops, size of the sampler-specific data following the loops +_SMPL_HEADER_FORMAT = "9I" +_SMPL_HEADER_SIZE = 36 +_SMPL_LOOP_COUNT_OFFSET = 28 +# Each loop: cue point ID, type, start, end (inclusive), fraction, play count (0 = infinite) +_SMPL_LOOP_FORMAT = "6I" +_SMPL_LOOP_SIZE = 24 +_SMPL_FORWARD_LOOP = 0 +_DEFAULT_MIDI_UNITY_NOTE = 60 -class _Chunk(NamedTuple): +class _ChunkHeader(NamedTuple): chunk_id: bytes - data: bytes + offset: int # Offset of the chunk's data in the file + size: int # Size of the chunk's data present in the file + + +def _read_riff_header(f: BinaryIO) -> Tuple[bytes, str]: + """Returns the RIFF identifier (RIFF or RIFX) and the struct endianness of the file's integers.""" + header = f.read(12) + if header[:4] not in (b"RIFF", b"RIFX") or header[8:12] != b"WAVE": + raise ValueError("Not a RIFF WAVE file.") + return header[:4], "<" if header[:4] == b"RIFF" else ">" -def _read_chunks(data: bytes, endian: str) -> List[_Chunk]: +def _read_chunk_headers(f: BinaryIO, endian: str) -> List[_ChunkHeader]: + """Lists the chunks of the file without reading their data, so that large files can be searched quickly.""" + file_size = f.seek(0, 2) chunks = [] offset = 12 - while offset + 8 <= len(data): - chunk_id = data[offset:offset + 4] - (size,) = struct.unpack_from(f"{endian}I", data, offset + 4) + while offset + 8 <= file_size: + f.seek(offset) + chunk_id, size = struct.unpack(f"{endian}4sI", f.read(8)) # Files written by interrupted or streaming encoders can leave the data size as a placeholder: # 0 (then the data runs to the end of the file), or a size past the end of the file if chunk_id == b"data" and size == 0: - size = len(data) - offset - 8 - chunk_data = data[offset + 8:offset + 8 + size] - chunks.append(_Chunk(chunk_id, chunk_data)) + size = file_size - offset - 8 + chunks.append(_ChunkHeader(chunk_id, offset + 8, min(size, file_size - offset - 8))) # Chunks are padded to an even size offset += 8 + size + (size & 1) return chunks +def _read_chunks(filepath: str) -> Tuple[bytes, str, List[Tuple[bytes, bytes]]]: + """Returns the RIFF identifier, the endianness and the (chunk ID, chunk data) of every chunk of the file.""" + with open(filepath, "rb") as f: + riff_id, endian = _read_riff_header(f) + chunks = [] + for chunk in _read_chunk_headers(f, endian): + f.seek(chunk.offset) + chunks.append((chunk.chunk_id, f.read(chunk.size))) + return riff_id, endian, chunks + + +def _write_riff(filepath: str, riff_id: bytes, endian: str, chunks: List[Tuple[bytes, bytes]]): + body = [] + for chunk_id, chunk_data in chunks: + body += [chunk_id, struct.pack(f"{endian}I", len(chunk_data)), chunk_data] + if len(chunk_data) & 1: + body.append(b"\x00") + body = b"".join(body) + + with open(filepath, "wb") as f: + f.write(riff_id + struct.pack(f"{endian}I", 4 + len(body)) + b"WAVE" + body) + + def trim_wav(src_filepath: str, dest_filepath: str, n_frames: int) -> int: """Writes a copy of a PCM/float WAV file that ends after `n_frames` frames, keeping all its other chunks. @@ -45,39 +92,110 @@ def trim_wav(src_filepath: str, dest_filepath: str, n_frames: int) -> int: Returns: int: The length of the trimmed file in frames. """ - with open(src_filepath, "rb") as f: - data = f.read() - - if data[:4] not in (b"RIFF", b"RIFX") or data[8:12] != b"WAVE": - raise ValueError("Not a RIFF WAVE file.") - endian = "<" if data[:4] == b"RIFF" else ">" + riff_id, endian, chunks = _read_chunks(src_filepath) - chunks = _read_chunks(data, endian) - fmt_chunk = next((chunk for chunk in chunks if chunk.chunk_id == b"fmt "), None) - if fmt_chunk is None or len(fmt_chunk.data) < 14 or not any(chunk.chunk_id == b"data" for chunk in chunks): + fmt_data = next((chunk_data for chunk_id, chunk_data in chunks if chunk_id == b"fmt "), None) + audio_idx = next((idx for idx, (chunk_id, _) in enumerate(chunks) if chunk_id == b"data"), None) + if fmt_data is None or len(fmt_data) < 14 or audio_idx is None: raise ValueError("Invalid WAVE file: missing format or data chunk.") - (block_align,) = struct.unpack_from(f"{endian}H", fmt_chunk.data, 12) + (block_align,) = struct.unpack_from(f"{endian}H", fmt_data, 12) if block_align == 0: raise ValueError("Invalid WAVE file: zero block alignment.") - audio_chunk = next(chunk for chunk in chunks if chunk.chunk_id == b"data") - kept_frames = min(max(0, n_frames), len(audio_chunk.data) // block_align) + kept_frames = min(max(0, n_frames), len(chunks[audio_idx][1]) // block_align) - body = [] - for chunk in chunks: - chunk_data = chunk.data - if chunk is audio_chunk: + trimmed_chunks = [] + for idx, (chunk_id, chunk_data) in enumerate(chunks): + if idx == audio_idx: chunk_data = chunk_data[:kept_frames * block_align] - elif chunk.chunk_id == b"fact" and len(chunk_data) >= 4: + elif chunk_id == b"fact" and len(chunk_data) >= 4: # The fact chunk starts with the number of frames (required for non-PCM formats, e.g. float) chunk_data = struct.pack(f"{endian}I", kept_frames) + chunk_data[4:] - body += [chunk.chunk_id, struct.pack(f"{endian}I", len(chunk_data)), chunk_data] - if len(chunk_data) & 1: - body.append(b"\x00") - body = b"".join(body) + trimmed_chunks.append((chunk_id, chunk_data)) - with open(dest_filepath, "wb") as f: - f.write(data[:4] + struct.pack(f"{endian}I", 4 + len(body)) + b"WAVE" + body) + _write_riff(dest_filepath, riff_id, endian, trimmed_chunks) return kept_frames + + +def read_smpl_loop(filepath: str) -> Optional[Tuple[int, int]]: + """Reads the first forward loop of the WAV file's sampler (smpl) chunk. + + Args: + filepath (str): Path to the audio file. + + Returns: + Optional[Tuple[int, int]]: (loop_start, loop_end) in samples, with loop_end exclusive like PyMusicLooper's + (the smpl chunk's loop end is the last sample played). None if the file is not a WAV file or has no such loop. + """ + with open(filepath, "rb") as f: + try: + _, endian = _read_riff_header(f) + except ValueError: + return None + smpl = next((chunk for chunk in _read_chunk_headers(f, endian) if chunk.chunk_id == b"smpl"), None) + if smpl is None or smpl.size < _SMPL_HEADER_SIZE: + return None + f.seek(smpl.offset) + smpl_data = f.read(smpl.size) + + (n_loops,) = struct.unpack_from(f"{endian}I", smpl_data, _SMPL_LOOP_COUNT_OFFSET) + for loop_idx in range(n_loops): + loop_offset = _SMPL_HEADER_SIZE + loop_idx * _SMPL_LOOP_SIZE + if loop_offset + _SMPL_LOOP_SIZE > len(smpl_data): + break + _, loop_type, loop_start, loop_end, _, _ = struct.unpack_from(f"{endian}{_SMPL_LOOP_FORMAT}", smpl_data, loop_offset) + if loop_type == _SMPL_FORWARD_LOOP: + return loop_start, loop_end + 1 + return None + + +def write_smpl_loop(filepath: str, loop_start: int, loop_end: int, rate: int): + """Writes the loop points to the WAV file's sampler (smpl) chunk, in place, as an infinitely repeating forward loop. + + An existing sampler chunk keeps everything but its first loop (its header, other loops and sampler-specific data); + otherwise, a new one is added. + + Args: + filepath (str): Path to the WAV file. + loop_start (int): Loop start in samples. + loop_end (int): Loop end in samples (exclusive, like everywhere in PyMusicLooper). + rate (int): Sample rate of the audio, used for the sample period of a new sampler chunk. + + Raises: + ValueError: if the file is not a RIFF/RIFX WAVE file. + """ + riff_id, endian, chunks = _read_chunks(filepath) + + smpl_idx = next((idx for idx, (chunk_id, _) in enumerate(chunks) if chunk_id == b"smpl"), None) + smpl_data = chunks[smpl_idx][1] if smpl_idx is not None else b"" + has_header = len(smpl_data) >= _SMPL_HEADER_SIZE + n_loops = struct.unpack_from(f"{endian}I", smpl_data, _SMPL_LOOP_COUNT_OFFSET)[0] if has_header else 0 + loop = struct.pack(f"{endian}{_SMPL_LOOP_FORMAT}", 0, _SMPL_FORWARD_LOOP, loop_start, loop_end - 1, 0, 0) + + if has_header and n_loops >= 1 and len(smpl_data) >= _SMPL_HEADER_SIZE + _SMPL_LOOP_SIZE: + # Keep the first loop's cue point ID, which cue labels may refer to + first_loop_end = _SMPL_HEADER_SIZE + _SMPL_LOOP_SIZE + smpl_data = smpl_data[:_SMPL_HEADER_SIZE + 4] + loop[4:] + smpl_data[first_loop_end:] + elif has_header and n_loops == 0: + smpl_data = ( + smpl_data[:_SMPL_LOOP_COUNT_OFFSET] + + struct.pack(f"{endian}I", 1) + + smpl_data[_SMPL_LOOP_COUNT_OFFSET + 4:_SMPL_HEADER_SIZE] + + loop + + smpl_data[_SMPL_HEADER_SIZE:] + ) + else: + # New (or unreadable) sampler chunk + sample_period_ns = round(1e9 / rate) + smpl_data = struct.pack( + f"{endian}{_SMPL_HEADER_FORMAT}", 0, 0, sample_period_ns, _DEFAULT_MIDI_UNITY_NOTE, 0, 0, 0, 1, 0 + ) + loop + + if smpl_idx is None: + chunks.append((b"smpl", smpl_data)) + else: + chunks[smpl_idx] = (b"smpl", smpl_data) + + _write_riff(filepath, riff_id, endian, chunks) diff --git a/tests/test_cli.py b/tests/test_cli.py index a4765e1..d96b831 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -91,3 +91,16 @@ def test_batch_export_skips_non_audio_files(run, track_path, stereo_track_path, 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"] + + +def test_trim_keeps_recommended_samples_after_loop_end_by_default(run, track_path, tmp_path): + import soundfile as sf + + from pymusiclooper.core import MusicLooper + from pymusiclooper.handler import RECOMMENDED_KEEP_AFTER + + result = run("trim", "--path", track_path, "--output-dir", tmp_path) + + assert result.exit_code == 0 + loop_end = MusicLooper(track_path).find_loop_pairs()[0].loop_end + assert sf.info(tmp_path / "track-trimmed.wav").frames == loop_end + RECOMMENDED_KEEP_AFTER diff --git a/tests/test_core.py b/tests/test_core.py index d55eb31..ba864ea 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -217,7 +217,7 @@ def _append_wav_chunk(path, chunk_id: bytes, chunk_data: bytes): def _wav_chunks(path) -> dict: from pymusiclooper.wav import _read_chunks - return {chunk.chunk_id: chunk.data for chunk in _read_chunks(path.read_bytes(), "<")} + return dict(_read_chunks(str(path))[2]) @pytest.mark.parametrize( @@ -549,3 +549,131 @@ def test_ogg_vorbis_start_granule_is_zero_for_regular_files(track, tmp_path): sf.write(source_path, track, SR, format="OGG", subtype="VORBIS") assert _start_granule(list(_read_pages(source_path.read_bytes()))) == 0 + + +# --- WAV sampler (smpl) chunk loop points --- + + +def _smpl_chunk(loops, sampler_data=b"", manufacturer=7): + """A sampler chunk with the given (cue point ID, type, start, inclusive end) loops.""" + header = struct.pack("<9I", manufacturer, 0, 45351, 60, 0, 0, 0, len(loops), len(sampler_data)) + return header + b"".join(struct.pack("<6I", cue_id, loop_type, start, end, 0, 0) for cue_id, loop_type, start, end in loops) + sampler_data + + +def test_smpl_loop_is_used_as_embedded_loop(track_path, tmp_path): + from pymusiclooper.wav import write_smpl_loop + + source_path = tmp_path / "track.wav" + source_path.write_bytes(open(track_path, "rb").read()) + write_smpl_loop(str(source_path), LOOP_START + 123, LOOP_END + 123, SR) + looper = MusicLooper(str(source_path)) + + assert looper.read_tags(None, None) == (LOOP_START + 123, LOOP_END + 123) + pair = looper.find_loop_pairs()[0] + assert pair.from_metadata + assert (pair.loop_start, pair.loop_end) == (LOOP_START + 123, LOOP_END + 123) + # Explicitly named tags are not replaced by the sampler loop + with pytest.raises(ValueError): + looper.read_tags("LOOP_START", "LOOP_END") + + +def test_loop_tags_take_precedence_over_smpl_loop(track_path, tmp_path): + from pymusiclooper.wav import write_smpl_loop + + source_path = tmp_path / "track.wav" + source_path.write_bytes(open(track_path, "rb").read()) + write_smpl_loop(str(source_path), 1000, 2000, SR) + _write_loop_tags(str(source_path), LOOP_START, LOOP_END) + + assert MusicLooper(str(source_path)).read_tags(None, None) == (LOOP_START, LOOP_END) + + +def test_read_smpl_loop_uses_first_forward_loop(track_path, tmp_path): + from pymusiclooper.wav import read_smpl_loop + + source_path = tmp_path / "track.wav" + source_path.write_bytes(open(track_path, "rb").read()) + # A ping-pong loop first, then a forward loop + _append_wav_chunk(source_path, b"smpl", _smpl_chunk([(1, 1, 10, 20), (2, 0, 1000, 1999)])) + + assert read_smpl_loop(str(source_path)) == (1000, 2000) + + +@pytest.mark.parametrize("filename", ["track.flac", "track.ogg"]) +def test_read_smpl_loop_ignores_other_formats(track, tmp_path, filename): + from pymusiclooper.wav import read_smpl_loop + + source_path = tmp_path / filename + sf.write(source_path, track, SR) + + assert read_smpl_loop(str(source_path)) is None + + +@pytest.mark.parametrize("trim", [False, True]) +def test_export_tags_writes_smpl_loop_to_wav(track_path, tmp_path, trim): + from pymusiclooper.wav import read_smpl_loop + + written = MusicLooper(track_path).export_tags( + LOOP_START, LOOP_END, "LOOPSTART", "LOOPLENGTH", output_dir=str(tmp_path), trim=trim, keep_after=100 + ) + + output_path = str(tmp_path / "track-tagged.wav") + assert written == (str(LOOP_START), str(LOOP_END - LOOP_START)) + assert read_smpl_loop(output_path) == (LOOP_START, LOOP_END) + assert MusicLooper(output_path).read_tags("LOOPSTART", "LOOPLENGTH") == (LOOP_START, LOOP_END) + assert sf.info(output_path).frames == (LOOP_END + 100 if trim else sf.info(track_path).frames) + smpl = _wav_chunks(Path(output_path))[b"smpl"] + # Sample period in nanoseconds, MIDI unity note, one infinitely repeating forward loop with an inclusive end + assert struct.unpack_from("<9I", smpl)[2:4] == (round(1e9 / SR), 60) + assert struct.unpack_from("<9I", smpl)[7] == 1 + assert struct.unpack_from("<6I", smpl, 36) == (0, 0, LOOP_START, LOOP_END - 1, 0, 0) + + +def test_export_tags_does_not_write_smpl_loop_to_other_formats(flac_track_path, tmp_path): + MusicLooper(flac_track_path).export_tags(LOOP_START, LOOP_END, "LOOP_START", "LOOP_END", output_dir=str(tmp_path)) + + assert b"smpl" not in (tmp_path / "track-tagged.flac").read_bytes() + + +def test_write_smpl_loop_keeps_existing_sampler_data(track_path, tmp_path): + from pymusiclooper.wav import write_smpl_loop + + source_path = tmp_path / "track.wav" + source_path.write_bytes(open(track_path, "rb").read()) + _append_wav_chunk(source_path, b"smpl", _smpl_chunk([(5, 0, 10, 20), (6, 1, 30, 40)], sampler_data=b"\x01\x02\x03\x04")) + + write_smpl_loop(str(source_path), LOOP_START, LOOP_END, SR) + + assert _wav_chunks(source_path)[b"smpl"] == _smpl_chunk( + [(5, 0, LOOP_START, LOOP_END - 1), (6, 1, 30, 40)], sampler_data=b"\x01\x02\x03\x04" + ) + assert sf.info(source_path).frames == sf.info(track_path).frames + + +def test_write_smpl_loop_adds_loop_to_sampler_chunk_without_loops(track_path, tmp_path): + from pymusiclooper.wav import write_smpl_loop + + source_path = tmp_path / "track.wav" + source_path.write_bytes(open(track_path, "rb").read()) + _append_wav_chunk(source_path, b"smpl", _smpl_chunk([], sampler_data=b"\x01\x02\x03\x04")) + + write_smpl_loop(str(source_path), LOOP_START, LOOP_END, SR) + + assert _wav_chunks(source_path)[b"smpl"] == _smpl_chunk( + [(0, 0, LOOP_START, LOOP_END - 1)], sampler_data=b"\x01\x02\x03\x04" + ) + + +def test_detected_loop_matching_embedded_loop_is_not_listed_twice(flac_track_path): + detected = MusicLooper(flac_track_path).find_loop_pairs(use_embedded_tags=False)[0] + _write_loop_tags(flac_track_path, detected.loop_start, detected.loop_end) + + pairs = MusicLooper(flac_track_path).find_loop_pairs() + + assert pairs[0].from_metadata + assert (pairs[0].loop_start, pairs[0].loop_end) == (detected.loop_start, detected.loop_end) + tolerance = int(0.005 * SR) + assert not any( + abs(pair.loop_start - detected.loop_start) <= tolerance and abs(pair.loop_end - detected.loop_end) <= tolerance + for pair in pairs[1:] + ) diff --git a/tests/test_handler.py b/tests/test_handler.py index 6010ef3..36bcd0d 100644 --- a/tests/test_handler.py +++ b/tests/test_handler.py @@ -120,13 +120,26 @@ def test_interactive_trim_prompt_declined(monkeypatch, fake_input, track, tmp_pa 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) + export_handler = _interactive_export_handler(monkeypatch, source_path, tmp_path / "out", to_txt=True) # Only the loop selection is answered; a trim prompt would exhaust the answers fake_input("0") export_handler.run() +def test_interactive_trim_prompt_skipped_when_printing_loop_points(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, to_stdout=True) + + # Only the loop selection is answered; a trim prompt would exhaust the answers + fake_input("0") + export_handler.run() + + assert not out_dir.exists() + + 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)