diff --git a/dimos/core/introspection/blueprint/dot.py b/dimos/core/introspection/blueprint/dot.py index ea66401033..35a460cf2e 100644 --- a/dimos/core/introspection/blueprint/dot.py +++ b/dimos/core/introspection/blueprint/dot.py @@ -51,6 +51,11 @@ class LayoutAlgo(Enum): # "FoxgloveBridge", } +# Modules only ignored when show_disconnected is False (compact view) +_COMPACT_ONLY_IGNORED_MODULES = { + "WebsocketVisModule", +} + def render( blueprint_set: Blueprint, @@ -58,6 +63,7 @@ def render( layout: set[LayoutAlgo] | None = None, ignored_streams: set[tuple[str, str]] | None = None, ignored_modules: set[str] | None = None, + show_disconnected: bool = False, ) -> str: """Generate a hub-style DOT graph from a Blueprint. @@ -69,6 +75,8 @@ def render( layout: Set of layout algorithms to apply. Default is none (let graphviz decide). ignored_streams: Set of (name, type_name) tuples to ignore. ignored_modules: Set of module names to ignore. + show_disconnected: If True, show streams that have a producer but no consumer + (or vice versa) as dashed stub nodes. Returns: A string in DOT format showing modules as nodes, type nodes as @@ -79,7 +87,10 @@ def render( if ignored_streams is None: ignored_streams = DEFAULT_IGNORED_CONNECTIONS if ignored_modules is None: - ignored_modules = DEFAULT_IGNORED_MODULES + if show_disconnected: + ignored_modules = DEFAULT_IGNORED_MODULES - _COMPACT_ONLY_IGNORED_MODULES + else: + ignored_modules = DEFAULT_IGNORED_MODULES # Collect all outputs: (name, type) -> list of producer modules producers: dict[tuple[str, type], list[type[Module]]] = defaultdict(list) @@ -116,6 +127,23 @@ def render( label = f"{name}:{type_name}" active_channels[key] = color_for_string(TYPE_COLORS, label) + # Find disconnected channels (producer-only or consumer-only) + disconnected_channels: dict[tuple[str, type], str] = {} + if show_disconnected: + all_keys = set(producers.keys()) | set(consumers.keys()) + for key in all_keys: + if key in active_channels: + continue + name, type_ = key + type_name = type_.__name__ + if (name, type_name) in ignored_streams: + continue + relevant_modules = producers.get(key, []) + consumers.get(key, []) + if all(m.__name__ in ignored_modules for m in relevant_modules): + continue + label = f"{name}:{type_name}" + disconnected_channels[key] = color_for_string(TYPE_COLORS, label) + # Group modules by package def get_group(mod_class: type[Module]) -> str: module_path = mod_class.__module__ @@ -218,6 +246,37 @@ def get_group(mod_class: type[Module]) -> str: continue lines.append(f' {node_id} -> {consumer.__name__} [color="{color}"];') + # Disconnected channels (dashed stub nodes) + if disconnected_channels: + lines.append("") + lines.append(" // Disconnected streams") + for key, color in sorted( + disconnected_channels.items(), key=lambda x: f"{x[0][0]}:{x[0][1].__name__}" + ): + name, type_ = key + type_name = type_.__name__ + node_id = sanitize_id(f"chan_{name}_{type_name}") + label = f"{name}:{type_name}" + lines.append( + f' {node_id} [label="{label}", shape=note, ' + f'style="filled,dashed", fillcolor="{color}15", color="{color}", ' + f'fontcolor="{color}", width=0, height=0, margin="0.1,0.05", fontsize=10];' + ) + + for producer in producers.get(key, []): + if producer.__name__ in ignored_modules: + continue + lines.append( + f" {producer.__name__} -> {node_id} " + f'[color="{color}", style=dashed, arrowhead=none];' + ) + for consumer in consumers.get(key, []): + if consumer.__name__ in ignored_modules: + continue + lines.append( + f' {node_id} -> {consumer.__name__} [color="{color}", style=dashed];' + ) + lines.append("}") return "\n".join(lines) @@ -227,6 +286,7 @@ def render_svg( output_path: str, *, layout: set[LayoutAlgo] | None = None, + show_disconnected: bool = False, ) -> None: """Generate an SVG file from a Blueprint using graphviz. @@ -234,13 +294,14 @@ def render_svg( blueprint_set: The blueprint set to visualize. output_path: Path to write the SVG file. layout: Set of layout algorithms to apply. + show_disconnected: If True, show streams with no matching counterpart. """ import subprocess if layout is None: layout = set() - dot_code = render(blueprint_set, layout=layout) + dot_code = render(blueprint_set, layout=layout, show_disconnected=show_disconnected) engine = "fdp" if LayoutAlgo.FDP in layout else "dot" result = subprocess.run( [engine, "-Tsvg", "-o", output_path], diff --git a/dimos/core/introspection/blueprint/mermaid.py b/dimos/core/introspection/blueprint/mermaid.py new file mode 100644 index 0000000000..bb28e2681b --- /dev/null +++ b/dimos/core/introspection/blueprint/mermaid.py @@ -0,0 +1,346 @@ +# Copyright 2025-2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Mermaid diagram renderer for blueprint visualization. + +Generates a Mermaid flowchart with direct labelled edges between modules: + + ModuleA -- "name:Type" --> ModuleB +""" + +from __future__ import annotations + +from collections import defaultdict +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from dimos.core.blueprints import Blueprint + from dimos.core.module import Module + +# --------------------------------------------------------------------------- +# Colour themes +# --------------------------------------------------------------------------- +# Each theme is a dict with "nodes" and "edges" colour lists. + +THEMES: dict[str, dict[str, list[str]]] = { + # Vivid — bold, high-contrast, maximally distinct + "vivid": { + "nodes": [ + "#1565c0", # blue + "#c62828", # red + "#2e7d32", # green + "#6a1b9a", # purple + "#d84315", # burnt orange + "#00838f", # teal + "#ad1457", # pink + "#4527a0", # deep purple + "#ef6c00", # orange + "#00695c", # dark teal + "#283593", # indigo + "#9e9d24", # olive + "#1565a0", # steel blue + "#b71c1c", # dark red + "#558b2f", # lime green + "#6d4c41", # brown + "#00796b", # sea green + "#7b1fa2", # violet + "#e65100", # deep orange + "#0277bd", # light blue + ], + "edges": [ + "#4cc9f0", # sky blue + "#f77f00", # orange + "#80ed99", # mint green + "#c77dff", # lavender + "#ffd166", # gold + "#ef476f", # coral red + "#06d6a0", # teal + "#3a86ff", # bright blue + "#ff9e00", # amber + "#e5383b", # red + "#2ec4b6", # cyan-teal + "#9b5de5", # purple + "#00f5d4", # aquamarine + "#fee440", # yellow + "#f15bb5", # magenta + "#00bbf9", # cerulean + "#8ac926", # lime green + "#ff595e", # salmon + "#1982c4", # steel blue + "#ffca3a", # sunflower + ], + }, + # Tailwind — coordinated palette based on Tailwind CSS colour system. + # Nodes use the 700 shade (rich, readable with white text). + # Edges use the 400 shade (bright, high-visibility on dark backgrounds). + "tailwind": { + "nodes": [ + "#3b82f6", # blue-500 + "#ef4444", # red-500 + "#22c55e", # green-500 + "#8b5cf6", # violet-500 + "#f97316", # orange-500 + "#06b6d4", # cyan-500 + "#ec4899", # pink-500 + "#6366f1", # indigo-500 + "#eab308", # yellow-500 + "#14b8a6", # teal-500 + "#f43f5e", # rose-500 + "#84cc16", # lime-500 + "#0ea5e9", # sky-500 + "#d946ef", # fuchsia-500 + "#10b981", # emerald-500 + "#a855f7", # purple-500 + "#f59e0b", # amber-500 + "#38bdf8", # sky-400 + "#fb7185", # rose-400 + "#a3e635", # lime-400 + ], + "edges": [ + "#60a5fa", # blue-400 + "#f87171", # red-400 + "#4ade80", # green-400 + "#a78bfa", # violet-400 + "#fb923c", # orange-400 + "#22d3ee", # cyan-400 + "#f472b6", # pink-400 + "#818cf8", # indigo-400 + "#facc15", # yellow-400 + "#2dd4bf", # teal-400 + "#fb7185", # rose-400 + "#a3e635", # lime-400 + "#38bdf8", # sky-400 + "#e879f9", # fuchsia-400 + "#34d399", # emerald-400 + "#c084fc", # purple-400 + "#fbbf24", # amber-400 + "#67e8f9", # cyan-300 + "#fda4af", # rose-300 + "#bef264", # lime-300 + ], + }, +} + +DEFAULT_THEME = "tailwind" + + +class _ColorAssigner: + """Assigns colours from a palette sequentially, cycling when exhausted.""" + + def __init__(self, palette: list[str]) -> None: + self._palette = palette + self._assigned: dict[str, str] = {} + self._next = 0 + + def __call__(self, key: str) -> str: + if key not in self._assigned: + self._assigned[key] = self._palette[self._next % len(self._palette)] + self._next += 1 + return self._assigned[key] + + +# Connections to ignore (too noisy/common) +DEFAULT_IGNORED_CONNECTIONS = {("odom", "PoseStamped")} + +DEFAULT_IGNORED_MODULES = { + "WebsocketVisModule", +} + +_COMPACT_ONLY_IGNORED_MODULES = { + "WebsocketVisModule", +} + + +def _mermaid_id(name: str) -> str: + """Sanitize a string into a valid Mermaid node id.""" + return name.replace(" ", "_").replace("-", "_") + + +def render( + blueprint_set: Blueprint, + *, + ignored_streams: set[tuple[str, str]] | None = None, + ignored_modules: set[str] | None = None, + show_disconnected: bool = False, + theme: str = DEFAULT_THEME, +) -> tuple[str, dict[str, str], set[str]]: + """Generate a Mermaid flowchart from a Blueprint. + + Returns ``(mermaid_code, label_color_map, disconnected_labels)`` where + *label_color_map* maps each edge label string to its hex colour and + *disconnected_labels* is the set of labels for dangling streams. + + Args: + theme: Colour theme name (one of ``THEMES`` keys). + """ + if ignored_streams is None: + ignored_streams = DEFAULT_IGNORED_CONNECTIONS + if ignored_modules is None: + if show_disconnected: + ignored_modules = DEFAULT_IGNORED_MODULES - _COMPACT_ONLY_IGNORED_MODULES + else: + ignored_modules = DEFAULT_IGNORED_MODULES + + # Collect producers/consumers + producers: dict[tuple[str, type], list[type[Module]]] = defaultdict(list) + consumers: dict[tuple[str, type], list[type[Module]]] = defaultdict(list) + module_names: set[str] = set() + + for bp in blueprint_set.blueprints: + if bp.module.__name__ in ignored_modules: + continue + module_names.add(bp.module.__name__) + for conn in bp.streams: + remapped_name = blueprint_set.remapping_map.get((bp.module, conn.name), conn.name) + key = (remapped_name, conn.type) + if conn.direction == "out": + producers[key].append(bp.module) + else: + consumers[key].append(bp.module) + + # Active channels: both producer and consumer exist + active_keys: list[tuple[str, type]] = [] + for key in producers: + name, type_ = key + if key not in consumers: + continue + if (name, type_.__name__) in ignored_streams: + continue + valid_p = [m for m in producers[key] if m.__name__ not in ignored_modules] + valid_c = [m for m in consumers[key] if m.__name__ not in ignored_modules] + if valid_p and valid_c: + active_keys.append(key) + + # Disconnected channels + disconnected_keys: list[tuple[str, type]] = [] + if show_disconnected: + all_keys = set(producers.keys()) | set(consumers.keys()) + for key in all_keys: + if key in active_keys: + continue + name, type_ = key + if (name, type_.__name__) in ignored_streams: + continue + relevant = producers.get(key, []) + consumers.get(key, []) + if all(m.__name__ in ignored_modules for m in relevant): + continue + disconnected_keys.append(key) + + palette = THEMES.get(theme, THEMES[DEFAULT_THEME]) + node_color = _ColorAssigner(palette["nodes"]) + edge_color = _ColorAssigner(palette["edges"]) + + lines = ["graph LR"] + + # Declare module nodes with rounded boxes + sorted_modules = sorted(module_names) + for mod_name in sorted_modules: + mid = _mermaid_id(mod_name) + lines.append(f" {mid}([{mod_name}]):::moduleNode") + + lines.append("") + + edge_idx = 0 + edge_colors: list[str] = [] + label_color_map: dict[str, str] = {} + stream_node_ids: dict[str, str] = {} # stream_node_id -> color + disconnected_labels: set[str] = set() + + # Active streams: producer -> stream-node -> consumers + lines.append(" %% Stream nodes and edges") + for key in sorted(active_keys, key=lambda k: f"{k[0]}:{k[1].__name__}"): + name, type_ = key + label = f"{name}:{type_.__name__}" + color = edge_color(label) + label_color_map[label] = color + + valid_producers = [m for m in producers[key] if m.__name__ not in ignored_modules] + valid_consumers = [m for m in consumers[key] if m.__name__ not in ignored_modules] + + for prod in valid_producers: + # Create a stream node per producer+stream pair + sn_id = _mermaid_id(f"{prod.__name__}_{name}_{type_.__name__}") + if sn_id not in stream_node_ids: + lines.append(f" {sn_id}[{label}]:::streamNode") + stream_node_ids[sn_id] = color + + # Edge: producer --- stream-node (no arrow, module color) + pid = _mermaid_id(prod.__name__) + lines.append(f" {pid} --- {sn_id}") + edge_colors.append(node_color(prod.__name__)) + edge_idx += 1 + + # Edges: stream-node -> each consumer + for cons in valid_consumers: + cid = _mermaid_id(cons.__name__) + lines.append(f" {sn_id} --> {cid}") + edge_colors.append(color) + edge_idx += 1 + + # Disconnected streams + if disconnected_keys: + lines.append("") + lines.append(" %% Disconnected streams") + for key in sorted(disconnected_keys, key=lambda k: f"{k[0]}:{k[1].__name__}"): + name, type_ = key + label = f"{name}:{type_.__name__}" + color = edge_color(label) + label_color_map[label] = color + disconnected_labels.add(label) + + for prod in producers.get(key, []): + if prod.__name__ in ignored_modules: + continue + sn_id = _mermaid_id(f"{prod.__name__}_{name}_{type_.__name__}") + if sn_id not in stream_node_ids: + lines.append(f" {sn_id}[{label}]:::streamNode") + stream_node_ids[sn_id] = color + pid = _mermaid_id(prod.__name__) + lines.append(f" {pid} -.- {sn_id}") + edge_colors.append(node_color(prod.__name__)) + edge_idx += 1 + + for cons in consumers.get(key, []): + if cons.__name__ in ignored_modules: + continue + # Consumer-only: create a standalone stream node + sn_id = _mermaid_id(f"dangling_{name}_{type_.__name__}") + if sn_id not in stream_node_ids: + lines.append(f" {sn_id}[{label}]:::streamNode") + stream_node_ids[sn_id] = color + cid = _mermaid_id(cons.__name__) + lines.append(f" {sn_id} -.-> {cid}") + edge_colors.append(color) + edge_idx += 1 + + # Module node styles (colored fill) + lines.append("") + for mod_name in sorted_modules: + mid = _mermaid_id(mod_name) + c = node_color(mod_name) + lines.append(f" style {mid} fill:{c}bf,stroke:{c},color:#eee,stroke-width:2px") + + # Stream node styles (no fill, colored text and border) + for sn_id, color in stream_node_ids.items(): + lines.append( + f" style {sn_id} fill:transparent,stroke:{color},color:{color},stroke-width:1px" + ) + + # Edge styles + if edge_colors: + lines.append("") + for i, c in enumerate(edge_colors): + lines.append(f" linkStyle {i} stroke:{c},stroke-width:2px") + + return "\n".join(lines), label_color_map, disconnected_labels diff --git a/dimos/core/introspection/blueprint/test_dot.py b/dimos/core/introspection/blueprint/test_dot.py new file mode 100644 index 0000000000..7eabd885b9 --- /dev/null +++ b/dimos/core/introspection/blueprint/test_dot.py @@ -0,0 +1,62 @@ +# Copyright 2025-2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +from dimos.core.blueprints import autoconnect +from dimos.core.introspection.blueprint.dot import render +from dimos.core.module import Module +from dimos.core.stream import In, Out + + +class MsgA: + pass + + +class MsgB: + pass + + +class ProducerModule(Module): + output_a: Out[MsgA] + output_b: Out[MsgB] + + +class ConsumerModule(Module): + output_a: In[MsgA] + + +# output_a connects (same name+type), output_b is disconnected (no consumer) +_combined = autoconnect(ProducerModule.blueprint(), ConsumerModule.blueprint()) + + +def test_render_without_disconnected() -> None: + dot = render(_combined, ignored_streams=set(), ignored_modules=set(), show_disconnected=False) + # Connected channel should be present + assert "output_a:MsgA" in dot + # Disconnected output_b should NOT appear + assert "output_b:MsgB" not in dot + + +def test_render_with_disconnected() -> None: + dot = render(_combined, ignored_streams=set(), ignored_modules=set(), show_disconnected=True) + # Connected channel should be present + assert "output_a:MsgA" in dot + # Disconnected output_b SHOULD appear with dashed style + assert "output_b:MsgB" in dot + assert "style=dashed" in dot + + +def test_disconnected_default_is_false() -> None: + dot = render(_combined, ignored_streams=set(), ignored_modules=set()) + assert "output_b:MsgB" not in dot diff --git a/dimos/core/introspection/svg.py b/dimos/core/introspection/svg.py index 57b88834e0..0aaed3a105 100644 --- a/dimos/core/introspection/svg.py +++ b/dimos/core/introspection/svg.py @@ -29,6 +29,7 @@ def to_svg( output_path: str, *, layout: set[LayoutAlgo] | None = None, + show_disconnected: bool = False, ) -> None: """Render a module or blueprint to SVG. @@ -40,6 +41,7 @@ def to_svg( target: Either a ModuleInfo (single module) or Blueprint (blueprint graph). output_path: Path to write the SVG file. layout: Layout algorithms (only used for blueprints). + show_disconnected: If True, show streams with no matching counterpart (blueprints only). """ # Avoid circular imports by importing here from dimos.core.blueprints import Blueprint @@ -52,6 +54,8 @@ def to_svg( elif isinstance(target, Blueprint): from dimos.core.introspection.blueprint import dot as blueprint_dot - blueprint_dot.render_svg(target, output_path, layout=layout) + blueprint_dot.render_svg( + target, output_path, layout=layout, show_disconnected=show_disconnected + ) else: raise TypeError(f"Expected ModuleInfo or Blueprint, got {type(target).__name__}") diff --git a/dimos/robot/cli/dimos.py b/dimos/robot/cli/dimos.py index 1137a612f3..224f8d5cce 100644 --- a/dimos/robot/cli/dimos.py +++ b/dimos/robot/cli/dimos.py @@ -550,6 +550,20 @@ def send( topic_send(topic, message_expr) +@main.command() +def graph( + python_file: str = typer.Argument(..., help="Python file containing Blueprint globals"), + no_disconnected: bool = typer.Option( + False, "--no-disconnected", help="Hide disconnected streams" + ), + port: int = typer.Option(0, "--port", help="HTTP server port (0 = random free port)"), +) -> None: + """Render blueprint graphs from a Python file and open in browser.""" + from dimos.utils.cli.graph import main as graph_main + + graph_main(python_file, show_disconnected=not no_disconnected, port=port) + + @main.command(name="rerun-bridge") def rerun_bridge_cmd( viewer_mode: str = typer.Option( diff --git a/dimos/utils/cli/graph.py b/dimos/utils/cli/graph.py new file mode 100644 index 0000000000..540f0dc99a --- /dev/null +++ b/dimos/utils/cli/graph.py @@ -0,0 +1,386 @@ +# Copyright 2025-2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Render Blueprint graphs from a Python file and open in the browser.""" + +from __future__ import annotations + +import importlib.util +import os +import shutil +import sys +import tempfile +from typing import TYPE_CHECKING +import webbrowser + +if TYPE_CHECKING: + from dimos.core.blueprints import Blueprint + + +def _find_package_root(filepath: str) -> str | None: + """Walk up from *filepath* looking for the outermost package directory. + + Returns the parent of that package (i.e. the directory that should be on + ``sys.path``), or ``None`` if the file is not inside a package. + """ + d = os.path.dirname(filepath) + root = None + while os.path.isfile(os.path.join(d, "__init__.py")): + root = d + parent = os.path.dirname(d) + if parent == d: + break + d = parent + if root is not None: + return os.path.dirname(root) + return None + + +def _load_blueprints(python_file: str) -> list[tuple[str, Blueprint]]: + """Import *python_file* and return ``[(name, Blueprint), ...]``.""" + filepath = os.path.abspath(python_file) + if not os.path.isfile(filepath): + raise FileNotFoundError(filepath) + + # Ensure the file's package root is importable so that relative imports + # like ``from smartnav.blueprints.foo import bar`` work. + pkg_root = _find_package_root(filepath) + if pkg_root and pkg_root not in sys.path: + sys.path.insert(0, pkg_root) + + spec = importlib.util.spec_from_file_location("_render_target", filepath) + if spec is None or spec.loader is None: + raise RuntimeError(f"Could not load {filepath}") + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + from dimos.core.blueprints import Blueprint + + blueprints: list[tuple[str, Blueprint]] = [] + for name, obj in vars(mod).items(): + if name.startswith("_"): + continue + if isinstance(obj, Blueprint): + blueprints.append((name, obj)) + + if not blueprints: + raise RuntimeError("No Blueprint instances found in module globals.") + + print(f"Found {len(blueprints)} blueprint(s): {', '.join(n for n, _ in blueprints)}") + return blueprints + + +def _build_html(python_file: str, *, show_disconnected: bool = True) -> str: + """Build an HTML page that renders blueprints as Mermaid diagrams.""" + from dimos.core.introspection.blueprint.mermaid import render as mermaid_render + + blueprints = _load_blueprints(python_file) + + import json + + sections = [] + all_label_colors: dict[str, str] = {} + all_disconnected: set[str] = set() + for name, bp in blueprints: + mermaid_code, label_colors, disconnected = mermaid_render( + bp, show_disconnected=show_disconnected + ) + all_label_colors.update(label_colors) + all_disconnected.update(disconnected) + sections.append( + f"

{name}

\n" + f'
' + f'
\n{mermaid_code}\n
' + f"
" + ) + label_colors_json = json.dumps(all_label_colors) + disconnected_json = json.dumps(sorted(all_disconnected)) + + return f"""\ + + + +Blueprint Diagrams + + +{"".join(sections)} +
+ + + +
+ +""" + + +def _build_html_graphviz(python_file: str, *, show_disconnected: bool = True) -> str: + """Build an HTML page that renders blueprints as Graphviz SVGs (requires ``dot``).""" + from dimos.core.introspection.svg import to_svg + + blueprints = _load_blueprints(python_file) + + if not shutil.which("dot"): + raise RuntimeError( + "graphviz is not installed (the 'dot' command was not found).\n" + "Install it with: brew install graphviz (macOS)\n" + " apt install graphviz (Debian/Ubuntu)" + ) + + sections = [] + for name, bp in blueprints: + fd, svg_path = tempfile.mkstemp(suffix=".svg", prefix=f"dimos_{name}_") + os.close(fd) + to_svg(bp, svg_path, show_disconnected=show_disconnected) + with open(svg_path) as f: + svg_content = f.read() + os.unlink(svg_path) + sections.append(f'

{name}

\n
{svg_content}
') + + return f"""\ + + + +Blueprint Diagrams + + +{"".join(sections)} +""" + + +def main(python_file: str, *, show_disconnected: bool = True, port: int = 0) -> None: + """Render Blueprint SVG diagrams and display them via a one-shot HTTP server.""" + from http.server import BaseHTTPRequestHandler, HTTPServer + + html = _build_html(python_file, show_disconnected=show_disconnected) + html_bytes = html.encode("utf-8") + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(html_bytes))) + self.end_headers() + self.wfile.write(html_bytes) + + def log_message(self, format: str, *args: object) -> None: + pass + + server = HTTPServer(("0.0.0.0", port), Handler) + actual_port = server.server_address[1] + url = f"http://localhost:{actual_port}" + print(f"Serving at {url} (will exit after first request)") + webbrowser.open(url) + server.handle_request() + print("Served. Exiting.") diff --git a/dimos/utils/cli/test_graph.py b/dimos/utils/cli/test_graph.py new file mode 100644 index 0000000000..4f1ceedfb2 --- /dev/null +++ b/dimos/utils/cli/test_graph.py @@ -0,0 +1,41 @@ +# Copyright 2025-2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +import pytest + +from dimos.utils.cli.graph import main + + +def test_file_not_found() -> None: + with pytest.raises(FileNotFoundError): + main("/nonexistent/path.py") + + +def test_no_blueprints(tmp_path: object) -> None: + import pathlib + + p = pathlib.Path(str(tmp_path)) / "empty.py" + p.write_text("x = 42\n") + with pytest.raises(RuntimeError, match="No Blueprint instances"): + main(str(p)) + + +def test_module_load_failure(tmp_path: object) -> None: + import pathlib + + p = pathlib.Path(str(tmp_path)) / "bad.py" + p.write_text("raise ImportError('boom')\n") + with pytest.raises(ImportError, match="boom"): + main(str(p))