Indexed gzip

Helpers for creating and detecting self-indexed .mzML.gz files. index_gzip is an alias of write_indexed_gzip.

mzmlpy.write_indexed_gzip

write_indexed_gzip(
    source: str | Path,
    output: str | Path,
    *,
    compression_level: int = 6
) -> IndexedGzipWriteResult

Create a deterministic, pyMZML-compatible self-indexed .mzML.gz file.

The input may be plain mzML or gzip-compressed mzML. Memory use depends on the largest XML section plus the embedded index. Gzip input is decompressed once to a temporary spool so the index can be sized before the output header is written.

Source code in src/mzmlpy/embedded_indexed_gzip.py

def write_indexed_gzip(
    source: str | Path,
    output: str | Path,
    *,
    compression_level: int = 6,
) -> IndexedGzipWriteResult:
    """Create a deterministic, pyMZML-compatible self-indexed ``.mzML.gz`` file.

    The input may be plain mzML or gzip-compressed mzML. Memory use depends on the largest
    XML section plus the embedded index. Gzip input is decompressed once to a temporary spool so
    the index can be sized before the output header is written.
    """
    if not -1 <= compression_level <= 9:
        raise MzmlError("compression_level must be between -1 and 9")
    output_path = str(output)
    if not output_path.lower().endswith((".gz", ".igz")):
        raise MzmlError("output path must end in .gz or .igz")

    with _decompressed_source(source) as plain_path:
        blocks: list[tuple[str, ...]] = []
        seen_aliases: set[str] = set()
        spectrum_count = 0
        chromatogram_count = 0
        with open(plain_path, "rb") as source_handler:
            for block in _iter_blocks(source_handler):
                aliases = []
                for alias in _aliases(block):
                    if "\x00" in alias:
                        raise MzmlOffsetIndexError("Embedded index identifiers cannot contain a null byte")
                    if alias in seen_aliases:
                        if alias.startswith(("s:", "c:", "si:", "ci:")):
                            raise MzmlError(f"Duplicate mzML identifier in input: {block.identifier}")
                        continue
                    seen_aliases.add(alias)
                    aliases.append(alias)
                blocks.append(tuple(aliases))
                spectrum_count += block.kind == "spectrum"
                chromatogram_count += block.kind == "chromatogram"

        encoded_aliases = [alias.encode("utf-8") for aliases in blocks for alias in aliases]
        identifier_width = max((len(alias) for alias in encoded_aliases), default=1)
        offset_width = 20
        if identifier_width > 255:
            raise MzmlOffsetIndexError("Embedded index identifiers cannot exceed 255 UTF-8 bytes")
        if len(encoded_aliases) > 10_000_000:
            raise MzmlOffsetIndexError("Embedded index contains too many entries")

        comment_size = 5 + len(encoded_aliases) * (identifier_width + offset_width)
        comment = bytearray(_FORMAT_MARKER + bytes((identifier_width, offset_width)))
        comment.extend(b"\x01" * (comment_size - len(comment)))

        index_entries: list[EmbeddedIndexEntry] = []
        with atomic_write_path(output_path) as temporary_output, open(temporary_output, "w+b") as output_handler:
            with open(plain_path, "rb") as source_handler:
                for block_number, block in enumerate(_iter_blocks(source_handler)):
                    payload_offset = _write_member(
                        output_handler,
                        block.data,
                        compression_level,
                        bytes(comment) if block_number == 0 else None,
                    )
                    for alias in blocks[block_number]:
                        index_entries.append(EmbeddedIndexEntry(alias, payload_offset))

            output_handler.seek(10 + len(_FORMAT_MARKER) + 2)
            for entry in index_entries:
                raw_identifier = entry.identifier.encode("utf-8")
                raw_offset = str(entry.offset).encode("ascii")
                if len(raw_offset) > offset_width:
                    raise MzmlOffsetIndexError("Compressed offset exceeds embedded index width")
                output_handler.write(raw_identifier.rjust(identifier_width, _PAD))
                output_handler.write(raw_offset.rjust(offset_width, _PAD))
            output_handler.flush()
            os.fsync(output_handler.fileno())

    return IndexedGzipWriteResult(
        output_path=Path(output_path),
        spectrum_count=spectrum_count,
        chromatogram_count=chromatogram_count,
        member_count=len(blocks),
        index_entry_count=len(index_entries),
    )

mzmlpy.is_embedded_indexed_gzip

is_embedded_indexed_gzip(path: str | Path) -> bool

Return whether a gzip file starts with the pyMZML FU version 1 marker.

Source code in src/mzmlpy/embedded_indexed_gzip.py

def is_embedded_indexed_gzip(path: str | Path) -> bool:
    """Return whether a gzip file starts with the pyMZML ``FU`` version 1 marker."""
    try:
        with open(path, "rb") as file_handler:
            if not _seek_comment(file_handler):
                return False
            return file_handler.read(len(_FORMAT_MARKER)) == _FORMAT_MARKER
    except (OSError, ValueError):
        return False

mzmlpy.IndexedGzipWriteResult dataclass

IndexedGzipWriteResult(
    output_path: Path,
    spectrum_count: int,
    chromatogram_count: int,
    member_count: int,
    index_entry_count: int,
)

Summary returned after creating a self-indexed gzip file.

AccessStrategy

The concrete storage route a reader chose, reported by Mzml.access_strategy.

mzmlpy.AccessStrategy

Bases: StrEnum

Concrete storage strategy selected for an mzML reader.