molclusters.report

report

Provides the format of the JSON report (molclusters.jsonl), and its reader.

The report, written by analysis.JsonReport, is JSON Lines: one JSON object per line, compressed with zstd (molclusters.jsonl.zst, the default), gzip (.jsonl.gz) or not at all (.jsonl), as the config's report_compression says. It's written frame by frame, so a run never holds it in memory, and an interrupted run leaves every frame it analysed readable.

The first line is the header::

{"format": "molclusters-report", "version": 4, "software": "MolClusters x.y.z",
 "trajectory": <absolute path, or null for an in-memory trajectory>,
 "topology": <absolute path, or null>, "n_frames": <frames in the run>,
 "decimals": 6, "units": {<quantity>: <unit>, ...},
 "connection_columns": ["i", "j", "distance", "angle", "n_hbonds"],
 "analyses": [<every analysis of the run, in the order they ran>],
 "contributors": [<the analyses that add fields, see below>],
 "config": <the effective configuration>}

Then one line per frame::

{"frame": <trajectory frame number>, "time": <ps>, "clusters": [<cluster>, ...]}

and each cluster is::

{
    "id": 3,
    "birth_time": 120.0,
    "size": 3,
    "resids": [107, 169, 171],
    "composition": {"MAL": 3},
    "mass": 402.260990,
    "volume": 834.555760,
    "radius": 5.840575,
    "radius_of_gyration": 3.981872,
    "diameter": 11.681149,
    "density": 0.800390,
    "charge": 0.0,
    "dipole_moment": 15.696198,
    "sphericity": 0.738690,
    "shape_parameter": 0.138796,
    "connections": [[169, 107, 2.684737, 171.434846, 1], ...],
}

with birth_time the time of the frame the cluster first appeared in (the run's first frame's for the clusters already there), and each connection a row of connection_columns: the two resids, then the connection's distance (and, for H-bonds, the D-H-A angle and the number of H-bonds between the pair; null for "cm" rules). radius is the equivalent sphere's (see MolGroup.radius), and radius_of_gyration the mass-weighted Rg it derives from.

Other analyses add fields of their own (see FrameAnalysis.report_frame and report_cluster), under their class name in a frame's or a cluster's record. The Nucleus analysis adds to each cluster "Nucleus": {"nuclei": [<group>, ...], "combined_dipole_moment": <dipole of all its nuclei together, or null>}, a group being a cluster without its id, birth_time and connections. The Lineage analysis adds to each frame how its clusters came from the previous frame's (see tracker.Transition)::

"Lineage": {"flows": [[<previous id>, <current id>, <molecules>], ...],
            "born": [<id>, ...], "merged": [[<absorbed id>, <into id>], ...],
            "dissolved": [<id>, ...]}

where id 0 in a flow stands for no cluster (free molecules), and on the first frame every cluster is born.

Floats are rounded to DECIMALS decimal places, what other analyses add included, and NaN is written as null (a bare NaN isn't valid JSON). Units are those of UNITS.

Classes:

• Report: A report file, read frame by frame, or as tables.

Functions:

• read_report: Open a report file.
• report_name: The report's file name for a compression.
• dumps: Encode a record as one line of the report.
• rounded: Round every float in a value, as the report's are.
• group_record: Encode a group of molecules and its properties.
• cluster_record: Encode a cluster, its properties and connections.

Report

Report(path: Path | str)

A report file, read frame by frame, or as tables.

The file is read again by each call, frame by frame, so even a report too large for memory can be gone through. Its compression is told from its first bytes, not its name.

Attributes:

Name	Type	Description
path	Path	The report file.
header	dict[str, Any]	The header: the software, inputs, units and configuration of the run.

Open a report, reading its header.

Parameters:

Name	Type	Description	Default
path	Path | str	The report, compressed or not.	required

Raises:

Type	Description
ValueError	If the file isn't a MolClusters report this version can read.

Source code in src/molclusters/report.py

def __init__(self, path: Path | str) -> None:
    """Open a report, reading its header.

    Parameters
    ----------
    path : Path | str
        The report, compressed or not.

    Raises
    ------
    ValueError
        If the file isn't a MolClusters report this version can read.
    """
    self.path = Path(path)
    with self.__lines() as lines:
        first = next(lines, b"")

    try:
        header = orjson.loads(first)
    except orjson.JSONDecodeError:
        header = None
    if not isinstance(header, dict) or header.get("format") != FORMAT:
        raise ValueError(
            f"{self.path} is not a MolClusters report. The molclusters.json of "
            "MolClusters 0.6 and earlier is a single JSON document: read it "
            "with the json module."
        )
    if header.get("version", 0) > VERSION:
        raise ValueError(
            f"{self.path} is a version {header['version']} report, and this "
            f"MolClusters reads up to version {VERSION}: update MolClusters."
        )
    self.header: dict[str, Any] = header

__lines

__lines() -> Iterator[Iterator[bytes]]

Open the file, decompressed, as lines.

Yields:

Type	Description
Iterator[bytes]	The file's lines, newlines included.

Source code in src/molclusters/report.py

@contextmanager
def __lines(self) -> Iterator[Iterator[bytes]]:
    """Open the file, decompressed, as lines.

    Yields
    ------
    Iterator[bytes]
        The file's lines, newlines included.
    """
    with self.path.open("rb") as raw:
        magic = raw.read(4)
        raw.seek(0)
        stream: BinaryIO
        if magic.startswith(_GZIP_MAGIC):
            stream = gzip.GzipFile(fileobj=raw)
        elif magic == _ZSTD_MAGIC:
            reader = zstandard.ZstdDecompressor().stream_reader(
                raw, read_across_frames=True, closefd=False
            )
            stream = io.BufferedReader(reader)
        else:
            stream = raw
        with stream:
            yield iter(stream)

frames

frames() -> Iterator[dict[str, Any]]

Read the frames, one at a time.

A report that ends early (the run was interrupted, or killed) gives the frames it has, with a warning.

Yields:

Type	Description
dict[str, Any]	Each frame's record (see the module's docstring).

Source code in src/molclusters/report.py

def frames(self) -> Iterator[dict[str, Any]]:
    """Read the frames, one at a time.

    A report that ends early (the run was interrupted, or killed) gives the
    frames it has, with a warning.

    Yields
    ------
    dict[str, Any]
        Each frame's record (see the module's docstring).
    """
    n_read = 0
    complete = True
    with self.__lines() as lines:
        next(lines)  # the header
        try:
            for line in lines:
                if not line.endswith(b"\n"):
                    complete = False  # cut off while being written
                    break
                yield orjson.loads(line)
                n_read += 1
        except EOFError:
            complete = False  # a gzip stream that was never finished

    expected = self.header.get("n_frames")
    if not complete or (expected is not None and n_read < expected):
        of = "" if expected is None else f" of the {expected} of the run"
        logger.warning(
            f"{self.path} ends early, after {n_read} frame(s){of}: the run "
            "was interrupted before writing the rest."
        )

__iter__

__iter__() -> Iterator[dict[str, Any]]

Read the frames, one at a time (see frames).

Returns:

Type	Description
Iterator[dict[str, Any]]	Each frame's record.

Source code in src/molclusters/report.py

def __iter__(self) -> Iterator[dict[str, Any]]:
    """Read the frames, one at a time (see `frames`).

    Returns
    -------
    Iterator[dict[str, Any]]
        Each frame's record.
    """
    return self.frames()

clusters

clusters() -> DataFrame

Tabulate the clusters: one row per cluster per frame.

Returns:

Type	Description
DataFrame	frame and time, then the cluster's fields but its connections (see connections); the composition gets a composition.<resname> column per residue name (NaN when absent), and the fields another analysis added a <Analysis>.<field> column each.

Source code in src/molclusters/report.py

def clusters(self) -> pd.DataFrame:
    """Tabulate the clusters: one row per cluster per frame.

    Returns
    -------
    pd.DataFrame
        ``frame`` and ``time``, then the cluster's fields but its connections
        (see `connections`); the composition gets a ``composition.<resname>``
        column per residue name (NaN when absent), and the fields another
        analysis added a ``<Analysis>.<field>`` column each.
    """
    rows = [
        {"frame": frame["frame"], "time": frame["time"], **cluster}
        for frame in self.frames()
        for cluster in frame["clusters"]
    ]
    for row in rows:
        del row["connections"]
    return pd.json_normalize(rows, max_level=1)

connections

connections() -> DataFrame

Tabulate the connections: one row per connection per frame.

Returns:

Type	Description
DataFrame	frame, time and cluster (its id), then the header's connection_columns; n_hbonds is a nullable integer (<NA> for "cm" rules).

Source code in src/molclusters/report.py

def connections(self) -> pd.DataFrame:
    """Tabulate the connections: one row per connection per frame.

    Returns
    -------
    pd.DataFrame
        ``frame``, ``time`` and ``cluster`` (its id), then the header's
        ``connection_columns``; ``n_hbonds`` is a nullable integer (``<NA>``
        for "cm" rules).
    """
    columns = ["frame", "time", "cluster", *self.header["connection_columns"]]
    rows = [
        [frame["frame"], frame["time"], cluster["id"], *row]
        for frame in self.frames()
        for cluster in frame["clusters"]
        for row in cluster["connections"]
    ]
    table = pd.DataFrame(rows, columns=columns)
    if "n_hbonds" in table:
        table["n_hbonds"] = table["n_hbonds"].astype("Int64")
    return table

report_name

report_name(compression: ReportCompression) -> str

Give the report's file name for a compression.

Parameters:

Name	Type	Description	Default
compression	ReportCompression	"zstd", "gzip" or "none".	required

Returns:

Type	Description
str	molclusters.jsonl with the compression's suffix, if any.

Source code in src/molclusters/report.py

def report_name(compression: ReportCompression) -> str:
    """Give the report's file name for a compression.

    Parameters
    ----------
    compression : ReportCompression
        "zstd", "gzip" or "none".

    Returns
    -------
    str
        ``molclusters.jsonl`` with the compression's suffix, if any.
    """
    return f"molclusters.jsonl{_SUFFIXES[compression]}"

dumps

dumps(record: dict[str, Any]) -> bytes

Encode a record as one line of the report.

Parameters:

Name	Type	Description	Default
record	dict[str, Any]	The record: JSON types, and numpy scalars or arrays.	required

Returns:

Type	Description
bytes	The record's JSON, newline included.

Source code in src/molclusters/report.py

def dumps(record: dict[str, Any]) -> bytes:
    """Encode a record as one line of the report.

    Parameters
    ----------
    record : dict[str, Any]
        The record: JSON types, and numpy scalars or arrays.

    Returns
    -------
    bytes
        The record's JSON, newline included.
    """
    return orjson.dumps(record, option=_OPTIONS)

rounded

rounded(value: Any) -> Any

Round every float in a value to DECIMALS decimal places, however nested.

For what analyses add to the report, whose floats are rounded as the report's own are.

Parameters:

Name	Type	Description	Default
value	Any	JSON types (mappings, sequences, numbers, strings, None), numpy scalars or arrays.	required

Returns:

Type	Description
Any	The value, mappings as dicts, other sequences as lists, float arrays rounded.

Raises:

Type	Description
TypeError	If the value holds something that can't be written as JSON.

Source code in src/molclusters/report.py

def rounded(value: Any) -> Any:  # noqa: ANN401 (any JSON-able value)
    """Round every float in a value to `DECIMALS` decimal places, however nested.

    For what analyses add to the report, whose floats are rounded as the report's
    own are.

    Parameters
    ----------
    value : Any
        JSON types (mappings, sequences, numbers, strings, None), numpy scalars or
        arrays.

    Returns
    -------
    Any
        The value, mappings as dicts, other sequences as lists, float arrays
        rounded.

    Raises
    ------
    TypeError
        If the value holds something that can't be written as JSON.
    """
    match value:
        case float() | np.floating():
            return _round(value)
        case None | bool() | int() | str() | np.integer() | np.bool_():
            return value
        case np.ndarray() if value.dtype.kind == "f":
            return np.round(value, DECIMALS)
        case np.ndarray() if value.dtype.kind in "biu":
            return np.ascontiguousarray(value)  # orjson takes contiguous arrays
        case Mapping():
            return {_key(key): rounded(item) for key, item in value.items()}
        case list() | tuple() | np.ndarray():
            return [rounded(item) for item in value]
    raise TypeError(f"A {type(value).__name__} can't be written to the report.")

group_record

group_record(group: MolGroup) -> dict[str, Any]

Encode a group of molecules and its properties.

Parameters:

Name	Type	Description	Default
group	MolGroup	The group, e.g. a nucleus.	required

Returns:

Type	Description
dict[str, Any]	Its size, resids, composition and properties (see the module's docstring).

Source code in src/molclusters/report.py

def group_record(group: MolGroup) -> dict[str, Any]:
    """Encode a group of molecules and its properties.

    Parameters
    ----------
    group : MolGroup
        The group, e.g. a nucleus.

    Returns
    -------
    dict[str, Any]
        Its size, resids, composition and properties (see the module's docstring).
    """
    return {
        "size": group.size,
        "resids": sorted(map(int, group.resids)),
        "composition": {str(name): n for name, n in group.composition.items()},
        "mass": _round(group.mass),
        "volume": _round(group.volume),
        "radius": _round(group.radius),
        "radius_of_gyration": _round(group.radius_of_gyration),
        "diameter": _round(group.diameter),
        "density": _round(group.density),
        "charge": _round(group.charge),
        "dipole_moment": _round(group.dipole_moment),
        "sphericity": _round(group.sphericity),
        "shape_parameter": _round(group.shape_parameter),
    }

connection_rows

connection_rows(cluster: Cluster) -> list[list[int | float | None]]

Encode the connections of a cluster, as rows of CONNECTION_COLUMNS.

Parameters:

Name	Type	Description	Default
cluster	Cluster	The cluster.	required

Returns:

Type	Description
list[list[int | float | None]]	One row per connection, in the order of the cluster's graph.

Source code in src/molclusters/report.py

def connection_rows(cluster: Cluster) -> list[list[int | float | None]]:
    """Encode the connections of a cluster, as rows of `CONNECTION_COLUMNS`.

    Parameters
    ----------
    cluster : Cluster
        The cluster.

    Returns
    -------
    list[list[int | float | None]]
        One row per connection, in the order of the cluster's graph.
    """
    attributes = CONNECTION_COLUMNS[2:]
    return [
        [int(i), int(j), *(_attribute(edge.get(name)) for name in attributes)]
        for i, j, edge in cluster.graph.edges.data()
    ]

cluster_record

cluster_record(cluster: Cluster) -> dict[str, Any]

Encode a cluster, its properties and connections.

Parameters:

Name	Type	Description	Default
cluster	Cluster	The cluster.	required

Returns:

Type	Description
dict[str, Any]	Its id and birth time, then group_record, then its connections.

Source code in src/molclusters/report.py

def cluster_record(cluster: Cluster) -> dict[str, Any]:
    """Encode a cluster, its properties and connections.

    Parameters
    ----------
    cluster : Cluster
        The cluster.

    Returns
    -------
    dict[str, Any]
        Its id and birth time, then `group_record`, then its connections.
    """
    return {
        "id": cluster.id,
        "birth_time": _round(cluster.birth_time),
        **group_record(cluster),
        "connections": connection_rows(cluster),
    }

read_report

read_report(path: Path | str) -> Report

Open a report file.

Parameters:

Name	Type	Description	Default
path	Path | str	The report, compressed or not.	required

Returns:

Type	Description
Report	The report, its header read.

Source code in src/molclusters/report.py

def read_report(path: Path | str) -> "Report":
    """Open a report file.

    Parameters
    ----------
    path : Path | str
        The report, compressed or not.

    Returns
    -------
    Report
        The report, its header read.
    """
    return Report(path)