molclusters.tracker

tracker

Provides the ClusterTracker class, which follows molecular clusters over a trajectory.

The tracker only keeps the clusters and their ids up to date, frame by frame; the analyses built on them (and all file output) live in molclusters.molclusters.

Classes:

• ClusterTracker: Finds the clusters of the current frame and keeps their ids stable across formation, growth, split, merge and dissolution events.
• Transition: How the clusters of a frame came from those of the frame before.

Transition dataclass

Transition(flows: Mapping[tuple[int, int], int], born: frozenset[int] = frozenset(), merged: Mapping[int, int] = MappingProxyType({}), dissolved: frozenset[int] = frozenset())

How the clusters of a frame came from those of the frame before.

Merges and splits may involve any number of clusters. A merge ends every cluster that merged maps to the same survivor. A split shows as the new clusters whose sources include a previous cluster, however many: a cluster breaking into several pieces has a new cluster per piece but the one it continues in, and a new cluster made of pieces of several clusters has each of them as a source.

For the first frame, every cluster is born, of free molecules.

Attributes:

Name	Type	Description
flows	Mapping[tuple[int, int], int]	(previous cluster id, current cluster id) -> how many molecules went from the one to the other, id 0 standing for no cluster (free molecules, or a group of an ignored composition). A cluster that carries on keeps its own molecules as a flow to itself; molecules free in both frames aren't counted.
born	frozenset[int]	The clusters new in this frame.
merged	Mapping[int, int]	Each previous cluster that ended in a merge -> the cluster that absorbed it (both chose the same group, and the other continues in it; see ClusterTracker.update).
dissolved	frozenset[int]	The previous clusters that ended without merging: no group held two or more of their molecules.

ended property

ended: frozenset[int]

The previous clusters that ended in this frame, merged or dissolved.

Returns:

Type	Description
frozenset[int]	Their ids.

sources

sources(cluster_id: int) -> dict[int, int]

Tell where the molecules of a current cluster came from.

Parameters:

Name	Type	Description	Default
cluster_id	int	A cluster of the current frame.	required

Returns:

Type	Description
dict[int, int]	Previous cluster id (0 for free molecules) -> how many of its molecules the cluster holds; a cluster that carries on counts its own.

Source code in src/molclusters/tracker.py

def sources(self, cluster_id: int) -> dict[int, int]:
    """Tell where the molecules of a current cluster came from.

    Parameters
    ----------
    cluster_id : int
        A cluster of the current frame.

    Returns
    -------
    dict[int, int]
        Previous cluster id (``0`` for free molecules) -> how many of its
        molecules the cluster holds; a cluster that carries on counts its own.
    """
    return {prev: n for (prev, cur), n in self.flows.items() if cur == cluster_id}

destinations

destinations(cluster_id: int) -> dict[int, int]

Tell where the molecules of a previous cluster went.

Parameters:

Name	Type	Description	Default
cluster_id	int	A cluster of the previous frame.	required

Returns:

Type	Description
dict[int, int]	Current cluster id (0 for none) -> how many of its molecules went there; a cluster that carries on counts its own.

Source code in src/molclusters/tracker.py

def destinations(self, cluster_id: int) -> dict[int, int]:
    """Tell where the molecules of a previous cluster went.

    Parameters
    ----------
    cluster_id : int
        A cluster of the previous frame.

    Returns
    -------
    dict[int, int]
        Current cluster id (``0`` for none) -> how many of its molecules went
        there; a cluster that carries on counts its own.
    """
    return {cur: n for (prev, cur), n in self.flows.items() if prev == cluster_id}

absorbed

absorbed(cluster_id: int) -> list[int]

Tell which previous clusters merged into a current one.

Parameters:

Name	Type	Description	Default
cluster_id	int	A cluster of the current frame.	required

Returns:

Type	Description
list[int]	The ids of the clusters it absorbed, in increasing order (empty if none).

Source code in src/molclusters/tracker.py

def absorbed(self, cluster_id: int) -> list[int]:
    """Tell which previous clusters merged into a current one.

    Parameters
    ----------
    cluster_id : int
        A cluster of the current frame.

    Returns
    -------
    list[int]
        The ids of the clusters it absorbed, in increasing order (empty if
        none).
    """
    return sorted(lost for lost, into in self.merged.items() if into == cluster_id)

ClusterTracker

ClusterTracker(universe: Universe, config: MolClsConfig)

Follows the molecular clusters of a trajectory, keeping their ids stable.

The clusters of the frame the Universe is on are found on creation; after moving the trajectory to another frame, update reconciles them with it.

Attributes:

Name	Type	Description
uni	Universe	The MDAnalysis Universe object associated with the simulation.
config	MolClsConfig	The analysis configuration (only its rules, distance backend and ignored compositions are used).
sels	dict[str, AtomGroup]	Atom groups for each residue type.
conntab	ConnectionTable	The connectivity table for molecular clusters.
clusters	dict[int, Cluster]	A dictionary of detected clusters, keyed by cluster ID.
mol_clt	dict[int, int]	A mapping of molecule IDs to their respective cluster IDs.
transition	Transition	How the current clusters came from the previous frame's (on the first frame, every cluster is born).

Initialize the tracker with the clusters of the current frame.

Parameters:

Name	Type	Description	Default
universe	Universe	The MDAnalysis Universe object associated with the simulation.	required
config	MolClsConfig	The analysis configuration.	required

Source code in src/molclusters/tracker.py

def __init__(self, universe: mda.Universe, config: MolClsConfig) -> None:
    """Initialize the tracker with the clusters of the current frame.

    Parameters
    ----------
    universe : mda.Universe
        The MDAnalysis Universe object associated with the simulation.
    config : MolClsConfig
        The analysis configuration.
    """
    self.uni = universe
    self.config = config

    self.sels: dict[str, core.groups.AtomGroup] = {
        res: self.uni.select_atoms(f"resname {res}")
        for res in config._rules.all_keys()
    }

    self.conntab = ConnectionTable(
        self.uni,
        self.config._rules,
        self.sels,
        backend=self.config.distance_backend,
    )
    self.clusters: dict[int, Cluster] = {}
    self.mol_clt: dict[int, int] = {}
    # each tracker numbers its own clusters, from 1
    self._ids = count(1)

    self._start_clusters()

update

update() -> None

Reconcile the current frame's connected groups with the previous clusters.

This is the dominance algorithm, which keeps cluster ids stable while clusters grow, shrink, split and merge.

A connected group (subconn in the code, a ConnectionTable._SubConnTable) is a set of molecules linked to each other, directly or through other molecules, by the config's rules in the current frame, and to nothing outside the set. Free (unlinked) molecules form no connected group. A connected group has no id yet; a cluster is what it becomes once this method gives it one, new or inherited from the previous frame.

The connection table is rebuilt for the current frame, and the previous clusters that the molecules of each connected group (ignored compositions excluded) came from are counted (see _count_origins). Ids are then assigned in two steps, each of which sees the whole frame:

1. Each previous cluster picks its best group: the one holding the most of its molecules, the purest on a tie (see _best_groups).
2. Each connected group chosen by one or more previous clusters continues the one that contributed the most molecules, the oldest on a tie (see _get_older_cluster). A group chosen by nobody is a new cluster.

The usual events follow from these two rules:

• formation: a group of free molecules is chosen by nobody -> new;
• growth or shrinking: a group chosen by one cluster continues it;
• split: the cluster continues in its best fragment and the other fragments, chosen by nobody, are new;
• merge: several clusters choose the same group, one continues and the others die, even if they also left a remnant elsewhere;
• dissolution: a cluster that contributed at most one molecule to every group has no best group and dies.

So a mixed dimer (one molecule from each origin) is always new, and each id continues in at most one group. Previous clusters that no connected group continues are dropped, along with the mol_clt entries of molecules that are now free. What happened is kept in transition: where each cluster's molecules came from, which clusters are new, and which ended, merged into another or dissolved.

Source code in src/molclusters/tracker.py

def update(self) -> None:
    """Reconcile the current frame's connected groups with the previous clusters.

    This is the dominance algorithm, which keeps cluster ids stable while
    clusters grow, shrink, split and merge.

    A *connected group* (``subconn`` in the code, a
    `ConnectionTable._SubConnTable`) is a set of molecules linked to each
    other, directly or through other molecules, by the config's rules in the
    current frame, and to nothing outside the set. Free (unlinked) molecules
    form no connected group. A connected group has no id yet; a *cluster* is
    what it becomes once this method gives it one, new or inherited from the
    previous frame.

    The connection table is rebuilt for the current frame, and the previous
    clusters that the molecules of each connected group (ignored compositions
    excluded) came from are counted (see `_count_origins`). Ids
    are then assigned in two steps, each of which sees the whole frame:

    1. Each previous cluster picks its best group: the one holding the most of
       its molecules, the purest on a tie (see `_best_groups`).
    2. Each connected group chosen by one or more previous clusters continues
       the one that contributed the most molecules, the oldest on a tie (see
       `_get_older_cluster`). A group chosen by nobody is a new cluster.

    The usual events follow from these two rules:

    - formation: a group of free molecules is chosen by nobody -> new;
    - growth or shrinking: a group chosen by one cluster continues it;
    - split: the cluster continues in its best fragment and the other fragments,
      chosen by nobody, are new;
    - merge: several clusters choose the same group, one continues and the
      others die, even if they also left a remnant elsewhere;
    - dissolution: a cluster that contributed at most one molecule to every
      group has no best group and dies.

    So a mixed dimer (one molecule from each origin) is always new, and each id
    continues in at most one group. Previous clusters that no connected group
    continues are dropped, along with the `mol_clt` entries of molecules that
    are now free. What happened is kept in `transition`: where each cluster's
    molecules came from, which clusters are new, and which ended, merged into
    another or dissolved.
    """
    self.conntab.update()

    modified_mols = set()
    modified_clusters = set()

    conn_info = [
        (sub, self._count_origins(sub))
        for sub in self.conntab.subconntables()
        if not self.config.is_ignored_composition(sub.resnames)
    ]

    candidates = defaultdict(list)
    for cls_id, i in self._best_groups(conn_info).items():
        candidates[i].append(cls_id)

    flows: Counter[tuple[int, int]] = Counter()
    born = set()
    merged = {}
    for i, (subconn, origin_clusters) in enumerate(conn_info):
        if candidates[i]:
            id = self._get_older_cluster(candidates[i], origin_clusters)
            self.clusters[id]._update(subconn)
            # every other cluster that chose this group merged into it
            merged.update((lost, id) for lost in candidates[i] if lost != id)
        else:
            id = self._create_new_cluster(subconn)
            born.add(id)

        for origin, n in origin_clusters.items():
            flows[origin, id] = n

        for mol in subconn:
            self.mol_clt[mol] = id

        modified_clusters.add(id)
        modified_mols.update(subconn)

    for mol in set(self.mol_clt.keys()).difference(modified_mols):
        flows[self.mol_clt.pop(mol), 0] += 1

    # the merged clusters continue in no group, so they end here too
    ended = set(self.clusters.keys()).difference(modified_clusters)
    for cls in ended:
        self.clusters.pop(cls)

    self.transition = Transition(
        MappingProxyType(dict(flows)),
        born=frozenset(born),
        merged=MappingProxyType(merged),
        dissolved=frozenset(ended.difference(merged)),
    )

find

find(mol: int) -> int | None

Find the cluster ID for a given molecule.

Parameters:

Name	Type	Description	Default
mol	int	The molecule ID to search for.	required

Returns:

Type	Description
int | None	The cluster ID, or None if the molecule is in no cluster.

Source code in src/molclusters/tracker.py

def find(self, mol: int) -> int | None:
    """Find the cluster ID for a given molecule.

    Parameters
    ----------
    mol : int
        The molecule ID to search for.

    Returns
    -------
    int | None
        The cluster ID, or None if the molecule is in no cluster.
    """
    return self.mol_clt.get(mol)