forked from genewildish/Mainline
David Gwilliam
406a58d292
Allow pipelines to be defined in TOML format with intuitive
node-and-connection syntax that's easy to read and edit.
- Add graph_toml.py with TOML parsing using tomllib
- Support simple format: "source": "headlines"
- Support full format: {"type": "camera", "mode": "scroll"}
- Parse connection strings in "A -> B -> C" chain format
- Add example pipeline_graph.toml demonstrating usage
Example TOML:
[nodes.source]
type = "source"
source = "headlines"
[nodes.camera]
type = "camera"
mode = "scroll"
[connections]
list = ["source -> camera -> display"]
113 lines
3.0 KiB
Python
113 lines
3.0 KiB
Python
"""TOML-based graph configuration loader."""
|
|
|
|
import tomllib
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
from engine.pipeline.graph import Graph, NodeType
|
|
from engine.pipeline.graph_adapter import graph_to_pipeline
|
|
|
|
|
|
def load_graph_from_toml(toml_path: str | Path) -> Graph:
|
|
"""Load a graph from a TOML file.
|
|
|
|
Args:
|
|
toml_path: Path to the TOML file
|
|
|
|
Returns:
|
|
Graph instance loaded from the TOML file
|
|
"""
|
|
with open(toml_path, "rb") as f:
|
|
data = tomllib.load(f)
|
|
|
|
return graph_from_dict(data)
|
|
|
|
|
|
def graph_from_dict(data: Dict[str, Any]) -> Graph:
|
|
"""Create a graph from a dictionary (TOML-compatible structure).
|
|
|
|
Args:
|
|
data: Dictionary with 'nodes' and 'connections' keys
|
|
|
|
Returns:
|
|
Graph instance
|
|
"""
|
|
graph = Graph()
|
|
|
|
# Parse nodes
|
|
nodes_data = data.get("nodes", {})
|
|
for name, node_info in nodes_data.items():
|
|
if isinstance(node_info, str):
|
|
# Simple format: "source": "headlines"
|
|
graph.node(name, NodeType.SOURCE, source=node_info)
|
|
elif isinstance(node_info, dict):
|
|
# Full format: {"type": "camera", "mode": "scroll"}
|
|
node_type = node_info.get("type", "custom")
|
|
config = {k: v for k, v in node_info.items() if k != "type"}
|
|
graph.node(name, node_type, **config)
|
|
|
|
# Parse connections
|
|
connections_data = data.get("connections", {})
|
|
if isinstance(connections_data, dict):
|
|
# Format: {"list": ["source -> camera -> display"]}
|
|
connections_list = connections_data.get("list", [])
|
|
else:
|
|
# Format: ["source -> camera -> display"]
|
|
connections_list = connections_data
|
|
|
|
for conn in connections_list:
|
|
if isinstance(conn, str):
|
|
# Parse "source -> target" format
|
|
parts = conn.split("->")
|
|
if len(parts) >= 2:
|
|
# Connect all nodes in the chain
|
|
for i in range(len(parts) - 1):
|
|
source = parts[i].strip()
|
|
target = parts[i + 1].strip()
|
|
graph.connect(source, target)
|
|
|
|
return graph
|
|
|
|
|
|
def load_pipeline_from_toml(
|
|
toml_path: str | Path, viewport_width: int = 80, viewport_height: int = 24
|
|
):
|
|
"""Load a pipeline from a TOML file.
|
|
|
|
Args:
|
|
toml_path: Path to the TOML file
|
|
viewport_width: Terminal width for the pipeline
|
|
viewport_height: Terminal height for the pipeline
|
|
|
|
Returns:
|
|
Pipeline instance loaded from the TOML file
|
|
"""
|
|
graph = load_graph_from_toml(toml_path)
|
|
return graph_to_pipeline(graph, viewport_width, viewport_height)
|
|
|
|
|
|
# Example TOML structure:
|
|
EXAMPLE_TOML = """
|
|
# Graph-based pipeline configuration
|
|
[nodes.source]
|
|
type = "source"
|
|
source = "headlines"
|
|
|
|
[nodes.camera]
|
|
type = "camera"
|
|
mode = "scroll"
|
|
speed = 1.0
|
|
|
|
[nodes.noise]
|
|
type = "effect"
|
|
effect = "noise"
|
|
intensity = 0.3
|
|
|
|
[nodes.display]
|
|
type = "display"
|
|
backend = "terminal"
|
|
|
|
[connections]
|
|
list = ["source -> camera -> noise -> display"]
|
|
"""
|