CrossIndex¶

CrossIndex manages sequences divided into named groups and computes cross-group pairwise comparisons. It is compatible with :class:~rusty_dot.dotplot.DotPlotter.

Workflow¶

Loading sequences and computing matches are separate explicit steps:

Load sequences with add_sequence() or load_fasta().
Call compute_matches() to compute k-mer matches between groups.
Call reorder_contigs() or reorder_for_colinearity() (requires step 2).
Plot with DotPlotter using query_group / target_group to specify groups.

Progress is logged at INFO level for each loading and computation step. Warnings are emitted when a sequence name already exists in the same or another group.

Alignment scope by number of groups¶

2 groups — compute_matches() compares the two groups.
3+ groups — compute_matches() computes all non-self ordered pairs by default. Use the query_group / target_group arguments to restrict to a specific pair.

Quick start¶

from rusty_dot.paf_io import CrossIndex
from rusty_dot.dotplot import DotPlotter

cross = CrossIndex(k=15)
cross.load_fasta("assembly_a.fasta", group="a")
cross.load_fasta("assembly_b.fasta", group="b")

# Explicitly compute k-mer matches (required before reorder_contigs)
cross.compute_matches()
print("Computed pairs:", cross.computed_group_pairs)

# Sort contigs for maximum collinearity
q_sorted, t_sorted = cross.reorder_contigs()

# Plot directly from CrossIndex — sequence names resolved via group params
plotter = DotPlotter(cross)
plotter.plot(
    query_group="a",   # sequences from group 'a' as rows
    target_group="b",  # sequences from group 'b' as columns
    output_path="cross_plot.png",
)

Custom group names¶

cross = CrossIndex(k=15)
cross.load_fasta("genome_a.fasta", group="Group_A")
cross.load_fasta("genome_b.fasta", group="Group_B")

cross.compute_matches()  # auto-detects the two groups
q_sorted, t_sorted = cross.reorder_contigs()

# Plot with explicit group names
plotter = DotPlotter(cross)
plotter.plot(
    query_group="Group_A",
    target_group="Group_B",
    output_path="cross_plot.png",
)

# Or rename groups and use explicitly
cross.rename_group("Group_A", "query")
cross.rename_group("Group_B", "target")
cross.compute_matches(query_group="query", target_group="target")
q_sorted, t_sorted = cross.reorder_contigs(query_group="query", target_group="target")

Class¶

`CrossIndex` ¶

Multi-group sequence index for cross-group pairwise comparisons.

Sequences are organised into named groups (e.g. 'assembly_a', 'assembly_b'). Each sequence is stored in a shared :class:~rusty_dot.SequenceIndex under a group:name internal key, which keeps names unique even when the same sequence identifier appears in multiple groups.

Workflow

Loading sequences and computing matches are separate, explicit steps:

Load sequences via :meth:add_sequence or :meth:load_fasta. Each sequence addition is logged at DEBUG level. A WARNING is emitted if the name already exists in the same group (FM-index overwritten) or in a different group.
Call :meth:compute_matches to find k-mer matches between groups. This must be done before calling :meth:reorder_contigs or :meth:reorder_for_colinearity.
Inspect :attr:computed_group_pairs to verify which pairs have been computed.

Alignment scope by number of groups

2 groups — :meth:compute_matches compares the two groups by default.
3+ groups — all non-self ordered pairs by default. Use the query_group / target_group arguments to restrict to a specific pair.

DotPlotter compatibility

CrossIndex exposes :meth:get_sequence_length, :meth:compare_sequences_stranded, and :meth:sequence_names so that it can be passed directly to :class:~rusty_dot.dotplot.DotPlotter::

cross = CrossIndex(k=15)
cross.load_fasta("assembly_a.fasta", group="a")
cross.load_fasta("assembly_b.fasta", group="b")
cross.compute_matches()

from rusty_dot.dotplot import DotPlotter
plotter = DotPlotter(cross)
plotter.plot(
    query_names=cross.sequence_names(group="a"),
    target_names=cross.sequence_names(group="b"),
    output_path="cross_plot.png",
)

Parameters:

Name	Type	Description	Default
`k`	`int`	K-mer length to use for indexing and comparison.	required

Examples:

>>> from rusty_dot.paf_io import CrossIndex
>>> cross = CrossIndex(k=10)
>>> cross.load_fasta("genome_a.fasta", group="a")
>>> cross.load_fasta("genome_b.fasta", group="b")
>>> cross.compute_matches()
>>> paf_lines = cross.get_paf()

Source code in rusty_dot/paf_io.py

class CrossIndex:
    """Multi-group sequence index for cross-group pairwise comparisons.

    Sequences are organised into named groups (e.g. ``'assembly_a'``,
    ``'assembly_b'``).  Each sequence is stored in a shared
    :class:`~rusty_dot.SequenceIndex` under a ``group:name`` internal key,
    which keeps names unique even when the same sequence identifier appears
    in multiple groups.

    **Workflow**

    Loading sequences and computing matches are separate, explicit steps:

    1. Load sequences via :meth:`add_sequence` or :meth:`load_fasta`.
       Each sequence addition is logged at ``DEBUG`` level.  A ``WARNING``
       is emitted if the name already exists in the same group (FM-index
       overwritten) or in a different group.
    2. Call :meth:`compute_matches` to find k-mer matches between groups.
       This must be done before calling :meth:`reorder_contigs` or
       :meth:`reorder_for_colinearity`.
    3. Inspect :attr:`computed_group_pairs` to verify which pairs have been
       computed.

    **Alignment scope by number of groups**

    * **2 groups** — :meth:`compute_matches` compares the two groups by
      default.
    * **3+ groups** — all non-self ordered pairs by default.  Use the
      *query_group* / *target_group* arguments to restrict to a specific pair.

    **DotPlotter compatibility**

    ``CrossIndex`` exposes :meth:`get_sequence_length`,
    :meth:`compare_sequences_stranded`, and :meth:`sequence_names` so that it
    can be passed directly to :class:`~rusty_dot.dotplot.DotPlotter`::

        cross = CrossIndex(k=15)
        cross.load_fasta("assembly_a.fasta", group="a")
        cross.load_fasta("assembly_b.fasta", group="b")
        cross.compute_matches()

        from rusty_dot.dotplot import DotPlotter
        plotter = DotPlotter(cross)
        plotter.plot(
            query_names=cross.sequence_names(group="a"),
            target_names=cross.sequence_names(group="b"),
            output_path="cross_plot.png",
        )

    Parameters
    ----------
    k : int
        K-mer length to use for indexing and comparison.

    Examples
    --------
    >>> from rusty_dot.paf_io import CrossIndex
    >>> cross = CrossIndex(k=10)
    >>> cross.load_fasta("genome_a.fasta", group="a")
    >>> cross.load_fasta("genome_b.fasta", group="b")
    >>> cross.compute_matches()
    >>> paf_lines = cross.get_paf()
    """

    def __init__(self, k: int) -> None:
        """Initialise an empty CrossIndex.

        Parameters
        ----------
        k : int
            K-mer length to use when building the sequence index.
        """
        self._k: int = k
        self._index: SequenceIndex = SequenceIndex(k=k)
        # group_label -> ordered list of original (un-prefixed) sequence names
        self._groups: dict[str, list[str]] = {}
        # group_label -> internal prefix used in _index (supports rename_group)
        self._internal_group: dict[str, str] = {}
        # (query_group, target_group) -> list[PafRecord] from compute_matches
        self._records_by_pair: dict[tuple[str, str], list[PafRecord]] = {}

    @property
    def _paf_records(self) -> list[PafRecord]:
        """All cached PAF records across every computed group pair (flat list).

        Note: this property rebuilds the list on every access.  Avoid calling
        it repeatedly in a tight loop; assign to a local variable instead.
        """
        result: list[PafRecord] = []
        for recs in self._records_by_pair.values():
            result.extend(recs)
        return result

    @property
    def computed_group_pairs(self) -> list[tuple[str, str]]:
        """Group pairs for which k-mer matches have been computed.

        Returns
        -------
        list[tuple[str, str]]
            List of ``(query_group, target_group)`` pairs that have had
            :meth:`compute_matches` run on them.  Use this to confirm that
            the required pair is ready before calling
            :meth:`reorder_contigs`.
        """
        return list(self._records_by_pair.keys())

    def get_records_for_pair(
        self, query_group: str, target_group: str
    ) -> list['PafRecord']:
        """Return the cached :class:`PafRecord` list for a computed group pair.

        Parameters
        ----------
        query_group : str
            Query group label.
        target_group : str
            Target group label.

        Returns
        -------
        list[PafRecord]
            Cached PAF records for the pair.  Returns an empty list if
            :meth:`compute_matches` has not been called for this pair.
        """
        return list(self._records_by_pair.get((query_group, target_group), []))

    def make_internal_name(self, group: str, name: str) -> str:
        """Construct the internal (``'group:name'``) identifier for a sequence.

        This is the public counterpart of the internal :meth:`_make_internal`
        helper and is suitable for use by external code such as
        :class:`~rusty_dot.dotplot.DotPlotter`.

        Parameters
        ----------
        group : str
            Group label.
        name : str
            Un-prefixed sequence name.

        Returns
        -------
        str
            Internal identifier in ``'group:name'`` form, with any
            ``rename_group`` remapping applied.
        """
        return self._make_internal(group, name)

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

    def _make_internal(self, group: str, name: str) -> str:
        """Format an internal (prefixed) name for use in SequenceIndex.

        Uses :attr:`_internal_group` to resolve the actual internal prefix,
        which may differ from *group* after a :meth:`rename_group` call.
        The ``get(group, group)`` fallback is safe because every call to
        :meth:`add_sequence` or :meth:`load_fasta` immediately registers the
        new group in ``_internal_group`` before any internal name is formed.
        """
        prefix = self._internal_group.get(group, group)
        return f'{prefix}:{name}'

    @staticmethod
    def _split_internal(internal: str) -> tuple[str, str]:
        """Split ``'group:name'`` into ``(group, name)``."""
        group, _, name = internal.partition(':')
        return group, name

    # ------------------------------------------------------------------
    # Adding sequences
    # ------------------------------------------------------------------

    def _check_name_collision(self, name: str, group: str) -> None:
        """Emit warnings if *name* already exists in the same or another group."""
        if name in self._groups.get(group, []):
            _log.warning(
                'CrossIndex: sequence %r already exists in group %r; '
                'its FM-index will be overwritten',
                name,
                group,
            )
        else:
            for other_g, other_names in self._groups.items():
                if other_g != group and name in other_names:
                    _log.warning(
                        'CrossIndex: sequence %r already exists in group %r; '
                        'adding the same name to group %r may cause confusion',
                        name,
                        other_g,
                        group,
                    )

    def add_sequence(self, name: str, seq: str, group: str = 'a') -> None:
        """Add a single sequence to the specified group.

        Logs a ``DEBUG``-level message for every sequence loaded, and a
        ``WARNING`` if *name* already exists in the same group (its FM-index
        will be overwritten) or in a different group (potential confusion).

        Parameters
        ----------
        name : str
            Sequence identifier (must be unique within the group).
        seq : str
            DNA sequence string.
        group : str, optional
            Group label.  Any non-empty string is accepted; ``':'`` is
            forbidden because it is used as an internal separator.
            Default is ``'a'``.

        Raises
        ------
        ValueError
            If *group* contains ``':'``.
        """
        if ':' in group:
            raise ValueError(f"Group name must not contain ':', got {group!r}")
        self._check_name_collision(name, group)
        _log.debug(
            'CrossIndex: adding sequence %r (len=%d) to group %r',
            name,
            len(seq),
            group,
        )
        if group not in self._groups:
            self._groups[group] = []
            self._internal_group[group] = group
        internal = self._make_internal(group, name)
        self._index.add_sequence(internal, seq)
        if name not in self._groups[group]:
            self._groups[group].append(name)

    def load_fasta(self, path: str, group: str = 'a') -> list[str]:
        """Load sequences from a FASTA file into the specified group.

        Logs an ``INFO``-level message when opening the file, a ``DEBUG``
        message for each sequence loaded (including sequence name and length),
        and a ``WARNING`` if any sequence name already exists in the same or
        another group.

        Parameters
        ----------
        path : str
            Path to a FASTA (``.fa`` / ``.fasta``) or gzipped FASTA file.
        group : str, optional
            Group label.  Default is ``'a'``.

        Returns
        -------
        list[str]
            The original (un-prefixed) sequence names that were loaded, in
            file order.

        Raises
        ------
        ValueError
            If *group* contains ``':'``, or if the file cannot be parsed, or
            if the FASTA file contains duplicate sequence names.
        """
        if ':' in group:
            raise ValueError(f"Group name must not contain ':', got {group!r}")
        from rusty_dot._rusty_dot import py_read_fasta

        _log.info('CrossIndex: loading sequences from %r into group %r', path, group)
        seqs = py_read_fasta(path)
        if group not in self._groups:
            self._groups[group] = []
            self._internal_group[group] = group
        names: list[str] = []
        for name, seq in seqs.items():
            self._check_name_collision(name, group)
            _log.debug(
                'CrossIndex: adding sequence %r (len=%d) to group %r',
                name,
                len(seq),
                group,
            )
            internal = self._make_internal(group, name)
            self._index.add_sequence(internal, seq)
            if name not in self._groups[group]:
                self._groups[group].append(name)
            names.append(name)
        _log.info(
            'CrossIndex: loaded %d sequence(s) from %r into group %r',
            len(names),
            path,
            group,
        )
        return names

    # ------------------------------------------------------------------
    # Properties and name helpers
    # ------------------------------------------------------------------

    @property
    def group_names(self) -> list[str]:
        """Return the list of group labels in insertion order.

        Returns
        -------
        list[str]
            Group labels.
        """
        return list(self._groups.keys())

    def sequence_names(self, group: str | None = None) -> list[str]:
        """Return internal (``group:name``) identifiers suitable for DotPlotter.

        Parameters
        ----------
        group : str or None, optional
            If given, return only names from that group.  If ``None``
            (default), return names from all groups.

        Returns
        -------
        list[str]
            Internal ``group:name`` strings in current :attr:`contig_order`.
        """
        if group is not None:
            return [self._make_internal(group, n) for n in self._groups.get(group, [])]
        result: list[str] = []
        for g, names in self._groups.items():
            result.extend(self._make_internal(g, n) for n in names)
        return result

    @property
    def contig_order(self) -> dict[str, list[str]]:
        """Current contig order per group as original (un-prefixed) names.

        Returns
        -------
        dict[str, list[str]]
            Mapping of group label → ordered list of sequence names.
            Updated in-place by :meth:`reorder_by_length` and
            :meth:`reorder_for_colinearity`.
        """
        return {g: list(names) for g, names in self._groups.items()}

    # ------------------------------------------------------------------
    # Backward-compatible properties (two-group a/b model)
    # ------------------------------------------------------------------

    @property
    def query_names(self) -> list[str]:
        """Un-prefixed names for group ``'a'`` (backward compatible).

        Returns
        -------
        list[str]
        """
        return list(self._groups.get('a', []))

    @property
    def target_names(self) -> list[str]:
        """Un-prefixed names for group ``'b'`` (backward compatible).

        Returns
        -------
        list[str]
        """
        return list(self._groups.get('b', []))

    # ------------------------------------------------------------------
    # DotPlotter-compatible interface
    # ------------------------------------------------------------------

    def get_sequence_length(self, name: str) -> int:
        """Return the length of the sequence identified by its internal name.

        Parameters
        ----------
        name : str
            Internal (``group:name``) identifier.

        Returns
        -------
        int
            Sequence length in bases.
        """
        return self._index.get_sequence_length(name)

    def compare_sequences_stranded(
        self, name1: str, name2: str, merge: bool = True
    ) -> list:
        """Compare two sequences by their internal names, returning stranded matches.

        Parameters
        ----------
        name1 : str
            Internal name of the query sequence.
        name2 : str
            Internal name of the target sequence.
        merge : bool, optional
            Whether to merge consecutive co-linear k-mer runs.
            Default is ``True``.

        Returns
        -------
        list of (int, int, int, int, str)
            List of ``(query_start, query_end, target_start, target_end, strand)``
            tuples.
        """
        return self._index.compare_sequences_stranded(name1, name2, merge)

    # ------------------------------------------------------------------
    # Contig reordering
    # ------------------------------------------------------------------

    def rename_group(self, old_name: str, new_name: str) -> None:
        """Rename a group label without re-indexing sequences.

        The internal prefix used in the underlying :class:`SequenceIndex` is
        preserved; only the public-facing group label is changed.

        Parameters
        ----------
        old_name : str
            Current group label to rename.
        new_name : str
            New group label.  Must not contain ``':'``.

        Raises
        ------
        KeyError
            If *old_name* is not a known group.
        ValueError
            If *new_name* contains ``':'`` or already exists as a group label.
        """
        if old_name not in self._groups:
            raise KeyError(f'Group {old_name!r} not found.')
        if ':' in new_name:
            raise ValueError(f"Group name must not contain ':', got {new_name!r}")
        if new_name in self._groups and new_name != old_name:
            raise ValueError(f'Group {new_name!r} already exists.')
        # Rebuild _groups preserving insertion order
        self._groups = {
            (new_name if k == old_name else k): v for k, v in self._groups.items()
        }
        # Update the internal-prefix mapping
        self._internal_group[new_name] = self._internal_group.pop(old_name)
        _log.info('CrossIndex: renamed group %r → %r', old_name, new_name)

    def set_group_members(self, group: str, names: list[str]) -> None:
        """Assign a custom list of sequence names to an existing group.

        Only the logical membership list is updated; sequences already indexed
        are not moved or removed from the underlying
        :class:`~rusty_dot.SequenceIndex`.

        Parameters
        ----------
        group : str
            Group label to update.  The group must already exist.
        names : list[str]
            New ordered list of un-prefixed sequence names for the group.

        Raises
        ------
        KeyError
            If *group* is not a known group.

        Warns
        -----
        Logs a warning for every name that is also present in another group.
        """
        if group not in self._groups:
            raise KeyError(f'Group {group!r} not found.')
        for n in names:
            for other_g, other_ns in self._groups.items():
                if other_g != group and n in other_ns:
                    _log.warning(
                        'CrossIndex: sequence %r is assigned to both group %r and group %r',
                        n,
                        other_g,
                        group,
                    )
        self._groups[group] = list(names)

    def reorder_by_length(self, group: str | None = None) -> None:
        """Reorder contigs within one or all groups by descending sequence length.

        Updates :attr:`contig_order` in-place.

        Parameters
        ----------
        group : str or None, optional
            Group to reorder.  If ``None`` (default), all groups are reordered.
        """
        groups_to_sort = [group] if group is not None else list(self._groups.keys())
        for g in groups_to_sort:
            self._groups[g].sort(
                key=lambda n: self._index.get_sequence_length(
                    self._make_internal(g, n)
                ),
                reverse=True,
            )

    def reorder_for_colinearity(self, query_group: str, target_group: str) -> None:
        """Reorder sequences in two groups to maximise dotplot collinearity.

        Uses the gravity-centre algorithm via
        :meth:`~rusty_dot.SequenceIndex.optimal_contig_order`.  Updates
        :attr:`contig_order` in-place for both groups.

        .. note::
            :meth:`compute_matches` must be called for ``(query_group,
            target_group)`` before calling this method.

        Parameters
        ----------
        query_group : str
            Group label for the query (y-axis / rows).
        target_group : str
            Group label for the target (x-axis / columns).

        Raises
        ------
        KeyError
            If either group label is not present in the index.
        ValueError
            If :meth:`compute_matches` has not been called for this group
            pair.
        """
        pair = (query_group, target_group)
        if pair not in self._records_by_pair:
            raise ValueError(
                f'No matches computed for group pair {pair!r}. '
                'Call compute_matches() for this pair first.'
            )
        q_internal = [
            self._make_internal(query_group, n) for n in self._groups[query_group]
        ]
        t_internal = [
            self._make_internal(target_group, n) for n in self._groups[target_group]
        ]
        sorted_q_int, sorted_t_int = self._index.optimal_contig_order(
            q_internal, t_internal
        )
        self._groups[query_group] = [self._split_internal(n)[1] for n in sorted_q_int]
        self._groups[target_group] = [self._split_internal(n)[1] for n in sorted_t_int]

    # ------------------------------------------------------------------
    # PAF output and match computation
    # ------------------------------------------------------------------

    def _get_default_group_pairs(self) -> list[tuple[str, str]]:
        """Return default group pairs for alignment.

        * 2 groups → one pair between the two groups.
        * 3+ groups → all non-self ordered pairs.
        """
        groups = list(self._groups.keys())
        if len(groups) == 2:
            return [(groups[0], groups[1])]
        return [
            (a, b) for i, a in enumerate(groups) for j, b in enumerate(groups) if i != j
        ]

    def compute_matches(
        self,
        query_group: str | None = None,
        target_group: str | None = None,
        merge: bool = True,
    ) -> None:
        """Compute k-mer matches between groups and cache the results.

        This is the primary computation step and must be called **before**
        :meth:`reorder_contigs` or :meth:`reorder_for_colinearity`.  Matches
        are computed only between groups — not within a single group.

        When *query_group* and *target_group* are both ``None``:

        * **2 groups** — the single cross-group pair is used.
        * **3+ groups** — all non-self ordered pairs are computed.

        The computed records are stored internally, keyed by
        ``(query_group, target_group)``, and the pair is added to
        :attr:`computed_group_pairs`.

        Parameters
        ----------
        query_group : str or None, optional
            Group label for query sequences.  When ``None`` (default) the
            groups are auto-detected (see above).
        target_group : str or None, optional
            Group label for target sequences.  When ``None`` (default) the
            groups are auto-detected.
        merge : bool, optional
            Whether to merge consecutive co-linear k-mer runs into single
            alignment blocks.  Default is ``True``.

        Raises
        ------
        ValueError
            If group auto-detection fails (≠2 groups, no explicit params), or
            if only one of *query_group* / *target_group* is supplied.
        KeyError
            If an explicit group label is not present in the index.
        """
        if query_group is None and target_group is None:
            pairs = self._get_default_group_pairs()
        elif (query_group is None) ^ (target_group is None):
            raise ValueError('Provide both query_group and target_group, or neither.')
        else:
            if query_group not in self._groups:
                raise KeyError(f'Group {query_group!r} not found.')
            if target_group not in self._groups:
                raise KeyError(f'Group {target_group!r} not found.')
            pairs = [(query_group, target_group)]

        for qg, tg in pairs:
            q_seqs = self._groups.get(qg, [])
            t_seqs = self._groups.get(tg, [])
            _log.info(
                'CrossIndex.compute_matches: computing matches between '
                'group %r (%d sequence(s)) and group %r (%d sequence(s))',
                qg,
                len(q_seqs),
                tg,
                len(t_seqs),
            )
            pair_records: list[PafRecord] = []
            for q_orig in q_seqs:
                for t_orig in t_seqs:
                    q_int = self._make_internal(qg, q_orig)
                    t_int = self._make_internal(tg, t_orig)
                    _log.debug(
                        'CrossIndex.compute_matches: comparing %r (group %r) '
                        'vs %r (group %r)',
                        q_orig,
                        qg,
                        t_orig,
                        tg,
                    )
                    lines = self._index.get_paf(q_int, t_int, merge)
                    for line in lines:
                        fields = line.split('\t')
                        fields[0] = q_orig
                        fields[5] = t_orig
                        pair_records.append(PafRecord.from_line('\t'.join(fields)))
            self._records_by_pair[(qg, tg)] = pair_records
            _log.info(
                'CrossIndex.compute_matches: stored %d record(s) for pair (%r, %r)',
                len(pair_records),
                qg,
                tg,
            )

    def get_paf(
        self,
        group_pairs: list[tuple[str, str]] | None = None,
        merge: bool = True,
    ) -> list[str]:
        """Return PAF lines for cross-group sequence comparisons.

        Parameters
        ----------
        group_pairs : list of (str, str) or None, optional
            Explicit list of ``(query_group, target_group)`` pairs to compare.
            If ``None`` (default):

            * 2 groups → the single cross-group pair.
            * 3+ groups → all non-self ordered pairs.
        merge : bool, optional
            Whether to merge consecutive co-linear k-mer runs before
            generating PAF lines.  Default is ``True``.

        Returns
        -------
        list[str]
            PAF-formatted lines (12 tab-separated columns each).
        """
        if group_pairs is None:
            group_pairs = self._get_default_group_pairs()

        paf_lines: list[str] = []
        for query_group, target_group in group_pairs:
            q_seqs = self._groups.get(query_group, [])
            t_seqs = self._groups.get(target_group, [])
            _log.info(
                'CrossIndex.get_paf: on-demand computation of %d x %d alignments '
                'between group %r and group %r '
                '(tip: call compute_matches() first to pre-cache results)',
                len(q_seqs),
                len(t_seqs),
                query_group,
                target_group,
            )
            for q_orig in q_seqs:
                for t_orig in t_seqs:
                    q_int = self._make_internal(query_group, q_orig)
                    t_int = self._make_internal(target_group, t_orig)
                    _log.debug(
                        'CrossIndex.get_paf: comparing %r (group %r) vs %r (group %r)',
                        q_orig,
                        query_group,
                        t_orig,
                        target_group,
                    )
                    lines = self._index.get_paf(q_int, t_int, merge)
                    for line in lines:
                        fields = line.split('\t')
                        fields[0] = q_orig
                        fields[5] = t_orig
                        paf_lines.append('\t'.join(fields))
        return paf_lines

    def run_merge(
        self,
        group_pairs: list[tuple[str, str]] | None = None,
    ) -> None:
        """Compute merged alignments and store the result as :attr:`_paf_records`.

        .. deprecated::
            Use :meth:`compute_matches` instead.  ``run_merge`` now delegates
            to ``compute_matches`` and is retained only for backward
            compatibility.

        Parameters
        ----------
        group_pairs : list of (str, str) or None, optional
            Group pairs to compare (same semantics as :meth:`compute_matches`).
            Defaults to all cross-group pairs.
        """
        if group_pairs is None:
            self.compute_matches(merge=True)
        else:
            for qg, tg in group_pairs:
                self.compute_matches(query_group=qg, target_group=tg, merge=True)

    # ------------------------------------------------------------------
    # Backward-compatible API (two-group a/b model)
    # ------------------------------------------------------------------

    def get_paf_all(self, merge: bool = True) -> list[str]:
        """Return PAF lines for all cross-group comparisons.

        Backward-compatible wrapper around :meth:`get_paf`.  When a group
        ``'b'`` is present, computes ``a`` vs ``b`` alignments; otherwise
        performs all-vs-all within group ``'a'``.

        Parameters
        ----------
        merge : bool, optional
            Whether to merge consecutive co-linear k-mer runs.
            Default is ``True``.

        Returns
        -------
        list[str]
            PAF-formatted lines.
        """
        if 'b' in self._groups and self._groups['b']:
            return self.get_paf(group_pairs=[('a', 'b')], merge=merge)
        # Single group or no group 'b': all-vs-all within group 'a'
        names_a = self._groups.get('a', [])
        _log.info(
            'CrossIndex: computing all-vs-all pairwise alignments '
            'within group a (%d sequences)',
            len(names_a),
        )
        paf_lines: list[str] = []
        for i, q_orig in enumerate(names_a):
            for j, t_orig in enumerate(names_a):
                if i == j:
                    continue
                q_int = self._make_internal('a', q_orig)
                t_int = self._make_internal('a', t_orig)
                lines = self._index.get_paf(q_int, t_int, merge)
                for line in lines:
                    fields = line.split('\t')
                    fields[0] = q_orig
                    fields[5] = t_orig
                    paf_lines.append('\t'.join(fields))
        return paf_lines

    def reorder_contigs(
        self,
        query_names: list[str] | None = None,
        target_names: list[str] | None = None,
        query_group: str | None = None,
        target_group: str | None = None,
    ) -> tuple[list[str], list[str]]:
        """Sort contigs for maximum collinearity.

        .. note::
            :meth:`compute_matches` must be called for the relevant group pair
            before calling this method.

        When *query_group* and *target_group* are not provided the method
        auto-detects the two groups to compare:

        * If there are **exactly two groups**, those two groups are used
          (regardless of their labels) and an info-level log message records
          which groups were selected and the order of the returned tuple.
        * Otherwise a :exc:`ValueError` is raised and the caller must supply
          explicit group labels via *query_group* / *target_group*.

        Parameters
        ----------
        query_names : list[str] or None, optional
            Explicit un-prefixed names within *query_group* to reorder.
            Defaults to all sequences in *query_group*.
        target_names : list[str] or None, optional
            Explicit un-prefixed names within *target_group* to reorder.
            Defaults to all sequences in *target_group*.
        query_group : str or None, optional
            Group label for the query (first element of the returned tuple).
            When ``None`` the group is auto-detected (requires exactly two
            groups).
        target_group : str or None, optional
            Group label for the target (second element of the returned tuple).
            When ``None`` the group is auto-detected (requires exactly two
            groups).

        Returns
        -------
        tuple[list[str], list[str]]
            ``(sorted_query_names, sorted_target_names)`` — both using
            original un-prefixed names.  The log output names the groups
            in the same order as the tuple elements.

        Raises
        ------
        ValueError
            If groups cannot be auto-detected (i.e. there are not exactly two
            groups and no explicit group labels were supplied), if only one of
            *query_group* / *target_group* is given, or if
            :meth:`compute_matches` has not been called for the resolved group
            pair.
        KeyError
            If an explicitly supplied group label is not present.
        """
        groups = list(self._groups.keys())

        if query_group is None and target_group is None:
            if len(groups) == 2:
                query_group, target_group = groups[0], groups[1]
                _log.info(
                    'CrossIndex.reorder_contigs: auto-selected groups '
                    '%r (query / first) and %r (target / second)',
                    query_group,
                    target_group,
                )
            else:
                raise ValueError(
                    'reorder_contigs requires exactly two groups when query_group '
                    'and target_group are not specified; '
                    f'found {len(groups)} group(s): {groups!r}. '
                    'Provide query_group and target_group explicitly, or use '
                    'reorder_for_colinearity for full control.'
                )
        elif (query_group is None) ^ (target_group is None):
            raise ValueError('Provide both query_group and target_group, or neither.')
        else:
            _log.info(
                'CrossIndex.reorder_contigs: using groups '
                '%r (query / first) and %r (target / second)',
                query_group,
                target_group,
            )

        pair = (query_group, target_group)
        if pair not in self._records_by_pair:
            raise ValueError(
                f'No matches computed for group pair {pair!r}. '
                'Call compute_matches() for this pair first.'
            )

        q_names = (
            query_names if query_names is not None else list(self._groups[query_group])
        )
        t_names = (
            target_names
            if target_names is not None
            else list(self._groups[target_group])
        )
        q_internal = [self._make_internal(query_group, n) for n in q_names]
        t_internal = [self._make_internal(target_group, n) for n in t_names]
        sorted_q_int, sorted_t_int = self._index.optimal_contig_order(
            q_internal, t_internal
        )
        sorted_q = [self._split_internal(n)[1] for n in sorted_q_int]
        sorted_t = [self._split_internal(n)[1] for n in sorted_t_int]
        return sorted_q, sorted_t

    # ------------------------------------------------------------------
    # Dunder methods
    # ------------------------------------------------------------------

    def __repr__(self) -> str:
        """Return a concise machine-readable representation.

        Returns
        -------
        str
            ``CrossIndex(k=<k>, groups={<label>=<n>, ...})``.
        """
        group_info = ', '.join(f'{g}={len(names)}' for g, names in self._groups.items())
        return f'CrossIndex(k={self._k}, groups={{{group_info}}})'

    def __str__(self) -> str:
        """Return a human-readable stats summary.

        Returns
        -------
        str
            Multi-line summary of groups, sequence counts, computed pairs,
            and cached PAF record count.
        """
        n_total = sum(len(v) for v in self._groups.values())
        lines = [f'CrossIndex (k={self._k})']
        lines.append(f'  Total sequences : {n_total}')
        for g, names in self._groups.items():
            lines.append(f'  Group {g!r:12s}: {len(names):>6d} sequences')
        if self._records_by_pair:
            for (qg, tg), recs in self._records_by_pair.items():
                lines.append(
                    f'  Computed pair   : ({qg!r}, {tg!r}) → {len(recs)} record(s)'
                )
        else:
            lines.append('  Computed pairs  : none (call compute_matches() first)')
        lines.append(f'  PAF records     : {len(self._paf_records)}')
        return '\n'.join(lines)

Attributes¶

`computed_group_pairs` `property` ¶

Group pairs for which k-mer matches have been computed.

Returns:

Type	Description
`list[tuple[str, str]]`	List of `(query_group, target_group)` pairs that have had :meth:`compute_matches` run on them. Use this to confirm that the required pair is ready before calling :meth:`reorder_contigs`.

`group_names` `property` ¶

Return the list of group labels in insertion order.

Returns:

Type	Description
`list[str]`	Group labels.

`contig_order` `property` ¶

Current contig order per group as original (un-prefixed) names.

Returns:

Type	Description
`dict[str, list[str]]`	Mapping of group label → ordered list of sequence names. Updated in-place by :meth:`reorder_by_length` and :meth:`reorder_for_colinearity`.

`query_names` `property` ¶

Un-prefixed names for group 'a' (backward compatible).

Returns:

Type	Description
`list[str]`

`target_names` `property` ¶

Un-prefixed names for group 'b' (backward compatible).

Returns:

Type	Description
`list[str]`

Functions¶

`init(k)` ¶

Initialise an empty CrossIndex.

Parameters:

Name	Type	Description	Default
`k`	`int`	K-mer length to use when building the sequence index.	required

Source code in rusty_dot/paf_io.py

def __init__(self, k: int) -> None:
    """Initialise an empty CrossIndex.

    Parameters
    ----------
    k : int
        K-mer length to use when building the sequence index.
    """
    self._k: int = k
    self._index: SequenceIndex = SequenceIndex(k=k)
    # group_label -> ordered list of original (un-prefixed) sequence names
    self._groups: dict[str, list[str]] = {}
    # group_label -> internal prefix used in _index (supports rename_group)
    self._internal_group: dict[str, str] = {}
    # (query_group, target_group) -> list[PafRecord] from compute_matches
    self._records_by_pair: dict[tuple[str, str], list[PafRecord]] = {}

`get_records_for_pair(query_group, target_group)` ¶

Return the cached :class:PafRecord list for a computed group pair.

Parameters:

Name	Type	Description	Default
`query_group`	`str`	Query group label.	required
`target_group`	`str`	Target group label.	required

Returns:

Type	Description
`list[PafRecord]`	Cached PAF records for the pair. Returns an empty list if :meth:`compute_matches` has not been called for this pair.

Source code in rusty_dot/paf_io.py

def get_records_for_pair(
    self, query_group: str, target_group: str
) -> list['PafRecord']:
    """Return the cached :class:`PafRecord` list for a computed group pair.

    Parameters
    ----------
    query_group : str
        Query group label.
    target_group : str
        Target group label.

    Returns
    -------
    list[PafRecord]
        Cached PAF records for the pair.  Returns an empty list if
        :meth:`compute_matches` has not been called for this pair.
    """
    return list(self._records_by_pair.get((query_group, target_group), []))

`make_internal_name(group, name)` ¶

Construct the internal ('group:name') identifier for a sequence.

This is the public counterpart of the internal :meth:_make_internal helper and is suitable for use by external code such as :class:~rusty_dot.dotplot.DotPlotter.

Parameters:

Name	Type	Description	Default
`group`	`str`	Group label.	required
`name`	`str`	Un-prefixed sequence name.	required

Returns:

Type	Description
`str`	Internal identifier in `'group:name'` form, with any `rename_group` remapping applied.

Source code in rusty_dot/paf_io.py

def make_internal_name(self, group: str, name: str) -> str:
    """Construct the internal (``'group:name'``) identifier for a sequence.

    This is the public counterpart of the internal :meth:`_make_internal`
    helper and is suitable for use by external code such as
    :class:`~rusty_dot.dotplot.DotPlotter`.

    Parameters
    ----------
    group : str
        Group label.
    name : str
        Un-prefixed sequence name.

    Returns
    -------
    str
        Internal identifier in ``'group:name'`` form, with any
        ``rename_group`` remapping applied.
    """
    return self._make_internal(group, name)

`add_sequence(name, seq, group='a')` ¶

Add a single sequence to the specified group.

Logs a DEBUG-level message for every sequence loaded, and a WARNING if name already exists in the same group (its FM-index will be overwritten) or in a different group (potential confusion).

Parameters:

Name	Type	Description	Default
`name`	`str`	Sequence identifier (must be unique within the group).	required
`seq`	`str`	DNA sequence string.	required
`group`	`str`	Group label. Any non-empty string is accepted; `':'` is forbidden because it is used as an internal separator. Default is `'a'`.	`'a'`

Raises:

Type	Description
`ValueError`	If group contains `':'`.

Source code in rusty_dot/paf_io.py

def add_sequence(self, name: str, seq: str, group: str = 'a') -> None:
    """Add a single sequence to the specified group.

    Logs a ``DEBUG``-level message for every sequence loaded, and a
    ``WARNING`` if *name* already exists in the same group (its FM-index
    will be overwritten) or in a different group (potential confusion).

    Parameters
    ----------
    name : str
        Sequence identifier (must be unique within the group).
    seq : str
        DNA sequence string.
    group : str, optional
        Group label.  Any non-empty string is accepted; ``':'`` is
        forbidden because it is used as an internal separator.
        Default is ``'a'``.

    Raises
    ------
    ValueError
        If *group* contains ``':'``.
    """
    if ':' in group:
        raise ValueError(f"Group name must not contain ':', got {group!r}")
    self._check_name_collision(name, group)
    _log.debug(
        'CrossIndex: adding sequence %r (len=%d) to group %r',
        name,
        len(seq),
        group,
    )
    if group not in self._groups:
        self._groups[group] = []
        self._internal_group[group] = group
    internal = self._make_internal(group, name)
    self._index.add_sequence(internal, seq)
    if name not in self._groups[group]:
        self._groups[group].append(name)

`load_fasta(path, group='a')` ¶

Load sequences from a FASTA file into the specified group.

Logs an INFO-level message when opening the file, a DEBUG message for each sequence loaded (including sequence name and length), and a WARNING if any sequence name already exists in the same or another group.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to a FASTA (`.fa` / `.fasta`) or gzipped FASTA file.	required
`group`	`str`	Group label. Default is `'a'`.	`'a'`

Returns:

Type	Description
`list[str]`	The original (un-prefixed) sequence names that were loaded, in file order.

Raises:

Type	Description
`ValueError`	If group contains `':'`, or if the file cannot be parsed, or if the FASTA file contains duplicate sequence names.

Source code in rusty_dot/paf_io.py

def load_fasta(self, path: str, group: str = 'a') -> list[str]:
    """Load sequences from a FASTA file into the specified group.

    Logs an ``INFO``-level message when opening the file, a ``DEBUG``
    message for each sequence loaded (including sequence name and length),
    and a ``WARNING`` if any sequence name already exists in the same or
    another group.

    Parameters
    ----------
    path : str
        Path to a FASTA (``.fa`` / ``.fasta``) or gzipped FASTA file.
    group : str, optional
        Group label.  Default is ``'a'``.

    Returns
    -------
    list[str]
        The original (un-prefixed) sequence names that were loaded, in
        file order.

    Raises
    ------
    ValueError
        If *group* contains ``':'``, or if the file cannot be parsed, or
        if the FASTA file contains duplicate sequence names.
    """
    if ':' in group:
        raise ValueError(f"Group name must not contain ':', got {group!r}")
    from rusty_dot._rusty_dot import py_read_fasta

    _log.info('CrossIndex: loading sequences from %r into group %r', path, group)
    seqs = py_read_fasta(path)
    if group not in self._groups:
        self._groups[group] = []
        self._internal_group[group] = group
    names: list[str] = []
    for name, seq in seqs.items():
        self._check_name_collision(name, group)
        _log.debug(
            'CrossIndex: adding sequence %r (len=%d) to group %r',
            name,
            len(seq),
            group,
        )
        internal = self._make_internal(group, name)
        self._index.add_sequence(internal, seq)
        if name not in self._groups[group]:
            self._groups[group].append(name)
        names.append(name)
    _log.info(
        'CrossIndex: loaded %d sequence(s) from %r into group %r',
        len(names),
        path,
        group,
    )
    return names

`sequence_names(group=None)` ¶

Return internal (group:name) identifiers suitable for DotPlotter.

Parameters:

Name	Type	Description	Default
`group`	`str or None`	If given, return only names from that group. If `None` (default), return names from all groups.	`None`

Returns:

Type	Description
`list[str]`	Internal `group:name` strings in current :attr:`contig_order`.

Source code in rusty_dot/paf_io.py

def sequence_names(self, group: str | None = None) -> list[str]:
    """Return internal (``group:name``) identifiers suitable for DotPlotter.

    Parameters
    ----------
    group : str or None, optional
        If given, return only names from that group.  If ``None``
        (default), return names from all groups.

    Returns
    -------
    list[str]
        Internal ``group:name`` strings in current :attr:`contig_order`.
    """
    if group is not None:
        return [self._make_internal(group, n) for n in self._groups.get(group, [])]
    result: list[str] = []
    for g, names in self._groups.items():
        result.extend(self._make_internal(g, n) for n in names)
    return result

`get_sequence_length(name)` ¶

Return the length of the sequence identified by its internal name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Internal (`group:name`) identifier.	required

Returns:

Type	Description
`int`	Sequence length in bases.

Source code in rusty_dot/paf_io.py

def get_sequence_length(self, name: str) -> int:
    """Return the length of the sequence identified by its internal name.

    Parameters
    ----------
    name : str
        Internal (``group:name``) identifier.

    Returns
    -------
    int
        Sequence length in bases.
    """
    return self._index.get_sequence_length(name)

`compare_sequences_stranded(name1, name2, merge=True)` ¶

Compare two sequences by their internal names, returning stranded matches.

Parameters:

Name	Type	Description	Default
`name1`	`str`	Internal name of the query sequence.	required
`name2`	`str`	Internal name of the target sequence.	required
`merge`	`bool`	Whether to merge consecutive co-linear k-mer runs. Default is `True`.	`True`

Returns:

Type	Description
`list of (int, int, int, int, str)`	List of `(query_start, query_end, target_start, target_end, strand)` tuples.

Source code in rusty_dot/paf_io.py

def compare_sequences_stranded(
    self, name1: str, name2: str, merge: bool = True
) -> list:
    """Compare two sequences by their internal names, returning stranded matches.

    Parameters
    ----------
    name1 : str
        Internal name of the query sequence.
    name2 : str
        Internal name of the target sequence.
    merge : bool, optional
        Whether to merge consecutive co-linear k-mer runs.
        Default is ``True``.

    Returns
    -------
    list of (int, int, int, int, str)
        List of ``(query_start, query_end, target_start, target_end, strand)``
        tuples.
    """
    return self._index.compare_sequences_stranded(name1, name2, merge)

`rename_group(old_name, new_name)` ¶

Rename a group label without re-indexing sequences.

The internal prefix used in the underlying :class:SequenceIndex is preserved; only the public-facing group label is changed.

Parameters:

Name	Type	Description	Default
`old_name`	`str`	Current group label to rename.	required
`new_name`	`str`	New group label. Must not contain `':'`.	required

Raises:

Type	Description
`KeyError`	If old_name is not a known group.
`ValueError`	If new_name contains `':'` or already exists as a group label.

Source code in rusty_dot/paf_io.py

def rename_group(self, old_name: str, new_name: str) -> None:
    """Rename a group label without re-indexing sequences.

    The internal prefix used in the underlying :class:`SequenceIndex` is
    preserved; only the public-facing group label is changed.

    Parameters
    ----------
    old_name : str
        Current group label to rename.
    new_name : str
        New group label.  Must not contain ``':'``.

    Raises
    ------
    KeyError
        If *old_name* is not a known group.
    ValueError
        If *new_name* contains ``':'`` or already exists as a group label.
    """
    if old_name not in self._groups:
        raise KeyError(f'Group {old_name!r} not found.')
    if ':' in new_name:
        raise ValueError(f"Group name must not contain ':', got {new_name!r}")
    if new_name in self._groups and new_name != old_name:
        raise ValueError(f'Group {new_name!r} already exists.')
    # Rebuild _groups preserving insertion order
    self._groups = {
        (new_name if k == old_name else k): v for k, v in self._groups.items()
    }
    # Update the internal-prefix mapping
    self._internal_group[new_name] = self._internal_group.pop(old_name)
    _log.info('CrossIndex: renamed group %r → %r', old_name, new_name)

`set_group_members(group, names)` ¶

Assign a custom list of sequence names to an existing group.

Only the logical membership list is updated; sequences already indexed are not moved or removed from the underlying :class:~rusty_dot.SequenceIndex.

Parameters:

Name	Type	Description	Default
`group`	`str`	Group label to update. The group must already exist.	required
`names`	`list[str]`	New ordered list of un-prefixed sequence names for the group.	required

Raises:

Type	Description
`KeyError`	If group is not a known group.

Warns:

Type	Description
`Logs a warning for every name that is also present in another group.`

Source code in rusty_dot/paf_io.py

def set_group_members(self, group: str, names: list[str]) -> None:
    """Assign a custom list of sequence names to an existing group.

    Only the logical membership list is updated; sequences already indexed
    are not moved or removed from the underlying
    :class:`~rusty_dot.SequenceIndex`.

    Parameters
    ----------
    group : str
        Group label to update.  The group must already exist.
    names : list[str]
        New ordered list of un-prefixed sequence names for the group.

    Raises
    ------
    KeyError
        If *group* is not a known group.

    Warns
    -----
    Logs a warning for every name that is also present in another group.
    """
    if group not in self._groups:
        raise KeyError(f'Group {group!r} not found.')
    for n in names:
        for other_g, other_ns in self._groups.items():
            if other_g != group and n in other_ns:
                _log.warning(
                    'CrossIndex: sequence %r is assigned to both group %r and group %r',
                    n,
                    other_g,
                    group,
                )
    self._groups[group] = list(names)

`reorder_by_length(group=None)` ¶

Reorder contigs within one or all groups by descending sequence length.

Updates :attr:contig_order in-place.

Parameters:

Name	Type	Description	Default
`group`	`str or None`	Group to reorder. If `None` (default), all groups are reordered.	`None`

Source code in rusty_dot/paf_io.py

def reorder_by_length(self, group: str | None = None) -> None:
    """Reorder contigs within one or all groups by descending sequence length.

    Updates :attr:`contig_order` in-place.

    Parameters
    ----------
    group : str or None, optional
        Group to reorder.  If ``None`` (default), all groups are reordered.
    """
    groups_to_sort = [group] if group is not None else list(self._groups.keys())
    for g in groups_to_sort:
        self._groups[g].sort(
            key=lambda n: self._index.get_sequence_length(
                self._make_internal(g, n)
            ),
            reverse=True,
        )

`reorder_for_colinearity(query_group, target_group)` ¶

Reorder sequences in two groups to maximise dotplot collinearity.

Uses the gravity-centre algorithm via :meth:~rusty_dot.SequenceIndex.optimal_contig_order. Updates :attr:contig_order in-place for both groups.

.. note:: :meth:compute_matches must be called for (query_group, target_group) before calling this method.

Parameters:

Name	Type	Description	Default
`query_group`	`str`	Group label for the query (y-axis / rows).	required
`target_group`	`str`	Group label for the target (x-axis / columns).	required

Raises:

Type	Description
`KeyError`	If either group label is not present in the index.
`ValueError`	If :meth:`compute_matches` has not been called for this group pair.

Source code in rusty_dot/paf_io.py

def reorder_for_colinearity(self, query_group: str, target_group: str) -> None:
    """Reorder sequences in two groups to maximise dotplot collinearity.

    Uses the gravity-centre algorithm via
    :meth:`~rusty_dot.SequenceIndex.optimal_contig_order`.  Updates
    :attr:`contig_order` in-place for both groups.

    .. note::
        :meth:`compute_matches` must be called for ``(query_group,
        target_group)`` before calling this method.

    Parameters
    ----------
    query_group : str
        Group label for the query (y-axis / rows).
    target_group : str
        Group label for the target (x-axis / columns).

    Raises
    ------
    KeyError
        If either group label is not present in the index.
    ValueError
        If :meth:`compute_matches` has not been called for this group
        pair.
    """
    pair = (query_group, target_group)
    if pair not in self._records_by_pair:
        raise ValueError(
            f'No matches computed for group pair {pair!r}. '
            'Call compute_matches() for this pair first.'
        )
    q_internal = [
        self._make_internal(query_group, n) for n in self._groups[query_group]
    ]
    t_internal = [
        self._make_internal(target_group, n) for n in self._groups[target_group]
    ]
    sorted_q_int, sorted_t_int = self._index.optimal_contig_order(
        q_internal, t_internal
    )
    self._groups[query_group] = [self._split_internal(n)[1] for n in sorted_q_int]
    self._groups[target_group] = [self._split_internal(n)[1] for n in sorted_t_int]

`compute_matches(query_group=None, target_group=None, merge=True)` ¶

Compute k-mer matches between groups and cache the results.

This is the primary computation step and must be called before :meth:reorder_contigs or :meth:reorder_for_colinearity. Matches are computed only between groups — not within a single group.

When query_group and target_group are both None:

2 groups — the single cross-group pair is used.
3+ groups — all non-self ordered pairs are computed.

The computed records are stored internally, keyed by (query_group, target_group), and the pair is added to :attr:computed_group_pairs.

Parameters:

Name	Type	Description	Default
`query_group`	`str or None`	Group label for query sequences. When `None` (default) the groups are auto-detected (see above).	`None`
`target_group`	`str or None`	Group label for target sequences. When `None` (default) the groups are auto-detected.	`None`
`merge`	`bool`	Whether to merge consecutive co-linear k-mer runs into single alignment blocks. Default is `True`.	`True`

Raises:

Type	Description
`ValueError`	If group auto-detection fails (≠2 groups, no explicit params), or if only one of query_group / target_group is supplied.
`KeyError`	If an explicit group label is not present in the index.

Source code in rusty_dot/paf_io.py

def compute_matches(
    self,
    query_group: str | None = None,
    target_group: str | None = None,
    merge: bool = True,
) -> None:
    """Compute k-mer matches between groups and cache the results.

    This is the primary computation step and must be called **before**
    :meth:`reorder_contigs` or :meth:`reorder_for_colinearity`.  Matches
    are computed only between groups — not within a single group.

    When *query_group* and *target_group* are both ``None``:

    * **2 groups** — the single cross-group pair is used.
    * **3+ groups** — all non-self ordered pairs are computed.

    The computed records are stored internally, keyed by
    ``(query_group, target_group)``, and the pair is added to
    :attr:`computed_group_pairs`.

    Parameters
    ----------
    query_group : str or None, optional
        Group label for query sequences.  When ``None`` (default) the
        groups are auto-detected (see above).
    target_group : str or None, optional
        Group label for target sequences.  When ``None`` (default) the
        groups are auto-detected.
    merge : bool, optional
        Whether to merge consecutive co-linear k-mer runs into single
        alignment blocks.  Default is ``True``.

    Raises
    ------
    ValueError
        If group auto-detection fails (≠2 groups, no explicit params), or
        if only one of *query_group* / *target_group* is supplied.
    KeyError
        If an explicit group label is not present in the index.
    """
    if query_group is None and target_group is None:
        pairs = self._get_default_group_pairs()
    elif (query_group is None) ^ (target_group is None):
        raise ValueError('Provide both query_group and target_group, or neither.')
    else:
        if query_group not in self._groups:
            raise KeyError(f'Group {query_group!r} not found.')
        if target_group not in self._groups:
            raise KeyError(f'Group {target_group!r} not found.')
        pairs = [(query_group, target_group)]

    for qg, tg in pairs:
        q_seqs = self._groups.get(qg, [])
        t_seqs = self._groups.get(tg, [])
        _log.info(
            'CrossIndex.compute_matches: computing matches between '
            'group %r (%d sequence(s)) and group %r (%d sequence(s))',
            qg,
            len(q_seqs),
            tg,
            len(t_seqs),
        )
        pair_records: list[PafRecord] = []
        for q_orig in q_seqs:
            for t_orig in t_seqs:
                q_int = self._make_internal(qg, q_orig)
                t_int = self._make_internal(tg, t_orig)
                _log.debug(
                    'CrossIndex.compute_matches: comparing %r (group %r) '
                    'vs %r (group %r)',
                    q_orig,
                    qg,
                    t_orig,
                    tg,
                )
                lines = self._index.get_paf(q_int, t_int, merge)
                for line in lines:
                    fields = line.split('\t')
                    fields[0] = q_orig
                    fields[5] = t_orig
                    pair_records.append(PafRecord.from_line('\t'.join(fields)))
        self._records_by_pair[(qg, tg)] = pair_records
        _log.info(
            'CrossIndex.compute_matches: stored %d record(s) for pair (%r, %r)',
            len(pair_records),
            qg,
            tg,
        )

`get_paf(group_pairs=None, merge=True)` ¶

Return PAF lines for cross-group sequence comparisons.

Parameters:

Name	Type	Description	Default
`group_pairs`	`list of (str, str) or None`	Explicit list of `(query_group, target_group)` pairs to compare. If `None` (default): 2 groups → the single cross-group pair. 3+ groups → all non-self ordered pairs.	`None`
`merge`	`bool`	Whether to merge consecutive co-linear k-mer runs before generating PAF lines. Default is `True`.	`True`

Returns:

Type	Description
`list[str]`	PAF-formatted lines (12 tab-separated columns each).

Source code in rusty_dot/paf_io.py

def get_paf(
    self,
    group_pairs: list[tuple[str, str]] | None = None,
    merge: bool = True,
) -> list[str]:
    """Return PAF lines for cross-group sequence comparisons.

    Parameters
    ----------
    group_pairs : list of (str, str) or None, optional
        Explicit list of ``(query_group, target_group)`` pairs to compare.
        If ``None`` (default):

        * 2 groups → the single cross-group pair.
        * 3+ groups → all non-self ordered pairs.
    merge : bool, optional
        Whether to merge consecutive co-linear k-mer runs before
        generating PAF lines.  Default is ``True``.

    Returns
    -------
    list[str]
        PAF-formatted lines (12 tab-separated columns each).
    """
    if group_pairs is None:
        group_pairs = self._get_default_group_pairs()

    paf_lines: list[str] = []
    for query_group, target_group in group_pairs:
        q_seqs = self._groups.get(query_group, [])
        t_seqs = self._groups.get(target_group, [])
        _log.info(
            'CrossIndex.get_paf: on-demand computation of %d x %d alignments '
            'between group %r and group %r '
            '(tip: call compute_matches() first to pre-cache results)',
            len(q_seqs),
            len(t_seqs),
            query_group,
            target_group,
        )
        for q_orig in q_seqs:
            for t_orig in t_seqs:
                q_int = self._make_internal(query_group, q_orig)
                t_int = self._make_internal(target_group, t_orig)
                _log.debug(
                    'CrossIndex.get_paf: comparing %r (group %r) vs %r (group %r)',
                    q_orig,
                    query_group,
                    t_orig,
                    target_group,
                )
                lines = self._index.get_paf(q_int, t_int, merge)
                for line in lines:
                    fields = line.split('\t')
                    fields[0] = q_orig
                    fields[5] = t_orig
                    paf_lines.append('\t'.join(fields))
    return paf_lines

`run_merge(group_pairs=None)` ¶

Compute merged alignments and store the result as :attr:_paf_records.

.. deprecated:: Use :meth:compute_matches instead. run_merge now delegates to compute_matches and is retained only for backward compatibility.

Parameters:

Name	Type	Description	Default
`group_pairs`	`list of (str, str) or None`	Group pairs to compare (same semantics as :meth:`compute_matches`). Defaults to all cross-group pairs.	`None`

Source code in rusty_dot/paf_io.py

def run_merge(
    self,
    group_pairs: list[tuple[str, str]] | None = None,
) -> None:
    """Compute merged alignments and store the result as :attr:`_paf_records`.

    .. deprecated::
        Use :meth:`compute_matches` instead.  ``run_merge`` now delegates
        to ``compute_matches`` and is retained only for backward
        compatibility.

    Parameters
    ----------
    group_pairs : list of (str, str) or None, optional
        Group pairs to compare (same semantics as :meth:`compute_matches`).
        Defaults to all cross-group pairs.
    """
    if group_pairs is None:
        self.compute_matches(merge=True)
    else:
        for qg, tg in group_pairs:
            self.compute_matches(query_group=qg, target_group=tg, merge=True)

`get_paf_all(merge=True)` ¶

Return PAF lines for all cross-group comparisons.

Backward-compatible wrapper around :meth:get_paf. When a group 'b' is present, computes a vs b alignments; otherwise performs all-vs-all within group 'a'.

Parameters:

Name	Type	Description	Default
`merge`	`bool`	Whether to merge consecutive co-linear k-mer runs. Default is `True`.	`True`

Returns:

Type	Description
`list[str]`	PAF-formatted lines.

Source code in rusty_dot/paf_io.py

def get_paf_all(self, merge: bool = True) -> list[str]:
    """Return PAF lines for all cross-group comparisons.

    Backward-compatible wrapper around :meth:`get_paf`.  When a group
    ``'b'`` is present, computes ``a`` vs ``b`` alignments; otherwise
    performs all-vs-all within group ``'a'``.

    Parameters
    ----------
    merge : bool, optional
        Whether to merge consecutive co-linear k-mer runs.
        Default is ``True``.

    Returns
    -------
    list[str]
        PAF-formatted lines.
    """
    if 'b' in self._groups and self._groups['b']:
        return self.get_paf(group_pairs=[('a', 'b')], merge=merge)
    # Single group or no group 'b': all-vs-all within group 'a'
    names_a = self._groups.get('a', [])
    _log.info(
        'CrossIndex: computing all-vs-all pairwise alignments '
        'within group a (%d sequences)',
        len(names_a),
    )
    paf_lines: list[str] = []
    for i, q_orig in enumerate(names_a):
        for j, t_orig in enumerate(names_a):
            if i == j:
                continue
            q_int = self._make_internal('a', q_orig)
            t_int = self._make_internal('a', t_orig)
            lines = self._index.get_paf(q_int, t_int, merge)
            for line in lines:
                fields = line.split('\t')
                fields[0] = q_orig
                fields[5] = t_orig
                paf_lines.append('\t'.join(fields))
    return paf_lines

`reorder_contigs(query_names=None, target_names=None, query_group=None, target_group=None)` ¶

Sort contigs for maximum collinearity.

.. note:: :meth:compute_matches must be called for the relevant group pair before calling this method.

When query_group and target_group are not provided the method auto-detects the two groups to compare:

If there are exactly two groups, those two groups are used (regardless of their labels) and an info-level log message records which groups were selected and the order of the returned tuple.
Otherwise a :exc:ValueError is raised and the caller must supply explicit group labels via query_group / target_group.

Parameters:

Name	Type	Description	Default
`query_names`	`list[str] or None`	Explicit un-prefixed names within query_group to reorder. Defaults to all sequences in query_group.	`None`
`target_names`	`list[str] or None`	Explicit un-prefixed names within target_group to reorder. Defaults to all sequences in target_group.	`None`
`query_group`	`str or None`	Group label for the query (first element of the returned tuple). When `None` the group is auto-detected (requires exactly two groups).	`None`
`target_group`	`str or None`	Group label for the target (second element of the returned tuple). When `None` the group is auto-detected (requires exactly two groups).	`None`

Returns:

Type	Description
`tuple[list[str], list[str]]`	`(sorted_query_names, sorted_target_names)` — both using original un-prefixed names. The log output names the groups in the same order as the tuple elements.

Raises:

Type	Description
`ValueError`	If groups cannot be auto-detected (i.e. there are not exactly two groups and no explicit group labels were supplied), if only one of query_group / target_group is given, or if :meth:`compute_matches` has not been called for the resolved group pair.
`KeyError`	If an explicitly supplied group label is not present.

Source code in rusty_dot/paf_io.py

def reorder_contigs(
    self,
    query_names: list[str] | None = None,
    target_names: list[str] | None = None,
    query_group: str | None = None,
    target_group: str | None = None,
) -> tuple[list[str], list[str]]:
    """Sort contigs for maximum collinearity.

    .. note::
        :meth:`compute_matches` must be called for the relevant group pair
        before calling this method.

    When *query_group* and *target_group* are not provided the method
    auto-detects the two groups to compare:

    * If there are **exactly two groups**, those two groups are used
      (regardless of their labels) and an info-level log message records
      which groups were selected and the order of the returned tuple.
    * Otherwise a :exc:`ValueError` is raised and the caller must supply
      explicit group labels via *query_group* / *target_group*.

    Parameters
    ----------
    query_names : list[str] or None, optional
        Explicit un-prefixed names within *query_group* to reorder.
        Defaults to all sequences in *query_group*.
    target_names : list[str] or None, optional
        Explicit un-prefixed names within *target_group* to reorder.
        Defaults to all sequences in *target_group*.
    query_group : str or None, optional
        Group label for the query (first element of the returned tuple).
        When ``None`` the group is auto-detected (requires exactly two
        groups).
    target_group : str or None, optional
        Group label for the target (second element of the returned tuple).
        When ``None`` the group is auto-detected (requires exactly two
        groups).

    Returns
    -------
    tuple[list[str], list[str]]
        ``(sorted_query_names, sorted_target_names)`` — both using
        original un-prefixed names.  The log output names the groups
        in the same order as the tuple elements.

    Raises
    ------
    ValueError
        If groups cannot be auto-detected (i.e. there are not exactly two
        groups and no explicit group labels were supplied), if only one of
        *query_group* / *target_group* is given, or if
        :meth:`compute_matches` has not been called for the resolved group
        pair.
    KeyError
        If an explicitly supplied group label is not present.
    """
    groups = list(self._groups.keys())

    if query_group is None and target_group is None:
        if len(groups) == 2:
            query_group, target_group = groups[0], groups[1]
            _log.info(
                'CrossIndex.reorder_contigs: auto-selected groups '
                '%r (query / first) and %r (target / second)',
                query_group,
                target_group,
            )
        else:
            raise ValueError(
                'reorder_contigs requires exactly two groups when query_group '
                'and target_group are not specified; '
                f'found {len(groups)} group(s): {groups!r}. '
                'Provide query_group and target_group explicitly, or use '
                'reorder_for_colinearity for full control.'
            )
    elif (query_group is None) ^ (target_group is None):
        raise ValueError('Provide both query_group and target_group, or neither.')
    else:
        _log.info(
            'CrossIndex.reorder_contigs: using groups '
            '%r (query / first) and %r (target / second)',
            query_group,
            target_group,
        )

    pair = (query_group, target_group)
    if pair not in self._records_by_pair:
        raise ValueError(
            f'No matches computed for group pair {pair!r}. '
            'Call compute_matches() for this pair first.'
        )

    q_names = (
        query_names if query_names is not None else list(self._groups[query_group])
    )
    t_names = (
        target_names
        if target_names is not None
        else list(self._groups[target_group])
    )
    q_internal = [self._make_internal(query_group, n) for n in q_names]
    t_internal = [self._make_internal(target_group, n) for n in t_names]
    sorted_q_int, sorted_t_int = self._index.optimal_contig_order(
        q_internal, t_internal
    )
    sorted_q = [self._split_internal(n)[1] for n in sorted_q_int]
    sorted_t = [self._split_internal(n)[1] for n in sorted_t_int]
    return sorted_q, sorted_t

`repr()` ¶

Return a concise machine-readable representation.

Returns:

Type	Description
`str`	`CrossIndex(k=<k>, groups={<label>=<n>, ...})`.

Source code in rusty_dot/paf_io.py

def __repr__(self) -> str:
    """Return a concise machine-readable representation.

    Returns
    -------
    str
        ``CrossIndex(k=<k>, groups={<label>=<n>, ...})``.
    """
    group_info = ', '.join(f'{g}={len(names)}' for g, names in self._groups.items())
    return f'CrossIndex(k={self._k}, groups={{{group_info}}})'

`str()` ¶

Return a human-readable stats summary.

Returns:

Type	Description
`str`	Multi-line summary of groups, sequence counts, computed pairs, and cached PAF record count.

Source code in rusty_dot/paf_io.py

def __str__(self) -> str:
    """Return a human-readable stats summary.

    Returns
    -------
    str
        Multi-line summary of groups, sequence counts, computed pairs,
        and cached PAF record count.
    """
    n_total = sum(len(v) for v in self._groups.values())
    lines = [f'CrossIndex (k={self._k})']
    lines.append(f'  Total sequences : {n_total}')
    for g, names in self._groups.items():
        lines.append(f'  Group {g!r:12s}: {len(names):>6d} sequences')
    if self._records_by_pair:
        for (qg, tg), recs in self._records_by_pair.items():
            lines.append(
                f'  Computed pair   : ({qg!r}, {tg!r}) → {len(recs)} record(s)'
            )
    else:
        lines.append('  Computed pairs  : none (call compute_matches() first)')
    lines.append(f'  PAF records     : {len(self._paf_records)}')
    return '\n'.join(lines)

CrossIndex¶

Workflow¶

Alignment scope by number of groups¶

Quick start¶

Custom group names¶

Class¶

CrossIndex ¶

Attributes¶

computed_group_pairs property ¶

group_names property ¶

contig_order property ¶

query_names property ¶

target_names property ¶

Functions¶

__init__(k) ¶

get_records_for_pair(query_group, target_group) ¶

make_internal_name(group, name) ¶

add_sequence(name, seq, group='a') ¶

load_fasta(path, group='a') ¶

sequence_names(group=None) ¶

get_sequence_length(name) ¶

compare_sequences_stranded(name1, name2, merge=True) ¶

rename_group(old_name, new_name) ¶

set_group_members(group, names) ¶

reorder_by_length(group=None) ¶

reorder_for_colinearity(query_group, target_group) ¶

compute_matches(query_group=None, target_group=None, merge=True) ¶

get_paf(group_pairs=None, merge=True) ¶

run_merge(group_pairs=None) ¶

get_paf_all(merge=True) ¶

reorder_contigs(query_names=None, target_names=None, query_group=None, target_group=None) ¶

__repr__() ¶

__str__() ¶

`CrossIndex` ¶

`computed_group_pairs` `property` ¶

`group_names` `property` ¶

`contig_order` `property` ¶

`query_names` `property` ¶

`target_names` `property` ¶

`init(k)` ¶

`get_records_for_pair(query_group, target_group)` ¶

`make_internal_name(group, name)` ¶

`add_sequence(name, seq, group='a')` ¶

`load_fasta(path, group='a')` ¶

`sequence_names(group=None)` ¶

`get_sequence_length(name)` ¶

`compare_sequences_stranded(name1, name2, merge=True)` ¶

`rename_group(old_name, new_name)` ¶

`set_group_members(group, names)` ¶

`reorder_by_length(group=None)` ¶

`reorder_for_colinearity(query_group, target_group)` ¶

`compute_matches(query_group=None, target_group=None, merge=True)` ¶

`get_paf(group_pairs=None, merge=True)` ¶

`run_merge(group_pairs=None)` ¶

`get_paf_all(merge=True)` ¶

`reorder_contigs(query_names=None, target_names=None, query_group=None, target_group=None)` ¶

`repr()` ¶

`str()` ¶