Skip to content

molclusters.analysis

The analyses run on the tracked clusters. FrameAnalysis, Run and Frame are in molclusters.analysis.base; the registry of built-ins, BUILTINS, in molclusters.analysis.builtins; each built-in in a module of its own. All of them can be imported from molclusters.analysis.

base

Provides FrameAnalysis, the base of the analyses run on the tracked clusters.

Classes:
  • FrameAnalysis: The base class of an analysis, run frame by frame.
  • Run: What an analysis sees of the run as a whole.
  • Frame: What an analysis sees of the current frame.

FrameAnalysis

Bases: ABC

An analysis of the tracked clusters, run frame by frame.

A run calls prepare once, then analyse for every frame (the first one included) once the clusters are up to date with it, and finally finish. The analyses of a run are called in order, so an analysis sees the results of the ones before it for the same frame.

Only analyse must be implemented. An analysis keeps its results as its own attributes; one that writes files does so through Run.output and declares them in outputs. The same analysis may be run more than once, so it should start its results over in prepare, not only in __init__.

An analysis logs with loguru's logger, as the rest of the package does; the run tags whatever is logged during its hooks with its class name (the analysis key of the record's extra), which the log shows. To keep the log readable over long trajectories:

  • analyse counts what's worth reporting (e.g. in a collections.Counter) instead of logging it, and finish logs a summary: one INFO line, or one warning saying how to fix the problem.
  • Anything logged per frame is DEBUG or TRACE, with loguru's own arguments (logger.debug("cluster {}: ...", cls.id)) rather than an f-string: a message filtered out by the log level is then never built.
  • A warning that could repeat is only logged once.

An analysis adds its own fields to the JSON report (JsonReport, which runs after every other analysis) by overriding report_frame, report_cluster, or both: what they return goes in each frame's, or each cluster's, record under the analysis' class name, its floats rounded as the report's are.

Attributes:

Name Type Description
outputs tuple[OutputFile, ...]

The files the analysis writes, to tell which results of an earlier run they overwrite (or leave behind), and to report them at the end. Usually a class attribute; an analysis whose files depend on its options sets it in __init__ instead, since it's read before prepare.

prepare 

prepare(run: Run) -> None

Get ready for a run, starting over any results. Does nothing unless overridden.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required
Source code in src/molclusters/analysis/base.py 
72
73
74
75
76
77
78
79
def prepare(self, run: "Run") -> None:  # noqa: B027 (optional hook)
    """Get ready for a run, starting over any results. Does nothing unless overridden.

    Parameters
    ----------
    run : Run
        The run about to start.
    """

analyse abstractmethod 

analyse(frame: Frame) -> None

Analyse the clusters of the current frame.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/base.py 
81
82
83
84
85
86
87
88
89
@abstractmethod
def analyse(self, frame: "Frame") -> None:
    """Analyse the clusters of the current frame.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """

finish 

finish(run: Run) -> None

Wrap up once every frame was analysed. Does nothing unless overridden.

Parameters:

Name Type Description Default
run Run

The run that just ended.

 required
Source code in src/molclusters/analysis/base.py 
91
92
93
94
95
96
97
98
def finish(self, run: "Run") -> None:  # noqa: B027 (optional hook)
    """Wrap up once every frame was analysed. Does nothing unless overridden.

    Parameters
    ----------
    run : Run
        The run that just ended.
    """

report_frame 

report_frame(frame: Frame) -> Mapping[str, Any] | None

Give what to add to the report's record of the current frame.

Called by JsonReport after every analysis has analysed the frame.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required

Returns:

Type Description
Mapping[str, Any] | None

The fields to add under the analysis' class name: JSON types, and numpy scalars or arrays. None adds nothing (the default).
Source code in src/molclusters/analysis/base.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def report_frame(self, frame: "Frame") -> Mapping[str, Any] | None:
    """Give what to add to the report's record of the current frame.

    Called by `JsonReport` after every analysis has analysed the frame.

    Parameters
    ----------
    frame : Frame
        The current frame.

    Returns
    -------
    Mapping[str, Any] | None
        The fields to add under the analysis' class name: JSON types, and numpy
        scalars or arrays. None adds nothing (the default).
    """
    return None

report_cluster 

report_cluster(frame: Frame, cluster: Cluster) -> Mapping[str, Any] | None

Give what to add to the report's record of a cluster of the current frame.

Called by JsonReport, for every cluster of the frame, after every analysis has analysed the frame.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
cluster Cluster

The cluster.

 required

Returns:

Type Description
Mapping[str, Any] | None

The fields to add under the analysis' class name: JSON types, and numpy scalars or arrays. None adds nothing (the default).
Source code in src/molclusters/analysis/base.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def report_cluster(
    self, frame: "Frame", cluster: Cluster
) -> Mapping[str, Any] | None:
    """Give what to add to the report's record of a cluster of the current frame.

    Called by `JsonReport`, for every cluster of the frame, after every analysis
    has analysed the frame.

    Parameters
    ----------
    frame : Frame
        The current frame.
    cluster : Cluster
        The cluster.

    Returns
    -------
    Mapping[str, Any] | None
        The fields to add under the analysis' class name: JSON types, and numpy
        scalars or arrays. None adds nothing (the default).
    """
    return None

Run 

Run(universe: Universe, config: MolClsConfig, n_frames: int, output: RunOutput, analyses: Sequence[FrameAnalysis])

What an analysis sees of the run as a whole.

Attributes:

Name Type Description
universe Universe

The Universe being analysed.
config MolClsConfig

The analysis configuration.
n_frames int

The number of frames of the run.
output RunOutput

Where the analyses write their files.
durations list[float]

The time spent in each analysis' hooks so far, in seconds, in the order of the analyses.

Initialize the context of a run.

Parameters:

Name Type Description Default
universe Universe

The Universe being analysed.

 required
config MolClsConfig

The analysis configuration.

 required
n_frames int

The number of frames of the run.

 required
output RunOutput

Where the analyses write their files.

 required
analyses Sequence[FrameAnalysis]

The analyses of the run, in the order they're called.

 required
Source code in src/molclusters/analysis/base.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def __init__(
    self,
    universe: mda.Universe,
    config: MolClsConfig,
    n_frames: int,
    output: RunOutput,
    analyses: Sequence[FrameAnalysis],
) -> None:
    """Initialize the context of a run.

    Parameters
    ----------
    universe : mda.Universe
        The Universe being analysed.
    config : MolClsConfig
        The analysis configuration.
    n_frames : int
        The number of frames of the run.
    output : RunOutput
        Where the analyses write their files.
    analyses : Sequence[FrameAnalysis]
        The analyses of the run, in the order they're called.
    """
    self.universe = universe
    self.config = config
    self.n_frames = n_frames
    self.output = output
    self.durations = [0.0] * len(analyses)
    self._analyses = analyses
    self._visible = len(analyses)

analyses property 

analyses: tuple[FrameAnalysis, ...]

The analyses of the run, in order; while preparing, only those before.

Returns:

Type Description
tuple[FrameAnalysis, ...]

The analyses whose results analysis finds (see there).

prepare_analyses 

prepare_analyses() -> None

Prepare every analysis of the run, in order (called by the runner).

Source code in src/molclusters/analysis/base.py 
201
202
203
204
205
206
207
def prepare_analyses(self) -> None:
    """Prepare every analysis of the run, in order (called by the runner)."""
    for i, analysis in enumerate(self._analyses):
        self._visible = i
        with self._hook(i, "prepare"):
            analysis.prepare(self)
    self._visible = len(self._analyses)

analyse_frame 

analyse_frame(frame: Frame) -> None

Run every analysis of the run on frame, in order (called by the runner).

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/base.py 
209
210
211
212
213
214
215
216
217
218
219
def analyse_frame(self, frame: "Frame") -> None:
    """Run every analysis of the run on `frame`, in order (called by the runner).

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    for i, analysis in enumerate(self._analyses):
        with self._hook(i, "analyse", frame.index):
            analysis.analyse(frame)

finish_analyses 

finish_analyses() -> None

Finish every analysis of the run, in order (called by the runner).

Source code in src/molclusters/analysis/base.py 
221
222
223
224
225
def finish_analyses(self) -> None:
    """Finish every analysis of the run, in order (called by the runner)."""
    for i, analysis in enumerate(self._analyses):
        with self._hook(i, "finish"):
            analysis.finish(self)

analysis 

analysis(kind: type[T]) -> T | None

Find the first analysis of the run of type kind, for its results.

While the analyses are being prepared, only those before the one being prepared are found, since only their results for a frame are ready when it analyses the frame.

Parameters:

Name Type Description Default
kind type[T]

The analysis class to look for (subclasses match too).

 required

Returns:

Type Description
T | None

The analysis, or None if the run has none (before this one).
Source code in src/molclusters/analysis/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def analysis[T: FrameAnalysis](self, kind: type[T]) -> T | None:
    """Find the first analysis of the run of type `kind`, for its results.

    While the analyses are being prepared, only those before the one being
    prepared are found, since only their results for a frame are ready when
    it analyses the frame.

    Parameters
    ----------
    kind : type[T]
        The analysis class to look for (subclasses match too).

    Returns
    -------
    T | None
        The analysis, or None if the run has none (before this one).
    """
    for analysis in self._analyses[: self._visible]:
        if isinstance(analysis, kind):
            return analysis
    return None

Frame 

Frame(index: int, tracker: ClusterTracker, output: RunOutput)

What an analysis sees of the current frame.

Attributes:

Name Type Description
index int

The index of the frame in the run (0 for its first frame).
time float

The time of the frame, as the trajectory reports it.
universe Universe

The Universe, positioned at this frame.
output RunOutput

Where the run's analyses write their files (the same as Run.output), e.g. to append this frame to a file.

Initialize the current frame of a run.

Parameters:

Name Type Description Default
index int

The index of the frame in the run.

 required
tracker ClusterTracker

The tracker, up to date with the frame.

 required
output RunOutput

Where the run's analyses write their files.

 required
Source code in src/molclusters/analysis/base.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def __init__(self, index: int, tracker: ClusterTracker, output: RunOutput) -> None:
    """Initialize the current frame of a run.

    Parameters
    ----------
    index : int
        The index of the frame in the run.
    tracker : ClusterTracker
        The tracker, up to date with the frame.
    output : RunOutput
        Where the run's analyses write their files.
    """
    self.index = index
    self.time: float = tracker.uni.coord.time
    self.universe = tracker.uni
    self.output = output
    self._tracker = tracker

clusters property 

clusters: Mapping[int, Cluster]

The clusters of the frame, keyed by cluster id (read-only).

Returns:

Type Description
Mapping[int, Cluster]

A read-only view of the clusters; the clusters must not be modified.

transition property 

transition: Transition

How the frame's clusters came from the previous frame's.

Returns:

Type Description
Transition

Where each cluster's molecules came from, which clusters are new, and which ended, merged or dissolved (on the run's first frame, every cluster is new).

find 

find(mol: int) -> int | None

Find the id of the cluster a molecule belongs to.

Parameters:

Name Type Description Default
mol int

The molecule's resid.

 required

Returns:

Type Description
int | None

The cluster id, or None if the molecule is in no cluster.
Source code in src/molclusters/analysis/base.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
def find(self, mol: int) -> int | None:
    """Find the id of the cluster a molecule belongs to.

    Parameters
    ----------
    mol : int
        The molecule's resid.

    Returns
    -------
    int | None
        The cluster id, or None if the molecule is in no cluster.
    """
    return self._tracker.find(mol)

builtins

Provides BUILTINS, the built-in analyses, in the order they run.

Classes:
  • Builtin: A built-in analysis, how the config builds it and what it needs.
Functions:
  • build_builtins: Build the built-in analyses a config runs.

A built-in runs unless the config's analyses turns it off (by class name), and only when the config options it needs are set. Adding one takes a line in BUILTINS; the config checks analyses against it. They run in its order, the report last (after the analyses given to MolClusters too).

Builtin dataclass 

Builtin(kind: type[FrameAnalysis], build: Callable[[MolClsConfig], FrameAnalysis], needs: tuple[str, ...] = (), last: bool = False)

A built-in analysis, how the config builds it and what it needs.

Attributes:

Name Type Description
kind type[FrameAnalysis]

The analysis class; its name turns it on or off in the config's analyses.
build Callable[[MolClsConfig], FrameAnalysis]

Builds the analysis from the config, once the options it needs are set.
needs tuple[str, ...]

The config options it needs, without which it doesn't run.
last bool

Whether it runs after every other analysis, the ones given to MolClusters included: the report, which the others add to.

name property 

name: str

The analysis' class name, as the config's analyses and the log name it.

Returns:

Type Description
str

E.g. "JsonReport".

runs 

runs(config: MolClsConfig) -> bool

Whether the config runs this analysis.

Parameters:

Name Type Description Default
config MolClsConfig

The analysis configuration.

 required

Returns:

Type Description
bool

True unless analyses turns it off or an option it needs isn't set.
Source code in src/molclusters/analysis/builtins.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def runs(self, config: MolClsConfig) -> bool:
    """Whether the config runs this analysis.

    Parameters
    ----------
    config : MolClsConfig
        The analysis configuration.

    Returns
    -------
    bool
        True unless `analyses` turns it off or an option it needs isn't set.
    """
    return config.analyses.get(self.name, True) and all(
        getattr(config, option) is not None for option in self.needs
    )

build_builtins 

build_builtins(config: MolClsConfig, extra: Iterable[FrameAnalysis] = ()) -> list[FrameAnalysis]

Build the built-in analyses a config runs, and order them with extra.

Parameters:

Name Type Description Default
config MolClsConfig

The analysis configuration, with the topology-dependent defaults filled in.

 required
extra Iterable[FrameAnalysis]

Other analyses to run, after the built-ins but those that run last.

 ()

Returns:

Type Description
list[FrameAnalysis]

The analyses in the order they run: the built-ins the config runs (see Builtin.runs), then extra, then the built-ins that run last.
Source code in src/molclusters/analysis/builtins.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def build_builtins(
    config: MolClsConfig, extra: Iterable[FrameAnalysis] = ()
) -> list[FrameAnalysis]:
    """Build the built-in analyses a config runs, and order them with `extra`.

    Parameters
    ----------
    config : MolClsConfig
        The analysis configuration, with the topology-dependent defaults filled in.
    extra : Iterable[FrameAnalysis]
        Other analyses to run, after the built-ins but those that run `last`.

    Returns
    -------
    list[FrameAnalysis]
        The analyses in the order they run: the built-ins the config runs (see
        `Builtin.runs`), then `extra`, then the built-ins that run last.
    """
    built = [(b.last, b.build(config)) for b in BUILTINS if b.runs(config)]
    return [
        *(analysis for last, analysis in built if not last),
        *extra,
        *(analysis for last, analysis in built if last),
    ]

Built-in analyses

size

Provides SizeEvolution, the number and sizes of the clusters over time (evo.txt).

SizeEvolution 

SizeEvolution()

Bases: FrameAnalysis

Records the number of clusters and their sizes, frame by frame.

Attributes:

Name Type Description
data ndarray

One row per frame: time, number of clusters, and the minimum, average and maximum cluster size (all 0 in a frame without clusters).

Initialize an empty record, sized by prepare.

Source code in src/molclusters/analysis/size.py 
28
29
30
def __init__(self) -> None:
    """Initialize an empty record, sized by `prepare`."""
    self.data = np.zeros((0, 5))
prepare 
prepare(run: Run) -> None

Make room for every frame of the run.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required
Source code in src/molclusters/analysis/size.py 
32
33
34
35
36
37
38
39
40
def prepare(self, run: Run) -> None:
    """Make room for every frame of the run.

    Parameters
    ----------
    run : Run
        The run about to start.
    """
    self.data = np.zeros((run.n_frames, 5))
analyse 
analyse(frame: Frame) -> None

Record the number of clusters and their sizes in the current frame.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/size.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def analyse(self, frame: Frame) -> None:
    """Record the number of clusters and their sizes in the current frame.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    sizes = [cls.size for cls in frame.clusters.values()]
    if len(sizes) == 0:
        avg, min_size, max_size = 0, 0, 0
    else:
        avg = np.average(sizes)
        min_size = min(sizes)
        max_size = max(sizes)

    self.data[frame.index][0] = frame.time
    self.data[frame.index][1] = len(sizes)
    self.data[frame.index][2] = min_size
    self.data[frame.index][3] = avg
    self.data[frame.index][4] = max_size
finish 
finish(run: Run) -> None

Log a summary of the cluster counts and sizes and write them to evo.txt.

Parameters:

Name Type Description Default
run Run

The run that just ended.

 required
Source code in src/molclusters/analysis/size.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def finish(self, run: Run) -> None:
    """Log a summary of the cluster counts and sizes and write them to evo.txt.

    Parameters
    ----------
    run : Run
        The run that just ended.
    """
    n_clusters = self.data[:, 1]
    logger.info(
        f"Found {n_clusters.mean():.1f} cluster(s) per frame on average "
        f"({n_clusters.min():.0f}-{n_clusters.max():.0f}); the largest held "
        f"{self.data[:, 4].max():.0f} molecule(s)."
    )

    np.savetxt(
        run.output.path("evo.txt"),
        self.data,
        header="Time NClusters MinSize AvgSize MaxSize",
    )

lineage

Provides Lineage, where the clusters came from, what became of them and when.

It writes cluster_events.csv, the clusters' formations, splits, merges and dissolutions frame by frame, and cluster_lifetimes.csv, each cluster's birth, end and lifetime.

Lineage 

Lineage()

Bases: FrameAnalysis

Records where each cluster came from, what became of it, and when.

Events are read from each frame's Frame.transition and stamped with the first frame that shows them, so a cluster's DeathTime is the time of the first frame it's gone from, and its lifetime, DeathTime - BirthTime, a whole number of frame spacings (a cluster seen in a single frame lived one spacing).

cluster_events.csv has one row per event (see EVENT_COLUMNS); Other is the other cluster involved, if any, and NMols the molecules it concerns:

  • formation: Cluster is new, made only of free molecules (NMols, its size);
  • split: Cluster is new, and took NMols molecules from cluster Other. A cluster made of pieces of several clusters has a row for each, and one breaking into several pieces a row for each new piece;
  • merge: cluster Other ended, merged into Cluster with NMols of its molecules; a merge of several clusters has a row for each one absorbed;
  • dissolution: Cluster ended without merging (NMols, its last size).

The clusters of the run's first frame have no event: whatever formed them happened before the run.

cluster_lifetimes.csv has one row per cluster (see LIFETIME_COLUMNS): Origin is "initial" (there at the first frame), "formation" or "split", Parents the clusters it took molecules from (";"-separated, most molecules first), Fate "merge", "dissolution" or "alive" (at the last frame), and MergedInto the cluster that absorbed it. The lifetimes of clusters born at the start or alive at the end are censored: they lived at least that long. One alive at the end has no DeathTime, and its Lifetime is up to the last frame.

Each frame's Frame.transition also goes in the report (see report_frame), for the flows of molecules between clusters.

Attributes:

Name Type Description
lives dict[int, _Life]

Cluster id -> what is known of the cluster so far, for every cluster seen.

Initialize an empty record, started over by prepare.

Source code in src/molclusters/analysis/lineage.py 
118
119
120
121
def __init__(self) -> None:
    """Initialize an empty record, started over by `prepare`."""
    self.lives: dict[int, _Life] = {}
    self._time = math.nan
prepare 
prepare(run: Run) -> None

Start the record over, and cluster_events.csv with its header.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required
Source code in src/molclusters/analysis/lineage.py 
123
124
125
126
127
128
129
130
131
132
133
def prepare(self, run: Run) -> None:
    """Start the record over, and cluster_events.csv with its header.

    Parameters
    ----------
    run : Run
        The run about to start.
    """
    self.lives = {}
    self._time = math.nan
    run.output.append(EVENTS, ",".join(EVENT_COLUMNS) + "\n")
analyse 
analyse(frame: Frame) -> None

Record the current frame's events, and the clusters' sizes.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/lineage.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def analyse(self, frame: Frame) -> None:
    """Record the current frame's events, and the clusters' sizes.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    transition = frame.transition
    time = float(frame.time)
    self._time = time
    rows: list[tuple[str, int, int | None, int]] = []

    for cid in sorted(transition.born):
        size = frame.clusters[cid].size
        parents = sorted(
            ((prev, n) for prev, n in transition.sources(cid).items() if prev),
            key=lambda source: (-source[1], source[0]),
        )
        if frame.index == 0:
            origin = "initial"
        elif parents:
            origin = "split"
            rows.extend(("split", cid, prev, n) for prev, n in parents)
        else:
            origin = "formation"
            rows.append(("formation", cid, None, size))
        self.lives[cid] = _Life(
            birth_time=time,
            born_at_start=frame.index == 0,
            origin=origin,
            parents=tuple(prev for prev, _ in parents),
            birth_size=size,
            max_size=size,
            last_size=size,
        )

    for lost, into in sorted(transition.merged.items()):
        rows.append(("merge", into, lost, transition.flows[lost, into]))
        self._end(lost, time, "merge", into)
    for cid in sorted(transition.dissolved):
        rows.append(("dissolution", cid, None, self.lives[cid].last_size))
        self._end(cid, time, "dissolution")

    for cid, cls in frame.clusters.items():
        if cid not in transition.born:
            life = self.lives[cid]
            life.n_frames += 1
            life.last_size = cls.size
            life.max_size = max(life.max_size, life.last_size)

    number = int(frame.universe.coord.frame)
    for event, cid, other, n in rows:
        logger.debug(
            "Frame {}: " + _DESCRIPTIONS[event],
            number,
            cluster=cid,
            other=other,
            n=n,
        )
        other_text = "" if other is None else other
        frame.output.append(
            EVENTS, f"{number},{time!r},{event},{cid},{other_text},{n}\n"
        )
report_frame 
report_frame(frame: Frame) -> dict[str, Any]

Give how the frame's clusters came from the previous frame's, for the report.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required

Returns:

Type Description
dict[str, Any]

The frame's Frame.transition, sorted: flows, rows of (previous id, current id, molecules), id 0 for no cluster; born; merged, rows of (absorbed id, id it merged into); and dissolved.
Source code in src/molclusters/analysis/lineage.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def report_frame(self, frame: Frame) -> dict[str, Any]:
    """Give how the frame's clusters came from the previous frame's, for the report.

    Parameters
    ----------
    frame : Frame
        The current frame.

    Returns
    -------
    dict[str, Any]
        The frame's `Frame.transition`, sorted: ``flows``, rows of (previous
        id, current id, molecules), id 0 for no cluster; ``born``; ``merged``,
        rows of (absorbed id, id it merged into); and ``dissolved``.
    """
    transition = frame.transition
    return {
        "flows": [
            [prev, cur, n] for (prev, cur), n in sorted(transition.flows.items())
        ],
        "born": sorted(transition.born),
        "merged": [list(pair) for pair in sorted(transition.merged.items())],
        "dissolved": sorted(transition.dissolved),
    }
lifetimes 
lifetimes() -> DataFrame

Tabulate every cluster seen so far, as cluster_lifetimes.csv has them.

Returns:

Type Description
DataFrame

One row per cluster, by id (see LIFETIME_COLUMNS).
Source code in src/molclusters/analysis/lineage.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def lifetimes(self) -> pd.DataFrame:
    """Tabulate every cluster seen so far, as cluster_lifetimes.csv has them.

    Returns
    -------
    pd.DataFrame
        One row per cluster, by id (see `LIFETIME_COLUMNS`).
    """
    rows = [
        [
            cid,
            life.birth_time,
            life.death_time,
            (self._time if life.fate == "alive" else life.death_time)
            - life.birth_time,
            life.n_frames,
            life.born_at_start,
            life.fate == "alive",
            life.origin,
            ";".join(map(str, life.parents)),
            life.fate,
            life.merged_into,
            life.birth_size,
            life.max_size,
            life.last_size,
        ]
        for cid, life in sorted(self.lives.items())
    ]
    table = pd.DataFrame(rows, columns=LIFETIME_COLUMNS)
    table["MergedInto"] = table["MergedInto"].astype("Int64")
    return table
finish 
finish(run: Run) -> None

Log a summary of the events and lifetimes, and write cluster_lifetimes.csv.

Parameters:

Name Type Description Default
run Run

The run that just ended.

 required
Source code in src/molclusters/analysis/lineage.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def finish(self, run: Run) -> None:
    """Log a summary of the events and lifetimes, and write cluster_lifetimes.csv.

    Parameters
    ----------
    run : Run
        The run that just ended.
    """
    table = self.lifetimes()
    table.to_csv(run.output.path(LIFETIMES), index=False)

    origins = Counter(table["Origin"])
    fates = Counter(table["Fate"])
    message = (
        f"Cluster lineage: {origins['formation']} cluster(s) formed of free "
        f"molecules and {origins['split']} split off others; "
        f"{fates['merge']} merged into another and {fates['dissolution']} "
        "dissolved."
    )
    # lifetimes neither censored at the start nor at the end
    whole = table[~table["BornAtStart"] & ~table["AliveAtEnd"]]
    if len(whole):
        single = int((whole["NFrames"] == 1).sum())
        message += (
            f" The {len(whole)} born and ended within the run lived "
            f"{np.median(whole['Lifetime']):.4g} ps (median); {single} of them "
            "for a single frame."
        )
    logger.info(message)

solute

Provides SoluteSolvent, the clusters holding both solutes and solvents (solute_solvent.csv).

SoluteSolvent 

SoluteSolvent(solutes: Iterable[str], solvents: Iterable[str])

Bases: FrameAnalysis

Records the clusters that hold both solute and solvent molecules, frame by frame.

Attributes:

Name Type Description
solutes set[str]

The residue names of the solutes.
solvents set[str]

The residue names of the solvents.
data ndarray

One row per frame (see COLUMNS): time, the number of such clusters, their average number of solutes and of solvents (0 without such clusters), and their average radius, density, charge, dipole moment, sphericity and shape parameter (NaN without such clusters).

Initialize an empty record, sized by prepare.

Parameters:

Name Type Description Default
solutes Iterable[str]

The residue names of the solutes.

 required
solvents Iterable[str]

The residue names of the solvents.

 required
Source code in src/molclusters/analysis/solute.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(self, solutes: Iterable[str], solvents: Iterable[str]) -> None:
    """Initialize an empty record, sized by `prepare`.

    Parameters
    ----------
    solutes : Iterable[str]
        The residue names of the solutes.
    solvents : Iterable[str]
        The residue names of the solvents.
    """
    self.solutes = set(solutes)
    self.solvents = set(solvents)
    self.data = np.zeros((0, len(COLUMNS)))
prepare 
prepare(run: Run) -> None

Make room for every frame of the run.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required
Source code in src/molclusters/analysis/solute.py 
62
63
64
65
66
67
68
69
70
def prepare(self, run: Run) -> None:
    """Make room for every frame of the run.

    Parameters
    ----------
    run : Run
        The run about to start.
    """
    self.data = np.zeros((run.n_frames, len(COLUMNS)))
analyse 
analyse(frame: Frame) -> None

Record the clusters holding both solutes and solvents in the current frame.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/solute.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def analyse(self, frame: Frame) -> None:
    """Record the clusters holding both solutes and solvents in the current frame.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    n_cls = 0
    n_solvents = []
    n_solutes = []
    radius = []
    dipole = []
    density = []
    sphericity = []
    shape = []
    charge = []

    for cls in self._solute_solvent_clusters(frame):
        n_cls += 1

        composition = cls.composition
        n_solvents.append(sum(composition.get(name, 0) for name in self.solvents))
        n_solutes.append(sum(composition.get(name, 0) for name in self.solutes))

        radius.append(cls.radius)
        dipole.append(cls.dipole_moment)
        density.append(cls.density)
        sphericity.append(cls.sphericity)
        shape.append(cls.shape_parameter)
        charge.append(cls.charge)

    row = self.data[frame.index]
    row[0] = frame.time
    row[1] = n_cls
    row[2] = 0 if n_cls == 0 else np.average(n_solutes)
    row[3] = 0 if n_cls == 0 else np.average(n_solvents)
    row[4] = np.nan if n_cls == 0 else np.average(radius)
    row[5] = np.nan if n_cls == 0 else np.average(density)
    row[6] = np.nan if n_cls == 0 else np.average(charge)
    row[7] = np.nan if n_cls == 0 else np.average(dipole)
    row[8] = np.nan if n_cls == 0 else np.average(sphericity)
    row[9] = np.nan if n_cls == 0 else np.average(shape)
finish 
finish(run: Run) -> None

Write the record to solute_solvent.csv.

Parameters:

Name Type Description Default
run Run

The run that just ended.

 required
Source code in src/molclusters/analysis/solute.py 
134
135
136
137
138
139
140
141
142
143
144
def finish(self, run: Run) -> None:
    """Write the record to solute_solvent.csv.

    Parameters
    ----------
    run : Run
        The run that just ended.
    """
    pd.DataFrame(self.data, columns=COLUMNS).to_csv(
        run.output.path("solute_solvent.csv"), index=False
    )

coordinates

Provides ClusterCoordinates, the coordinates of the clusters holding solutes (.gro files).

ClusterCoordinates 

ClusterCoordinates(solutes: Iterable[str], *, follow: bool = False, folder: str = 'coordinates')

Bases: FrameAnalysis

Writes the coordinates of every cluster holding a solute, frame by frame.

Each such cluster, made whole, is appended to cls-n<size>.gro (the clusters of one size, pooled) and cls-id<id>.gro (one cluster's own trajectory). When following the solutes, a cluster holding exactly one solute is also appended to solute-<resid>.gro. The files go in the folder of the output directory.

Attributes:

Name Type Description
solutes set[str]

The residue names of the solutes.
follow bool

Whether to write each solute's cluster to its own file.
folder str

The folder of the output directory the files go in.
solute_ids set[int]

The resids of the solute molecules, found by prepare.
follow_skipped Counter[int]

Cluster id -> number of frames it wasn't followed in, for holding more than one solute.

Initialize the analysis.

Parameters:

Name Type Description Default
solutes Iterable[str]

The residue names of the solutes.

 required
follow bool

Whether to write each solute's cluster to its own file.

 False
folder str

The folder of the output directory the files go in ("" for the output directory itself).

 'coordinates'
Source code in src/molclusters/analysis/coordinates.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    solutes: Iterable[str],
    *,
    follow: bool = False,
    folder: str = "coordinates",
) -> None:
    """Initialize the analysis.

    Parameters
    ----------
    solutes : Iterable[str]
        The residue names of the solutes.
    follow : bool
        Whether to write each solute's cluster to its own file.
    folder : str
        The folder of the output directory the files go in ("" for the
        output directory itself).
    """
    self.solutes = set(solutes)
    self.follow = follow
    self.folder = folder
    self.solute_ids: set[int] = set()
    self.follow_skipped: Counter[int] = Counter()

    self.outputs = (
        OutputFile(self._file("cls-n<size>.gro")),
        OutputFile(self._file("cls-id<id>.gro")),
    )
    if follow:
        self.outputs += (OutputFile(self._file("solute-<resid>.gro")),)
prepare 
prepare(run: Run) -> None

Find the solute molecules.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required
Source code in src/molclusters/analysis/coordinates.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def prepare(self, run: Run) -> None:
    """Find the solute molecules.

    Parameters
    ----------
    run : Run
        The run about to start.
    """
    self.solute_ids = {
        int(resid)
        for name in sorted(self.solutes)
        for resid in run.universe.select_atoms(f"resname {name}").residues.resids
    }
    self.follow_skipped = Counter()
analyse 
analyse(frame: Frame) -> None

Append the clusters of the current frame that hold a solute.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/coordinates.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def analyse(self, frame: Frame) -> None:
    """Append the clusters of the current frame that hold a solute.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    for cls in frame.clusters.values():
        sol_ids = self.solute_ids.intersection(cls.resids)
        if not sol_ids:
            continue

        text = self._render(cls, frame.time)

        # pooled by size: an ensemble of what an N-mer looks like, across all
        # clusters that were ever that size, independent of cluster identity
        frame.output.append(self._file(f"cls-n{cls.size}.gro"), text)

        # pooled by identity: this specific cluster's own trajectory, tracked
        # across frames via the dominance algorithm regardless of size changes
        frame.output.append(self._file(f"cls-id{cls.id}.gro"), text)

        # TODO: change to support merges
        # FIXME: with changes in config this needs to be updated
        if self.follow:
            if len(sol_ids) > 1:
                # TODO: make more feature-rich follow procedure
                # counted, not logged: `finish` sums them up
                self.follow_skipped[cls.id] += 1
                continue

            (sol_id,) = sol_ids

            frame.output.append(self._file(f"solute-{sol_id}.gro"), text)
finish 
finish(run: Run) -> None

Warn about the frames in which solutes weren't followed.

Parameters:

Name Type Description Default
run Run

The run that just ended.

 required
Source code in src/molclusters/analysis/coordinates.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def finish(self, run: Run) -> None:
    """Warn about the frames in which solutes weren't followed.

    Parameters
    ----------
    run : Run
        The run that just ended.
    """
    if self.follow_skipped:
        logger.warning(
            f"Solutes were not followed in {self.follow_skipped.total()} "
            f"frame(s) of {len(self.follow_skipped)} cluster(s) holding more than "
            "one solute: those frames are missing from the solute-<resid>.gro "
            "files. The cluster ids are logged at DEBUG level (--log-level DEBUG)."
        )
        logger.debug(f"Frames not followed, by cluster id: {self.follow_skipped}")

nucleus

Provides Nucleus, the nuclei inside the clusters (nucleus_data.csv).

Nucleus 

Nucleus(resnames: Iterable[str])

Bases: FrameAnalysis

Finds the nuclei inside the clusters, frame by frame.

A nucleus is a connected group, inside a cluster, of molecules with one of the nucleus residue names. Each cluster's nuclei also go in the report (see report_cluster).

Attributes:

Name Type Description
resnames set[str]

The residue names of the molecules that make up nuclei.
nuclei dict[int, list[MolGroup]]

Cluster id -> the cluster's nuclei in the current frame (an empty list for a cluster without nuclei), for the analyses after this one.
data ndarray

One row per frame (see COLUMNS): time, the average number of nuclei of the clusters that have any (0 without nuclei), and the nuclei's average size, radius, density, charge, dipole moment, sphericity and shape parameter (NaN without nuclei).

Initialize an empty record, sized by prepare.

Parameters:

Name Type Description Default
resnames Iterable[str]

The residue names of the molecules that make up nuclei.

 required
Source code in src/molclusters/analysis/nucleus.py 
57
58
59
60
61
62
63
64
65
66
67
def __init__(self, resnames: Iterable[str]) -> None:
    """Initialize an empty record, sized by `prepare`.

    Parameters
    ----------
    resnames : Iterable[str]
        The residue names of the molecules that make up nuclei.
    """
    self.resnames = set(resnames)
    self.nuclei: dict[int, list[MolGroup]] = {}
    self.data = np.full((0, len(COLUMNS)), np.nan)
prepare 
prepare(run: Run) -> None

Make room for every frame of the run.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required
Source code in src/molclusters/analysis/nucleus.py 
69
70
71
72
73
74
75
76
77
def prepare(self, run: Run) -> None:
    """Make room for every frame of the run.

    Parameters
    ----------
    run : Run
        The run about to start.
    """
    self.data = np.full((run.n_frames, len(COLUMNS)), np.nan)
analyse 
analyse(frame: Frame) -> None

Find the nuclei of the current frame and record their averages.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/nucleus.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def analyse(self, frame: Frame) -> None:
    """Find the nuclei of the current frame and record their averages.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    self.nuclei = {}
    n_nucleus = []
    sizes = []
    radius = []
    dipole = []
    density = []
    sphericity = []
    shape = []
    charge = []
    for cid, cls in frame.clusters.items():
        possible_nucleus = [
            rid
            for rnm, rid in zip(cls.resnames, cls.resids, strict=True)
            if rnm in self.resnames
        ]

        self.nuclei[cid] = []
        subcomps = nx.induced_subgraph(cls.graph, possible_nucleus)
        n_nuc = 0
        for sg in nx.connected_components(subcomps):
            tp = MolGroup(frame.universe, list(sg))

            n_nuc += 1
            sizes.append(tp.size)
            radius.append(tp.radius)
            dipole.append(tp.dipole_moment)
            density.append(tp.density)
            sphericity.append(tp.sphericity)
            shape.append(tp.shape_parameter)
            charge.append(tp.charge)

            self.nuclei[cid].append(tp)

        if n_nuc != 0:
            n_nucleus.append(n_nuc)

    row = self.data[frame.index]
    row[0] = frame.time
    row[1] = 0 if len(n_nucleus) == 0 else np.average(n_nucleus)
    row[2] = np.nan if len(n_nucleus) == 0 else np.average(sizes)
    row[3] = np.nan if len(n_nucleus) == 0 else np.average(radius)
    row[4] = np.nan if len(n_nucleus) == 0 else np.average(density)
    row[5] = np.nan if len(n_nucleus) == 0 else np.average(charge)
    row[6] = np.nan if len(n_nucleus) == 0 else np.average(dipole)
    row[7] = np.nan if len(n_nucleus) == 0 else np.average(sphericity)
    row[8] = np.nan if len(n_nucleus) == 0 else np.average(shape)
report_cluster 
report_cluster(frame: Frame, cluster: Cluster) -> dict[str, Any]

Give the nuclei of a cluster, for the report.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
cluster Cluster

The cluster.

 required

Returns:

Type Description
dict[str, Any]

nuclei, one report.group_record per nucleus, and combined_dipole_moment, the dipole moment of all of them together (None without nuclei).
Source code in src/molclusters/analysis/nucleus.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def report_cluster(self, frame: Frame, cluster: Cluster) -> dict[str, Any]:
    """Give the nuclei of a cluster, for the report.

    Parameters
    ----------
    frame : Frame
        The current frame.
    cluster : Cluster
        The cluster.

    Returns
    -------
    dict[str, Any]
        ``nuclei``, one `report.group_record` per nucleus, and
        ``combined_dipole_moment``, the dipole moment of all of them together
        (None without nuclei).
    """
    nuclei = self.nuclei.get(cluster.id, [])
    combined = reduce(operator.add, nuclei).dipole_moment if nuclei else None
    return {
        "nuclei": [group_record(nucleus) for nucleus in nuclei],
        "combined_dipole_moment": combined,
    }
finish 
finish(run: Run) -> None

Write the record to nucleus_data.csv.

Parameters:

Name Type Description Default
run Run

The run that just ended.

 required
Source code in src/molclusters/analysis/nucleus.py 
158
159
160
161
162
163
164
165
166
167
168
def finish(self, run: Run) -> None:
    """Write the record to nucleus_data.csv.

    Parameters
    ----------
    run : Run
        The run that just ended.
    """
    pd.DataFrame(self.data, columns=COLUMNS).to_csv(
        run.output.path("nucleus_data.csv"), index=False
    )

report

Provides JsonReport, every cluster of every frame and its properties (molclusters.jsonl).

JsonReport 

JsonReport(compression: ReportCompression = 'zstd')

Bases: FrameAnalysis

Records every cluster of every frame, with its properties.

The report is written frame by frame, as JSON Lines, compressed as asked; its format is described in molclusters.report, which also reads it back. Every other analysis of the run can add fields to it (see FrameAnalysis.report_frame and report_cluster), so it runs after them all.

Attributes:

Name Type Description
compression ReportCompression

How the report is compressed: "zstd", "gzip" or "none".
name str

The report's file name, e.g. molclusters.jsonl.zst.

Initialize the report, started by prepare.

Parameters:

Name Type Description Default
compression ReportCompression

How to compress the report: "zstd", "gzip" or "none".

 'zstd'
Source code in src/molclusters/analysis/report.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(self, compression: ReportCompression = "zstd") -> None:
    """Initialize the report, started by `prepare`.

    Parameters
    ----------
    compression : ReportCompression
        How to compress the report: "zstd", "gzip" or "none".
    """
    self.compression = compression
    self.name = report_name(compression)
    self.outputs = (OutputFile(self.name),)
    self._frame_contributors: list[tuple[str, FrameAnalysis]] = []
    self._cluster_contributors: list[tuple[str, FrameAnalysis]] = []
prepare 
prepare(run: Run) -> None

Start the report with its header: the run's inputs and configuration.

Parameters:

Name Type Description Default
run Run

The run about to start.

 required

Raises:

Type Description
ValueError

If two analyses of the same class name would add to the report.
Source code in src/molclusters/analysis/report.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def prepare(self, run: Run) -> None:
    """Start the report with its header: the run's inputs and configuration.

    Parameters
    ----------
    run : Run
        The run about to start.

    Raises
    ------
    ValueError
        If two analyses of the same class name would add to the report.
    """
    contributors = [
        analysis
        for analysis in run.analyses
        if _overrides(analysis, "report_frame")
        or _overrides(analysis, "report_cluster")
    ]
    names = [type(analysis).__name__ for analysis in contributors]
    repeated = sorted({name for name in names if names.count(name) > 1})
    if repeated:
        # their fields would go under the same key
        raise ValueError(
            f"More than one {' and '.join(repeated)} analysis would add to the "
            "report: run only one of each."
        )

    self._frame_contributors = [
        (name, analysis)
        for name, analysis in zip(names, contributors, strict=True)
        if _overrides(analysis, "report_frame")
    ]
    self._cluster_contributors = [
        (name, analysis)
        for name, analysis in zip(names, contributors, strict=True)
        if _overrides(analysis, "report_cluster")
    ]

    header = {
        "format": FORMAT,
        "version": VERSION,
        "software": f"MolClusters {__version__}",
        "trajectory": _source(run.universe.trajectory),
        "topology": _source(run.universe),
        "n_frames": run.n_frames,
        "decimals": DECIMALS,
        "units": UNITS,
        "connection_columns": CONNECTION_COLUMNS,
        # this one runs last, so the run's analyses are those before it, and it
        "analyses": [type(a).__name__ for a in (*run.analyses, self)],
        "contributors": names,
        "config": run.config.model_dump(mode="json"),
    }
    run.output.append(self.name, dumps(header))
analyse 
analyse(frame: Frame) -> None

Add the clusters of the current frame to the report.

Parameters:

Name Type Description Default
frame Frame

The current frame.

 required
Source code in src/molclusters/analysis/report.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def analyse(self, frame: Frame) -> None:
    """Add the clusters of the current frame to the report.

    Parameters
    ----------
    frame : Frame
        The current frame.
    """
    clusters = []
    for cls in frame.clusters.values():
        record = cluster_record(cls)
        for name, analysis in self._cluster_contributors:
            fields = _contribution(analysis, "report_cluster", frame, cls)
            if fields is not None:
                record[name] = fields
        clusters.append(record)

    record = {
        "frame": int(frame.universe.coord.frame),
        "time": round(float(frame.time), DECIMALS),
        "clusters": clusters,
    }
    for name, analysis in self._frame_contributors:
        fields = _contribution(analysis, "report_frame", frame)
        if fields is not None:
            record[name] = fields
    frame.output.append(self.name, dumps(record))