"""Dependency-free data model for a FlowView run graph."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Mapping
[docs]
class NodeKind(str, Enum):
"""The role a node plays in a pipeline graph."""
INPUT = "input"
PROCESS = "process"
OUTPUT = "output"
[docs]
class NodeState(str, Enum):
"""Lifecycle state of a pipeline node."""
PENDING = "pending"
RUNNING = "running"
DONE = "done"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass(frozen=True)
[docs]
class Node:
"""One immutable snapshot of a pipeline stage or artifact.
:param id: stable identifier used by events and graph edges.
:param label: human-readable stage or artifact name shown in FlowView.
:param kind: input, process, or output role controlling the node's meaning
and presentation.
:param state: current lifecycle state, normalized to :class:`NodeState`.
:param started_at: start time in seconds on the producer's clock, if known.
:param ended_at: completion, failure, or skip time on that clock, if known.
:param progress: completed and total work-unit counts, or ``None`` before a
stage reports progress.
:param metrics: named scalar results displayed on the node and retained in
exported provenance.
:param thumbnail: path of a cached representative image, if provided.
:param params: stage configuration or provenance values displayed in the
inspector and retained in exports.
:param error: formatted failure diagnostic, or ``None`` when no failure was
recorded.
"""
id: str
label: str
kind: NodeKind
state: NodeState = NodeState.PENDING
started_at: float | None = None
ended_at: float | None = None
progress: tuple[int, int] | None = None
metrics: dict[str, float | int | str] = field(default_factory=dict)
thumbnail: str | None = None
params: dict[str, Any] = field(default_factory=dict)
error: str | None = None
[docs]
def __post_init__(self) -> None:
"""Normalise enums and detach caller-owned mutable dictionaries."""
object.__setattr__(self, "kind", NodeKind(self.kind))
object.__setattr__(self, "state", NodeState(self.state))
object.__setattr__(self, "metrics", dict(self.metrics))
object.__setattr__(self, "params", dict(self.params))
@dataclass(frozen=True)
[docs]
class Edge:
"""A directed transfer between two nodes.
:param src: identifier of the node producing the transfer.
:param dst: identifier of the node receiving the transfer.
:param label: optional relationship or payload caption, such as
``"consumes"`` or ``"produces"``.
:param volume: optional number of objects, files, or rows transferred; the
renderer uses it to scale edge width.
"""
src: str
dst: str
label: str | None = None
volume: int | None = None
@dataclass(frozen=True)
[docs]
class RunGraph:
"""A serialisable snapshot of one FlowView run.
:param run_id: stable identifier for the captured run.
:param started_at: run start time in seconds on the producer's clock.
:param nodes: node identifiers mapped to their immutable snapshots.
:param edges: directed transfers connecting the recorded nodes.
:param spacr_version: spaCR version that produced the run.
:param settings_digest: hash of the settings used for the run.
"""
run_id: str
started_at: float
nodes: dict[str, Node]
edges: list[Edge]
spacr_version: str
settings_digest: str
[docs]
def __post_init__(self) -> None:
"""Detach the graph's containers from caller-owned containers."""
object.__setattr__(self, "nodes", dict(self.nodes))
object.__setattr__(self, "edges", list(self.edges))
[docs]
def to_dict(self) -> dict[str, Any]:
"""Return a JSON-compatible canonical representation."""
nodes = {
node_id: {
"id": node.id,
"label": node.label,
"kind": node.kind.value,
"state": node.state.value,
"started_at": node.started_at,
"ended_at": node.ended_at,
"progress": node.progress,
"metrics": node.metrics,
"thumbnail": node.thumbnail,
"params": node.params,
"error": node.error,
}
for node_id, node in sorted(self.nodes.items())
}
edges = [
{
"src": edge.src,
"dst": edge.dst,
"label": edge.label,
"volume": edge.volume,
}
for edge in sorted(
self.edges,
key=lambda item: json.dumps(
[item.src, item.dst, item.label, item.volume],
ensure_ascii=False,
separators=(",", ":"),
),
)
]
return {
"run_id": self.run_id,
"started_at": self.started_at,
"nodes": nodes,
"edges": edges,
"spacr_version": self.spacr_version,
"settings_digest": self.settings_digest,
}
[docs]
def to_json(self) -> str:
"""Serialise the graph deterministically as strict JSON."""
return json.dumps(
self.to_dict(),
ensure_ascii=False,
allow_nan=False,
sort_keys=True,
separators=(",", ":"),
)
@classmethod
[docs]
def from_dict(cls, payload: Mapping[str, Any]) -> "RunGraph":
"""Restore a graph from :meth:`to_dict` output.
:param payload: mapping with the ``run_id``, ``started_at``,
``nodes``, ``edges``, ``spacr_version`` and ``settings_digest``
keys :meth:`to_dict` writes; every key is required and a missing
one raises ``KeyError``.
"""
nodes = {
node_id: Node(
id=node_payload["id"],
label=node_payload["label"],
kind=NodeKind(node_payload["kind"]),
state=NodeState(node_payload["state"]),
started_at=node_payload["started_at"],
ended_at=node_payload["ended_at"],
progress=(
tuple(node_payload["progress"])
if node_payload["progress"] is not None
else None
),
metrics=node_payload["metrics"],
thumbnail=node_payload["thumbnail"],
params=node_payload["params"],
error=node_payload["error"],
)
for node_id, node_payload in payload["nodes"].items()
}
edges = [
Edge(
src=edge_payload["src"],
dst=edge_payload["dst"],
label=edge_payload["label"],
volume=edge_payload["volume"],
)
for edge_payload in payload["edges"]
]
return cls(
run_id=payload["run_id"],
started_at=payload["started_at"],
nodes=nodes,
edges=edges,
spacr_version=payload["spacr_version"],
settings_digest=payload["settings_digest"],
)
@classmethod
[docs]
def from_json(cls, payload: str | bytes | bytearray) -> "RunGraph":
"""Restore a graph from its deterministic JSON record.
:param payload: JSON text (``str``, ``bytes`` or ``bytearray``) as
written by :meth:`to_json`; it is decoded with ``json.loads`` and
passed to :meth:`from_dict`.
"""
return cls.from_dict(json.loads(payload))