Coverage for src/loman/planning.py: 100%
130 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-07 00:32 +0000
1"""Non-mutating validation and execution planning for computation graphs."""
3from collections.abc import Collection, Mapping, Sequence
4from dataclasses import dataclass
6import networkx as nx
7import pandas as pd
9from .consts import NodeAttributes, States
10from .nodekey import NodeKey
12_CURRENT_STATES = {States.UPTODATE, States.PINNED}
15@dataclass(frozen=True)
16class ValidationReport:
17 """Structural and readiness findings for a computation graph."""
19 cycles: tuple[tuple[NodeKey, ...], ...]
20 placeholders: tuple[NodeKey, ...]
21 uninitialized_inputs: tuple[NodeKey, ...]
22 error_nodes: tuple[NodeKey, ...]
23 missing_executors: tuple[tuple[NodeKey, str], ...]
25 @property
26 def is_valid(self) -> bool:
27 """Return whether the graph is structurally valid."""
28 return not self.cycles and not self.placeholders and not self.missing_executors
30 @property
31 def is_ready(self) -> bool:
32 """Return whether the graph is valid and has all required inputs."""
33 return self.is_valid and not self.uninitialized_inputs and not self.error_nodes
35 def to_df(self) -> pd.DataFrame:
36 """Return validation findings as one row per issue."""
37 rows: list[dict[str, str]] = []
38 for component in self.cycles:
39 nodes = ", ".join(str(node) for node in component)
40 rows.append({"issue": "cycle", "node": nodes, "detail": f"Dependency cycle: {nodes}"})
41 rows.extend(
42 {"issue": "placeholder", "node": str(node), "detail": "Node has no definition"}
43 for node in self.placeholders
44 )
45 rows.extend(
46 {"issue": "uninitialized_input", "node": str(node), "detail": "Input value has not been supplied"}
47 for node in self.uninitialized_inputs
48 )
49 rows.extend(
50 {"issue": "error", "node": str(node), "detail": "Previous calculation failed"} for node in self.error_nodes
51 )
52 rows.extend(
53 {
54 "issue": "missing_executor",
55 "node": str(node),
56 "detail": f"Executor {executor_name!r} is not configured",
57 }
58 for node, executor_name in self.missing_executors
59 )
60 return pd.DataFrame(rows, columns=["issue", "node", "detail"])
63@dataclass(frozen=True)
64class ExecutionPlan:
65 """A non-mutating description of work required by a computation."""
67 targets: tuple[NodeKey, ...] | None
68 execution_order: tuple[NodeKey, ...]
69 current_nodes: tuple[NodeKey, ...]
70 blocked_nodes: tuple[NodeKey, ...]
71 node_states: tuple[tuple[NodeKey, States], ...]
72 blocked_by: tuple[tuple[NodeKey, tuple[NodeKey, ...]], ...]
73 blocker_reasons: tuple[tuple[NodeKey, str], ...]
74 executor_assignments: tuple[tuple[NodeKey, str], ...]
75 validation: ValidationReport
77 @property
78 def is_feasible(self) -> bool:
79 """Return whether all requested targets can be brought up to date."""
80 return self.validation.is_ready and not self.blocked_nodes
82 def to_df(self) -> pd.DataFrame:
83 """Return current, runnable, and blocked nodes as an execution table."""
84 order = {node: index + 1 for index, node in enumerate(self.execution_order)}
85 executors = dict(self.executor_assignments)
86 states = dict(self.node_states)
87 blocked_by = dict(self.blocked_by)
88 reasons = dict(self.blocker_reasons)
89 rows = []
90 for node in (*self.current_nodes, *self.execution_order, *self.blocked_nodes):
91 blockers = blocked_by.get(node, ())
92 blocker_details = tuple(dict.fromkeys(reasons[blocker] for blocker in blockers))
93 plan_status = "blocked" if blockers else "current" if node in self.current_nodes else "pending"
94 rows.append(
95 {
96 "node": str(node),
97 "state": states[node],
98 "plan_status": plan_status,
99 "order": order.get(node),
100 "executor": executors.get(node),
101 "blocked_by": ", ".join(str(blocker) for blocker in blockers) or None,
102 "reason": "; ".join(blocker_details) or None,
103 }
104 )
105 return pd.DataFrame(
106 rows,
107 columns=["node", "state", "plan_status", "order", "executor", "blocked_by", "reason"],
108 ).set_index("node")
111def _ordered_nodes(dag: nx.DiGraph, nodes: Collection[NodeKey]) -> tuple[NodeKey, ...]:
112 """Return *nodes* in graph insertion order."""
113 return tuple(node for node in dag.nodes if node in nodes)
116def _find_cycles(dag: nx.DiGraph, nodes: Collection[NodeKey]) -> tuple[tuple[NodeKey, ...], ...]:
117 """Return cyclic strongly connected components in graph insertion order."""
118 node_order = {node: index for index, node in enumerate(dag.nodes)}
119 graph = dag.subgraph(nodes)
120 components = []
121 for component in nx.strongly_connected_components(graph):
122 if len(component) > 1 or any(graph.has_edge(node, node) for node in component):
123 components.append(tuple(sorted(component, key=node_order.__getitem__)))
124 components.sort(key=lambda component: node_order[component[0]])
125 return tuple(components)
128def validate_graph(
129 dag: nx.DiGraph,
130 executor_map: Mapping[str, object],
131 *,
132 nodes: Collection[NodeKey] | None = None,
133 executor_nodes: Collection[NodeKey] | None = None,
134) -> ValidationReport:
135 """Inspect a graph, or a subset of it, without changing it."""
136 selected = set(dag.nodes if nodes is None else nodes)
137 checked_executors = selected if executor_nodes is None else set(executor_nodes)
139 placeholders: set[NodeKey] = set()
140 uninitialized_inputs: set[NodeKey] = set()
141 error_nodes: set[NodeKey] = set()
142 missing_executors: list[tuple[NodeKey, str]] = []
144 for node_key in dag.nodes:
145 if node_key not in selected:
146 continue
147 node = dag.nodes[node_key]
148 state = node.get(NodeAttributes.STATE)
149 if state == States.PLACEHOLDER:
150 placeholders.add(node_key)
151 elif state != States.ERROR and state not in _CURRENT_STATES and node.get(NodeAttributes.FUNC) is None:
152 uninitialized_inputs.add(node_key)
153 if state == States.ERROR:
154 error_nodes.add(node_key)
156 executor_name = node.get(NodeAttributes.EXECUTOR)
157 if (
158 node_key in checked_executors
159 and node.get(NodeAttributes.FUNC) is not None
160 and executor_name is not None
161 and executor_name not in executor_map
162 ):
163 missing_executors.append((node_key, executor_name))
165 return ValidationReport(
166 cycles=_find_cycles(dag, selected),
167 placeholders=_ordered_nodes(dag, placeholders),
168 uninitialized_inputs=_ordered_nodes(dag, uninitialized_inputs),
169 error_nodes=_ordered_nodes(dag, error_nodes),
170 missing_executors=tuple(missing_executors),
171 )
174def _select_required_nodes(dag: nx.DiGraph, targets: Sequence[NodeKey]) -> tuple[set[NodeKey], set[NodeKey]]:
175 """Select target dependencies, stopping traversal at nodes with current values."""
176 selected: set[NodeKey] = set()
177 current: set[NodeKey] = set()
178 to_visit = list(reversed(targets))
180 while to_visit:
181 node_key = to_visit.pop()
182 if node_key in selected:
183 continue
184 selected.add(node_key)
185 if dag.nodes[node_key].get(NodeAttributes.STATE) in _CURRENT_STATES:
186 current.add(node_key)
187 continue
188 to_visit.extend(reversed(list(dag.predecessors(node_key))))
190 return selected, current
193def create_execution_plan(
194 dag: nx.DiGraph,
195 executor_map: Mapping[str, object],
196 targets: Sequence[NodeKey] | None,
197) -> ExecutionPlan:
198 """Build a deterministic execution plan without running or mutating nodes."""
199 plan_targets = tuple(dag.nodes) if targets is None else tuple(dict.fromkeys(targets))
200 selected, current = _select_required_nodes(dag, plan_targets)
201 pending = selected - current
202 validation = validate_graph(dag, executor_map, nodes=selected, executor_nodes=pending)
204 blocker_reasons: dict[NodeKey, str] = {}
205 blocker_reasons.update((node, "Node has no definition") for node in validation.placeholders)
206 blocker_reasons.update((node, "Input value has not been supplied") for node in validation.uninitialized_inputs)
207 blocker_reasons.update((node, "Previous calculation failed") for node in validation.error_nodes)
208 for component in validation.cycles:
209 detail = f"Dependency cycle: {', '.join(str(node) for node in component)}"
210 blocker_reasons.update((node, detail) for node in component)
211 blocker_reasons.update(
212 (node, f"Executor {executor_name!r} is not configured") for node, executor_name in validation.missing_executors
213 )
215 pending_graph = dag.subgraph(pending)
216 blocked_by: dict[NodeKey, set[NodeKey]] = {node: {node} for node in blocker_reasons if node in pending}
217 for blocker in blocker_reasons:
218 if blocker in pending:
219 for descendent in nx.descendants(pending_graph, blocker):
220 blocked_by.setdefault(descendent, set()).add(blocker)
222 blocked = set(blocked_by)
223 node_order = {node: index for index, node in enumerate(dag.nodes)}
225 runnable = pending - blocked
226 runnable_graph = dag.subgraph(runnable)
227 execution_order = tuple(nx.topological_sort(runnable_graph))
228 assignments = tuple(
229 (
230 node_key,
231 "default"
232 if dag.nodes[node_key].get(NodeAttributes.EXECUTOR) is None
233 else dag.nodes[node_key][NodeAttributes.EXECUTOR],
234 )
235 for node_key in execution_order
236 )
238 return ExecutionPlan(
239 targets=None if targets is None else tuple(dict.fromkeys(targets)),
240 execution_order=execution_order,
241 current_nodes=_ordered_nodes(dag, current),
242 blocked_nodes=_ordered_nodes(dag, blocked),
243 node_states=tuple((node, dag.nodes[node][NodeAttributes.STATE]) for node in dag.nodes if node in selected),
244 blocked_by=tuple(
245 (node, tuple(sorted(blocked_by[node], key=node_order.__getitem__)))
246 for node in dag.nodes
247 if node in blocked_by
248 ),
249 blocker_reasons=tuple((node, blocker_reasons[node]) for node in dag.nodes if node in blocker_reasons),
250 executor_assignments=assignments,
251 validation=validation,
252 )