feat: add networkx dependency graph construction
- build_graph() creates a DAG from project config with validation - get_build_order() wraps topological_generations() for parallel batches - get_subgraph_for_target() extracts transitive deps for single-target builds - Validates missing dependencies and cycle detection
This commit is contained in:
parent
ce2160bd6c
commit
bedc0cc9ec
1 changed files with 64 additions and 0 deletions
64
bulkgen/graph.py
Normal file
64
bulkgen/graph.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
"""Dependency graph construction and traversal using networkx."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import networkx as nx
|
||||
|
||||
from bulkgen.config import ProjectConfig
|
||||
|
||||
|
||||
def build_graph(config: ProjectConfig, project_dir: Path) -> nx.DiGraph:
|
||||
"""Build a dependency DAG from the project configuration.
|
||||
|
||||
Nodes are filenames: target names (keys in ``config.targets``) and
|
||||
external files that exist on disk. Edges point from dependency to
|
||||
dependent (``A -> B`` means *A must exist before B*).
|
||||
|
||||
Raises :class:`ValueError` if a dependency is neither a defined target
|
||||
nor an existing file, or if the graph contains a cycle.
|
||||
"""
|
||||
graph = nx.DiGraph()
|
||||
target_names = set(config.targets)
|
||||
|
||||
for target_name, target_cfg in config.targets.items():
|
||||
graph.add_node(target_name)
|
||||
|
||||
deps: list[str] = list(target_cfg.inputs)
|
||||
if target_cfg.reference_image is not None:
|
||||
deps.append(target_cfg.reference_image)
|
||||
deps.extend(target_cfg.control_images)
|
||||
|
||||
for dep in deps:
|
||||
if dep not in target_names and not (project_dir / dep).exists():
|
||||
msg = (
|
||||
f"Target '{target_name}' depends on '{dep}', "
|
||||
f"which is neither a defined target nor an existing file"
|
||||
)
|
||||
raise ValueError(msg)
|
||||
graph.add_edge(dep, target_name)
|
||||
|
||||
if not nx.is_directed_acyclic_graph(graph):
|
||||
cycles = list(nx.simple_cycles(graph))
|
||||
msg = f"Dependency cycle detected: {cycles}"
|
||||
raise ValueError(msg)
|
||||
|
||||
return graph
|
||||
|
||||
|
||||
def get_build_order(graph: nx.DiGraph) -> list[list[str]]:
|
||||
"""Return targets grouped into generations for parallel execution.
|
||||
|
||||
Each inner list contains nodes with no inter-dependencies that can
|
||||
be built concurrently.
|
||||
"""
|
||||
return [list(gen) for gen in nx.topological_generations(graph)]
|
||||
|
||||
|
||||
def get_subgraph_for_target(graph: nx.DiGraph, target: str) -> nx.DiGraph:
|
||||
"""Return the subgraph containing *target* and all its transitive dependencies."""
|
||||
ancestors = nx.ancestors(graph, target)
|
||||
ancestors.add(target)
|
||||
subgraph = nx.DiGraph(graph.subgraph(ancestors))
|
||||
return subgraph
|
||||
Loading…
Add table
Add a link
Reference in a new issue