import argparse import html import json from pathlib import Path import yaml DEFAULT_GRAPH_YAML = Path(__file__).resolve().parent / "graph.yaml" # --- ODPG relationship model (ODPG spec literals; graphs may add domain-specific types) --- # Edge `type` values are shown exactly as in YAML (e.g. contributesTo, governedBy, uses). ODPG_CORE_EDGE_TYPES_ORDERED: tuple[str, ...] = ( "uses", "supports", "contributesTo", "measures", "tracks", "dependsOn", "produce", "Consumes", "governedBy", "ownedBy", "alignsWith", "alignWith", "relatedTo", "impacts", "derivedFrom", "exposes", "monitors", "identifies", ) ODPG_EDGE_DESCRIPTIONS: dict[str, str] = { "uses": "A node uses another node as part of execution or operation", "supports": "A node supports a business objective", "contributesTo": "A node contributes toward an outcome or objective", "measures": "A KPI measures an objective or outcome", "tracks": "A node tracks or provides KPI-related information", "dependsOn": "A node depends on another node", "produce": "A node produces data, outputs, or services", "Consumes": "A node consumes data, APIs, or outputs", "governedBy": "A node is governed by a policy or control", "ownedBy": "A node is owned by a person, team, or domain", "alignsWith": "A node aligns strategically or semantically with another node", "alignWith": "A node aligns strategically or semantically with another node", "relatedTo": "A generic semantic relationship", "impacts": "A node impacts another node", "derivedFrom": "A node originates from another node", "exposes": "A node exposes an API or interface", "monitors": "A node monitors another node", "identifies": "A node identifies an opportunity or condition", } ODPG_EDGE_DESCRIPTIONS_LOWER: dict[str, str] = {k.lower(): v for k, v in ODPG_EDGE_DESCRIPTIONS.items()} ODPG_CORE_EDGE_TYPES_LOWER: frozenset[str] = frozenset(s.lower() for s in ODPG_CORE_EDGE_TYPES_ORDERED) _ODPG_LEGEND_COLOR_BY_LOWER: dict[str, str] = { "uses": "#16a34a", "supports": "#ea580c", "contributesto": "#0284c7", "measures": "#0d9488", "tracks": "#14b8a6", "dependson": "#6366f1", "produce": "#059669", "consumes": "#a855f7", "governedby": "#db2777", "ownedby": "#ca8a04", "alignswith": "#f97316", "alignwith": "#f97316", "relatedto": "#64748b", "impacts": "#dc2626", "derivedfrom": "#78716c", "exposes": "#7c3aed", "monitors": "#0891b2", "identifies": "#475569", } def load_graph_from_yaml(path: Path | str | None = None) -> dict: graph_path = Path(path) if path is not None else DEFAULT_GRAPH_YAML if not graph_path.is_file(): raise FileNotFoundError(f"Graph YAML not found: {graph_path}") return yaml.safe_load(graph_path.read_text(encoding="utf-8")) def _graph_payload(document: dict) -> dict: payload = document.get("graph") return payload if isinstance(payload, dict) else document def _graph_metadata(document: dict) -> dict: payload = _graph_payload(document) metadata = payload.get("metadata") if isinstance(metadata, dict): return metadata return { "id": document.get("id"), "name": document.get("name"), "description": document.get("description"), } def _node_ref(node: dict) -> str: return node.get("$ref") or node.get("ref") or "" def validate_graph(graph): required_root_fields = ["schema", "version", "kind"] for field in required_root_fields: if field not in graph: raise ValueError(f"Missing required root field: {field}") if graph["kind"] not in ("Graph", "DataProductGraph"): raise ValueError("Invalid kind. Expected: Graph") payload = _graph_payload(graph) metadata = _graph_metadata(graph) if "graph" in graph and not isinstance(graph["graph"], dict): raise ValueError("Root field graph must be an object") if "graph" in graph: for field in ["metadata", "nodes", "edges"]: if field not in payload: raise ValueError(f"Missing required graph field: graph.{field}") for field in ["id", "name", "description"]: if field not in metadata: raise ValueError(f"Missing required metadata field: graph.metadata.{field}") else: for field in ["id", "name", "nodes", "edges"]: if field not in payload: raise ValueError(f"Missing required graph field: {field}") node_ids = set() for node in payload["nodes"]: for field in ["id", "type"]: if field not in node: raise ValueError(f"Node is missing required field: {field}") if not _node_ref(node): raise ValueError("Node is missing required field: $ref") if node["id"] in node_ids: raise ValueError(f"Duplicate node id found: {node['id']}") node_ids.add(node["id"]) for edge in payload["edges"]: for field in ["from", "to", "type", "confidence"]: if field not in edge: raise ValueError(f"Edge is missing required field: {field}") if edge["from"] not in node_ids: raise ValueError(f"Edge source does not match any node id: {edge['from']}") if edge["to"] not in node_ids: raise ValueError(f"Edge target does not match any node id: {edge['to']}") return True def _ref_to_display_name(ref: str) -> str: base = Path(ref).name if base.endswith((".yaml", ".yml")): base = base.rsplit(".", 1)[0] base = base.replace("-", " ").replace("_", " ") return " ".join(w.capitalize() for w in base.split()) if base else "" def _edge_type_raw(edge: dict) -> str: return str(edge.get("type", "")).strip() def collect_relationship_types(graph: dict) -> list[str]: """Types present in the graph: ODPG spec order (camelCase literals), then other types A–Z.""" payload = _graph_payload(graph) seen_lower_to_raw: dict[str, str] = {} for edge in payload.get("edges") or []: raw = _edge_type_raw(edge) if not raw: continue lo = raw.lower() if lo not in seen_lower_to_raw: seen_lower_to_raw[lo] = raw ordered_core: list[str] = [] for spec in ODPG_CORE_EDGE_TYPES_ORDERED: lo = spec.lower() if lo in seen_lower_to_raw: ordered_core.append(seen_lower_to_raw[lo]) extras = sorted( (seen_lower_to_raw[lo] for lo in seen_lower_to_raw if lo not in ODPG_CORE_EDGE_TYPES_LOWER), key=str.lower, ) return ordered_core + extras def _edge_relationship_label(edge: dict) -> str: """Edge label exactly as authored in YAML `type` (ODPG stabilization spec).""" return _edge_type_raw(edge) def _edge_vis_tooltip(edge: dict, display: str) -> str: lines = [f"Type: {display}", f"Confidence: {edge['confidence']}"] desc = ODPG_EDGE_DESCRIPTIONS_LOWER.get(display.lower(), "") if desc: lines.append(desc) return "\n".join(lines) def _edge_legend_color(type_display: str) -> str: return _ODPG_LEGEND_COLOR_BY_LOWER.get(type_display.lower(), "#94a3b8") def _edge_line_dashed(type_display: str, edge: dict) -> bool: if edge.get("dashed") is True: return True return type_display.lower() == "governedby" def _build_legend_relationship_html(relationship_types: list[str], graph: dict) -> str: payload = _graph_payload(graph) edges = payload.get("edges") or [] def sample_edge(label_raw: str) -> dict: lo = label_raw.lower() for e in edges: if _edge_type_raw(e).lower() == lo: return e return {} blocks: list[str] = [] for label_raw in relationship_types: edge_sample = sample_edge(label_raw) color = _edge_legend_color(label_raw) dash = ' stroke-dasharray="4 3"' if _edge_line_dashed(label_raw, edge_sample) else "" safe_label = html.escape(label_raw) blocks.append( f'
\n' f' \n' f" {safe_label}\n" f"
" ) if not blocks: blocks.append( '

No edges in graph.

' ) return "\n".join(blocks) def build_html(graph: dict) -> str: payload = _graph_payload(graph) metadata = _graph_metadata(graph) relationship_types = collect_relationship_types(graph) node_types = sorted({str(node.get("type", "")) for node in payload["nodes"] if node.get("type")}) confidence_levels = sorted({str(edge.get("confidence", "")) for edge in payload["edges"] if edge.get("confidence")}) odpg_supported_ordered_json = json.dumps(list(ODPG_CORE_EDGE_TYPES_ORDERED), ensure_ascii=False) odpg_descriptions_json = json.dumps(ODPG_EDGE_DESCRIPTIONS_LOWER, ensure_ascii=False) graph_title = metadata.get("name", {}).get("en", metadata.get("id", "ODPG Graph Explorer")) graph_meta = f"{metadata.get('id')} · ODPG {graph.get('version')} · {graph.get('kind')}" vis_nodes = [] for node in payload["nodes"]: ref = _node_ref(node) display_name = _ref_to_display_name(ref) vis_nodes.append( { "id": node["id"], "label": node["id"], "title": ( f"ID: {node['id']}\n" f"Type: {node['type']}\n" f"Ref: {ref}" ), "group": node["type"], "ref": ref, "type": node["type"], "displayName": display_name, } ) vis_edges = [] for idx, edge in enumerate(payload["edges"]): display = _edge_relationship_label(edge) conf = str(edge["confidence"]).lower() ec = _edge_legend_color(display) vis_edges.append( { "id": f"edge-{idx}", "from": edge["from"], "to": edge["to"], "label": f"{display}\n({conf})", "type": display, "title": _edge_vis_tooltip(edge, display), "arrows": "to", "confidence": edge["confidence"], "dashes": _edge_line_dashed(display, edge), "color": { "color": ec, "highlight": ec, "hover": ec, "inherit": False, }, } ) legend_edges_html = _build_legend_relationship_html(relationship_types, graph) return f""" ODPG Graph Explorer — {html.escape(graph_title)}
ODPG Logo

Node Types

UseCase
DataProduct
API
BusinessObjective
KPI
Policy
Agent
StrategicOpportunity

Relationship Types

{legend_edges_html}
""" def generate_graph_explorer( graph_yaml: Path | str | None = None, output_file: Path | str = "graph-explorer.html", ) -> None: graph = load_graph_from_yaml(graph_yaml) validate_graph(graph) html_out = build_html(graph) Path(output_file).write_text(html_out, encoding="utf-8") print(f"Graph Explorer generated successfully: {output_file}") if __name__ == "__main__": parser = argparse.ArgumentParser( description="Generate graph-explorer.html from an ODPG graph YAML file.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument( "-i", "--input", dest="graph_yaml", type=Path, default=DEFAULT_GRAPH_YAML, metavar="PATH", help="Path to graph YAML file", ) parser.add_argument( "-o", "--output", type=Path, default=Path("graph-explorer.html"), metavar="PATH", help="Output HTML file path", ) args = parser.parse_args() generate_graph_explorer(args.graph_yaml, args.output)