API Reference

The API reference is generated automatically from the source docstrings.

Parser and data classes

WAVParser

Pure Python WAV file parser.

Source code in src/riffy/wav.py

class WAVParser:
    """Pure Python WAV file parser."""

    def __init__(self, file_path: str | Path):
        """Initialize parser with file path and automatically parse the file."""
        self.file_path = Path(file_path)
        self.format_info: WAVFormat | None = None
        # chunks maps a 4-character chunk ID to the ordered list of every chunk
        # with that ID, so files carrying duplicate IDs (e.g. multiple ``LIST``
        # chunks) preserve all occurrences in file order. See ``get_chunk`` /
        # ``get_chunks`` for ergonomic access.
        self.chunks: dict[str, list[WAVChunk]] = {}
        self.audio_data: bytes | None = None
        self._file_size = 0
        # RIFF form of the file on disk: "RIFF" (classic), or "RF64"/"BW64" for
        # the 64-bit large-file variants.
        self.riff_form: str = _CLASSIC_FORM
        self._ds64: _Ds64 | None = None

        # Automatically parse the file on initialization
        self.parse()

    @property
    def is_rf64(self) -> bool:
        """Whether the file on disk uses the RF64/BW64 (64-bit) large-file form."""
        return self.riff_form in _RF64_FORMS

    def parse(self) -> dict:
        """Parse the WAV file and return comprehensive information.

        Re-parsing re-reads the file from disk and discards any in-memory
        modifications previously made via ``add_chunk``, ``replace_chunk``,
        or ``set_chunk``.
        """
        if not self.file_path.exists():
            raise FileNotFoundError(f"WAV file not found: {self.file_path}")

        # Reset state so re-parsing reflects the file on disk, not prior edits.
        self.format_info = None
        self.chunks = {}
        self.audio_data = None
        self.riff_form = _CLASSIC_FORM
        self._ds64 = None

        self._file_size = self.file_path.stat().st_size

        with open(self.file_path, "rb") as f:
            self._parse_riff_header(f)
            self._parse_chunks(f)
            self._calculate_duration()
            self._validate_format()

        return self.get_info()

    def _parse_riff_header(self, f: BinaryIO) -> None:
        """Parse the RIFF/RF64/BW64 header."""
        riff_header = f.read(12)
        if len(riff_header) != 12:
            raise CorruptedFileError("File too small to be a valid WAV file")

        riff_id, file_size, wave_id = struct.unpack("<4sI4s", riff_header)

        form = riff_id.decode("latin-1")
        if form != _CLASSIC_FORM and form not in _RF64_FORMS:
            raise InvalidWAVFormatError("Not a valid RIFF file")
        self.riff_form = form

        if wave_id != b"WAVE":
            raise InvalidWAVFormatError("Not a valid WAV file")

    def _parse_chunks(self, f: BinaryIO) -> None:
        """Parse all chunks in the WAV file."""
        # An RF64/BW64 file must lead with a ds64 chunk carrying the 64-bit sizes.
        if self.is_rf64:
            self._parse_ds64(f)

        while True:
            chunk_header = f.read(8)
            if len(chunk_header) < 8:
                break

            chunk_id, chunk_size = struct.unpack("<4sI", chunk_header)

            # Decode chunk ID with strict ASCII validation
            try:
                chunk_id = chunk_id.decode("ascii")
            except UnicodeDecodeError as e:
                raise InvalidChunkError(f"Invalid chunk ID (non-ASCII bytes): {chunk_id!r}") from e
            if len(chunk_id) != 4:  # pragma: no cover - defensive; 4 bytes always decode to 4 chars
                raise InvalidChunkError(f"Invalid chunk ID length: {len(chunk_id)}")

            # In RF64/BW64, a 0xFFFFFFFF size is a sentinel: the real 64-bit size
            # comes from the ds64 chunk (dedicated field for 'data', table for
            # other chunks).
            if self.is_rf64 and chunk_size == _SIZE32_MAX:
                chunk_size = self._resolve_ds64_size(chunk_id)

            chunk_offset = f.tell()
            chunk_data = f.read(chunk_size)
            if len(chunk_data) != chunk_size:
                raise CorruptedFileError(f"Incomplete chunk: {chunk_id}")

            self.chunks.setdefault(chunk_id, []).append(
                WAVChunk(id=chunk_id, size=chunk_size, data=chunk_data, offset=chunk_offset)
            )

            # Format/audio state is taken from the first occurrence of each ID.
            if chunk_id == "fmt " and self.format_info is None:
                self._parse_format_chunk(chunk_data)
            elif chunk_id == "data" and self.audio_data is None:
                self.audio_data = chunk_data

            if chunk_size % 2:
                f.read(1)

    def _parse_ds64(self, f: BinaryIO) -> None:
        """Parse the mandatory leading ds64 chunk of an RF64/BW64 file."""
        header = f.read(8)
        if len(header) < 8 or header[:4] != b"ds64":
            raise InvalidWAVFormatError(f"{self.riff_form} file must begin with a 'ds64' chunk")
        (ds64_size,) = struct.unpack("<I", header[4:8])
        data = f.read(ds64_size)
        if len(data) != ds64_size or ds64_size < 28:
            raise CorruptedFileError("Incomplete or undersized 'ds64' chunk")

        riff_size, data_size, sample_count, table_length = struct.unpack("<QQQI", data[:28])
        table: dict[str, int] = {}
        offset = 28
        for _ in range(table_length):
            if offset + 12 > len(data):
                raise CorruptedFileError("Truncated 'ds64' size table")
            entry_id = data[offset : offset + 4].decode("latin-1")
            (entry_size,) = struct.unpack("<Q", data[offset + 4 : offset + 12])
            table[entry_id] = entry_size
            offset += 12

        self._ds64 = _Ds64(
            riff_size=riff_size,
            data_size=data_size,
            sample_count=sample_count,
            table=table,
        )
        if ds64_size % 2:  # honor even-byte padding like any other chunk
            f.read(1)

    def _resolve_ds64_size(self, chunk_id: str) -> int:
        """Resolve a 0xFFFFFFFF sentinel size to its real 64-bit value from ds64."""
        if self._ds64 is None:  # pragma: no cover - guarded by is_rf64 caller
            raise CorruptedFileError("Size sentinel encountered without a ds64 chunk")
        if chunk_id == "data":
            return self._ds64.data_size
        if chunk_id in self._ds64.table:
            return self._ds64.table[chunk_id]
        raise CorruptedFileError(
            f"Chunk '{chunk_id}' uses the 64-bit size sentinel but has no ds64 table entry"
        )

    def _parse_format_chunk(self, data: bytes) -> None:
        """Parse the format chunk."""
        if len(data) < 16:
            raise InvalidWAVFormatError("Format chunk too small")

        fmt_data = struct.unpack("<HHIIHH", data[:16])
        audio_format = fmt_data[0]

        # Non-PCM formats have extra fields (cbSize + extension data)
        if audio_format != 1 and len(data) < 18:
            raise InvalidWAVFormatError(
                f"Non-PCM format (type {audio_format}) requires at least 18 bytes in format chunk"
            )

        # For non-PCM formats, read cbSize to validate extension data
        if audio_format != 1:
            cb_size = struct.unpack("<H", data[16:18])[0]
            expected_size = 18 + cb_size
            if len(data) < expected_size:
                raise InvalidWAVFormatError(
                    f"Format chunk size ({len(data)} bytes) is smaller than expected "
                    f"({expected_size} bytes) for format type {audio_format}"
                )

        self.format_info = WAVFormat(
            audio_format=audio_format,
            channels=fmt_data[1],
            sample_rate=fmt_data[2],
            byte_rate=fmt_data[3],
            block_align=fmt_data[4],
            bits_per_sample=fmt_data[5],
        )

    def _calculate_duration(self) -> None:
        """Calculate audio duration after all chunks are parsed."""
        if self.audio_data and self.format_info and self.format_info.byte_rate > 0:
            self.format_info.duration_seconds = len(self.audio_data) / self.format_info.byte_rate

    def _validate_format(self) -> None:
        """Validate the parsed format."""
        if not self.format_info:
            raise MissingChunkError("No 'fmt ' chunk found")

        if not self.format_info.is_pcm:
            raise UnsupportedFormatError(
                f"Unsupported audio format: {self.format_info.audio_format} (only PCM is supported)"
            )

        if self.format_info.channels == 0:
            raise InvalidWAVFormatError("Invalid number of channels")

        if self.format_info.sample_rate == 0:
            raise InvalidWAVFormatError("Invalid sample rate")

        if self.audio_data is None:
            raise MissingChunkError("No 'data' chunk found")

    def get_info(self) -> dict:
        """Get comprehensive information about the WAV file."""
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        info = {
            "file_path": str(self.file_path),
            "file_size": self._file_size,
            "format": {
                "audio_format": self.format_info.audio_format,
                "channels": self.format_info.channels,
                "sample_rate": self.format_info.sample_rate,
                "byte_rate": self.format_info.byte_rate,
                "block_align": self.format_info.block_align,
                "bits_per_sample": self.format_info.bits_per_sample,
                "is_pcm": self.format_info.is_pcm,
            },
            "duration_seconds": self.format_info.duration_seconds,
            "audio_data_size": len(self.audio_data) if self.audio_data else 0,
            "sample_count": self._calculate_sample_count(),
            "chunks": {
                chunk_id: [chunk.size for chunk in chunk_list]
                for chunk_id, chunk_list in self.chunks.items()
            },
        }

        return info

    def get_chunks(self, chunk_id: str) -> list[WAVChunk]:
        """Return every chunk with the given ID, in file order (empty if none)."""
        return self.chunks.get(chunk_id, [])

    def get_chunk(self, chunk_id: str) -> WAVChunk | None:
        """Return the first chunk with the given ID, or ``None`` if absent.

        Convenience accessor for the common single-occurrence case, so callers
        do not have to index into the per-ID list returned by ``chunks``.
        """
        chunk_list = self.chunks.get(chunk_id)
        return chunk_list[0] if chunk_list else None

    def get_chunk_bytes(self, chunk_id: str) -> bytes | None:
        """Return the raw payload of the first chunk with the given ID, or ``None``.

        This is the raw-bytes accessor the metadata layer decodes from.
        """
        chunk = self.get_chunk(chunk_id)
        return chunk.data if chunk else None

    def _calculate_sample_count(self) -> int:
        """Calculate total number of samples."""
        if not self.audio_data or not self.format_info:
            return 0

        bytes_per_sample = self.format_info.bits_per_sample // 8
        return len(self.audio_data) // (bytes_per_sample * self.format_info.channels)

    def export_chunk(self, chunk_id: str, output_path: str | Path) -> int:
        """
        Export a specific chunk's data to a binary file.

        Args:
            chunk_id: The ID of the chunk to export (e.g., 'fmt ', 'data')
            output_path: Path where the chunk data will be written

        Returns:
            Number of bytes written

        Raises:
            WAVError: If file hasn't been parsed yet or file write errors occur
            MissingChunkError: If the specified chunk doesn't exist

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> parser.export_chunk('data', 'audio_data.bin')
            176400
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        chunk = self.get_chunk(chunk_id)
        if chunk is None:
            available_chunks = ", ".join(self.chunks.keys())
            raise MissingChunkError(
                f"Chunk '{chunk_id}' not found. Available chunks: {available_chunks}"
            )

        output_path = Path(output_path)

        try:
            with open(output_path, "wb") as f:
                f.write(chunk.data)
        except OSError as e:
            raise WAVError(f"Failed to write chunk to {output_path}: {e}") from e

        return len(chunk.data)

    def export_audio_data(self, output_path: str | Path) -> int:
        """
        Export raw audio data to a binary file (convenience method).

        This is equivalent to export_chunk('data', output_path) but provides
        a more intuitive interface for the common use case of extracting audio.

        Args:
            output_path: Path where the audio data will be written

        Returns:
            Number of bytes written

        Raises:
            WAVError: If file hasn't been parsed yet or no audio data exists

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> parser.export_audio_data('raw_audio.bin')
            176400
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        if not self.audio_data:
            raise WAVError("No audio data available to export.")

        return self.export_chunk("data", output_path)

    def list_chunks(self) -> dict[str, list[dict[str, int]]]:
        """
        List all chunks in the WAV file with their sizes and offsets.

        Because a WAV file may contain multiple chunks with the same ID, each ID
        maps to a list of ``{"size", "offset"}`` entries in file order.

        Returns:
            Dictionary mapping chunk IDs to a list of their occurrences' metadata

        Raises:
            WAVError: If file hasn't been parsed yet

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> chunks = parser.list_chunks()
            >>> print(chunks)
            {'fmt ': [{'size': 16, 'offset': 12}], 'data': [{'size': 176400, 'offset': 36}]}
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        return {
            chunk_id: [{"size": chunk.size, "offset": chunk.offset} for chunk in chunk_list]
            for chunk_id, chunk_list in self.chunks.items()
        }

    def replace_chunk(self, chunk_id: str, new_data: bytes) -> None:
        """
        Replace an existing chunk's data with new data.

        When a file contains multiple chunks with the same ID, this replaces the
        first occurrence and leaves the others untouched.

        Args:
            chunk_id: The ID of the chunk to replace (e.g., 'fmt ', 'data')
            new_data: The new chunk data (raw bytes)

        Raises:
            WAVError: If file hasn't been parsed yet
            MissingChunkError: If the specified chunk doesn't exist

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> with open("new_data.bin", "rb") as f:
            ...     parser.replace_chunk('data', f.read())
            >>> parser.write_wav("modified.wav")
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        chunk_list = self.chunks.get(chunk_id)
        if not chunk_list:
            available_chunks = ", ".join(self.chunks.keys())
            raise MissingChunkError(
                f"Chunk '{chunk_id}' not found. Available chunks: {available_chunks}"
            )

        # Update the first occurrence's data, preserving its offset.
        old_chunk = chunk_list[0]
        chunk_list[0] = WAVChunk(
            id=chunk_id,
            size=len(new_data),
            data=new_data,
            offset=old_chunk.offset,  # Offset will be recalculated on write
        )

        # Update audio_data if this is the data chunk
        if chunk_id == "data":
            self.audio_data = new_data
            self._calculate_duration()

    def add_chunk(self, chunk_id: str, chunk_data: bytes) -> None:
        """
        Add a chunk to the WAV file, appending it after any existing chunks with
        the same ID.

        Unlike v0.2.x, adding a chunk whose ID already exists is allowed and
        appends a new occurrence (the chunk store now keeps every occurrence).
        Use ``replace_chunk`` to overwrite an existing occurrence instead.

        Args:
            chunk_id: The ID of the new chunk (must be exactly 4 ASCII characters)
            chunk_data: The chunk data (raw bytes)

        Raises:
            WAVError: If file hasn't been parsed yet
            InvalidChunkError: If chunk_id is not exactly 4 ASCII characters

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> parser.add_chunk('INFO', b'Artist: Example')
            >>> parser.write_wav("modified.wav")
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        # Validate chunk_id
        if len(chunk_id) != 4:
            raise InvalidChunkError(f"Chunk ID must be exactly 4 characters, got {len(chunk_id)}")

        try:
            chunk_id.encode("ascii")
        except UnicodeEncodeError as e:
            raise InvalidChunkError(
                f"Chunk ID must contain only ASCII characters: {chunk_id!r}"
            ) from e

        # Append a new occurrence, preserving any existing chunks with this ID.
        self.chunks.setdefault(chunk_id, []).append(
            WAVChunk(
                id=chunk_id,
                size=len(chunk_data),
                data=chunk_data,
                offset=0,  # Will be calculated on write
            )
        )

    def set_chunk(self, chunk_id: str, chunk_data: bytes) -> None:
        """
        Set a chunk's data, replacing it if it exists or adding it if it doesn't.

        This is a convenience method that combines add_chunk and replace_chunk.

        Args:
            chunk_id: The ID of the chunk (must be exactly 4 ASCII characters)
            chunk_data: The chunk data (raw bytes)

        Raises:
            WAVError: If file hasn't been parsed yet
            InvalidChunkError: If chunk_id is not exactly 4 ASCII characters

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> parser.set_chunk('INFO', b'Artist: Example')
            >>> parser.write_wav("modified.wav")
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        # Validate chunk_id
        if len(chunk_id) != 4:
            raise InvalidChunkError(f"Chunk ID must be exactly 4 characters, got {len(chunk_id)}")

        try:
            chunk_id.encode("ascii")
        except UnicodeEncodeError as e:
            raise InvalidChunkError(
                f"Chunk ID must contain only ASCII characters: {chunk_id!r}"
            ) from e

        if chunk_id in self.chunks:
            self.replace_chunk(chunk_id, chunk_data)
        else:
            self.add_chunk(chunk_id, chunk_data)

    def copy_chunk_from_parser(self, chunk_id: str, source_parser: "WAVParser") -> None:
        """
        Copy a chunk from another WAVParser instance.

        Args:
            chunk_id: The ID of the chunk to copy
            source_parser: The source WAVParser instance to copy from

        Raises:
            WAVError: If either file hasn't been parsed yet
            MissingChunkError: If the chunk doesn't exist in the source parser

        Example:
            >>> parser1 = WAVParser("audio1.wav")
            >>> parser2 = WAVParser("audio2.wav")
            >>> parser2.copy_chunk_from_parser('data', parser1)
            >>> parser2.write_wav("modified.wav")
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        if not source_parser.format_info:
            raise WAVError("Source parser hasn't been parsed yet.")

        source_chunk = source_parser.get_chunk(chunk_id)
        if source_chunk is None:
            available_chunks = ", ".join(source_parser.chunks.keys())
            raise MissingChunkError(
                f"Chunk '{chunk_id}' not found in source. Available chunks: {available_chunks}"
            )

        self.set_chunk(chunk_id, source_chunk.data)

    def remove_chunk(self, chunk_id: str, index: int | None = None) -> None:
        """
        Remove a chunk from the in-memory chunk store.

        When the file carries multiple chunks with the same ID, ``index`` selects
        which occurrence to remove (in file order); with the default
        ``index=None`` every occurrence of ``chunk_id`` is removed. Call
        ``write_wav(...)`` afterwards to persist the change to disk.

        Removing the ``fmt `` or ``data`` chunk is allowed but leaves the file
        un-writable until it is restored, since ``write_wav`` requires both.

        Args:
            chunk_id: The ID of the chunk to remove (e.g., 'guan', 'LIST')
            index: The occurrence to remove, or ``None`` to remove all of them

        Raises:
            WAVError: If file hasn't been parsed yet
            MissingChunkError: If the chunk ID (or the given index) is absent

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> parser.remove_chunk('guan')
            >>> parser.write_wav("stripped.wav")
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        chunk_list = self.chunks.get(chunk_id)
        if not chunk_list:
            available_chunks = ", ".join(self.chunks.keys())
            raise MissingChunkError(
                f"Chunk '{chunk_id}' not found. Available chunks: {available_chunks}"
            )

        if index is None:
            del self.chunks[chunk_id]
        else:
            if index < 0 or index >= len(chunk_list):
                raise MissingChunkError(
                    f"Chunk '{chunk_id}' has no occurrence at index {index} "
                    f"(found {len(chunk_list)})"
                )
            del chunk_list[index]
            if not chunk_list:
                del self.chunks[chunk_id]

        # Keep audio_data consistent if the data chunk is now gone entirely.
        if chunk_id == "data" and "data" not in self.chunks:
            self.audio_data = None

    def write_wav(
        self,
        output_path: str | Path,
        overwrite: bool = False,
        force_rf64: bool = False,
    ) -> int:
        """
        Write the WAV file with all modifications to disk.

        This method reconstructs the entire WAV file, properly updating all
        chunk offsets and the RIFF file size.

        Classic 32-bit WAV output is used whenever the file fits, byte-for-byte
        identical to prior versions. The RF64/BW64 large-file form (with a
        ``ds64`` chunk carrying 64-bit sizes) is emitted only when a size crosses
        the 4 GB 32-bit limit, or when ``force_rf64`` is set.

        Args:
            output_path: Path where the WAV file will be written
            overwrite: If True, allow overwriting the source file. Default is False
                      for safety.
            force_rf64: If True, always emit the RF64/BW64 form even when the
                      sizes would fit in classic WAV. Useful for workflows that
                      require BW64 output.

        Returns:
            Number of bytes written

        Raises:
            WAVError: If file hasn't been parsed yet
            FileExistsError: If output_path is the same as input and overwrite=False
            MissingChunkError: If required chunks are missing

        Example:
            >>> parser = WAVParser("audio.wav")
            >>> parser.replace_chunk('data', new_audio_data)
            >>> parser.write_wav("modified.wav")
            176444
        """
        if not self.format_info:
            raise WAVError("File not parsed yet. Call parse() first.")

        output_path = Path(output_path)

        # Check if we're trying to overwrite the source file
        if output_path.resolve() == self.file_path.resolve() and not overwrite:
            raise FileExistsError(
                "Output path is the same as input file. Set overwrite=True to allow this operation."
            )

        # Ensure required chunks exist
        if "fmt " not in self.chunks:
            raise MissingChunkError("Cannot write WAV file without 'fmt ' chunk")

        if "data" not in self.chunks:
            raise MissingChunkError("Cannot write WAV file without 'data' chunk")

        ordered = self._write_order()

        # Classic RIFF size (excludes the ds64 chunk, which only exists in RF64).
        classic_riff_size = 4 + sum(8 + c.size + (c.size % 2) for c in ordered)
        need_rf64 = force_rf64 or _rf64_required(classic_riff_size, [c.size for c in ordered])

        if need_rf64:
            return self._write_rf64(output_path, ordered)
        return self._write_classic(output_path, ordered, classic_riff_size)

    def _write_order(self) -> list[WAVChunk]:
        """Return every chunk in write order: ``fmt `` first, ``data`` next, then
        remaining IDs sorted, with each ID's occurrences in file order."""
        order_ids: list[str] = []
        if "fmt " in self.chunks:
            order_ids.append("fmt ")
        if "data" in self.chunks:
            order_ids.append("data")
        for chunk_id in sorted(self.chunks.keys()):
            if chunk_id not in order_ids:
                order_ids.append(chunk_id)

        ordered: list[WAVChunk] = []
        for chunk_id in order_ids:
            ordered.extend(self.chunks[chunk_id])
        return ordered

    def _write_classic(self, output_path: Path, ordered: list[WAVChunk], riff_size: int) -> int:
        """Write a classic 32-bit RIFF/WAVE file (byte-for-byte stable output)."""
        bytes_written = 0
        with open(output_path, "wb") as f:
            f.write(b"RIFF")
            f.write(struct.pack("<I", riff_size))
            f.write(b"WAVE")
            bytes_written += 12

            for chunk in ordered:
                f.write(chunk.id.encode("ascii"))
                f.write(struct.pack("<I", chunk.size))
                f.write(chunk.data)
                bytes_written += 8 + chunk.size
                if chunk.size % 2:
                    f.write(b"\x00")
                    bytes_written += 1

        return bytes_written

    def _write_rf64(self, output_path: Path, ordered: list[WAVChunk]) -> int:
        """Write an RF64/BW64 file, moving 64-bit sizes into a leading ds64 chunk."""
        # Classify sizes: the first 'data' chunk gets the dedicated ds64 field;
        # any non-'data' chunk over 4 GB goes in the ds64 size table.
        data_size = 0
        first_data_seen = False
        table: dict[str, int] = {}
        for chunk in ordered:
            if chunk.id == "data" and not first_data_seen:
                data_size = chunk.size
                first_data_seen = True
            elif chunk.size > _SIZE32_MAX:
                table[chunk.id] = chunk.size

        block_align = self.format_info.block_align if self.format_info else 0
        sample_count = data_size // block_align if block_align else 0

        ds64_payload = self._build_ds64_payload(0, data_size, sample_count, table)
        ds64_chunk_len = len(ds64_payload)  # 28 + 12*len(table); always even
        riff_size = (
            4  # 'WAVE'
            + (8 + ds64_chunk_len)  # the ds64 chunk
            + sum(8 + c.size + (c.size % 2) for c in ordered)
        )
        ds64_payload = self._build_ds64_payload(riff_size, data_size, sample_count, table)

        form = self.riff_form if self.riff_form in _RF64_FORMS else "RF64"

        bytes_written = 0
        with open(output_path, "wb") as f:
            f.write(form.encode("ascii"))
            f.write(struct.pack("<I", _SIZE32_MAX))  # size sentinel; real size in ds64
            f.write(b"WAVE")
            f.write(b"ds64")
            f.write(struct.pack("<I", ds64_chunk_len))
            f.write(ds64_payload)
            bytes_written += 12 + 8 + ds64_chunk_len

            first_data_written = False
            for chunk in ordered:
                if chunk.id == "data" and not first_data_written:
                    size_field = _SIZE32_MAX
                    first_data_written = True
                elif chunk.size > _SIZE32_MAX:
                    size_field = _SIZE32_MAX
                else:
                    size_field = chunk.size

                f.write(chunk.id.encode("ascii"))
                f.write(struct.pack("<I", size_field))
                f.write(chunk.data)
                bytes_written += 8 + chunk.size
                if chunk.size % 2:
                    f.write(b"\x00")
                    bytes_written += 1

        return bytes_written

    @staticmethod
    def _build_ds64_payload(
        riff_size: int, data_size: int, sample_count: int, table: dict[str, int]
    ) -> bytes:
        """Build a ds64 chunk payload from 64-bit sizes and an optional size table."""
        payload = struct.pack("<QQQI", riff_size, data_size, sample_count, len(table))
        for chunk_id, size in table.items():
            payload += chunk_id.encode("latin-1") + struct.pack("<Q", size)
        return payload

is_rf64 property

is_rf64

Whether the file on disk uses the RF64/BW64 (64-bit) large-file form.

__init__

__init__(file_path)

Initialize parser with file path and automatically parse the file.

Source code in src/riffy/wav.py

def __init__(self, file_path: str | Path):
    """Initialize parser with file path and automatically parse the file."""
    self.file_path = Path(file_path)
    self.format_info: WAVFormat | None = None
    # chunks maps a 4-character chunk ID to the ordered list of every chunk
    # with that ID, so files carrying duplicate IDs (e.g. multiple ``LIST``
    # chunks) preserve all occurrences in file order. See ``get_chunk`` /
    # ``get_chunks`` for ergonomic access.
    self.chunks: dict[str, list[WAVChunk]] = {}
    self.audio_data: bytes | None = None
    self._file_size = 0
    # RIFF form of the file on disk: "RIFF" (classic), or "RF64"/"BW64" for
    # the 64-bit large-file variants.
    self.riff_form: str = _CLASSIC_FORM
    self._ds64: _Ds64 | None = None

    # Automatically parse the file on initialization
    self.parse()

parse

parse()

Parse the WAV file and return comprehensive information.

Re-parsing re-reads the file from disk and discards any in-memory modifications previously made via add_chunk, replace_chunk, or set_chunk.

Source code in src/riffy/wav.py

def parse(self) -> dict:
    """Parse the WAV file and return comprehensive information.

    Re-parsing re-reads the file from disk and discards any in-memory
    modifications previously made via ``add_chunk``, ``replace_chunk``,
    or ``set_chunk``.
    """
    if not self.file_path.exists():
        raise FileNotFoundError(f"WAV file not found: {self.file_path}")

    # Reset state so re-parsing reflects the file on disk, not prior edits.
    self.format_info = None
    self.chunks = {}
    self.audio_data = None
    self.riff_form = _CLASSIC_FORM
    self._ds64 = None

    self._file_size = self.file_path.stat().st_size

    with open(self.file_path, "rb") as f:
        self._parse_riff_header(f)
        self._parse_chunks(f)
        self._calculate_duration()
        self._validate_format()

    return self.get_info()

get_info

get_info()

Get comprehensive information about the WAV file.

Source code in src/riffy/wav.py

def get_info(self) -> dict:
    """Get comprehensive information about the WAV file."""
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    info = {
        "file_path": str(self.file_path),
        "file_size": self._file_size,
        "format": {
            "audio_format": self.format_info.audio_format,
            "channels": self.format_info.channels,
            "sample_rate": self.format_info.sample_rate,
            "byte_rate": self.format_info.byte_rate,
            "block_align": self.format_info.block_align,
            "bits_per_sample": self.format_info.bits_per_sample,
            "is_pcm": self.format_info.is_pcm,
        },
        "duration_seconds": self.format_info.duration_seconds,
        "audio_data_size": len(self.audio_data) if self.audio_data else 0,
        "sample_count": self._calculate_sample_count(),
        "chunks": {
            chunk_id: [chunk.size for chunk in chunk_list]
            for chunk_id, chunk_list in self.chunks.items()
        },
    }

    return info

get_chunks

get_chunks(chunk_id)

Return every chunk with the given ID, in file order (empty if none).

Source code in src/riffy/wav.py

def get_chunks(self, chunk_id: str) -> list[WAVChunk]:
    """Return every chunk with the given ID, in file order (empty if none)."""
    return self.chunks.get(chunk_id, [])

get_chunk

get_chunk(chunk_id)

Return the first chunk with the given ID, or None if absent.

Convenience accessor for the common single-occurrence case, so callers do not have to index into the per-ID list returned by chunks.

Source code in src/riffy/wav.py

def get_chunk(self, chunk_id: str) -> WAVChunk | None:
    """Return the first chunk with the given ID, or ``None`` if absent.

    Convenience accessor for the common single-occurrence case, so callers
    do not have to index into the per-ID list returned by ``chunks``.
    """
    chunk_list = self.chunks.get(chunk_id)
    return chunk_list[0] if chunk_list else None

get_chunk_bytes

get_chunk_bytes(chunk_id)

Return the raw payload of the first chunk with the given ID, or None.

This is the raw-bytes accessor the metadata layer decodes from.

Source code in src/riffy/wav.py

def get_chunk_bytes(self, chunk_id: str) -> bytes | None:
    """Return the raw payload of the first chunk with the given ID, or ``None``.

    This is the raw-bytes accessor the metadata layer decodes from.
    """
    chunk = self.get_chunk(chunk_id)
    return chunk.data if chunk else None

export_chunk

export_chunk(chunk_id, output_path)

Export a specific chunk's data to a binary file.

Parameters:

Name	Type	Description	Default
chunk_id	str	The ID of the chunk to export (e.g., 'fmt ', 'data')	required
output_path	str | Path	Path where the chunk data will be written	required

Returns:

Type	Description
int	Number of bytes written

Raises:

Type	Description
WAVError	If file hasn't been parsed yet or file write errors occur
MissingChunkError	If the specified chunk doesn't exist

Example

parser = WAVParser("audio.wav") parser.export_chunk('data', 'audio_data.bin') 176400

Source code in src/riffy/wav.py

def export_chunk(self, chunk_id: str, output_path: str | Path) -> int:
    """
    Export a specific chunk's data to a binary file.

    Args:
        chunk_id: The ID of the chunk to export (e.g., 'fmt ', 'data')
        output_path: Path where the chunk data will be written

    Returns:
        Number of bytes written

    Raises:
        WAVError: If file hasn't been parsed yet or file write errors occur
        MissingChunkError: If the specified chunk doesn't exist

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> parser.export_chunk('data', 'audio_data.bin')
        176400
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    chunk = self.get_chunk(chunk_id)
    if chunk is None:
        available_chunks = ", ".join(self.chunks.keys())
        raise MissingChunkError(
            f"Chunk '{chunk_id}' not found. Available chunks: {available_chunks}"
        )

    output_path = Path(output_path)

    try:
        with open(output_path, "wb") as f:
            f.write(chunk.data)
    except OSError as e:
        raise WAVError(f"Failed to write chunk to {output_path}: {e}") from e

    return len(chunk.data)

export_audio_data

export_audio_data(output_path)

Export raw audio data to a binary file (convenience method).

This is equivalent to export_chunk('data', output_path) but provides a more intuitive interface for the common use case of extracting audio.

Parameters:

Name	Type	Description	Default
output_path	str | Path	Path where the audio data will be written	required

Returns:

Type	Description
int	Number of bytes written

Raises:

Type	Description
WAVError	If file hasn't been parsed yet or no audio data exists

Example

parser = WAVParser("audio.wav") parser.export_audio_data('raw_audio.bin') 176400

Source code in src/riffy/wav.py

def export_audio_data(self, output_path: str | Path) -> int:
    """
    Export raw audio data to a binary file (convenience method).

    This is equivalent to export_chunk('data', output_path) but provides
    a more intuitive interface for the common use case of extracting audio.

    Args:
        output_path: Path where the audio data will be written

    Returns:
        Number of bytes written

    Raises:
        WAVError: If file hasn't been parsed yet or no audio data exists

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> parser.export_audio_data('raw_audio.bin')
        176400
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    if not self.audio_data:
        raise WAVError("No audio data available to export.")

    return self.export_chunk("data", output_path)

list_chunks

list_chunks()

List all chunks in the WAV file with their sizes and offsets.

Because a WAV file may contain multiple chunks with the same ID, each ID maps to a list of {"size", "offset"} entries in file order.

Returns:

Type	Description
dict[str, list[dict[str, int]]]	Dictionary mapping chunk IDs to a list of their occurrences' metadata

Raises:

Type	Description
WAVError	If file hasn't been parsed yet

Example

parser = WAVParser("audio.wav") chunks = parser.list_chunks() print(chunks) {'fmt ': [{'size': 16, 'offset': 12}], 'data': [{'size': 176400, 'offset': 36}]}

Source code in src/riffy/wav.py

def list_chunks(self) -> dict[str, list[dict[str, int]]]:
    """
    List all chunks in the WAV file with their sizes and offsets.

    Because a WAV file may contain multiple chunks with the same ID, each ID
    maps to a list of ``{"size", "offset"}`` entries in file order.

    Returns:
        Dictionary mapping chunk IDs to a list of their occurrences' metadata

    Raises:
        WAVError: If file hasn't been parsed yet

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> chunks = parser.list_chunks()
        >>> print(chunks)
        {'fmt ': [{'size': 16, 'offset': 12}], 'data': [{'size': 176400, 'offset': 36}]}
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    return {
        chunk_id: [{"size": chunk.size, "offset": chunk.offset} for chunk in chunk_list]
        for chunk_id, chunk_list in self.chunks.items()
    }

replace_chunk

replace_chunk(chunk_id, new_data)

Replace an existing chunk's data with new data.

When a file contains multiple chunks with the same ID, this replaces the first occurrence and leaves the others untouched.

Parameters:

Name	Type	Description	Default
chunk_id	str	The ID of the chunk to replace (e.g., 'fmt ', 'data')	required
new_data	bytes	The new chunk data (raw bytes)	required

Raises:

Type	Description
WAVError	If file hasn't been parsed yet
MissingChunkError	If the specified chunk doesn't exist

Example

parser = WAVParser("audio.wav") with open("new_data.bin", "rb") as f: ... parser.replace_chunk('data', f.read()) parser.write_wav("modified.wav")

Source code in src/riffy/wav.py

def replace_chunk(self, chunk_id: str, new_data: bytes) -> None:
    """
    Replace an existing chunk's data with new data.

    When a file contains multiple chunks with the same ID, this replaces the
    first occurrence and leaves the others untouched.

    Args:
        chunk_id: The ID of the chunk to replace (e.g., 'fmt ', 'data')
        new_data: The new chunk data (raw bytes)

    Raises:
        WAVError: If file hasn't been parsed yet
        MissingChunkError: If the specified chunk doesn't exist

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> with open("new_data.bin", "rb") as f:
        ...     parser.replace_chunk('data', f.read())
        >>> parser.write_wav("modified.wav")
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    chunk_list = self.chunks.get(chunk_id)
    if not chunk_list:
        available_chunks = ", ".join(self.chunks.keys())
        raise MissingChunkError(
            f"Chunk '{chunk_id}' not found. Available chunks: {available_chunks}"
        )

    # Update the first occurrence's data, preserving its offset.
    old_chunk = chunk_list[0]
    chunk_list[0] = WAVChunk(
        id=chunk_id,
        size=len(new_data),
        data=new_data,
        offset=old_chunk.offset,  # Offset will be recalculated on write
    )

    # Update audio_data if this is the data chunk
    if chunk_id == "data":
        self.audio_data = new_data
        self._calculate_duration()

add_chunk

add_chunk(chunk_id, chunk_data)

Add a chunk to the WAV file, appending it after any existing chunks with the same ID.

Unlike v0.2.x, adding a chunk whose ID already exists is allowed and appends a new occurrence (the chunk store now keeps every occurrence). Use replace_chunk to overwrite an existing occurrence instead.

Parameters:

Name	Type	Description	Default
chunk_id	str	The ID of the new chunk (must be exactly 4 ASCII characters)	required
chunk_data	bytes	The chunk data (raw bytes)	required

Raises:

Type	Description
WAVError	If file hasn't been parsed yet
InvalidChunkError	If chunk_id is not exactly 4 ASCII characters

Example

parser = WAVParser("audio.wav") parser.add_chunk('INFO', b'Artist: Example') parser.write_wav("modified.wav")

Source code in src/riffy/wav.py

def add_chunk(self, chunk_id: str, chunk_data: bytes) -> None:
    """
    Add a chunk to the WAV file, appending it after any existing chunks with
    the same ID.

    Unlike v0.2.x, adding a chunk whose ID already exists is allowed and
    appends a new occurrence (the chunk store now keeps every occurrence).
    Use ``replace_chunk`` to overwrite an existing occurrence instead.

    Args:
        chunk_id: The ID of the new chunk (must be exactly 4 ASCII characters)
        chunk_data: The chunk data (raw bytes)

    Raises:
        WAVError: If file hasn't been parsed yet
        InvalidChunkError: If chunk_id is not exactly 4 ASCII characters

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> parser.add_chunk('INFO', b'Artist: Example')
        >>> parser.write_wav("modified.wav")
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    # Validate chunk_id
    if len(chunk_id) != 4:
        raise InvalidChunkError(f"Chunk ID must be exactly 4 characters, got {len(chunk_id)}")

    try:
        chunk_id.encode("ascii")
    except UnicodeEncodeError as e:
        raise InvalidChunkError(
            f"Chunk ID must contain only ASCII characters: {chunk_id!r}"
        ) from e

    # Append a new occurrence, preserving any existing chunks with this ID.
    self.chunks.setdefault(chunk_id, []).append(
        WAVChunk(
            id=chunk_id,
            size=len(chunk_data),
            data=chunk_data,
            offset=0,  # Will be calculated on write
        )
    )

set_chunk

set_chunk(chunk_id, chunk_data)

Set a chunk's data, replacing it if it exists or adding it if it doesn't.

This is a convenience method that combines add_chunk and replace_chunk.

Parameters:

Name	Type	Description	Default
chunk_id	str	The ID of the chunk (must be exactly 4 ASCII characters)	required
chunk_data	bytes	The chunk data (raw bytes)	required

Raises:

Type	Description
WAVError	If file hasn't been parsed yet
InvalidChunkError	If chunk_id is not exactly 4 ASCII characters

Example

parser = WAVParser("audio.wav") parser.set_chunk('INFO', b'Artist: Example') parser.write_wav("modified.wav")

Source code in src/riffy/wav.py

def set_chunk(self, chunk_id: str, chunk_data: bytes) -> None:
    """
    Set a chunk's data, replacing it if it exists or adding it if it doesn't.

    This is a convenience method that combines add_chunk and replace_chunk.

    Args:
        chunk_id: The ID of the chunk (must be exactly 4 ASCII characters)
        chunk_data: The chunk data (raw bytes)

    Raises:
        WAVError: If file hasn't been parsed yet
        InvalidChunkError: If chunk_id is not exactly 4 ASCII characters

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> parser.set_chunk('INFO', b'Artist: Example')
        >>> parser.write_wav("modified.wav")
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    # Validate chunk_id
    if len(chunk_id) != 4:
        raise InvalidChunkError(f"Chunk ID must be exactly 4 characters, got {len(chunk_id)}")

    try:
        chunk_id.encode("ascii")
    except UnicodeEncodeError as e:
        raise InvalidChunkError(
            f"Chunk ID must contain only ASCII characters: {chunk_id!r}"
        ) from e

    if chunk_id in self.chunks:
        self.replace_chunk(chunk_id, chunk_data)
    else:
        self.add_chunk(chunk_id, chunk_data)

copy_chunk_from_parser

copy_chunk_from_parser(chunk_id, source_parser)

Copy a chunk from another WAVParser instance.

Parameters:

Name	Type	Description	Default
chunk_id	str	The ID of the chunk to copy	required
source_parser	WAVParser	The source WAVParser instance to copy from	required

Raises:

Type	Description
WAVError	If either file hasn't been parsed yet
MissingChunkError	If the chunk doesn't exist in the source parser

Example

parser1 = WAVParser("audio1.wav") parser2 = WAVParser("audio2.wav") parser2.copy_chunk_from_parser('data', parser1) parser2.write_wav("modified.wav")

Source code in src/riffy/wav.py

def copy_chunk_from_parser(self, chunk_id: str, source_parser: "WAVParser") -> None:
    """
    Copy a chunk from another WAVParser instance.

    Args:
        chunk_id: The ID of the chunk to copy
        source_parser: The source WAVParser instance to copy from

    Raises:
        WAVError: If either file hasn't been parsed yet
        MissingChunkError: If the chunk doesn't exist in the source parser

    Example:
        >>> parser1 = WAVParser("audio1.wav")
        >>> parser2 = WAVParser("audio2.wav")
        >>> parser2.copy_chunk_from_parser('data', parser1)
        >>> parser2.write_wav("modified.wav")
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    if not source_parser.format_info:
        raise WAVError("Source parser hasn't been parsed yet.")

    source_chunk = source_parser.get_chunk(chunk_id)
    if source_chunk is None:
        available_chunks = ", ".join(source_parser.chunks.keys())
        raise MissingChunkError(
            f"Chunk '{chunk_id}' not found in source. Available chunks: {available_chunks}"
        )

    self.set_chunk(chunk_id, source_chunk.data)

remove_chunk

remove_chunk(chunk_id, index=None)

Remove a chunk from the in-memory chunk store.

When the file carries multiple chunks with the same ID, index selects which occurrence to remove (in file order); with the default index=None every occurrence of chunk_id is removed. Call write_wav(...) afterwards to persist the change to disk.

Removing the fmt or data chunk is allowed but leaves the file un-writable until it is restored, since write_wav requires both.

Parameters:

Name	Type	Description	Default
chunk_id	str	The ID of the chunk to remove (e.g., 'guan', 'LIST')	required
index	int | None	The occurrence to remove, or None to remove all of them	None

Raises:

Type	Description
WAVError	If file hasn't been parsed yet
MissingChunkError	If the chunk ID (or the given index) is absent

Example

parser = WAVParser("audio.wav") parser.remove_chunk('guan') parser.write_wav("stripped.wav")

Source code in src/riffy/wav.py

def remove_chunk(self, chunk_id: str, index: int | None = None) -> None:
    """
    Remove a chunk from the in-memory chunk store.

    When the file carries multiple chunks with the same ID, ``index`` selects
    which occurrence to remove (in file order); with the default
    ``index=None`` every occurrence of ``chunk_id`` is removed. Call
    ``write_wav(...)`` afterwards to persist the change to disk.

    Removing the ``fmt `` or ``data`` chunk is allowed but leaves the file
    un-writable until it is restored, since ``write_wav`` requires both.

    Args:
        chunk_id: The ID of the chunk to remove (e.g., 'guan', 'LIST')
        index: The occurrence to remove, or ``None`` to remove all of them

    Raises:
        WAVError: If file hasn't been parsed yet
        MissingChunkError: If the chunk ID (or the given index) is absent

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> parser.remove_chunk('guan')
        >>> parser.write_wav("stripped.wav")
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    chunk_list = self.chunks.get(chunk_id)
    if not chunk_list:
        available_chunks = ", ".join(self.chunks.keys())
        raise MissingChunkError(
            f"Chunk '{chunk_id}' not found. Available chunks: {available_chunks}"
        )

    if index is None:
        del self.chunks[chunk_id]
    else:
        if index < 0 or index >= len(chunk_list):
            raise MissingChunkError(
                f"Chunk '{chunk_id}' has no occurrence at index {index} "
                f"(found {len(chunk_list)})"
            )
        del chunk_list[index]
        if not chunk_list:
            del self.chunks[chunk_id]

    # Keep audio_data consistent if the data chunk is now gone entirely.
    if chunk_id == "data" and "data" not in self.chunks:
        self.audio_data = None

write_wav

write_wav(output_path, overwrite=False, force_rf64=False)

Write the WAV file with all modifications to disk.

This method reconstructs the entire WAV file, properly updating all chunk offsets and the RIFF file size.

Classic 32-bit WAV output is used whenever the file fits, byte-for-byte identical to prior versions. The RF64/BW64 large-file form (with a ds64 chunk carrying 64-bit sizes) is emitted only when a size crosses the 4 GB 32-bit limit, or when force_rf64 is set.

Parameters:

Name	Type	Description	Default
output_path	str | Path	Path where the WAV file will be written	required
overwrite	bool	If True, allow overwriting the source file. Default is False for safety.	False
force_rf64	bool	If True, always emit the RF64/BW64 form even when the sizes would fit in classic WAV. Useful for workflows that require BW64 output.	False

Returns:

Type	Description
int	Number of bytes written

Raises:

Type	Description
WAVError	If file hasn't been parsed yet
FileExistsError	If output_path is the same as input and overwrite=False
MissingChunkError	If required chunks are missing

Example

parser = WAVParser("audio.wav") parser.replace_chunk('data', new_audio_data) parser.write_wav("modified.wav") 176444

Source code in src/riffy/wav.py

def write_wav(
    self,
    output_path: str | Path,
    overwrite: bool = False,
    force_rf64: bool = False,
) -> int:
    """
    Write the WAV file with all modifications to disk.

    This method reconstructs the entire WAV file, properly updating all
    chunk offsets and the RIFF file size.

    Classic 32-bit WAV output is used whenever the file fits, byte-for-byte
    identical to prior versions. The RF64/BW64 large-file form (with a
    ``ds64`` chunk carrying 64-bit sizes) is emitted only when a size crosses
    the 4 GB 32-bit limit, or when ``force_rf64`` is set.

    Args:
        output_path: Path where the WAV file will be written
        overwrite: If True, allow overwriting the source file. Default is False
                  for safety.
        force_rf64: If True, always emit the RF64/BW64 form even when the
                  sizes would fit in classic WAV. Useful for workflows that
                  require BW64 output.

    Returns:
        Number of bytes written

    Raises:
        WAVError: If file hasn't been parsed yet
        FileExistsError: If output_path is the same as input and overwrite=False
        MissingChunkError: If required chunks are missing

    Example:
        >>> parser = WAVParser("audio.wav")
        >>> parser.replace_chunk('data', new_audio_data)
        >>> parser.write_wav("modified.wav")
        176444
    """
    if not self.format_info:
        raise WAVError("File not parsed yet. Call parse() first.")

    output_path = Path(output_path)

    # Check if we're trying to overwrite the source file
    if output_path.resolve() == self.file_path.resolve() and not overwrite:
        raise FileExistsError(
            "Output path is the same as input file. Set overwrite=True to allow this operation."
        )

    # Ensure required chunks exist
    if "fmt " not in self.chunks:
        raise MissingChunkError("Cannot write WAV file without 'fmt ' chunk")

    if "data" not in self.chunks:
        raise MissingChunkError("Cannot write WAV file without 'data' chunk")

    ordered = self._write_order()

    # Classic RIFF size (excludes the ds64 chunk, which only exists in RF64).
    classic_riff_size = 4 + sum(8 + c.size + (c.size % 2) for c in ordered)
    need_rf64 = force_rf64 or _rf64_required(classic_riff_size, [c.size for c in ordered])

    if need_rf64:
        return self._write_rf64(output_path, ordered)
    return self._write_classic(output_path, ordered, classic_riff_size)

WAVFormat dataclass

WAV format information.

Source code in src/riffy/wav.py

@dataclass
class WAVFormat:
    """WAV format information."""

    audio_format: int
    channels: int
    sample_rate: int
    byte_rate: int
    block_align: int
    bits_per_sample: int
    duration_seconds: float = 0.0

    @property
    def is_pcm(self) -> bool:
        """Check if format is PCM (uncompressed)."""
        return self.audio_format == 1

is_pcm property

is_pcm

Check if format is PCM (uncompressed).

WAVChunk dataclass

Represents a RIFF chunk in the WAV file.

Source code in src/riffy/wav.py

@dataclass
class WAVChunk:
    """Represents a RIFF chunk in the WAV file."""

    id: str
    size: int
    data: bytes
    offset: int

Recorder metadata

The unified entry point and per-standard classes. See the Recorder Metadata guide for worked examples.

Unified metadata view: surface whichever standards a file contains.

:class:RecordingMetadata inspects a WAV file and exposes each recorder-metadata standard it finds (GUANO, RIFF INFO, Broadcast Wave bext, AudioMoth) side by side, each kept close to its own raw parsed form.

Deliberately, this view does not perform cross-standard reconciliation: it does not pick a "best-available" timestamp, location, or device by precedence across standards. That merging is policy, and it belongs downstream (in bioamla), which can decide precedence for its own workflows. Keeping riffy close to the raw standards keeps its behavior predictable and avoids baking one consumer's opinions into the library.

(The AudioMoth view is a decoded read of the same INFO block that info exposes raw, so a file may legitimately report both info and audiomoth — that is within-standard convenience, not cross-standard merging.)

RecordingMetadata dataclass

The recorder-metadata standards found in one WAV file, side by side.

Each attribute is the parsed view for that standard, or None if the file does not contain it.

Source code in src/riffy/metadata/recording.py

@dataclass
class RecordingMetadata:
    """The recorder-metadata standards found in one WAV file, side by side.

    Each attribute is the parsed view for that standard, or ``None`` if the file
    does not contain it.
    """

    guano: GuanoMetadata | None = None
    info: InfoMetadata | None = None
    bext: BextMetadata | None = None
    audiomoth: AudioMothMetadata | None = None
    wamd: WamdMetadata | None = None

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "RecordingMetadata":
        """Detect and parse every supported standard from a parsed WAV file."""
        return cls(
            guano=GuanoMetadata.from_parser(parser),
            info=InfoMetadata.from_parser(parser),
            bext=BextMetadata.from_parser(parser),
            audiomoth=AudioMothMetadata.from_parser(parser),
            wamd=WamdMetadata.from_parser(parser),
        )

    @property
    def sources(self) -> tuple[str, ...]:
        """The names of the standards present, in a stable order."""
        present = []
        if self.guano is not None:
            present.append("guano")
        if self.info is not None:
            present.append("info")
        if self.bext is not None:
            present.append("bext")
        if self.audiomoth is not None:
            present.append("audiomoth")
        if self.wamd is not None:
            present.append("wamd")
        return tuple(present)

sources property

sources

The names of the standards present, in a stable order.

from_parser classmethod

from_parser(parser)

Detect and parse every supported standard from a parsed WAV file.

Source code in src/riffy/metadata/recording.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "RecordingMetadata":
    """Detect and parse every supported standard from a parsed WAV file."""
    return cls(
        guano=GuanoMetadata.from_parser(parser),
        info=InfoMetadata.from_parser(parser),
        bext=BextMetadata.from_parser(parser),
        audiomoth=AudioMothMetadata.from_parser(parser),
        wamd=WamdMetadata.from_parser(parser),
    )

read_metadata

read_metadata(path)

Parse a WAV file and return the recorder metadata it contains.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a WAV file.	required

Returns:

Name	Type	Description
A	RecordingMetadata	class:RecordingMetadata exposing each detected standard.

Example

meta = read_metadata("recording.wav") meta.sources ('guano',) meta.guano.make 'Wildlife Acoustics, Inc.'

Source code in src/riffy/metadata/recording.py

def read_metadata(path: str | Path) -> RecordingMetadata:
    """Parse a WAV file and return the recorder metadata it contains.

    Args:
        path: Path to a WAV file.

    Returns:
        A :class:`RecordingMetadata` exposing each detected standard.

    Example:
        >>> meta = read_metadata("recording.wav")
        >>> meta.sources
        ('guano',)
        >>> meta.guano.make
        'Wildlife Acoustics, Inc.'
    """
    return RecordingMetadata.from_parser(WAVParser(path))

dump_metadata

dump_metadata(path)

Parse a WAV file and return its recorder metadata as a plain, JSON-serializable dict.

A convenience for inspection and tooling (it backs python -m riffy). The returned dict has file, riff_form, format, sources, and one entry per standard (guano, info, bext, audiomoth, wamd, ixml), each None when absent. All values are JSON-safe: datetimes become ISO strings and binary fields (e.g. bext umid) become hex strings.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a WAV file.	required

Returns:

Type	Description
dict	A JSON-serializable dict describing the file's metadata.

Source code in src/riffy/metadata/recording.py

def dump_metadata(path: str | Path) -> dict:
    """Parse a WAV file and return its recorder metadata as a plain, JSON-serializable dict.

    A convenience for inspection and tooling (it backs ``python -m riffy``). The
    returned dict has ``file``, ``riff_form``, ``format``, ``sources``, and one
    entry per standard (``guano``, ``info``, ``bext``, ``audiomoth``, ``wamd``,
    ``ixml``), each ``None`` when absent. All values are JSON-safe: datetimes
    become ISO strings and binary fields (e.g. bext ``umid``) become hex strings.

    Args:
        path: Path to a WAV file.

    Returns:
        A JSON-serializable dict describing the file's metadata.
    """
    parser = WAVParser(path)
    meta = RecordingMetadata.from_parser(parser)
    ixml = IXmlMetadata.from_parser(parser)

    sources = list(meta.sources)
    if ixml is not None:
        sources.append("ixml")

    return {
        "file": str(path),
        "riff_form": parser.riff_form,
        "format": parser.get_info()["format"],
        "sources": sources,
        "guano": _guano_to_dict(meta.guano),
        "info": dict(meta.info.tags) if meta.info is not None else None,
        "bext": _bext_to_dict(meta.bext),
        "audiomoth": _audiomoth_to_dict(meta.audiomoth),
        "wamd": meta.wamd.to_dict() if meta.wamd is not None else None,
        "ixml": ixml.to_dict() if ixml is not None else None,
    }

GUANO metadata: read and write the guan chunk.

GUANO (Grand Unified Acoustic Notation Ontology) is the closest thing the bioacoustics field has to a vendor-neutral metadata standard. It is embedded in a WAV sub-chunk with ID guan holding UTF-8 text, one [Namespace|]Key: Value field per line, with GUANO|Version required as the first field.

Design highlights (see the v0.3.0 plan §5.1):

• Attribute-first. Well-known fields are exposed as typed attributes (timestamp is a datetime with its offset preserved, loc_position is a (lat, lon) tuple, species_manual_id is a list, ...). Vendor and arbitrary fields remain reachable through :meth:~GuanoMetadata.get / :meth:~GuanoMetadata.set / :attr:~GuanoMetadata.fields.
• Round-trip safety. Unknown fields and vendor namespaces are preserved verbatim and in order — a hard requirement of the GUANO spec, not a nicety.
• Fail soft. Non-UTF-8 payloads fall back to latin-1 with a warning; a single malformed line is skipped with a warning rather than abandoning the whole chunk.

GuanoMetadata

Typed, round-trip-safe view of a guan chunk's GUANO metadata.

Construct from an existing file via :meth:from_parser / :meth:from_bytes, or build a fresh one with GuanoMetadata(version="1.0"). Well-known fields are typed attributes; everything else is reachable via :meth:get / :meth:set and :attr:fields.

Source code in src/riffy/metadata/guano.py

class GuanoMetadata:
    """Typed, round-trip-safe view of a ``guan`` chunk's GUANO metadata.

    Construct from an existing file via :meth:`from_parser` / :meth:`from_bytes`,
    or build a fresh one with ``GuanoMetadata(version="1.0")``. Well-known fields
    are typed attributes; everything else is reachable via :meth:`get` /
    :meth:`set` and :attr:`fields`.
    """

    def __init__(self, version: str | None = None) -> None:
        # Ordered mapping preserves file order for round-trip fidelity. Python
        # dicts preserve insertion order, which is exactly what we need.
        self._fields: dict[FieldKey, str] = {}
        if version is not None:
            self.version = version

    # ------------------------------------------------------------------ #
    # Construction
    # ------------------------------------------------------------------ #

    @classmethod
    def from_bytes(cls, data: bytes) -> "GuanoMetadata":
        """Parse GUANO metadata from a raw ``guan`` chunk payload.

        Args:
            data: The raw bytes of the ``guan`` chunk.

        Returns:
            A populated :class:`GuanoMetadata`.
        """
        self = cls()
        text = decode_text(data, context="guan")
        for raw_line in text.split("\n"):
            # Strip surrounding whitespace and NUL bytes: some writers (e.g.
            # Wildlife Acoustics) pad the chunk with NULs rather than spaces.
            line = raw_line.strip(_STRIP_CHARS)
            if not line:
                continue  # empty lines and whitespace/NUL padding are ignored
            if ":" not in line:
                warnings.warn(
                    f"guan: skipping malformed line without ':' separator: {line!r}",
                    stacklevel=2,
                )
                continue
            name_part, value_part = line.split(":", 1)
            namespace, key = _split_namespace(name_part)
            key = key.strip()
            if not key:
                warnings.warn(
                    f"guan: skipping line with empty key: {line!r}",
                    stacklevel=2,
                )
                continue
            # Values are stored unescaped (real newlines); re-escaped on write.
            value = _unescape(value_part.strip())
            self._fields[(namespace, key)] = value
        return self

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "GuanoMetadata | None":
        """Parse GUANO metadata from a parser's ``guan`` chunk.

        Args:
            parser: A parsed :class:`~riffy.WAVParser`.

        Returns:
            A :class:`GuanoMetadata`, or ``None`` if the file has no ``guan``
            chunk.
        """
        data = parser.get_chunk_bytes(CHUNK_ID)
        if data is None:
            return None
        return cls.from_bytes(data)

    # ------------------------------------------------------------------ #
    # Generic (namespace, key) access — preserves unknown/vendor fields
    # ------------------------------------------------------------------ #

    def get(self, namespace: str, key: str, default: str | None = None) -> str | None:
        """Return the raw string value for ``(namespace, key)``, or ``default``.

        Use the empty string for the base (un-prefixed) namespace.
        """
        return self._fields.get((namespace, key), default)

    def set(self, namespace: str, key: str, value: str) -> None:
        """Set the raw string value for ``(namespace, key)``.

        Newlines in ``value`` are allowed and re-escaped on write.
        """
        self._fields[(namespace, key)] = value

    def remove(self, namespace: str, key: str) -> None:
        """Delete ``(namespace, key)`` if present (no error if absent)."""
        self._fields.pop((namespace, key), None)

    def __contains__(self, key: object) -> bool:
        return key in self._fields

    def __iter__(self) -> Iterator[FieldKey]:
        return iter(self._fields)

    @property
    def fields(self) -> dict[FieldKey, str]:
        """A copy of the full ``(namespace, key) -> raw string`` mapping."""
        return dict(self._fields)

    def get_binary(self, namespace: str, key: str) -> bytes | None:
        """Return a Base64-decoded (RFC 4648) value, or ``None`` if absent."""
        raw = self._fields.get((namespace, key))
        if raw is None:
            return None
        return base64.b64decode(raw)

    def set_binary(self, namespace: str, key: str, value: bytes) -> None:
        """Store ``value`` as a Base64-encoded (RFC 4648) string."""
        self._fields[(namespace, key)] = base64.b64encode(value).decode("ascii")

    # ------------------------------------------------------------------ #
    # Typed string helpers
    # ------------------------------------------------------------------ #

    def _get_str(self, key: str) -> str | None:
        return self._fields.get(("", key))

    def _set_str(self, key: str, value: str | None) -> None:
        if value is None:
            self._fields.pop(("", key), None)
        else:
            self._fields[("", key)] = value

    def _get_float(self, key: str) -> float | None:
        raw = self._fields.get(("", key))
        if raw is None:
            return None
        try:
            return float(raw)
        except ValueError:
            warnings.warn(f"guan: {key!r} is not a valid float: {raw!r}", stacklevel=2)
            return None

    def _set_float(self, key: str, value: float | None) -> None:
        if value is None:
            self._fields.pop(("", key), None)
        else:
            self._fields[("", key)] = repr(value)

    def _get_int(self, key: str) -> int | None:
        raw = self._fields.get(("", key))
        if raw is None:
            return None
        try:
            return int(raw)
        except ValueError:
            warnings.warn(f"guan: {key!r} is not a valid int: {raw!r}", stacklevel=2)
            return None

    def _set_int(self, key: str, value: int | None) -> None:
        if value is None:
            self._fields.pop(("", key), None)
        else:
            self._fields[("", key)] = str(value)

    def _get_list(self, key: str) -> list[str]:
        raw = self._fields.get(("", key))
        if not raw:
            return []
        return [item.strip() for item in raw.split(",") if item.strip()]

    def _set_list(self, key: str, value: Iterable[str] | None) -> None:
        if value is None:
            self._fields.pop(("", key), None)
        else:
            self._fields[("", key)] = ", ".join(value)

    # ------------------------------------------------------------------ #
    # Well-known typed attributes
    # ------------------------------------------------------------------ #

    @property
    def version(self) -> str | None:
        """``GUANO|Version`` — the format version (required on write)."""
        return self._fields.get((GUANO_NAMESPACE, "Version"))

    @version.setter
    def version(self, value: str | None) -> None:
        if value is None:
            self._fields.pop((GUANO_NAMESPACE, "Version"), None)
        else:
            self._fields[(GUANO_NAMESPACE, "Version")] = value

    @property
    def make(self) -> str | None:
        return self._get_str("Make")

    @make.setter
    def make(self, value: str | None) -> None:
        self._set_str("Make", value)

    @property
    def model(self) -> str | None:
        return self._get_str("Model")

    @model.setter
    def model(self, value: str | None) -> None:
        self._set_str("Model", value)

    @property
    def serial(self) -> str | None:
        return self._get_str("Serial")

    @serial.setter
    def serial(self, value: str | None) -> None:
        self._set_str("Serial", value)

    @property
    def firmware_version(self) -> str | None:
        return self._get_str("Firmware Version")

    @firmware_version.setter
    def firmware_version(self, value: str | None) -> None:
        self._set_str("Firmware Version", value)

    @property
    def hardware_version(self) -> str | None:
        return self._get_str("Hardware Version")

    @hardware_version.setter
    def hardware_version(self, value: str | None) -> None:
        self._set_str("Hardware Version", value)

    @property
    def original_filename(self) -> str | None:
        return self._get_str("Original Filename")

    @original_filename.setter
    def original_filename(self, value: str | None) -> None:
        self._set_str("Original Filename", value)

    @property
    def note(self) -> str | None:
        """``Note`` — free text, may contain newlines."""
        return self._get_str("Note")

    @note.setter
    def note(self, value: str | None) -> None:
        self._set_str("Note", value)

    @property
    def length(self) -> float | None:
        return self._get_float("Length")

    @length.setter
    def length(self, value: float | None) -> None:
        self._set_float("Length", value)

    @property
    def filter_hp(self) -> float | None:
        return self._get_float("Filter HP")

    @filter_hp.setter
    def filter_hp(self, value: float | None) -> None:
        self._set_float("Filter HP", value)

    @property
    def filter_lp(self) -> float | None:
        return self._get_float("Filter LP")

    @filter_lp.setter
    def filter_lp(self, value: float | None) -> None:
        self._set_float("Filter LP", value)

    @property
    def humidity(self) -> float | None:
        return self._get_float("Humidity")

    @humidity.setter
    def humidity(self, value: float | None) -> None:
        self._set_float("Humidity", value)

    @property
    def temperature_int(self) -> float | None:
        return self._get_float("Temperature Int")

    @temperature_int.setter
    def temperature_int(self, value: float | None) -> None:
        self._set_float("Temperature Int", value)

    @property
    def temperature_ext(self) -> float | None:
        return self._get_float("Temperature Ext")

    @temperature_ext.setter
    def temperature_ext(self, value: float | None) -> None:
        self._set_float("Temperature Ext", value)

    @property
    def loc_accuracy(self) -> float | None:
        return self._get_float("Loc Accuracy")

    @loc_accuracy.setter
    def loc_accuracy(self, value: float | None) -> None:
        self._set_float("Loc Accuracy", value)

    @property
    def loc_elevation(self) -> float | None:
        return self._get_float("Loc Elevation")

    @loc_elevation.setter
    def loc_elevation(self, value: float | None) -> None:
        self._set_float("Loc Elevation", value)

    @property
    def samplerate(self) -> int | None:
        return self._get_int("Samplerate")

    @samplerate.setter
    def samplerate(self, value: int | None) -> None:
        self._set_int("Samplerate", value)

    @property
    def te(self) -> int:
        """``TE`` — time-expansion factor (defaults to 1 when absent)."""
        value = self._get_int("TE")
        return 1 if value is None else value

    @te.setter
    def te(self, value: int | None) -> None:
        self._set_int("TE", value)

    @property
    def species_auto_id(self) -> list[str]:
        return self._get_list("Species Auto ID")

    @species_auto_id.setter
    def species_auto_id(self, value: Iterable[str] | None) -> None:
        self._set_list("Species Auto ID", value)

    @property
    def species_manual_id(self) -> list[str]:
        return self._get_list("Species Manual ID")

    @species_manual_id.setter
    def species_manual_id(self, value: Iterable[str] | None) -> None:
        self._set_list("Species Manual ID", value)

    @property
    def tags(self) -> list[str]:
        return self._get_list("Tags")

    @tags.setter
    def tags(self, value: Iterable[str] | None) -> None:
        self._set_list("Tags", value)

    @property
    def timestamp(self) -> datetime | None:
        """``Timestamp`` — parsed ISO 8601 / RFC 3339, offset preserved."""
        raw = self._fields.get(("", "Timestamp"))
        if raw is None:
            return None
        return _parse_timestamp(raw)

    @timestamp.setter
    def timestamp(self, value: datetime | str | None) -> None:
        if value is None:
            self._fields.pop(("", "Timestamp"), None)
        elif isinstance(value, datetime):
            self._fields[("", "Timestamp")] = value.isoformat()
        else:
            self._fields[("", "Timestamp")] = value

    @property
    def loc_position(self) -> tuple[float, float] | None:
        """``Loc Position`` — ``(latitude, longitude)`` WGS84 float tuple."""
        raw = self._fields.get(("", "Loc Position"))
        if raw is None:
            return None
        parts = raw.split()
        if len(parts) != 2:
            warnings.warn(f"guan: 'Loc Position' is not 'lat lon': {raw!r}", stacklevel=2)
            return None
        try:
            return (float(parts[0]), float(parts[1]))
        except ValueError:
            warnings.warn(f"guan: 'Loc Position' has non-numeric parts: {raw!r}", stacklevel=2)
            return None

    @loc_position.setter
    def loc_position(self, value: tuple[float, float] | None) -> None:
        if value is None:
            self._fields.pop(("", "Loc Position"), None)
        else:
            lat, lon = value
            self._fields[("", "Loc Position")] = f"{lat} {lon}"

    # ------------------------------------------------------------------ #
    # Serialization
    # ------------------------------------------------------------------ #

    def to_chunk_bytes(self) -> bytes:
        """Serialize to a ``guan`` chunk payload (UTF-8, Version-first, even).

        Raises:
            ValueError: If ``GUANO|Version`` has not been set. The spec requires
                it as the mandatory first field.
        """
        version_key = (GUANO_NAMESPACE, "Version")
        if version_key not in self._fields:
            raise ValueError("GUANO requires 'GUANO|Version'; set .version before serializing")

        # GUANO|Version must come first; everything else keeps insertion order.
        ordered: list[FieldKey] = [version_key]
        ordered += [key for key in self._fields if key != version_key]

        lines = []
        for namespace, key in ordered:
            prefix = f"{namespace}|{key}" if namespace else key
            lines.append(f"{prefix}: {_escape(self._fields[(namespace, key)])}")

        encoded = "\n".join(lines).encode("utf-8")
        # Even-byte padding with a space, per the GUANO whitespace convention;
        # the trailing space is stripped on read.
        return pad_to_even(encoded, pad_byte=0x20)

    def write_to_parser(self, parser: WAVParser) -> None:
        """Write this metadata into ``parser`` as the ``guan`` chunk.

        Replaces any existing ``guan`` chunk. Call ``parser.write_wav(...)``
        afterwards to persist to disk.
        """
        parser.set_chunk(CHUNK_ID, self.to_chunk_bytes())

fields property

fields

A copy of the full (namespace, key) -> raw string mapping.

version property writable

version

GUANO|Version — the format version (required on write).

note property writable

note

Note — free text, may contain newlines.

te property writable

te

TE — time-expansion factor (defaults to 1 when absent).

timestamp property writable

timestamp

Timestamp — parsed ISO 8601 / RFC 3339, offset preserved.

loc_position property writable

loc_position

Loc Position — (latitude, longitude) WGS84 float tuple.

from_bytes classmethod

from_bytes(data)

Parse GUANO metadata from a raw guan chunk payload.

Parameters:

Name	Type	Description	Default
data	bytes	The raw bytes of the guan chunk.	required

Returns:

Type	Description
GuanoMetadata	A populated :class:GuanoMetadata.

Source code in src/riffy/metadata/guano.py

@classmethod
def from_bytes(cls, data: bytes) -> "GuanoMetadata":
    """Parse GUANO metadata from a raw ``guan`` chunk payload.

    Args:
        data: The raw bytes of the ``guan`` chunk.

    Returns:
        A populated :class:`GuanoMetadata`.
    """
    self = cls()
    text = decode_text(data, context="guan")
    for raw_line in text.split("\n"):
        # Strip surrounding whitespace and NUL bytes: some writers (e.g.
        # Wildlife Acoustics) pad the chunk with NULs rather than spaces.
        line = raw_line.strip(_STRIP_CHARS)
        if not line:
            continue  # empty lines and whitespace/NUL padding are ignored
        if ":" not in line:
            warnings.warn(
                f"guan: skipping malformed line without ':' separator: {line!r}",
                stacklevel=2,
            )
            continue
        name_part, value_part = line.split(":", 1)
        namespace, key = _split_namespace(name_part)
        key = key.strip()
        if not key:
            warnings.warn(
                f"guan: skipping line with empty key: {line!r}",
                stacklevel=2,
            )
            continue
        # Values are stored unescaped (real newlines); re-escaped on write.
        value = _unescape(value_part.strip())
        self._fields[(namespace, key)] = value
    return self

from_parser classmethod

from_parser(parser)

Parse GUANO metadata from a parser's guan chunk.

Parameters:

Name	Type	Description	Default
parser	WAVParser	A parsed :class:~riffy.WAVParser.	required

Returns:

Name	Type	Description
A	GuanoMetadata | None	class:GuanoMetadata, or None if the file has no guan
	GuanoMetadata | None	chunk.

Source code in src/riffy/metadata/guano.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "GuanoMetadata | None":
    """Parse GUANO metadata from a parser's ``guan`` chunk.

    Args:
        parser: A parsed :class:`~riffy.WAVParser`.

    Returns:
        A :class:`GuanoMetadata`, or ``None`` if the file has no ``guan``
        chunk.
    """
    data = parser.get_chunk_bytes(CHUNK_ID)
    if data is None:
        return None
    return cls.from_bytes(data)

get

get(namespace, key, default=None)

Return the raw string value for (namespace, key), or default.

Use the empty string for the base (un-prefixed) namespace.

Source code in src/riffy/metadata/guano.py

def get(self, namespace: str, key: str, default: str | None = None) -> str | None:
    """Return the raw string value for ``(namespace, key)``, or ``default``.

    Use the empty string for the base (un-prefixed) namespace.
    """
    return self._fields.get((namespace, key), default)

set

set(namespace, key, value)

Set the raw string value for (namespace, key).

Newlines in value are allowed and re-escaped on write.

Source code in src/riffy/metadata/guano.py

def set(self, namespace: str, key: str, value: str) -> None:
    """Set the raw string value for ``(namespace, key)``.

    Newlines in ``value`` are allowed and re-escaped on write.
    """
    self._fields[(namespace, key)] = value

remove

remove(namespace, key)

Delete (namespace, key) if present (no error if absent).

Source code in src/riffy/metadata/guano.py

def remove(self, namespace: str, key: str) -> None:
    """Delete ``(namespace, key)`` if present (no error if absent)."""
    self._fields.pop((namespace, key), None)

get_binary

get_binary(namespace, key)

Return a Base64-decoded (RFC 4648) value, or None if absent.

Source code in src/riffy/metadata/guano.py

def get_binary(self, namespace: str, key: str) -> bytes | None:
    """Return a Base64-decoded (RFC 4648) value, or ``None`` if absent."""
    raw = self._fields.get((namespace, key))
    if raw is None:
        return None
    return base64.b64decode(raw)

set_binary

set_binary(namespace, key, value)

Store value as a Base64-encoded (RFC 4648) string.

Source code in src/riffy/metadata/guano.py

def set_binary(self, namespace: str, key: str, value: bytes) -> None:
    """Store ``value`` as a Base64-encoded (RFC 4648) string."""
    self._fields[(namespace, key)] = base64.b64encode(value).decode("ascii")

to_chunk_bytes

to_chunk_bytes()

Serialize to a guan chunk payload (UTF-8, Version-first, even).

Raises:

Type	Description
ValueError	If GUANO|Version has not been set. The spec requires it as the mandatory first field.

Source code in src/riffy/metadata/guano.py

def to_chunk_bytes(self) -> bytes:
    """Serialize to a ``guan`` chunk payload (UTF-8, Version-first, even).

    Raises:
        ValueError: If ``GUANO|Version`` has not been set. The spec requires
            it as the mandatory first field.
    """
    version_key = (GUANO_NAMESPACE, "Version")
    if version_key not in self._fields:
        raise ValueError("GUANO requires 'GUANO|Version'; set .version before serializing")

    # GUANO|Version must come first; everything else keeps insertion order.
    ordered: list[FieldKey] = [version_key]
    ordered += [key for key in self._fields if key != version_key]

    lines = []
    for namespace, key in ordered:
        prefix = f"{namespace}|{key}" if namespace else key
        lines.append(f"{prefix}: {_escape(self._fields[(namespace, key)])}")

    encoded = "\n".join(lines).encode("utf-8")
    # Even-byte padding with a space, per the GUANO whitespace convention;
    # the trailing space is stripped on read.
    return pad_to_even(encoded, pad_byte=0x20)

write_to_parser

write_to_parser(parser)

Write this metadata into parser as the guan chunk.

Replaces any existing guan chunk. Call parser.write_wav(...) afterwards to persist to disk.

Source code in src/riffy/metadata/guano.py

def write_to_parser(self, parser: WAVParser) -> None:
    """Write this metadata into ``parser`` as the ``guan`` chunk.

    Replaces any existing ``guan`` chunk. Call ``parser.write_wav(...)``
    afterwards to persist to disk.
    """
    parser.set_chunk(CHUNK_ID, self.to_chunk_bytes())

RIFF INFO metadata: read and write the LIST/INFO chunk.

A RIFF INFO block is a LIST chunk whose list-type is INFO, containing NUL-terminated (ZSTR) subchunks each keyed by a 4-character FOURCC (INAM for title, IART for artist, ICMT for comment, ...). This module decodes that payload into friendly typed attributes while keeping raw FOURCC access, and rebuilds it on write.

Practical details handled here (see the v0.3.0 plan §5.2):

• Each value is a NUL-terminated string, but real-world producers are sloppy — the reader tolerates both terminated and non-terminated values.
• Each subchunk is padded to an even length; padding is honored on read and emitted on write.
• A file may carry several LIST chunks (e.g. an adtl list alongside the INFO one); this module targets the INFO list specifically and leaves the others untouched.

Values are treated as latin-1, which losslessly round-trips any byte and matches the ASCII/codepage heritage of RIFF INFO. (UTF-8 metadata belongs in GUANO.)

This module is also the carrier the AudioMoth comment decoder reads from.

InfoMetadata

Typed, round-trip-safe view of a LIST/INFO block.

Construct from a file via :meth:from_parser / :meth:from_bytes, or build a fresh one with InfoMetadata(). Well-known tags are exposed as friendly str | None attributes (title, artist, comment, ...); every tag, known or not, is reachable by FOURCC via :meth:get / :meth:set and :attr:tags.

Source code in src/riffy/metadata/info.py

class InfoMetadata:
    """Typed, round-trip-safe view of a ``LIST``/``INFO`` block.

    Construct from a file via :meth:`from_parser` / :meth:`from_bytes`, or build
    a fresh one with ``InfoMetadata()``. Well-known tags are exposed as friendly
    ``str | None`` attributes (``title``, ``artist``, ``comment``, ...); every
    tag, known or not, is reachable by FOURCC via :meth:`get` / :meth:`set` and
    :attr:`tags`.
    """

    def __init__(self) -> None:
        # Ordered FOURCC -> value; dict preserves insertion order for round-trip.
        self._tags: dict[str, str] = {}

    # ------------------------------------------------------------------ #
    # Construction
    # ------------------------------------------------------------------ #

    @classmethod
    def from_bytes(cls, data: bytes) -> "InfoMetadata":
        """Parse an INFO block from a raw ``LIST`` chunk payload.

        Args:
            data: The full ``LIST`` chunk payload, including the leading 4-byte
                list-type (which must be ``INFO``).

        Returns:
            A populated :class:`InfoMetadata`.

        Raises:
            ValueError: If the payload is not an ``INFO`` list.
        """
        if data[:4] != LIST_TYPE:
            raise ValueError(f"Not an INFO list (list-type is {data[:4]!r}, expected b'INFO')")

        self = cls()
        offset = 4
        n = len(data)
        while offset + 8 <= n:
            fourcc = data[offset : offset + 4].decode("latin-1")
            size = struct.unpack_from("<I", data, offset + 4)[0]
            offset += 8
            if offset + size > n:
                warnings.warn(
                    f"INFO: subchunk {fourcc!r} claims {size} bytes but only "
                    f"{n - offset} remain; truncating",
                    stacklevel=2,
                )
                size = n - offset
            value, _ = read_zstr(data[offset : offset + size])
            self._tags[fourcc] = value
            offset += size
            if size % 2:  # skip the even-alignment pad byte
                offset += 1
        return self

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "InfoMetadata | None":
        """Parse the INFO block from a parser's ``LIST`` chunks.

        Scans every ``LIST`` chunk and decodes the first one whose list-type is
        ``INFO``.

        Args:
            parser: A parsed :class:`~riffy.WAVParser`.

        Returns:
            An :class:`InfoMetadata`, or ``None`` if the file has no INFO list.
        """
        for chunk in parser.get_chunks(CHUNK_ID):
            if chunk.data[:4] == LIST_TYPE:
                return cls.from_bytes(chunk.data)
        return None

    # ------------------------------------------------------------------ #
    # Raw FOURCC access
    # ------------------------------------------------------------------ #

    def get(self, fourcc: str, default: str | None = None) -> str | None:
        """Return the value for a FOURCC tag, or ``default`` if absent."""
        return self._tags.get(fourcc, default)

    def set(self, fourcc: str, value: str | None) -> None:
        """Set (or, when ``value`` is ``None``, remove) a FOURCC tag.

        Raises:
            InvalidChunkError: If ``fourcc`` is not exactly four ASCII characters.
        """
        validate_fourcc(fourcc)
        if value is None:
            self._tags.pop(fourcc, None)
        else:
            self._tags[fourcc] = value

    def remove(self, fourcc: str) -> None:
        """Delete a FOURCC tag if present (no error if absent)."""
        self._tags.pop(fourcc, None)

    def __contains__(self, fourcc: object) -> bool:
        return fourcc in self._tags

    @property
    def tags(self) -> dict[str, str]:
        """A copy of the full ``FOURCC -> value`` mapping, in order."""
        return dict(self._tags)

    # ------------------------------------------------------------------ #
    # Friendly typed attributes
    # ------------------------------------------------------------------ #

    @property
    def title(self) -> str | None:
        return self._tags.get("INAM")

    @title.setter
    def title(self, value: str | None) -> None:
        self.set("INAM", value)

    @property
    def artist(self) -> str | None:
        return self._tags.get("IART")

    @artist.setter
    def artist(self, value: str | None) -> None:
        self.set("IART", value)

    @property
    def comment(self) -> str | None:
        return self._tags.get("ICMT")

    @comment.setter
    def comment(self, value: str | None) -> None:
        self.set("ICMT", value)

    @property
    def creation_date(self) -> str | None:
        return self._tags.get("ICRD")

    @creation_date.setter
    def creation_date(self, value: str | None) -> None:
        self.set("ICRD", value)

    @property
    def copyright(self) -> str | None:
        return self._tags.get("ICOP")

    @copyright.setter
    def copyright(self, value: str | None) -> None:
        self.set("ICOP", value)

    @property
    def software(self) -> str | None:
        return self._tags.get("ISFT")

    @software.setter
    def software(self, value: str | None) -> None:
        self.set("ISFT", value)

    @property
    def engineer(self) -> str | None:
        return self._tags.get("IENG")

    @engineer.setter
    def engineer(self, value: str | None) -> None:
        self.set("IENG", value)

    @property
    def genre(self) -> str | None:
        return self._tags.get("IGNR")

    @genre.setter
    def genre(self, value: str | None) -> None:
        self.set("IGNR", value)

    @property
    def product(self) -> str | None:
        return self._tags.get("IPRD")

    @product.setter
    def product(self, value: str | None) -> None:
        self.set("IPRD", value)

    @property
    def source(self) -> str | None:
        return self._tags.get("ISRC")

    @source.setter
    def source(self, value: str | None) -> None:
        self.set("ISRC", value)

    @property
    def technician(self) -> str | None:
        return self._tags.get("ITCH")

    @technician.setter
    def technician(self, value: str | None) -> None:
        self.set("ITCH", value)

    @property
    def keywords(self) -> str | None:
        return self._tags.get("IKEY")

    @keywords.setter
    def keywords(self, value: str | None) -> None:
        self.set("IKEY", value)

    @property
    def subject(self) -> str | None:
        return self._tags.get("ISBJ")

    @subject.setter
    def subject(self, value: str | None) -> None:
        self.set("ISBJ", value)

    # ------------------------------------------------------------------ #
    # Serialization
    # ------------------------------------------------------------------ #

    def to_chunk_bytes(self) -> bytes:
        """Serialize to a ``LIST`` chunk payload (``INFO`` type + subchunks).

        Each value is written NUL-terminated and padded to even length. FOURCCs
        and values are encoded as latin-1, byte-for-byte, so anything read back
        (including non-ASCII tags produced by sloppy encoders) round-trips
        without loss. New tags added via :meth:`set` are already ASCII-validated
        on the way in.
        """
        out = bytearray(LIST_TYPE)
        for fourcc, value in self._tags.items():
            payload = value.encode("latin-1") + b"\x00"
            out += fourcc.encode("latin-1")
            out += struct.pack("<I", len(payload))
            out += payload
            if len(payload) % 2:
                out += b"\x00"
        return bytes(out)

    def write_to_parser(self, parser: WAVParser) -> None:
        """Write this INFO block into ``parser`` as a ``LIST`` chunk.

        Replaces the existing ``INFO`` ``LIST`` chunk if present (leaving any
        other ``LIST`` chunks, such as ``adtl``, untouched), otherwise appends a
        new one. Call ``parser.write_wav(...)`` afterwards to persist to disk.
        """
        payload = self.to_chunk_bytes()
        occurrences = parser.chunks.get(CHUNK_ID, [])
        for i, chunk in enumerate(occurrences):
            if chunk.data[:4] == LIST_TYPE:
                occurrences[i] = WAVChunk(
                    id=CHUNK_ID, size=len(payload), data=payload, offset=chunk.offset
                )
                return
        parser.add_chunk(CHUNK_ID, payload)

tags property

tags

A copy of the full FOURCC -> value mapping, in order.

from_bytes classmethod

from_bytes(data)

Parse an INFO block from a raw LIST chunk payload.

Parameters:

Name	Type	Description	Default
data	bytes	The full LIST chunk payload, including the leading 4-byte list-type (which must be INFO).	required

Returns:

Type	Description
InfoMetadata	A populated :class:InfoMetadata.

Raises:

Type	Description
ValueError	If the payload is not an INFO list.

Source code in src/riffy/metadata/info.py

@classmethod
def from_bytes(cls, data: bytes) -> "InfoMetadata":
    """Parse an INFO block from a raw ``LIST`` chunk payload.

    Args:
        data: The full ``LIST`` chunk payload, including the leading 4-byte
            list-type (which must be ``INFO``).

    Returns:
        A populated :class:`InfoMetadata`.

    Raises:
        ValueError: If the payload is not an ``INFO`` list.
    """
    if data[:4] != LIST_TYPE:
        raise ValueError(f"Not an INFO list (list-type is {data[:4]!r}, expected b'INFO')")

    self = cls()
    offset = 4
    n = len(data)
    while offset + 8 <= n:
        fourcc = data[offset : offset + 4].decode("latin-1")
        size = struct.unpack_from("<I", data, offset + 4)[0]
        offset += 8
        if offset + size > n:
            warnings.warn(
                f"INFO: subchunk {fourcc!r} claims {size} bytes but only "
                f"{n - offset} remain; truncating",
                stacklevel=2,
            )
            size = n - offset
        value, _ = read_zstr(data[offset : offset + size])
        self._tags[fourcc] = value
        offset += size
        if size % 2:  # skip the even-alignment pad byte
            offset += 1
    return self

from_parser classmethod

from_parser(parser)

Parse the INFO block from a parser's LIST chunks.

Scans every LIST chunk and decodes the first one whose list-type is INFO.

Parameters:

Name	Type	Description	Default
parser	WAVParser	A parsed :class:~riffy.WAVParser.	required

Returns:

Name	Type	Description
An	InfoMetadata | None	class:InfoMetadata, or None if the file has no INFO list.

Source code in src/riffy/metadata/info.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "InfoMetadata | None":
    """Parse the INFO block from a parser's ``LIST`` chunks.

    Scans every ``LIST`` chunk and decodes the first one whose list-type is
    ``INFO``.

    Args:
        parser: A parsed :class:`~riffy.WAVParser`.

    Returns:
        An :class:`InfoMetadata`, or ``None`` if the file has no INFO list.
    """
    for chunk in parser.get_chunks(CHUNK_ID):
        if chunk.data[:4] == LIST_TYPE:
            return cls.from_bytes(chunk.data)
    return None

get

get(fourcc, default=None)

Return the value for a FOURCC tag, or default if absent.

Source code in src/riffy/metadata/info.py

def get(self, fourcc: str, default: str | None = None) -> str | None:
    """Return the value for a FOURCC tag, or ``default`` if absent."""
    return self._tags.get(fourcc, default)

set

set(fourcc, value)

Set (or, when value is None, remove) a FOURCC tag.

Raises:

Type	Description
InvalidChunkError	If fourcc is not exactly four ASCII characters.

Source code in src/riffy/metadata/info.py

def set(self, fourcc: str, value: str | None) -> None:
    """Set (or, when ``value`` is ``None``, remove) a FOURCC tag.

    Raises:
        InvalidChunkError: If ``fourcc`` is not exactly four ASCII characters.
    """
    validate_fourcc(fourcc)
    if value is None:
        self._tags.pop(fourcc, None)
    else:
        self._tags[fourcc] = value

remove

remove(fourcc)

Delete a FOURCC tag if present (no error if absent).

Source code in src/riffy/metadata/info.py

def remove(self, fourcc: str) -> None:
    """Delete a FOURCC tag if present (no error if absent)."""
    self._tags.pop(fourcc, None)

to_chunk_bytes

to_chunk_bytes()

Serialize to a LIST chunk payload (INFO type + subchunks).

Each value is written NUL-terminated and padded to even length. FOURCCs and values are encoded as latin-1, byte-for-byte, so anything read back (including non-ASCII tags produced by sloppy encoders) round-trips without loss. New tags added via :meth:set are already ASCII-validated on the way in.

Source code in src/riffy/metadata/info.py

def to_chunk_bytes(self) -> bytes:
    """Serialize to a ``LIST`` chunk payload (``INFO`` type + subchunks).

    Each value is written NUL-terminated and padded to even length. FOURCCs
    and values are encoded as latin-1, byte-for-byte, so anything read back
    (including non-ASCII tags produced by sloppy encoders) round-trips
    without loss. New tags added via :meth:`set` are already ASCII-validated
    on the way in.
    """
    out = bytearray(LIST_TYPE)
    for fourcc, value in self._tags.items():
        payload = value.encode("latin-1") + b"\x00"
        out += fourcc.encode("latin-1")
        out += struct.pack("<I", len(payload))
        out += payload
        if len(payload) % 2:
            out += b"\x00"
    return bytes(out)

write_to_parser

write_to_parser(parser)

Write this INFO block into parser as a LIST chunk.

Replaces the existing INFO LIST chunk if present (leaving any other LIST chunks, such as adtl, untouched), otherwise appends a new one. Call parser.write_wav(...) afterwards to persist to disk.

Source code in src/riffy/metadata/info.py

def write_to_parser(self, parser: WAVParser) -> None:
    """Write this INFO block into ``parser`` as a ``LIST`` chunk.

    Replaces the existing ``INFO`` ``LIST`` chunk if present (leaving any
    other ``LIST`` chunks, such as ``adtl``, untouched), otherwise appends a
    new one. Call ``parser.write_wav(...)`` afterwards to persist to disk.
    """
    payload = self.to_chunk_bytes()
    occurrences = parser.chunks.get(CHUNK_ID, [])
    for i, chunk in enumerate(occurrences):
        if chunk.data[:4] == LIST_TYPE:
            occurrences[i] = WAVChunk(
                id=CHUNK_ID, size=len(payload), data=payload, offset=chunk.offset
            )
            return
    parser.add_chunk(CHUNK_ID, payload)

Broadcast Wave bext metadata: read and write the fixed binary layout.

Unlike GUANO and RIFF INFO, the bext chunk (EBU Tech 3285) is a fixed C-struct binary layout, not key-value text, so it is parsed with :mod:struct. The layout grows across versions — v0 has the base fields, v1 adds the 64-byte UMID, v2 adds five loudness fields — but the fixed portion is always 602 bytes (a Reserved region absorbs the difference), with CodingHistory filling the remainder of the chunk.

This module reads and writes all three versions. The writer is table-driven off the version field and gates the version-specific regions accordingly; the v2 loudness fields are preserved verbatim on write but never computed (per the v0.3.0 plan §5.3).

BextMetadata dataclass

Typed view of a bext (Broadcast Wave) chunk.

Version-specific fields are None when the chunk's version does not include them: umid requires v1+, and the five loudness fields require v2+.

Source code in src/riffy/metadata/bext.py

@dataclass
class BextMetadata:
    """Typed view of a ``bext`` (Broadcast Wave) chunk.

    Version-specific fields are ``None`` when the chunk's ``version`` does not
    include them: ``umid`` requires v1+, and the five loudness fields require
    v2+.
    """

    description: str = ""
    originator: str = ""
    originator_reference: str = ""
    origination_date: str = ""
    origination_time: str = ""
    time_reference: int = 0
    version: int = 0
    umid: bytes | None = None
    loudness_value: int | None = None
    loudness_range: int | None = None
    max_true_peak_level: int | None = None
    max_momentary_loudness: int | None = None
    max_short_term_loudness: int | None = None
    coding_history: str = ""

    # ------------------------------------------------------------------ #
    # Construction
    # ------------------------------------------------------------------ #

    @classmethod
    def from_bytes(cls, data: bytes) -> "BextMetadata":
        """Parse a ``bext`` chunk payload.

        Tolerates a truncated fixed header (a short chunk is zero-padded to the
        602-byte layout before unpacking), so malformed producers degrade rather
        than raise.
        """
        fixed = data[:FIXED_SIZE].ljust(FIXED_SIZE, b"\x00")
        (
            description,
            originator,
            originator_reference,
            origination_date,
            origination_time,
            tr_low,
            tr_high,
            version,
            umid,
            loud_value,
            loud_range,
            max_true_peak,
            max_momentary,
            max_short_term,
            _reserved,
        ) = _HEADER.unpack(fixed)

        return cls(
            description=_decode_fixed(description),
            originator=_decode_fixed(originator),
            originator_reference=_decode_fixed(originator_reference),
            origination_date=_decode_fixed(origination_date),
            origination_time=_decode_fixed(origination_time),
            time_reference=tr_low | (tr_high << 32),
            version=version,
            # Version-gated: only interpret regions the declared version defines.
            umid=umid if version >= 1 else None,
            loudness_value=loud_value if version >= 2 else None,
            loudness_range=loud_range if version >= 2 else None,
            max_true_peak_level=max_true_peak if version >= 2 else None,
            max_momentary_loudness=max_momentary if version >= 2 else None,
            max_short_term_loudness=max_short_term if version >= 2 else None,
            coding_history=data[FIXED_SIZE:].split(b"\x00", 1)[0].decode("latin-1"),
        )

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "BextMetadata | None":
        """Parse the ``bext`` chunk from a parser, or ``None`` if absent."""
        data = parser.get_chunk_bytes(CHUNK_ID)
        if data is None:
            return None
        return cls.from_bytes(data)

    # ------------------------------------------------------------------ #
    # Serialization
    # ------------------------------------------------------------------ #

    def to_chunk_bytes(self) -> bytes:
        """Serialize to a ``bext`` chunk payload, gated by ``version``.

        Version-specific regions are emitted only when ``version`` includes
        them; otherwise they are zero-filled. Loudness values are written
        verbatim (never computed).
        """
        umid = (self.umid or b"").ljust(64, b"\x00")[:64] if self.version >= 1 else b"\x00" * 64

        if self.version >= 2:
            loudness = (
                self.loudness_value or 0,
                self.loudness_range or 0,
                self.max_true_peak_level or 0,
                self.max_momentary_loudness or 0,
                self.max_short_term_loudness or 0,
            )
        else:
            loudness = (0, 0, 0, 0, 0)

        header = _HEADER.pack(
            _encode_fixed(self.description, 256),
            _encode_fixed(self.originator, 32),
            _encode_fixed(self.originator_reference, 32),
            _encode_fixed(self.origination_date, 10),
            _encode_fixed(self.origination_time, 8),
            self.time_reference & 0xFFFFFFFF,
            (self.time_reference >> 32) & 0xFFFFFFFF,
            self.version,
            umid,
            *loudness,
            b"\x00" * 180,
        )
        return header + self.coding_history.encode("latin-1")

    def write_to_parser(self, parser: WAVParser) -> None:
        """Write this metadata into ``parser`` as the ``bext`` chunk.

        Replaces an existing ``bext`` chunk or appends one. Call
        ``parser.write_wav(...)`` afterwards to persist to disk.
        """
        payload = self.to_chunk_bytes()
        existing = parser.chunks.get(CHUNK_ID)
        if existing:
            existing[0] = WAVChunk(
                id=CHUNK_ID, size=len(payload), data=payload, offset=existing[0].offset
            )
        else:
            parser.add_chunk(CHUNK_ID, payload)

from_bytes classmethod

from_bytes(data)

Parse a bext chunk payload.

Tolerates a truncated fixed header (a short chunk is zero-padded to the 602-byte layout before unpacking), so malformed producers degrade rather than raise.

Source code in src/riffy/metadata/bext.py

@classmethod
def from_bytes(cls, data: bytes) -> "BextMetadata":
    """Parse a ``bext`` chunk payload.

    Tolerates a truncated fixed header (a short chunk is zero-padded to the
    602-byte layout before unpacking), so malformed producers degrade rather
    than raise.
    """
    fixed = data[:FIXED_SIZE].ljust(FIXED_SIZE, b"\x00")
    (
        description,
        originator,
        originator_reference,
        origination_date,
        origination_time,
        tr_low,
        tr_high,
        version,
        umid,
        loud_value,
        loud_range,
        max_true_peak,
        max_momentary,
        max_short_term,
        _reserved,
    ) = _HEADER.unpack(fixed)

    return cls(
        description=_decode_fixed(description),
        originator=_decode_fixed(originator),
        originator_reference=_decode_fixed(originator_reference),
        origination_date=_decode_fixed(origination_date),
        origination_time=_decode_fixed(origination_time),
        time_reference=tr_low | (tr_high << 32),
        version=version,
        # Version-gated: only interpret regions the declared version defines.
        umid=umid if version >= 1 else None,
        loudness_value=loud_value if version >= 2 else None,
        loudness_range=loud_range if version >= 2 else None,
        max_true_peak_level=max_true_peak if version >= 2 else None,
        max_momentary_loudness=max_momentary if version >= 2 else None,
        max_short_term_loudness=max_short_term if version >= 2 else None,
        coding_history=data[FIXED_SIZE:].split(b"\x00", 1)[0].decode("latin-1"),
    )

from_parser classmethod

from_parser(parser)

Parse the bext chunk from a parser, or None if absent.

Source code in src/riffy/metadata/bext.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "BextMetadata | None":
    """Parse the ``bext`` chunk from a parser, or ``None`` if absent."""
    data = parser.get_chunk_bytes(CHUNK_ID)
    if data is None:
        return None
    return cls.from_bytes(data)

to_chunk_bytes

to_chunk_bytes()

Serialize to a bext chunk payload, gated by version.

Version-specific regions are emitted only when version includes them; otherwise they are zero-filled. Loudness values are written verbatim (never computed).

Source code in src/riffy/metadata/bext.py

def to_chunk_bytes(self) -> bytes:
    """Serialize to a ``bext`` chunk payload, gated by ``version``.

    Version-specific regions are emitted only when ``version`` includes
    them; otherwise they are zero-filled. Loudness values are written
    verbatim (never computed).
    """
    umid = (self.umid or b"").ljust(64, b"\x00")[:64] if self.version >= 1 else b"\x00" * 64

    if self.version >= 2:
        loudness = (
            self.loudness_value or 0,
            self.loudness_range or 0,
            self.max_true_peak_level or 0,
            self.max_momentary_loudness or 0,
            self.max_short_term_loudness or 0,
        )
    else:
        loudness = (0, 0, 0, 0, 0)

    header = _HEADER.pack(
        _encode_fixed(self.description, 256),
        _encode_fixed(self.originator, 32),
        _encode_fixed(self.originator_reference, 32),
        _encode_fixed(self.origination_date, 10),
        _encode_fixed(self.origination_time, 8),
        self.time_reference & 0xFFFFFFFF,
        (self.time_reference >> 32) & 0xFFFFFFFF,
        self.version,
        umid,
        *loudness,
        b"\x00" * 180,
    )
    return header + self.coding_history.encode("latin-1")

write_to_parser

write_to_parser(parser)

Write this metadata into parser as the bext chunk.

Replaces an existing bext chunk or appends one. Call parser.write_wav(...) afterwards to persist to disk.

Source code in src/riffy/metadata/bext.py

def write_to_parser(self, parser: WAVParser) -> None:
    """Write this metadata into ``parser`` as the ``bext`` chunk.

    Replaces an existing ``bext`` chunk or appends one. Call
    ``parser.write_wav(...)`` afterwards to persist to disk.
    """
    payload = self.to_chunk_bytes()
    existing = parser.chunks.get(CHUNK_ID)
    if existing:
        existing[0] = WAVChunk(
            id=CHUNK_ID, size=len(payload), data=payload, offset=existing[0].offset
        )
    else:
        parser.add_chunk(CHUNK_ID, payload)

WAMD metadata: read and write the Wildlife Acoustics wamd chunk.

WAMD (Wildlife Acoustics Metadata) is the vendor format Song Meter recorders embed alongside — or, on older firmware, instead of — GUANO. Unlike GUANO's line-oriented text, WAMD is a packed binary stream of length-prefixed entries::

uint16 LE  id        # which field (see WAMD_IDS)
uint32 LE  length    # value length in bytes
length     value     # the field's raw bytes

Entries sit back-to-back with no inter-entry padding; odd-length values are followed by an explicit 0xFFFF alignment entry rather than an implicit pad byte. Most values are UTF-8 text; version and time_expansion are 16-bit integers, and a few (voice notes, program image, run-state) are opaque blobs.

Design highlights (mirroring the GUANO/bext modules):

• Round-trip safety. Entries are kept as an ordered (id, raw-bytes) list, so every untouched field — including alignment padding, duplicate ids, and blobs riffy does not interpret — re-serializes byte-for-byte. Only the field you edit changes.
• Typed accessors. Well-known fields are exposed by name; loc_position decodes the GPS waypoint to a (lat, lon) tuple. Everything else stays reachable through :meth:get_raw / :meth:set_raw.
• Fail soft. A length that overruns the chunk is clamped with a warning; non-UTF-8 text degrades to latin-1 rather than raising.

The junk variant: older firmware wrote the same binary stream under a junk chunk id. :meth:from_parser treats a junk chunk as WAMD only when it carries the unmistakable WAMD signature (a leading 2-byte version entry that parses cleanly to the end), so ordinary junk padding is never mistaken for metadata.

WamdMetadata

Typed, round-trip-safe view of a wamd chunk's WAMD metadata.

Construct from a file via :meth:from_parser / :meth:from_bytes, or build a fresh one with WamdMetadata(). Well-known fields are exposed as attributes; every entry, known or not, is reachable via :meth:get_raw / :meth:set_raw and :attr:entries.

Source code in src/riffy/metadata/wamd.py

class WamdMetadata:
    """Typed, round-trip-safe view of a ``wamd`` chunk's WAMD metadata.

    Construct from a file via :meth:`from_parser` / :meth:`from_bytes`, or build a
    fresh one with ``WamdMetadata()``. Well-known fields are exposed as
    attributes; every entry, known or not, is reachable via :meth:`get_raw` /
    :meth:`set_raw` and :attr:`entries`.
    """

    def __init__(self, chunk_id: str = CHUNK_ID) -> None:
        # Ordered (id, raw-value-bytes). A list — not a dict — because WAMD uses
        # repeated 0xFFFF alignment entries, and only a list preserves them (and
        # any other duplicate id) for a byte-exact round trip.
        self._entries: list[tuple[int, bytes]] = []
        #: The chunk id this metadata was read from (``"wamd"`` or a ``junk``
        #: variant); writes go back to the same chunk.
        self._chunk_id = chunk_id

    # ------------------------------------------------------------------ #
    # Construction
    # ------------------------------------------------------------------ #

    @classmethod
    def from_bytes(cls, data: bytes, chunk_id: str = CHUNK_ID) -> "WamdMetadata":
        """Parse WAMD metadata from a raw chunk payload.

        A length field that overruns the buffer is clamped (with a warning) so a
        truncated or malformed chunk degrades rather than raising.
        """
        self = cls(chunk_id=chunk_id)
        offset = 0
        n = len(data)
        while offset + _HEADER_SIZE <= n:
            tag, length = _ENTRY_HEADER.unpack_from(data, offset)
            start = offset + _HEADER_SIZE
            if start + length > n:
                warnings.warn(
                    f"wamd: entry 0x{tag:04x} claims {length} bytes but only "
                    f"{n - start} remain; truncating",
                    stacklevel=2,
                )
                length = n - start
            self._entries.append((tag, data[start : start + length]))
            offset = start + length
        return self

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "WamdMetadata | None":
        """Parse WAMD metadata from a parser's ``wamd`` (or ``junk``) chunk.

        Prefers a real ``wamd`` chunk; failing that, a ``junk`` chunk is used
        only when it carries the WAMD signature. Returns ``None`` if neither is
        present.
        """
        data = parser.get_chunk_bytes(CHUNK_ID)
        if data is not None:
            return cls.from_bytes(data, chunk_id=CHUNK_ID)
        for junk_id in _JUNK_IDS:
            for chunk in parser.get_chunks(junk_id):
                if _looks_like_wamd(chunk.data):
                    return cls.from_bytes(chunk.data, chunk_id=junk_id)
        return None

    # ------------------------------------------------------------------ #
    # Generic entry access — preserves unknown/blob/padding entries
    # ------------------------------------------------------------------ #

    def get_raw(self, tag: int) -> bytes | None:
        """Return the raw value bytes for the first entry with ``tag``, or ``None``."""
        for entry_tag, value in self._entries:
            if entry_tag == tag:
                return value
        return None

    def set_raw(self, tag: int, value: bytes) -> None:
        """Set the raw value for ``tag``, replacing the first entry in place or appending."""
        for i, (entry_tag, _) in enumerate(self._entries):
            if entry_tag == tag:
                self._entries[i] = (tag, value)
                return
        self._entries.append((tag, value))

    def remove(self, tag: int) -> None:
        """Delete every entry with ``tag`` (no error if absent)."""
        self._entries = [entry for entry in self._entries if entry[0] != tag]

    def __contains__(self, tag: object) -> bool:
        return any(entry_tag == tag for entry_tag, _ in self._entries)

    @property
    def entries(self) -> list[tuple[int, bytes]]:
        """A copy of the full ordered ``(id, raw-bytes)`` entry list."""
        return list(self._entries)

    # ------------------------------------------------------------------ #
    # Text helpers
    # ------------------------------------------------------------------ #

    def get_text(self, tag: int) -> str | None:
        """Return the UTF-8 (fail-soft latin-1) decoded value for ``tag``, or ``None``."""
        value = self.get_raw(tag)
        if value is None:
            return None
        return decode_text(value, context="wamd")

    def set_text(self, tag: int, value: str) -> None:
        """Set ``tag`` to the UTF-8 encoding of ``value``."""
        self.set_raw(tag, value.encode("utf-8"))

    # ------------------------------------------------------------------ #
    # Well-known typed attributes
    # ------------------------------------------------------------------ #

    @property
    def version(self) -> int | None:
        """The WAMD format version (a 16-bit integer), or ``None`` if absent."""
        return self._get_uint16(VERSION_ID)

    @property
    def model(self) -> str | None:
        return self.get_text(MODEL_ID)

    @model.setter
    def model(self, value: str | None) -> None:
        self._set_or_remove_text(MODEL_ID, value)

    @property
    def serial(self) -> str | None:
        return self.get_text(SERIAL_ID)

    @serial.setter
    def serial(self, value: str | None) -> None:
        self._set_or_remove_text(SERIAL_ID, value)

    @property
    def firmware(self) -> str | None:
        return self.get_text(FIRMWARE_ID)

    @firmware.setter
    def firmware(self, value: str | None) -> None:
        self._set_or_remove_text(FIRMWARE_ID, value)

    @property
    def prefix(self) -> str | None:
        return self.get_text(PREFIX_ID)

    @prefix.setter
    def prefix(self, value: str | None) -> None:
        self._set_or_remove_text(PREFIX_ID, value)

    @property
    def timestamp(self) -> str | None:
        """The recording timestamp as the recorder wrote it (raw text)."""
        return self.get_text(TIMESTAMP_ID)

    @timestamp.setter
    def timestamp(self, value: str | None) -> None:
        self._set_or_remove_text(TIMESTAMP_ID, value)

    @property
    def notes(self) -> str | None:
        return self.get_text(NOTES_ID)

    @notes.setter
    def notes(self, value: str | None) -> None:
        self._set_or_remove_text(NOTES_ID, value)

    @property
    def time_expansion(self) -> int | None:
        """The time-expansion factor (a 16-bit integer), or ``None`` if absent."""
        return self._get_uint16(TIME_EXPANSION_ID)

    @property
    def loc_position(self) -> tuple[float, float] | None:
        """``(latitude, longitude)`` WGS84 tuple decoded from the GPS waypoint.

        Handles both Wildlife Acoustics dialects: the Song Meter
        ``datum, lat, N|S, lon, E|W`` form and the EMTouch signed
        ``datum, lat, lon`` form. Altitude, if present, is preserved on write
        but not surfaced here.
        """
        raw = self.get_text(GPSFIRST_ID)
        if raw is None:
            return None
        parsed = _parse_gps(raw)
        if parsed is None:
            return None
        return parsed[1], parsed[2]

    @loc_position.setter
    def loc_position(self, value: tuple[float, float] | None) -> None:
        if value is None:
            self.remove(GPSFIRST_ID)
            return
        lat, lon = value
        # Preserve the datum and altitude of any existing waypoint; only the
        # coordinates change. Emit the canonical EMTouch signed form.
        datum, altitude = "WGS84", None
        existing = self.get_text(GPSFIRST_ID)
        if existing is not None:
            parsed = _parse_gps(existing)
            if parsed is not None:
                datum, altitude = parsed[0], parsed[3]
        waypoint = f"{datum},{lat},{lon}"
        if altitude is not None:
            waypoint += f",{altitude}"
        self.set_text(GPSFIRST_ID, waypoint)

    # ------------------------------------------------------------------ #
    # Internal typed helpers
    # ------------------------------------------------------------------ #

    def _get_uint16(self, tag: int) -> int | None:
        value = self.get_raw(tag)
        if value is None:
            return None
        if len(value) != 2:
            warnings.warn(
                f"wamd: entry 0x{tag:04x} is not a 2-byte integer: {value!r}", stacklevel=2
            )
            return None
        return struct.unpack("<H", value)[0]

    def _set_or_remove_text(self, tag: int, value: str | None) -> None:
        if value is None:
            self.remove(tag)
        else:
            self.set_text(tag, value)

    # ------------------------------------------------------------------ #
    # Serialization
    # ------------------------------------------------------------------ #

    def to_chunk_bytes(self) -> bytes:
        """Serialize to a ``wamd`` chunk payload (entries back-to-back, no padding).

        The chunk-level even-byte padding is left to the writer, matching ``bext``
        and ``guan``; the pad byte lives outside the declared chunk size, so a
        re-read never mistakes it for a truncated entry.
        """
        out = bytearray()
        for tag, value in self._entries:
            out += _ENTRY_HEADER.pack(tag, len(value))
            out += value
        return bytes(out)

    def write_to_parser(self, parser: WAVParser) -> None:
        """Write this metadata back into ``parser`` under its source chunk id.

        Replaces the chunk it was read from (the ``wamd`` chunk, or the specific
        WAMD-bearing ``junk`` chunk) or appends a new ``wamd`` chunk. Call
        ``parser.write_wav(...)`` afterwards to persist to disk.
        """
        payload = self.to_chunk_bytes()
        occurrences = parser.chunks.get(self._chunk_id, [])
        target: int | None = 0
        if self._chunk_id != CHUNK_ID:
            # A junk variant: replace the specific occurrence that holds WAMD.
            target = next(
                (i for i, c in enumerate(occurrences) if _looks_like_wamd(c.data)),
                None,
            )
        if occurrences and target is not None:
            existing = occurrences[target]
            occurrences[target] = WAVChunk(
                id=self._chunk_id, size=len(payload), data=payload, offset=existing.offset
            )
        else:
            parser.add_chunk(self._chunk_id, payload)

    # ------------------------------------------------------------------ #
    # Inspection
    # ------------------------------------------------------------------ #

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe view: named fields decoded, blobs/unknowns as hex.

        Alignment padding is omitted. ``loc_position`` is added as a ``[lat, lon]``
        pair when the GPS waypoint parses.
        """
        out: dict[str, object] = {}
        for tag, value in self._entries:
            if tag == ALIGNMENT_ID:
                continue
            name = WAMD_IDS.get(tag, f"0x{tag:04x}")
            if tag in _UINT16_IDS and len(value) == 2:
                out[name] = struct.unpack("<H", value)[0]
            elif tag in _BLOB_IDS:
                out[name] = value.hex()
            else:
                out[name] = decode_text(value, context="wamd")
        position = self.loc_position
        if position is not None:
            out["loc_position"] = [position[0], position[1]]
        return out

entries property

entries

A copy of the full ordered (id, raw-bytes) entry list.

version property

version

The WAMD format version (a 16-bit integer), or None if absent.

timestamp property writable

timestamp

The recording timestamp as the recorder wrote it (raw text).

time_expansion property

time_expansion

The time-expansion factor (a 16-bit integer), or None if absent.

loc_position property writable

loc_position

(latitude, longitude) WGS84 tuple decoded from the GPS waypoint.

Handles both Wildlife Acoustics dialects: the Song Meter datum, lat, N|S, lon, E|W form and the EMTouch signed datum, lat, lon form. Altitude, if present, is preserved on write but not surfaced here.

from_bytes classmethod

from_bytes(data, chunk_id=CHUNK_ID)

Parse WAMD metadata from a raw chunk payload.

A length field that overruns the buffer is clamped (with a warning) so a truncated or malformed chunk degrades rather than raising.

Source code in src/riffy/metadata/wamd.py

@classmethod
def from_bytes(cls, data: bytes, chunk_id: str = CHUNK_ID) -> "WamdMetadata":
    """Parse WAMD metadata from a raw chunk payload.

    A length field that overruns the buffer is clamped (with a warning) so a
    truncated or malformed chunk degrades rather than raising.
    """
    self = cls(chunk_id=chunk_id)
    offset = 0
    n = len(data)
    while offset + _HEADER_SIZE <= n:
        tag, length = _ENTRY_HEADER.unpack_from(data, offset)
        start = offset + _HEADER_SIZE
        if start + length > n:
            warnings.warn(
                f"wamd: entry 0x{tag:04x} claims {length} bytes but only "
                f"{n - start} remain; truncating",
                stacklevel=2,
            )
            length = n - start
        self._entries.append((tag, data[start : start + length]))
        offset = start + length
    return self

from_parser classmethod

from_parser(parser)

Parse WAMD metadata from a parser's wamd (or junk) chunk.

Prefers a real wamd chunk; failing that, a junk chunk is used only when it carries the WAMD signature. Returns None if neither is present.

Source code in src/riffy/metadata/wamd.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "WamdMetadata | None":
    """Parse WAMD metadata from a parser's ``wamd`` (or ``junk``) chunk.

    Prefers a real ``wamd`` chunk; failing that, a ``junk`` chunk is used
    only when it carries the WAMD signature. Returns ``None`` if neither is
    present.
    """
    data = parser.get_chunk_bytes(CHUNK_ID)
    if data is not None:
        return cls.from_bytes(data, chunk_id=CHUNK_ID)
    for junk_id in _JUNK_IDS:
        for chunk in parser.get_chunks(junk_id):
            if _looks_like_wamd(chunk.data):
                return cls.from_bytes(chunk.data, chunk_id=junk_id)
    return None

get_raw

get_raw(tag)

Return the raw value bytes for the first entry with tag, or None.

Source code in src/riffy/metadata/wamd.py

def get_raw(self, tag: int) -> bytes | None:
    """Return the raw value bytes for the first entry with ``tag``, or ``None``."""
    for entry_tag, value in self._entries:
        if entry_tag == tag:
            return value
    return None

set_raw

set_raw(tag, value)

Set the raw value for tag, replacing the first entry in place or appending.

Source code in src/riffy/metadata/wamd.py

def set_raw(self, tag: int, value: bytes) -> None:
    """Set the raw value for ``tag``, replacing the first entry in place or appending."""
    for i, (entry_tag, _) in enumerate(self._entries):
        if entry_tag == tag:
            self._entries[i] = (tag, value)
            return
    self._entries.append((tag, value))

remove

remove(tag)

Delete every entry with tag (no error if absent).

Source code in src/riffy/metadata/wamd.py

def remove(self, tag: int) -> None:
    """Delete every entry with ``tag`` (no error if absent)."""
    self._entries = [entry for entry in self._entries if entry[0] != tag]

get_text

get_text(tag)

Return the UTF-8 (fail-soft latin-1) decoded value for tag, or None.

Source code in src/riffy/metadata/wamd.py

def get_text(self, tag: int) -> str | None:
    """Return the UTF-8 (fail-soft latin-1) decoded value for ``tag``, or ``None``."""
    value = self.get_raw(tag)
    if value is None:
        return None
    return decode_text(value, context="wamd")

set_text

set_text(tag, value)

Set tag to the UTF-8 encoding of value.

Source code in src/riffy/metadata/wamd.py

def set_text(self, tag: int, value: str) -> None:
    """Set ``tag`` to the UTF-8 encoding of ``value``."""
    self.set_raw(tag, value.encode("utf-8"))

to_chunk_bytes

to_chunk_bytes()

Serialize to a wamd chunk payload (entries back-to-back, no padding).

The chunk-level even-byte padding is left to the writer, matching bext and guan; the pad byte lives outside the declared chunk size, so a re-read never mistakes it for a truncated entry.

Source code in src/riffy/metadata/wamd.py

def to_chunk_bytes(self) -> bytes:
    """Serialize to a ``wamd`` chunk payload (entries back-to-back, no padding).

    The chunk-level even-byte padding is left to the writer, matching ``bext``
    and ``guan``; the pad byte lives outside the declared chunk size, so a
    re-read never mistakes it for a truncated entry.
    """
    out = bytearray()
    for tag, value in self._entries:
        out += _ENTRY_HEADER.pack(tag, len(value))
        out += value
    return bytes(out)

write_to_parser

write_to_parser(parser)

Write this metadata back into parser under its source chunk id.

Replaces the chunk it was read from (the wamd chunk, or the specific WAMD-bearing junk chunk) or appends a new wamd chunk. Call parser.write_wav(...) afterwards to persist to disk.

Source code in src/riffy/metadata/wamd.py

def write_to_parser(self, parser: WAVParser) -> None:
    """Write this metadata back into ``parser`` under its source chunk id.

    Replaces the chunk it was read from (the ``wamd`` chunk, or the specific
    WAMD-bearing ``junk`` chunk) or appends a new ``wamd`` chunk. Call
    ``parser.write_wav(...)`` afterwards to persist to disk.
    """
    payload = self.to_chunk_bytes()
    occurrences = parser.chunks.get(self._chunk_id, [])
    target: int | None = 0
    if self._chunk_id != CHUNK_ID:
        # A junk variant: replace the specific occurrence that holds WAMD.
        target = next(
            (i for i, c in enumerate(occurrences) if _looks_like_wamd(c.data)),
            None,
        )
    if occurrences and target is not None:
        existing = occurrences[target]
        occurrences[target] = WAVChunk(
            id=self._chunk_id, size=len(payload), data=payload, offset=existing.offset
        )
    else:
        parser.add_chunk(self._chunk_id, payload)

to_dict

to_dict()

Return a JSON-safe view: named fields decoded, blobs/unknowns as hex.

Alignment padding is omitted. loc_position is added as a [lat, lon] pair when the GPS waypoint parses.

Source code in src/riffy/metadata/wamd.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe view: named fields decoded, blobs/unknowns as hex.

    Alignment padding is omitted. ``loc_position`` is added as a ``[lat, lon]``
    pair when the GPS waypoint parses.
    """
    out: dict[str, object] = {}
    for tag, value in self._entries:
        if tag == ALIGNMENT_ID:
            continue
        name = WAMD_IDS.get(tag, f"0x{tag:04x}")
        if tag in _UINT16_IDS and len(value) == 2:
            out[name] = struct.unpack("<H", value)[0]
        elif tag in _BLOB_IDS:
            out[name] = value.hex()
        else:
            out[name] = decode_text(value, context="wamd")
    position = self.loc_position
    if position is not None:
        out["loc_position"] = [position[0], position[1]]
    return out

AudioMoth comment-string decoder.

AudioMoth recorders pack their metadata into a single free-text comment stored in the RIFF INFO ICMT field (with the device string in IART), in an undocumented format that drifts between firmware versions. A typical string::

Recorded at 19:30:00 12/11/2021 (UTC) by AudioMoth 248D9B045EC9EE79 at
medium gain while battery was 4.1V and temperature was 14.0C.

Rather than switch on firmware version (the clause order and wording differ between the metamoth reference parser and the AudioMoth firmware source, and change repeatedly across versions), this decoder extracts each field with its own tolerant, independently-anchored regex. That makes it partial: if one field (say the timestamp) fails to match, every other field that did match is still returned, rather than discarding the whole string — a concrete pain point reported against existing parsers.

Format knowledge here is derived from the metamoth library (github.com/mbsantiago/metamoth, src/metamoth/parsing.py) and the AudioMoth firmware source (github.com/OpenAcousticDevices/AudioMoth-Firmware-Basic, setHeaderComment in src/main.c). Parsing is best-effort and firmware-dependent by nature.

Known limitation: the firmware 1.8+ Frequency trigger (...) clause is not decoded into a dedicated field (no real sample was available to validate it against). It is left in the raw comment string for downstream consumers, and it is deliberately not mistaken for a frequency filter.

AudioMothMetadata dataclass

Structured, best-effort view of an AudioMoth comment string.

Every field is optional: an attribute is left unset when the corresponding clause is absent or unparseable, so a partially-recognized comment still yields whatever could be extracted.

Source code in src/riffy/metadata/audiomoth.py

@dataclass
class AudioMothMetadata:
    """Structured, best-effort view of an AudioMoth comment string.

    Every field is optional: an attribute is left unset when the corresponding
    clause is absent or unparseable, so a partially-recognized comment still
    yields whatever could be extracted.
    """

    comment: str | None = None
    artist: str | None = None
    timestamp: datetime | None = None
    device_id: str | None = None
    deployment_id: str | None = None
    gain: str | None = None
    #: The numeric voltage figure mentioned in the comment. For a threshold
    #: reading ("less than 2.5V") this is the threshold, not a measurement; the
    #: full phrase is preserved in ``battery_state``.
    battery_voltage: float | None = None
    battery_state: str | None = None
    temperature_c: float | None = None
    external_microphone: bool = False
    amplitude_threshold: str | None = None
    min_trigger_duration_s: int | None = None
    filter_type: str | None = None
    filter_frequencies_khz: list[float] = field(default_factory=list)
    recording_stopped_reason: str | None = None

    # ------------------------------------------------------------------ #
    # Construction
    # ------------------------------------------------------------------ #

    @classmethod
    def from_comment(cls, comment: str, artist: str | None = None) -> "AudioMothMetadata":
        """Decode an AudioMoth comment string (and optional artist string).

        Extraction is partial: each field is parsed independently, so a failure
        in one clause never prevents the others from being returned.
        """
        self = cls(comment=comment, artist=artist)

        self.timestamp = _parse_timestamp(comment)

        device = _DEVICE_RE.search(comment)
        if device:
            self.device_id = device.group(1)
        deployment = _DEPLOYMENT_RE.search(comment)
        if deployment:
            self.deployment_id = deployment.group(1)

        self.gain = _parse_gain(comment)

        battery = _BATTERY_RE.search(comment)
        if battery:
            self.battery_state = battery.group(1)
            self.battery_voltage = float(battery.group(2))

        temperature = _TEMPERATURE_RE.search(comment)
        if temperature:
            self.temperature_c = float(temperature.group(1))

        self.external_microphone = _EXTERNAL_MIC_RE.search(comment) is not None

        threshold = _THRESHOLD_RE.search(comment)
        if threshold:
            self.amplitude_threshold = threshold.group(1).strip()
        duration = _TRIGGER_DURATION_RE.search(comment)
        if duration:
            self.min_trigger_duration_s = int(duration.group(1))

        filter_match = _FILTER_RE.search(comment)
        if filter_match:
            self.filter_type = filter_match.group(1)
            self.filter_frequencies_khz = [
                float(f) for f in _FILTER_FREQ_RE.findall(filter_match.group(0))
            ]

        stopped = _STOPPED_RE.search(comment)
        if stopped:
            self.recording_stopped_reason = stopped.group(1).strip()

        return self

    @classmethod
    def from_info(cls, info: InfoMetadata) -> "AudioMothMetadata | None":
        """Decode from a parsed :class:`~riffy.metadata.InfoMetadata`.

        Reads the comment from ``ICMT`` and the device string from ``IART``.
        Returns ``None`` if the INFO block shows no sign of being AudioMoth's.
        """
        comment = info.comment
        artist = info.artist
        if not _looks_like_audiomoth(comment, artist):
            return None
        return cls.from_comment(comment or "", artist)

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "AudioMothMetadata | None":
        """Decode from a parser's INFO block, or ``None`` if not AudioMoth."""
        info = InfoMetadata.from_parser(parser)
        if info is None:
            return None
        return cls.from_info(info)

    # ------------------------------------------------------------------ #
    # Within-standard normalization
    # ------------------------------------------------------------------ #

    def to_guano(self) -> GuanoMetadata:
        """Map the extracted fields onto GUANO-equivalent keys.

        A within-standard convenience (not cross-standard reconciliation): it
        normalizes an AudioMoth recording into the same shape as a GUANO-native
        one, so downstream code can treat them uniformly. Device-specific fields
        with no GUANO well-known equivalent go under the ``AudioMoth`` namespace.
        """
        g = GuanoMetadata(version="1.0")
        g.make = "Open Acoustic Devices"
        g.model = "AudioMoth"
        if self.device_id is not None:
            g.serial = self.device_id
        if self.timestamp is not None:
            g.timestamp = self.timestamp
        if self.temperature_c is not None:
            g.temperature_int = self.temperature_c
        if self.gain is not None:
            g.set("AudioMoth", "Gain", self.gain)
        if self.battery_voltage is not None:
            g.set("AudioMoth", "Battery Voltage", f"{self.battery_voltage}")
        if self.deployment_id is not None:
            g.set("AudioMoth", "Deployment ID", self.deployment_id)
        return g

from_comment classmethod

from_comment(comment, artist=None)

Decode an AudioMoth comment string (and optional artist string).

Extraction is partial: each field is parsed independently, so a failure in one clause never prevents the others from being returned.

Source code in src/riffy/metadata/audiomoth.py

@classmethod
def from_comment(cls, comment: str, artist: str | None = None) -> "AudioMothMetadata":
    """Decode an AudioMoth comment string (and optional artist string).

    Extraction is partial: each field is parsed independently, so a failure
    in one clause never prevents the others from being returned.
    """
    self = cls(comment=comment, artist=artist)

    self.timestamp = _parse_timestamp(comment)

    device = _DEVICE_RE.search(comment)
    if device:
        self.device_id = device.group(1)
    deployment = _DEPLOYMENT_RE.search(comment)
    if deployment:
        self.deployment_id = deployment.group(1)

    self.gain = _parse_gain(comment)

    battery = _BATTERY_RE.search(comment)
    if battery:
        self.battery_state = battery.group(1)
        self.battery_voltage = float(battery.group(2))

    temperature = _TEMPERATURE_RE.search(comment)
    if temperature:
        self.temperature_c = float(temperature.group(1))

    self.external_microphone = _EXTERNAL_MIC_RE.search(comment) is not None

    threshold = _THRESHOLD_RE.search(comment)
    if threshold:
        self.amplitude_threshold = threshold.group(1).strip()
    duration = _TRIGGER_DURATION_RE.search(comment)
    if duration:
        self.min_trigger_duration_s = int(duration.group(1))

    filter_match = _FILTER_RE.search(comment)
    if filter_match:
        self.filter_type = filter_match.group(1)
        self.filter_frequencies_khz = [
            float(f) for f in _FILTER_FREQ_RE.findall(filter_match.group(0))
        ]

    stopped = _STOPPED_RE.search(comment)
    if stopped:
        self.recording_stopped_reason = stopped.group(1).strip()

    return self

from_info classmethod

from_info(info)

Decode from a parsed :class:~riffy.metadata.InfoMetadata.

Reads the comment from ICMT and the device string from IART. Returns None if the INFO block shows no sign of being AudioMoth's.

Source code in src/riffy/metadata/audiomoth.py

@classmethod
def from_info(cls, info: InfoMetadata) -> "AudioMothMetadata | None":
    """Decode from a parsed :class:`~riffy.metadata.InfoMetadata`.

    Reads the comment from ``ICMT`` and the device string from ``IART``.
    Returns ``None`` if the INFO block shows no sign of being AudioMoth's.
    """
    comment = info.comment
    artist = info.artist
    if not _looks_like_audiomoth(comment, artist):
        return None
    return cls.from_comment(comment or "", artist)

from_parser classmethod

from_parser(parser)

Decode from a parser's INFO block, or None if not AudioMoth.

Source code in src/riffy/metadata/audiomoth.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "AudioMothMetadata | None":
    """Decode from a parser's INFO block, or ``None`` if not AudioMoth."""
    info = InfoMetadata.from_parser(parser)
    if info is None:
        return None
    return cls.from_info(info)

to_guano

to_guano()

Map the extracted fields onto GUANO-equivalent keys.

A within-standard convenience (not cross-standard reconciliation): it normalizes an AudioMoth recording into the same shape as a GUANO-native one, so downstream code can treat them uniformly. Device-specific fields with no GUANO well-known equivalent go under the AudioMoth namespace.

Source code in src/riffy/metadata/audiomoth.py

def to_guano(self) -> GuanoMetadata:
    """Map the extracted fields onto GUANO-equivalent keys.

    A within-standard convenience (not cross-standard reconciliation): it
    normalizes an AudioMoth recording into the same shape as a GUANO-native
    one, so downstream code can treat them uniformly. Device-specific fields
    with no GUANO well-known equivalent go under the ``AudioMoth`` namespace.
    """
    g = GuanoMetadata(version="1.0")
    g.make = "Open Acoustic Devices"
    g.model = "AudioMoth"
    if self.device_id is not None:
        g.serial = self.device_id
    if self.timestamp is not None:
        g.timestamp = self.timestamp
    if self.temperature_c is not None:
        g.temperature_int = self.temperature_c
    if self.gain is not None:
        g.set("AudioMoth", "Gain", self.gain)
    if self.battery_voltage is not None:
        g.set("AudioMoth", "Battery Voltage", f"{self.battery_voltage}")
    if self.deployment_id is not None:
        g.set("AudioMoth", "Deployment ID", self.deployment_id)
    return g

iXML metadata: read the iXML chunk (stretch goal, read-only).

iXML is a UTF-8 XML document embedded in an iXML chunk, used by production recorders (Sound Devices, Zoom, ...) to carry structured take/scene/track metadata. This module parses it into a nested dict / element tree for inspection. Authoring iXML is out of scope for v0.3.0.

Security note: parsing uses the standard library xml.etree.ElementTree, which does not fetch external entities but is not hardened against maliciously crafted XML (e.g. entity-expansion attacks). iXML from field recorders is effectively trusted input; do not point this at hostile XML without your own hardening (riffy keeps the zero-dependency promise, so it does not bundle defusedxml).

IXmlMetadata

Read-only view of an iXML chunk's XML document.

Source code in src/riffy/metadata/ixml.py

class IXmlMetadata:
    """Read-only view of an ``iXML`` chunk's XML document."""

    def __init__(self, root: Element) -> None:
        #: The parsed XML root element (``xml.etree.ElementTree.Element``).
        self.root = root

    @classmethod
    def from_bytes(cls, data: bytes) -> "IXmlMetadata":
        """Parse iXML from a raw ``iXML`` chunk payload.

        Raises:
            xml.etree.ElementTree.ParseError: If the payload is not valid XML.
        """
        # iXML is UTF-8 and commonly NUL-padded to an even length; trim padding
        # and surrounding whitespace before handing it to the XML parser.
        text = decode_text(data, context="iXML").strip("\x00 \t\r\n")
        return cls(fromstring(text))

    @classmethod
    def from_parser(cls, parser: WAVParser) -> "IXmlMetadata | None":
        """Parse the ``iXML`` chunk from a parser.

        Returns ``None`` if the file has no ``iXML`` chunk, or if the chunk is
        present but does not contain parseable XML (a warning is emitted).
        """
        raw = parser.get_chunk_bytes(CHUNK_ID)
        if raw is None:
            return None
        try:
            return cls.from_bytes(raw)
        except ParseError:
            warnings.warn("iXML: chunk does not contain valid XML; skipping", stacklevel=2)
            return None

    def find(self, path: str) -> str | None:
        """Return the text of the first element matching an ElementTree ``path``."""
        return self.root.findtext(path)

    def to_dict(self) -> dict[str, object]:
        """Return the document as a nested dict keyed by the root tag.

        Leaf elements become their stripped text; repeated child tags become a
        list of values.
        """
        return {self.root.tag: _element_to_dict(self.root)}

from_bytes classmethod

from_bytes(data)

Parse iXML from a raw iXML chunk payload.

Raises:

Type	Description
ParseError	If the payload is not valid XML.

Source code in src/riffy/metadata/ixml.py

@classmethod
def from_bytes(cls, data: bytes) -> "IXmlMetadata":
    """Parse iXML from a raw ``iXML`` chunk payload.

    Raises:
        xml.etree.ElementTree.ParseError: If the payload is not valid XML.
    """
    # iXML is UTF-8 and commonly NUL-padded to an even length; trim padding
    # and surrounding whitespace before handing it to the XML parser.
    text = decode_text(data, context="iXML").strip("\x00 \t\r\n")
    return cls(fromstring(text))

from_parser classmethod

from_parser(parser)

Parse the iXML chunk from a parser.

Returns None if the file has no iXML chunk, or if the chunk is present but does not contain parseable XML (a warning is emitted).

Source code in src/riffy/metadata/ixml.py

@classmethod
def from_parser(cls, parser: WAVParser) -> "IXmlMetadata | None":
    """Parse the ``iXML`` chunk from a parser.

    Returns ``None`` if the file has no ``iXML`` chunk, or if the chunk is
    present but does not contain parseable XML (a warning is emitted).
    """
    raw = parser.get_chunk_bytes(CHUNK_ID)
    if raw is None:
        return None
    try:
        return cls.from_bytes(raw)
    except ParseError:
        warnings.warn("iXML: chunk does not contain valid XML; skipping", stacklevel=2)
        return None

find

find(path)

Return the text of the first element matching an ElementTree path.

Source code in src/riffy/metadata/ixml.py

def find(self, path: str) -> str | None:
    """Return the text of the first element matching an ElementTree ``path``."""
    return self.root.findtext(path)

to_dict

to_dict()

Return the document as a nested dict keyed by the root tag.

Leaf elements become their stripped text; repeated child tags become a list of values.

Source code in src/riffy/metadata/ixml.py

def to_dict(self) -> dict[str, object]:
    """Return the document as a nested dict keyed by the root tag.

    Leaf elements become their stripped text; repeated child tags become a
    list of values.
    """
    return {self.root.tag: _element_to_dict(self.root)}

Diffing

Compare two WAV files for verification and validation.

riffy.diff(a, b) reports the differences between two files at two levels:

• Chunks — which RIFF chunks were added, removed, or changed (by raw bytes), or left unchanged. Multi-occurrence IDs are compared occurrence by occurrence.
• Metadata fields — a decoded, per-standard diff (GUANO, RIFF INFO, Broadcast Wave bext) so you can confirm exactly which fields changed — e.g. that a batch edit touched only Loc Position and left everything else intact.

Both files are read fully into memory (as the parser already does), so diffing a pair of very large recordings uses memory for both at once.

WavDiff dataclass

The structured difference between two WAV files.

Source code in src/riffy/diff.py

@dataclass
class WavDiff:
    """The structured difference between two WAV files."""

    path_a: str
    path_b: str
    form_a: str  # RIFF | RF64 | BW64
    form_b: str
    chunks: list[ChunkDelta] = field(default_factory=list)
    fields: list[FieldDelta] = field(default_factory=list)

    @property
    def changed_chunks(self) -> list[ChunkDelta]:
        """Chunk deltas that are not ``unchanged``."""
        return [c for c in self.chunks if c.status != UNCHANGED]

    @property
    def identical(self) -> bool:
        """True when the two files have the same form and no chunk differs."""
        return self.form_a == self.form_b and not self.changed_chunks

changed_chunks property

changed_chunks

Chunk deltas that are not unchanged.

identical property

identical

True when the two files have the same form and no chunk differs.

ChunkDelta dataclass

One chunk occurrence's difference between file A and file B.

Source code in src/riffy/diff.py

@dataclass
class ChunkDelta:
    """One chunk occurrence's difference between file A and file B."""

    chunk_id: str
    index: int  # occurrence index within this chunk ID (0 for the common case)
    status: str  # added | removed | changed | unchanged
    size_a: int | None  # None when absent from A
    size_b: int | None  # None when absent from B

FieldDelta dataclass

One decoded metadata field's difference between file A and file B.

Source code in src/riffy/diff.py

@dataclass
class FieldDelta:
    """One decoded metadata field's difference between file A and file B."""

    standard: str  # "guano" | "info" | "bext" | "wamd"
    key: str
    status: str  # added | removed | changed
    old: str | None  # value in A (None when absent)
    new: str | None  # value in B (None when absent)

diff

diff(a, b, *, include_unchanged=False)

Compare two WAV files (paths or parsers) and return a :class:WavDiff.

Parameters:

Name	Type	Description	Default
a	str | Path | WAVParser	The first file — a path or an already-parsed :class:~riffy.WAVParser.	required
b	str | Path | WAVParser	The second file.	required
include_unchanged	bool	If True, include unchanged chunk deltas too (default: only report chunks that differ).	False

Source code in src/riffy/diff.py

def diff(
    a: "str | Path | WAVParser",
    b: "str | Path | WAVParser",
    *,
    include_unchanged: bool = False,
) -> WavDiff:
    """Compare two WAV files (paths or parsers) and return a :class:`WavDiff`.

    Args:
        a: The first file — a path or an already-parsed :class:`~riffy.WAVParser`.
        b: The second file.
        include_unchanged: If True, include ``unchanged`` chunk deltas too
            (default: only report chunks that differ).
    """
    pa = a if isinstance(a, WAVParser) else WAVParser(a)
    pb = b if isinstance(b, WAVParser) else WAVParser(b)
    return WavDiff(
        path_a=str(pa.file_path),
        path_b=str(pb.file_path),
        form_a=pa.riff_form,
        form_b=pb.riff_form,
        chunks=_diff_chunks(pa, pb, include_unchanged),
        fields=_diff_metadata(pa, pb),
    )

Exceptions

Custom exceptions for the riffy RIFF parser library.

RiffyError

Bases: Exception

Base exception for all riffy errors.

Source code in src/riffy/exceptions.py

class RiffyError(Exception):
    """Base exception for all riffy errors."""

    pass

WAVError

Bases: RiffyError

Base exception for WAV-related errors.

Source code in src/riffy/exceptions.py

class WAVError(RiffyError):
    """Base exception for WAV-related errors."""

    pass

InvalidWAVFormatError

Bases: WAVError

Raised when a WAV file has an invalid format.

Source code in src/riffy/exceptions.py

class InvalidWAVFormatError(WAVError):
    """Raised when a WAV file has an invalid format."""

    pass

CorruptedFileError

Bases: WAVError

Raised when a file is corrupted or incomplete.

Source code in src/riffy/exceptions.py

class CorruptedFileError(WAVError):
    """Raised when a file is corrupted or incomplete."""

    pass

UnsupportedFormatError

Bases: WAVError

Raised when a WAV format is not supported.

Source code in src/riffy/exceptions.py

class UnsupportedFormatError(WAVError):
    """Raised when a WAV format is not supported."""

    pass

ChunkError

Bases: RiffyError

Base exception for RIFF chunk-related errors.

Source code in src/riffy/exceptions.py

class ChunkError(RiffyError):
    """Base exception for RIFF chunk-related errors."""

    pass

InvalidChunkError

Bases: ChunkError

Raised when a RIFF chunk is invalid.

Source code in src/riffy/exceptions.py

class InvalidChunkError(ChunkError):
    """Raised when a RIFF chunk is invalid."""

    pass

MissingChunkError

Bases: ChunkError

Raised when a required RIFF chunk is missing.

Source code in src/riffy/exceptions.py

class MissingChunkError(ChunkError):
    """Raised when a required RIFF chunk is missing."""

    pass