molclusters

The package exports the names most code needs (see the overview). MolClusters, the runner, is defined in molclusters.molclusters:

molclusters

Provides the MolClusters class for analyzing molecular clusters in molecular dynamics simulations.

Classes:

• MolClusters: Runs the analyses the config enables (and any given ones) on the clusters of every frame of a trajectory.

The clusters themselves, and their ids, are tracked by tracker.ClusterTracker; the analyses are in the analysis package.

Dependencies:

• MDAnalysis: For molecular dynamics trajectory and structure analysis.
• NumPy: For numerical computations.
• tqdm: For progress tracking during analysis.

MolClusters

MolClusters(universe: Universe, config: MolClsConfig, analyses: Iterable[FrameAnalysis] = ())

A class for analyzing molecular clusters in molecular dynamics simulations.

It tracks the clusters frame by frame and runs the analyses on them: the built-ins the config enables (cluster sizes, lineage, solute-solvent, coordinates, nuclei; see analysis.builtins), then the ones given to it, then the JSON report, which every other analysis can add to. A built-in runs when the config options it needs are set, unless the config's analyses turns it off.

Attributes:

Name	Type	Description
uni	Universe	The MDAnalysis Universe object associated with the simulation.
config	MolClsConfig	The analysis configuration.
tracker	ClusterTracker | None	Follows the clusters frame by frame, keeping their ids stable; created by run (None before), and left at the last frame.
analyses	list[FrameAnalysis]	The analyses run on the clusters of every frame: the built-ins the config enables, then the given ones, then the report (see analysis to find one).
output	RunOutput	Where the last run wrote its files.

Initialize the MolClusters object.

The residues must be numbered 1 to N in topology order; otherwise a ValueError says so (see conntable.check_resids).

Parameters:

Name	Type	Description	Default
universe	Universe	The MDAnalysis Universe object associated with the simulation.	required
config	MolClsConfig	The analysis configuration.	required
analyses	Iterable[FrameAnalysis]	Analyses to run in addition to the ones the config enables, after them (but before the report) and in the order given.	()

Raises:

Type	Description
TypeError	If one of analyses isn't a FrameAnalysis instance.

Source code in src/molclusters/molclusters.py

def __init__(
    self,
    universe: mda.Universe,
    config: MolClsConfig,
    analyses: Iterable[FrameAnalysis] = (),
) -> None:
    """Initialize the MolClusters object.

    The residues must be numbered 1 to N in topology order; otherwise a
    ValueError says so (see `conntable.check_resids`).

    Parameters
    ----------
    universe : mda.Universe
        The MDAnalysis Universe object associated with the simulation.
    config : MolClsConfig
        The analysis configuration.
    analyses : Iterable[FrameAnalysis]
        Analyses to run in addition to the ones the config enables, after them
        (but before the report) and in the order given.

    Raises
    ------
    TypeError
        If one of `analyses` isn't a `FrameAnalysis` instance.
    """
    check_resids(universe)
    extra = list(analyses)
    for analysis in extra:
        if not isinstance(analysis, FrameAnalysis):
            hint = (
                " (pass an instance, not the class)"
                if isinstance(analysis, type)
                else ""
            )
            raise TypeError(
                f"Analyses must be FrameAnalysis instances, got {analysis!r}{hint}."
            )

    self.uni = universe
    self.config = config

    self.__check_resnames()
    logger.info(f"Effective configuration:\n{config.describe()}")

    self.tracker: ClusterTracker | None = None  # created by run()

    self.analyses: list[FrameAnalysis] = build_builtins(config, extra)
    if not self.analyses:
        logger.warning(
            "Every analysis is turned off ('analyses' in the config): the clusters "
            "will be tracked, but no results written."
        )

analysis

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

Find the first of the analyses of type kind, for its results.

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 there's none (e.g. the config doesn't enable that built-in, or turns it off).

Source code in src/molclusters/molclusters.py

def analysis[T: FrameAnalysis](self, kind: type[T]) -> T | None:
    """Find the first of the analyses of type `kind`, for its results.

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

    Returns
    -------
    T | None
        The analysis, or None if there's none (e.g. the config doesn't enable
        that built-in, or turns it off).
    """
    for analysis in self.analyses:
        if isinstance(analysis, kind):
            return analysis
    return None

__check_resnames

__check_resnames() -> None

Warn about residue names in the config that the topology doesn't have.

A misspelled or missing name selects no atoms, so everything built on it would silently come out empty.

Source code in src/molclusters/molclusters.py

def __check_resnames(self) -> None:
    """Warn about residue names in the config that the topology doesn't have.

    A misspelled or missing name selects no atoms, so everything built on it
    would silently come out empty.
    """
    present = set(self.uni.residues.resnames)
    comps = self.config.ignore_composition or []
    fields = {
        "rules": self.config._rules.all_keys(),
        "solute": self.config.solute or [],
        "solvent": self.config.solvent or [],
        "nucleus": self.config.nucleus or [],
        "ignore_composition": {name for comp in comps for name in comp},
    }
    for field, names in fields.items():
        missing = sorted(set(names) - present)
        if missing:
            logger.warning(
                f"'{field}' names residue(s) {missing} that are not in the "
                f"topology (residue names found: {sorted(present)})."
            )

__check_previous_outputs

__check_previous_outputs(directory: Path) -> None

Tell which results of an earlier run in the output directory get overwritten.

Parameters:

Name	Type	Description	Default
directory	Path	The directory this run writes to.	required

Source code in src/molclusters/molclusters.py

def __check_previous_outputs(self, directory: path.Path) -> None:
    """Tell which results of an earlier run in the output directory get overwritten.

    Parameters
    ----------
    directory : path.Path
        The directory this run writes to.
    """
    overwritten = []
    for out in self.__declared_outputs():
        files = out.existing(directory)
        if files:
            overwritten.append(
                f"{out.name} ({len(files)} file(s))" if out.is_pattern else out.name
            )
    if overwritten:
        logger.info(f"Overwriting results of an earlier run: {overwritten}")

__check_leftover_outputs

__check_leftover_outputs() -> None

Warn about an earlier run's output files that this run didn't overwrite.

Families of files (e.g. one .gro file per cluster id) needn't have the same members from one run to the next, so some of the files in the output directory may still be an earlier run's.

Source code in src/molclusters/molclusters.py

def __check_leftover_outputs(self) -> None:
    """Warn about an earlier run's output files that this run didn't overwrite.

    Families of files (e.g. one .gro file per cluster id) needn't have the same
    members from one run to the next, so some of the files in the output
    directory may still be an earlier run's.
    """
    directory = self.output.directory
    leftovers = sorted(
        {
            name
            for out in self.__declared_outputs()
            for file in out.existing(directory)
            if (name := file.relative_to(directory).as_posix())
            not in self.output.written
        }
    )
    if leftovers:
        shown = ", ".join(leftovers[:5]) + (", ..." if len(leftovers) > 5 else "")
        logger.warning(
            f"{len(leftovers)} file(s) of an earlier run are left in {directory} "
            f"({shown}): this run didn't write them. Delete them to keep only "
            "this run's results."
        )

__declared_outputs

__declared_outputs() -> list[OutputFile]

List the files this run writes, in the order they're reported.

Returns:

Type	Description
list[OutputFile]	The analyses' outputs, in the order of the analyses.

Source code in src/molclusters/molclusters.py

def __declared_outputs(self) -> list[OutputFile]:
    """List the files this run writes, in the order they're reported.

    Returns
    -------
    list[OutputFile]
        The analyses' outputs, in the order of the analyses.
    """
    return [out for analysis in self.analyses for out in analysis.outputs]

__log_rule_connections

__log_rule_connections(n_frames: int) -> None

Log how many connections each rule found, warning about unused rules.

Parameters:

Name	Type	Description	Default
n_frames	int	The number of frames analysed.	required

Source code in src/molclusters/molclusters.py

def __log_rule_connections(self, n_frames: int) -> None:
    """Log how many connections each rule found, warning about unused rules.

    Parameters
    ----------
    n_frames : int
        The number of frames analysed.
    """
    rules = self.config._rules
    counts = self.tracker.conntab.rule_connections
    present = set(self.uni.residues.resnames)

    per_frame = ", ".join(
        f"{mi} - {mj} ({rules[mi, mj]}) {counts[mi, mj] / n_frames:.1f}"
        for mi, mj in sorted(rules)
    )
    logger.info(f"Connections per frame, by rule: {per_frame}")

    # rules naming residues missing from the topology were warned about already
    unused = [
        f"{mi} - {mj} ({rules[mi, mj]})"
        for mi, mj in sorted(rules)
        if counts[mi, mj] == 0 and {mi, mj} <= present
    ]
    if unused:
        logger.warning(
            f"Rule(s) {', '.join(unused)} never connected any molecules in "
            f"{n_frames} frame(s): check their cutoffs (distances in angstrom, "
            "angles in degrees)."
        )

run

run(output_dir: Path | str | None = None) -> None

Track the clusters over the whole trajectory and run the analyses on them.

Every run starts over from the first frame with a new tracker (cluster ids start from 1 again), and the analyses start over in prepare, so running again gives the same results.

Parameters:

Name	Type	Description	Default
output_dir	Path | str | None	The directory to write the results to, created if missing; the current directory when None.	None

Source code in src/molclusters/molclusters.py

def run(self, output_dir: path.Path | str | None = None) -> None:
    """Track the clusters over the whole trajectory and run the analyses on them.

    Every run starts over from the first frame with a new tracker (cluster ids
    start from 1 again), and the analyses start over in `prepare`, so running
    again gives the same results.

    Parameters
    ----------
    output_dir : path.Path | str | None
        The directory to write the results to, created if missing; the current
        directory when None.
    """
    directory = (
        path.Path.cwd() if output_dir is None else path.Path(output_dir).absolute()
    )
    directory.mkdir(parents=True, exist_ok=True)

    n_frames = len(self.uni.trajectory)
    self.__check_previous_outputs(directory)
    logger.info(f"Tracking clusters over {n_frames} frame(s)")
    start = time.perf_counter()
    # the bar only shows on the terminal, so the log files get a line every 10%
    progress_step = max(1, n_frames // 10)

    self.uni.trajectory[0]
    self.tracker = ClusterTracker(self.uni, self.config)
    tracking = time.perf_counter() - start  # the analyses' time is kept by `run`

    self.output = RunOutput(directory)
    run = Run(self.uni, self.config, n_frames, self.output, self.analyses)

    # the appended files are buffered, so write what was already rendered even
    # if the run is interrupted, as appending it frame by frame used to; the
    # output is closed after `finish`, which may append too
    with self.output:
        run.prepare_analyses()
        run.analyse_frame(Frame(0, self.tracker, self.output))

        with tqdm(total=n_frames, initial=1, mininterval=5, miniters=10) as pbar:
            for i, _ in enumerate(self.uni.trajectory[1:], start=1):
                update_start = time.perf_counter()
                self.tracker.update()
                tracking += time.perf_counter() - update_start
                run.analyse_frame(Frame(i, self.tracker, self.output))
                pbar.update()

                done = i + 1
                if done % progress_step == 0 and done < n_frames:
                    logger.bind(**FILE_ONLY).info(
                        f"Frame {done}/{n_frames} ({done / n_frames:.0%}), "
                        f"{format_duration(time.perf_counter() - start)} elapsed"
                    )

        elapsed = time.perf_counter() - start
        logger.info(
            f"Tracked {n_frames} frame(s) in {format_duration(elapsed)} "
            f"({n_frames / max(elapsed, 1e-9):.1f} frames/s)"
        )
        self.__log_rule_connections(n_frames)

        run.finish_analyses()

    durations = ", ".join(
        f"{type(analysis).__name__} {format_duration(seconds)}"
        for analysis, seconds in zip(self.analyses, run.durations, strict=True)
    )
    logger.info(
        f"Time spent tracking the clusters: {format_duration(tracking)}; "
        f"in each analysis: {durations}"
    )

    outputs = ", ".join(out.name for out in self.__declared_outputs())
    logger.info(f"Results written to {self.output.directory}: {outputs}")
    self.__check_leftover_outputs()