kaptive.core.pairwise¶

def __call__(self, queries: Sequences, targets: Sequences, seeds: Seeds | None = None) -> PairwiseAlignments:
    r"""Perform pairwise alignment of query and target sequences.

    Args:
        queries (Sequences): Collection of query sequences.
        targets (Sequences): Collection of target sequences (must match query batch size).
        seeds (Seeds | None): Optional alignment seeds guiding diagonal alignment.

    Returns:
        PairwiseAlignments: Alignment scores, statistics, and coordinates for each sequence pair.

    Raises:
        ValueError: If query and target batches have different numbers of sequences.
    """
    if len(queries.offsets) != len(targets.offsets):
        raise ValueError("Query and target batches must have the same number of sequences.")

    n = len(queries.offsets)
    if n == 0:
        return PairwiseAlignments.empty()

    # Handle the unified routing
    if seeds is not None:
        is_seeded = True
        offsets_arr = seeds.offsets
    else:
        is_seeded = False
        offsets_arr = np.zeros(n, dtype=np.int32)

    out_scores = np.empty(n, dtype=np.int32)
    out_matches = np.empty(n, dtype=np.int32)
    out_mismatches = np.empty(n, dtype=np.int32)
    out_gaps = np.empty(n, dtype=np.int32)
    out_q_starts = np.empty(n, dtype=np.int32)
    out_q_ends = np.empty(n, dtype=np.int32)
    out_t_starts = np.empty(n, dtype=np.int32)
    out_t_ends = np.empty(n, dtype=np.int32)

    _batched_banded_gotoh(
        queries.seqs,
        queries.offsets,
        queries.lengths,
        targets.seqs,
        targets.offsets,
        targets.lengths,
        _blosum62_matrix(),
        self.k,
        self.gap_open,
        self.gap_extend,
        is_seeded,
        offsets_arr,
        out_scores,
        out_matches,
        out_mismatches,
        out_gaps,
        out_q_starts,
        out_q_ends,
        out_t_starts,
        out_t_ends,
    )

    return PairwiseAlignments(
        out_scores,
        out_matches,
        out_mismatches,
        out_gaps,
        out_q_starts,
        out_q_ends,
        out_t_starts,
        out_t_ends,
    )

def align_seeds(self, queries: Sequences, targets: Sequences, seeds: Seeds) -> PairwiseAlignments:
    r"""Convenience method to extract and align specific sequence pairs mapped by seeds.

    Args:
        queries (Sequences): Full collection of query sequences.
        targets (Sequences): Full collection of target sequences.
        seeds (Seeds): Seed collection mapping specific query sequences to target sequences.

    Returns:
        PairwiseAlignments: Alignment results parallel to the provided seeds.
    """
    paired_queries, paired_targets = seeds.extract_sequences(queries, targets)
    return self(paired_queries, paired_targets, seeds)

def __getitem__(self, item: Any) -> PairwiseAlignment | PairwiseAlignments:
    r"""Access alignment results by index, slice, or boolean array mask.

    Args:
        item (Any): Integer index, slice, or boolean/integer array.

    Returns:
        PairwiseAlignment | PairwiseAlignments: Scalar record or sliced batch collection.

    Raises:
        IndexError: If integer index is out of range.
    """
    if isinstance(item, (int, np.integer)):
        if item < 0:
            item += len(self)
        if item < 0 or item >= len(self):
            raise IndexError("Batch index out of range")
        return PairwiseAlignment(
            score=int(self.scores[item]),
            matches=int(self.matches[item]),
            mismatches=int(self.mismatches[item]),
            gaps=int(self.gaps[item]),
            q_start=int(self.q_starts[item]),
            q_end=int(self.q_ends[item]),
            t_start=int(self.t_starts[item]),
            t_end=int(self.t_ends[item]),
        )
    return PairwiseAlignments(
        scores=self.scores[item],
        matches=self.matches[item],
        mismatches=self.mismatches[item],
        gaps=self.gaps[item],
        q_starts=self.q_starts[item],
        q_ends=self.q_ends[item],
        t_starts=self.t_starts[item],
        t_ends=self.t_ends[item],
    )

def __len__(self) -> int:
    r"""Return the number of alignments in the batch.

    Returns:
        int: Total count of alignments.
    """
    return len(self.scores)

@classmethod
def concat(cls, batches: Iterable[Self]) -> Self:  # type: ignore
    r"""Concatenate multiple PairwiseAlignments collections into a single batch.

    Args:
        batches (Iterable[PairwiseAlignments]): Iterable of alignment collections.

    Returns:
        PairwiseAlignments: Single concatenated alignment collection.
    """
    batches_list = list(batches)
    if not batches_list:
        return cls.empty()  # type: ignore
    return cls(
        scores=np.concatenate([b.scores for b in batches_list]),
        matches=np.concatenate([b.matches for b in batches_list]),
        mismatches=np.concatenate([b.mismatches for b in batches_list]),
        gaps=np.concatenate([b.gaps for b in batches_list]),
        q_starts=np.concatenate([b.q_starts for b in batches_list]),
        q_ends=np.concatenate([b.q_ends for b in batches_list]),
        t_starts=np.concatenate([b.t_starts for b in batches_list]),
        t_ends=np.concatenate([b.t_ends for b in batches_list]),
    )

@classmethod
def empty(cls) -> PairwiseAlignments:
    r"""Create an empty PairwiseAlignments collection with zero-length int32 arrays.

    Returns:
        PairwiseAlignments: Empty alignment collection.
    """
    return cls(
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
        np.empty(0, dtype=np.int32),
    )

@classmethod
def from_dict(cls, d: dict[str, Any]) -> PairwiseAlignments:
    r"""Deserialize a PairwiseAlignments batch from a dictionary of array-like data.

    Args:
        d (dict[str, Any]): Dictionary containing alignment field arrays.

    Returns:
        PairwiseAlignments: Deserialized pairwise alignments container.
    """
    return cls(
        np.array(d["scores"], dtype=np.int32),
        np.array(d["matches"], dtype=np.int32),
        np.array(d["mismatches"], dtype=np.int32),
        np.array(d["gaps"], dtype=np.int32),
        np.array(d["q_starts"], dtype=np.int32),
        np.array(d["q_ends"], dtype=np.int32),
        np.array(d["t_starts"], dtype=np.int32),
        np.array(d["t_ends"], dtype=np.int32),
    )

def to_dict(self) -> dict[str, npt.NDArray[np.int32]]:
    r"""Convert the alignment batch to a dictionary of NumPy arrays for serialization.

    Returns:
        dict[str, npt.NDArray[np.int32]]: Dictionary mapping attribute names to arrays.
    """
    return {
        "scores": self.scores,
        "matches": self.matches,
        "mismatches": self.mismatches,
        "gaps": self.gaps,
        "q_starts": self.q_starts,
        "q_ends": self.q_ends,
        "t_starts": self.t_starts,
        "t_ends": self.t_ends,
    }