kaptive.core.genome¶

def __del__(self) -> None:
    r"""Clean up resources by closing the underlying binary handle if still open."""
    self._handle.close()

def __enter__(self) -> Self:
    r"""Enter the runtime context for the FASTA reader.

    Returns:
        FastaReader: The current reader context manager instance.
    """
    return self

def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    r"""Exit the runtime context and close the underlying binary stream.

    Args:
        exc_type (Any): Exception type if an exception was raised inside context.
        exc_val (Any): Exception instance if an exception was raised.
        exc_tb (Any): Traceback object if an exception was raised.
    """
    self._handle.close()

def __iter__(self) -> Self:
    r"""Return the iterator object itself.

    Returns:
        Iterator[SeqRecord]: Iterator yielding parsed [`SeqRecord`][kaptive.core.seq.SeqRecord] objects.
    """
    return self

def __next__(self) -> SeqRecord:
    r"""Fetch the next sequence record from the parsed FASTA stream.

    Returns:
        SeqRecord: The next FASTA record as a [`SeqRecord`][kaptive.core.seq.SeqRecord].

    Raises:
        StopIteration: When all FASTA records have been consumed.
    """
    return next(self._generator)

def __getitem__(self, item: str) -> bytes:
    r"""Retrieve raw sequence bytes for a contig by its identifier.

    Args:
        item (str): Contig identifier string.

    Returns:
        bytes: Contig sequence bytes.

    Raises:
        KeyError: If `item` is not a recognized contig identifier in `id_map`.
    """
    idx = self.id_map[item]
    offset_val = self.contigs.offsets[idx]
    length_val = self.contigs.lengths[idx]
    return self.contigs.seqs[offset_val : offset_val + length_val].tobytes()

def __iter__(self) -> Iterator[SeqRecord]:
    r"""Iterate over contigs in the assembly.

    Returns:
        Iterator[SeqRecord]: An iterator over [`SeqRecord`][kaptive.core.seq.SeqRecord] contigs.
    """
    return iter(self.contigs)

def __len__(self) -> int:
    r"""Calculate the total number of base pairs across all contigs in the assembly.

    Returns:
        int: Cumulative sequence length in base pairs.
    """
    return len(self.contigs.seqs)

def __post_init__(self) -> None:
    r"""Initialize derived fields such as `id_map` after dataclass creation."""
    object.__setattr__(self, "id_map", {name: i for i, name in enumerate(self.contigs.ids)})

def __str__(self) -> str:
    r"""Return the string representation of the assembly.

    Returns:
        str: Assembly identifier string (`id`).
    """
    return self.id

@classmethod
def ensure(cls, genome: Self | str | Path | IO[bytes]) -> Self:
    r"""Ensure the input object is coerced into a [`GenomeAssembly`][kaptive.core.genome.GenomeAssembly].

    Args:
        genome (Self | str | Path | IO[bytes]): Existing assembly object, file path,
            or binary FASTA stream.

    Returns:
        GenomeAssembly: Validated or loaded [`GenomeAssembly`][kaptive.core.genome.GenomeAssembly] instance.
    """
    if isinstance(genome, cls):
        return genome
    elif isinstance(genome, (str, Path)):
        return cls.from_file(genome)
    return cls.from_stream(genome)  # type: ignore

@classmethod
def from_file(cls, filepath: str | Path) -> Self:
    r"""Load a genome assembly from a FASTA file path.

    Supports plain `.fasta` / `.fa` / `.fna` files as well as compressed
    `.gz`, `.bz2`, and `.xz` files.

    Args:
        filepath (str | Path): Path to the target FASTA file.

    Returns:
        GenomeAssembly: Loaded [`GenomeAssembly`][kaptive.core.genome.GenomeAssembly] instance.

    Raises:
        NotImplementedError: If the file format or compression extension is unsupported.
    """
    filepath = Path(filepath)
    if not (m := cls._SEQUENCE_FILE_REGEX.search(filepath.name)):
        raise NotImplementedError(f"Unsupported format: {filepath}")

    with cls._OPENERS.get(m.group("compression"), open)(filepath, mode="rb") as handle:
        return cls.from_stream(handle, filepath.name.removesuffix(m.group()))

@classmethod
def from_records(cls, id_: str, records: Iterable[SeqRecord]) -> Self:
    r"""Construct a [`GenomeAssembly`][kaptive.core.genome.GenomeAssembly] from sequence records.

    Args:
        id_ (str): Genome assembly identifier.
        records (Iterable[SeqRecord]): Iterable of [`SeqRecord`][kaptive.core.seq.SeqRecord] objects.

    Returns:
        GenomeAssembly: Constructed [`GenomeAssembly`][kaptive.core.genome.GenomeAssembly] instance.
    """
    return cls(id_, Sequences.from_records(list(records)))

@classmethod
def from_stream(cls, handle: IO[bytes], id_: str | None = None) -> Self:
    r"""Load a genome assembly from an open binary stream.

    Args:
        handle (IO[bytes]): Open binary FASTA stream handle.
        id_ (str | None): Custom assembly identifier. If None, derived from
            handle name or defaults to `'unknown'`.

    Returns:
        GenomeAssembly: Loaded [`GenomeAssembly`][kaptive.core.genome.GenomeAssembly] instance.
    """
    with FastaReader(handle) as records:
        return cls.from_records(id_ or getattr(handle, "name", "unknown"), records)

def get_rammappy_index(self) -> Any:
    r"""Lazily build and return a thread-safe cached `rammappy.Index` for the assembly.

    Returns:
        Any: The compiled `rammappy.Index` instance.
    """
    if self.rammappy_index is None:
        with self._index_lock:
            if self.rammappy_index is None:
                import rammappy

                contig_seqs = [(c.id.encode(), c.seq) for c in self.contigs]
                idx = rammappy.Index.build(contig_seqs)
                object.__setattr__(self, "rammappy_index", idx)
    return self.rammappy_index