From 2215068649e1403138f1dd594f454db87b791154 Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Sat, 3 Oct 2026 00:39:53 +0200 Subject: [PATCH] feat(architecture): extract code2flow package for Mermaid diagrams and visualizations Co-authored-by: Koru Agent --- .../modularization-and-rust-engine-plan.md | 6 +- packages/code2flow/README.md | 13 + packages/code2flow/code2flow/__init__.py | 22 ++ packages/code2flow/code2flow/calls.py | 70 +++++ packages/code2flow/code2flow/compact.py | 52 +++ packages/code2flow/code2flow/fix.py | 152 +++++++++ packages/code2flow/code2flow/flow_compact.py | 180 +++++++++++ packages/code2flow/code2flow/flow_detailed.py | 84 +++++ packages/code2flow/code2flow/flow_full.py | 84 +++++ packages/code2flow/code2flow/flow_helpers.py | 295 ++++++++++++++++++ packages/code2flow/code2flow/utils.py | 101 ++++++ packages/code2flow/code2flow/validation.py | 138 ++++++++ packages/code2flow/pyproject.toml | 28 ++ packages/code2flow/tests/test_code2flow.py | 47 +++ pyproject.toml | 4 + 15 files changed, 1273 insertions(+), 3 deletions(-) create mode 100644 packages/code2flow/README.md create mode 100644 packages/code2flow/code2flow/__init__.py create mode 100644 packages/code2flow/code2flow/calls.py create mode 100644 packages/code2flow/code2flow/compact.py create mode 100644 packages/code2flow/code2flow/fix.py create mode 100644 packages/code2flow/code2flow/flow_compact.py create mode 100644 packages/code2flow/code2flow/flow_detailed.py create mode 100644 packages/code2flow/code2flow/flow_full.py create mode 100644 packages/code2flow/code2flow/flow_helpers.py create mode 100644 packages/code2flow/code2flow/utils.py create mode 100644 packages/code2flow/code2flow/validation.py create mode 100644 packages/code2flow/pyproject.toml create mode 100644 packages/code2flow/tests/test_code2flow.py diff --git a/docs/refactoring/modularization-and-rust-engine-plan.md b/docs/refactoring/modularization-and-rust-engine-plan.md index aa729e8..5b216e7 100644 --- a/docs/refactoring/modularization-and-rust-engine-plan.md +++ b/docs/refactoring/modularization-and-rust-engine-plan.md @@ -164,7 +164,7 @@ graph TD - Implemented parallel cache validation (`PersistentCache`) with Rayon & SHA-256 in Rust. - Implemented native call graph resolution with $O(1)$ candidate lookups and module affinity matching. -4. **Phase 4 (In Progress - Ecosystem Modularization)**: +4. **Phase 4 (Complete - Ecosystem Modularization)**: - **`code2graph` (Complete)**: Extracted graph data models (`AnalysisResult`, `FunctionInfo`, `FlowNode`, `FlowEdge`, `GraphQuery`) into standalone package at `packages/code2graph` and published to `https://github.com/semcod/code2graph`. - - **`code2toon` (Complete)**: Extracted TOON format specification, parser, validator, and diff engine into standalone package at `packages/code2toon`. - - **`code2flow`**: Extract visual diagrams and heavy rendering dependencies into standalone package. + - **`code2toon` (Complete)**: Extracted TOON format specification, parser, validator, and diff engine into standalone package at `packages/code2toon` and published to `https://github.com/semcod/code2toon`. + - **`code2flow` (Complete)**: Extracted Mermaid visual flow diagrams, call graph visualizations, and syntax validators into standalone package at `packages/code2flow`. diff --git a/packages/code2flow/README.md b/packages/code2flow/README.md new file mode 100644 index 0000000..ff5cbe1 --- /dev/null +++ b/packages/code2flow/README.md @@ -0,0 +1,13 @@ +# code2flow + +Mermaid flow diagrams, call graph visualizations, and architectural diagrams for static code analysis. + +## Features +- **Mermaid Exporters**: Generate compact architectural diagrams, full function-level flowcharts, and filtered call graphs (`flow.mmd`, `compact_flow.mmd`, `calls.mmd`). +- **Syntax Validation & Auto-fix**: Robust syntax validation for Mermaid diagrams with automated sanitization of labels, node names, and subgraph delimiters. +- **Pure Topology Input**: Accepts `AnalysisResult` from `code2graph` without coupling to file system crawlers or AST parsers. + +## Installation +```bash +pip install code2flow +``` diff --git a/packages/code2flow/code2flow/__init__.py b/packages/code2flow/code2flow/__init__.py new file mode 100644 index 0000000..8d73795 --- /dev/null +++ b/packages/code2flow/code2flow/__init__.py @@ -0,0 +1,22 @@ +"""code2flow: Mermaid flow diagrams, call graph visualizations, and validation.""" + +from .compact import export_compact +from .flow_compact import export_flow_compact +from .flow_detailed import export_flow_detailed +from .flow_full import export_flow_full +from .calls import export_calls +from .validation import validate_mermaid_file +from .fix import fix_mermaid_file + +__version__ = "0.1.0" + +__all__ = [ + "export_compact", + "export_flow_compact", + "export_flow_detailed", + "export_flow_full", + "export_calls", + "validate_mermaid_file", + "fix_mermaid_file", + "__version__", +] diff --git a/packages/code2flow/code2flow/calls.py b/packages/code2flow/code2flow/calls.py new file mode 100644 index 0000000..d44df7b --- /dev/null +++ b/packages/code2flow/code2flow/calls.py @@ -0,0 +1,70 @@ +"""Mermaid calls export — calls.mmd simplified call graph.""" + +from typing import Dict, List, Set, Tuple + +from code2graph import AnalysisResult + +from .utils import ( + readable_id, + safe_module, + resolve_callee, + write_file, + build_name_index, +) + + +def _collect_call_edges( + functions: dict, name_index: dict +) -> Tuple[Set[str], List[Tuple[str, str]]]: + """Collect connected nodes and edges from call graph (max 500 edges).""" + connected: Set[str] = set() + edges: List[Tuple[str, str]] = [] + for func_name, fi in functions.items(): + for callee in fi.calls[:10]: + resolved = resolve_callee(callee, functions, name_index) + if resolved and resolved != func_name: + connected.add(func_name) + connected.add(resolved) + edges.append((func_name, resolved)) + if len(edges) >= 500: + return connected, edges + return connected, edges + + +def export_calls(result: AnalysisResult, output_path: str) -> None: + """Export simplified call graph — only connected nodes.""" + lines = ["flowchart LR"] + name_index = build_name_index(result.functions) + connected, edges = _collect_call_edges(result.functions, name_index) + + # Group connected nodes by module + from .utils import module_of + + modules: Dict[str, list] = {} + for fn in connected: + module = module_of(fn) + fi = result.functions.get(fn) + if fi: + modules.setdefault(module, []).append((fn, fi)) + + for module, funcs in sorted(modules.items()): + safe_mod = safe_module(module) + lines.append(f" subgraph {safe_mod}") + for func_name, fi in funcs: + sid = readable_id(func_name) + short = fi.name[:30] + lines.append(f' {sid}["{short}"]') + lines.append(" end") + + # Edges + seen: Set[Tuple[str, str]] = set() + for src, dst in edges: + pair = (readable_id(src), readable_id(dst)) + if pair not in seen: + seen.add(pair) + lines.append(f" {pair[0]} --> {pair[1]}") + + write_file(output_path, lines) + + +__all__ = ["export_calls"] diff --git a/packages/code2flow/code2flow/compact.py b/packages/code2flow/code2flow/compact.py new file mode 100644 index 0000000..e6cdaf1 --- /dev/null +++ b/packages/code2flow/code2flow/compact.py @@ -0,0 +1,52 @@ +"""Mermaid compact export — compact_flow.mmd module-level aggregation.""" + +from collections import defaultdict +from typing import Dict, Set, Tuple + +from code2graph import AnalysisResult + +from .utils import safe_module, resolve_callee, write_file, build_name_index + + +def _compute_module_stats(functions: dict, module_of) -> Tuple[Dict, Dict]: + """Return (mod_funcs, mod_lines) counters keyed by module name.""" + mod_funcs: Dict[str, int] = defaultdict(int) + mod_lines: Dict[str, int] = defaultdict(int) + for func_name, fi in functions.items(): + mod = module_of(func_name) + mod_funcs[mod] += 1 + mod_lines[mod] += (fi.end_line - fi.line if getattr(fi, "end_line", None) else 0) + return mod_funcs, mod_lines + + +def _compute_cross_edges(functions: dict, name_index: dict, module_of) -> Dict[Tuple[str, str], int]: + """Return weighted cross-module edge counter.""" + cross_edges: Dict[Tuple[str, str], int] = defaultdict(int) + for func_name, fi in functions.items(): + src_mod = module_of(func_name) + for callee in fi.calls: + resolved = resolve_callee(callee, functions, name_index) + if resolved: + dst_mod = module_of(resolved) + if dst_mod != src_mod: + cross_edges[(src_mod, dst_mod)] += 1 + return cross_edges + + +def export_compact(result: AnalysisResult, output_path: str) -> None: + """Export module-level graph: one node per module, weighted edges.""" + from .utils import module_of + lines = ["flowchart TD"] + name_index = build_name_index(result.functions) + mod_funcs, _ = _compute_module_stats(result.functions, module_of) + cross_edges = _compute_cross_edges(result.functions, name_index, module_of) + active_mods: Set[str] = {m for pair in cross_edges for m in pair} or set(mod_funcs) + for mod in sorted(active_mods): + lines.append(f' {safe_module(mod)}["{mod}
{mod_funcs.get(mod, 0)} funcs"]') + for (src, dst), weight in sorted(cross_edges.items(), key=lambda x: -x[1]): + arrow = "==>" if weight > 3 else "-->" + lines.append(f" {safe_module(src)} {arrow}|{weight}| {safe_module(dst)}") + write_file(output_path, lines) + + +__all__ = ["export_compact"] diff --git a/packages/code2flow/code2flow/fix.py b/packages/code2flow/code2flow/fix.py new file mode 100644 index 0000000..df40c2a --- /dev/null +++ b/packages/code2flow/code2flow/fix.py @@ -0,0 +1,152 @@ +"""Mermaid file fixer — auto-repair common syntax errors.""" + +from pathlib import Path +from typing import List, Optional, Union + + +def _sanitize_label_text(txt: str) -> str: + """Replace Mermaid syntax chars in labels with HTML entities.""" + return ( + txt.replace("&", "&") + .replace('"', """) + .replace("[", "[") + .replace("]", "]") + .replace("(", "(") + .replace(")", ")") + .replace("{", "{") + .replace("}", "}") + .replace("|", "|") + ) + + +def _sanitize_node_id(node_id: str) -> str: + """Make a Mermaid-safe node identifier.""" + import re + + node_id = (node_id or "").strip() + node_id = re.split(r"[\[\]\(\)\{\}\"\|\s]", node_id, maxsplit=1)[0] + node_id = re.sub(r"[^A-Za-z0-9_]", "_", node_id) + return node_id or "N" + + +def fix_mermaid_file(mmd_path: Union[str, Path]) -> bool: + """Attempt to fix common Mermaid syntax errors.""" + mmd_path = Path(mmd_path) + try: + content = mmd_path.read_text(encoding="utf-8") + lines = content.split("\n") + fixed_lines = [] + + for line in lines: + line = _fix_edge_line(line) + line = _fix_subgraph_line(line) + result = _fix_class_line(line) + if result is not None: + fixed_lines.extend(result) + else: + fixed_lines.append(line) + + fixed_content = "\n".join(fixed_lines) + if fixed_content != content: + mmd_path.write_text(fixed_content, encoding="utf-8") + return True + + except Exception as e: + print(f"Error fixing {mmd_path}: {e}") + + return False + + +def _fix_edge_line(line: str) -> str: + """Fix edge labels and endpoint issues.""" + import re + + if "-->" not in line: + return line + + # Fix edge labels with pipe issues + line = _fix_edge_label_pipes(line) + + # Fix stray trailing '|' after node IDs + line = re.sub(r"(\b[A-Za-z]\w*)\|\s*$", r"\1", line) + + # Sanitize edge label content inside |...| + def _sanitize_edge_label(m): + """Replace unsafe characters inside a Mermaid |label| with safe equivalents.""" + return f"|{_sanitize_label_text(m.group(1))}|" + + if "|" in line: + line = re.sub(r"\|([^|]{1,200})\|", _sanitize_edge_label, line) + + # Sanitize edge endpoints (lines without labels) + if "|" not in line: + m = re.match(r"^(\s*)([^\s-]+)\s*-->\s*([^\s]+)\s*$", line) + if m: + indent, src, dst = m.groups() + line = f"{indent}{_sanitize_node_id(src)} --> {_sanitize_node_id(dst)}" + + return line + + +def _fix_edge_label_pipes(line: str) -> str: + """Fix edge labels with pipe/parenthesis issues.""" + if "|" not in line or "-->|" not in line: + return line + parts = line.split("-->|", 1) + if len(parts) != 2: + return line + label_and_target = parts[1] + if "|" not in label_and_target: + return line + parts2 = label_and_target.split("|", 1) + if len(parts2) != 2: + return line + label_content, target = parts2 + label_content = label_content.strip("|") + if label_content.endswith("("): + label_content = label_content[:-1] + elif label_content.count("(") > label_content.count(")"): + missing = label_content.count("(") - label_content.count(")") + label_content += ")" * missing + return f"{parts[0]}-->|{label_content}|{target}" + + +def _fix_subgraph_line(line: str) -> str: + """Fix malformed subgraph IDs.""" + if line.strip().startswith("subgraph "): + subgraph_part = line.strip()[9:].split("(", 1) + if len(subgraph_part) == 2: + subgraph_id, rest = subgraph_part + subgraph_id = ( + subgraph_id.replace(".", "_").replace("-", "_").replace(":", "_") + ) + line = f" subgraph {subgraph_id}({rest}" + return line + + +def _fix_class_line(line: str) -> Optional[List[str]]: + """Fix class definitions with too many nodes. Returns list of lines or None.""" + if line.strip().startswith("class ") and "," in line: + class_parts = line.split(" ", 1) + if len(class_parts) == 2: + nodes_and_class = class_parts[1] + nodes, class_name = nodes_and_class.rsplit(" ", 1) + node_list = nodes.split(",") + if len(node_list) > 10: + result = [] + for i in range(0, len(node_list), 10): + chunk = ",".join(node_list[i : i + 10]) + result.append(f" class {chunk} {class_name}") + return result + return None + + +__all__ = [ + "fix_mermaid_file", + "_sanitize_label_text", + "_sanitize_node_id", + "_fix_edge_line", + "_fix_edge_label_pipes", + "_fix_subgraph_line", + "_fix_class_line", +] diff --git a/packages/code2flow/code2flow/flow_compact.py b/packages/code2flow/code2flow/flow_compact.py new file mode 100644 index 0000000..2317670 --- /dev/null +++ b/packages/code2flow/code2flow/flow_compact.py @@ -0,0 +1,180 @@ +"""Mermaid flow compact export — flow.mmd architectural view (~50 nodes).""" + +from collections import defaultdict +from typing import Dict, List, Set + +from code2graph import AnalysisResult + +from .utils import readable_id, resolve_callee, write_file, get_cc, build_name_index + + +# Default skip patterns for noise reduction +SKIP_PATTERNS = ("examples", "benchmarks", "demo_langs", "tests", "scripts") + + +def should_skip_module(module: str, include_examples: bool = False) -> bool: + """Check if module should be skipped (examples, benchmarks, etc.).""" + if include_examples: + return False + mod_lower = module.lower() + return any(pat in mod_lower for pat in SKIP_PATTERNS) + + +# Pre-computed set of functions that have at least one incoming call +_called_funcs_cache: Dict[int, Set[str]] = {} + + +def _get_called_funcs(result: AnalysisResult) -> Set[str]: + """Build (and cache) the set of all functions called by other functions.""" + rid = id(result) + if rid not in _called_funcs_cache: + called: Set[str] = set() + for fi in result.functions.values(): + called.update(fi.calls) + _called_funcs_cache[rid] = called + return _called_funcs_cache[rid] + + +def is_entry_point(func_name: str, fi, result: AnalysisResult) -> bool: + """Detect if function is an entry point (main, cli, api entry).""" + name = fi.name + # Common entry point patterns + if name in ("main", "cli_main", "run", "start", "serve"): + return True + if name.startswith("main_") or name.endswith("_main"): + return True + # CLI commands + if "cli" in func_name.lower() and name not in ("__init__", "create_parser"): + return True + # API handlers + if func_name.startswith("code2llm.api."): + return True + # Entry points have no incoming calls from within the project + called_funcs = _get_called_funcs(result) + if func_name not in called_funcs and name not in ("__init__", "__getattr__"): + return True + return False + + +def build_callers_graph( + result: AnalysisResult, name_index: Dict[str, List[str]] +) -> Dict[str, Set[str]]: + """Build reverse graph: map each function to its callers.""" + callers: Dict[str, Set[str]] = defaultdict(set) + for func_name, fi in result.functions.items(): + for callee in fi.calls: + resolved = resolve_callee(callee, result.functions, name_index) + if resolved: + callers[resolved].add(func_name) + return callers + + +def find_leaves(result: AnalysisResult, name_index: Dict[str, List[str]]) -> Set[str]: + """Find leaf nodes (functions that don't call other project functions).""" + leaves = set() + for func_name, fi in result.functions.items(): + has_internal_call = any( + resolve_callee(c, result.functions, name_index) for c in fi.calls + ) + if not has_internal_call: + leaves.add(func_name) + return leaves + + +def _longest_path_dfs( + result: AnalysisResult, + start: str, + visited: Set[str], + name_index: Dict[str, List[str]], +) -> List[str]: + """DFS to find longest path from start node.""" + if start in visited: + return [] + visited = visited | {start} + fi = result.functions.get(start) + if not fi: + return [start] + + longest: List[str] = [] + for callee in fi.calls: + resolved = resolve_callee(callee, result.functions, name_index) + if resolved and resolved not in visited: + path = _longest_path_dfs(result, resolved, visited, name_index) + if len(path) > len(longest): + longest = path + + return [start] + longest + + +def _select_longest_path( + result: AnalysisResult, entry_points: List[str], name_index: Dict[str, List[str]] +) -> List[str]: + """Select the longest path from all entry points.""" + max_path: List[str] = [] + for ep in entry_points: + if ep in result.functions: + path = _longest_path_dfs(result, ep, set(), name_index) + if len(path) > len(max_path): + max_path = path + return max_path + + +def find_critical_path(result: AnalysisResult, entry_points: List[str]) -> Set[str]: + """Find the longest path from entry points (critical path).""" + if not entry_points: + return set() + # Build name index for O(1) resolution + name_index = build_name_index(result.functions) + # Build data structures + build_callers_graph(result, name_index) + find_leaves(result, name_index) + # Find longest path from each entry point + max_path = _select_longest_path(result, entry_points, name_index) + return set(max_path) + + +def export_flow_compact( + result: AnalysisResult, output_path: str, include_examples: bool = False +) -> None: + """Export compact architectural view (~50 nodes). + + Shows entry points, high-level modules, and critical path. + """ + from .flow_helpers import ( + _entry_points, + _filtered_functions, + _render_architecture_view, + ) + from .utils import module_of + + lines = ["flowchart TD"] + lines.append("") + lines.append(" %% Entry points (blue)") + lines.append(" classDef entry fill:#4dabf7,stroke:#1971c2,color:#fff") + lines.append("") + filtered_funcs = _filtered_functions( + result, + module_of, + should_skip_module, + include_examples, + ) + entry_points = _entry_points(filtered_funcs, result, is_entry_point) + critical_path = find_critical_path(result, entry_points) + _render_architecture_view( + lines, + filtered_funcs, + entry_points, + critical_path, + module_of, + readable_id, + get_cc, + ) + write_file(output_path, lines) + + +__all__ = [ + "export_flow_compact", + "should_skip_module", + "is_entry_point", + "find_critical_path", +] diff --git a/packages/code2flow/code2flow/flow_detailed.py b/packages/code2flow/code2flow/flow_detailed.py new file mode 100644 index 0000000..038fb46 --- /dev/null +++ b/packages/code2flow/code2flow/flow_detailed.py @@ -0,0 +1,84 @@ +"""Mermaid flow detailed export — flow_detailed.mmd per-module view (~150 nodes).""" + +from code2graph import AnalysisResult + +from .utils import ( + readable_id, + safe_module, + resolve_callee, + write_file, + get_cc, + build_name_index, +) +from .flow_helpers import ( + _entry_points, + _filtered_functions, + _group_functions_by_module, + _render_module_subgraphs, + _render_flow_edges, + _render_flow_styles, +) +from .flow_compact import should_skip_module, is_entry_point + + +def export_flow_detailed( + result: AnalysisResult, output_path: str, include_examples: bool = False +) -> None: + """Export detailed per-module view (~150 nodes). + + Shows all significant functions per module with CC annotations. + """ + from .utils import module_of + + lines = ["flowchart TD"] + + # Build name index for O(1) callee resolution + name_index = build_name_index(result.functions) + lines.append("") + lines.append(" %% Styling definitions") + lines.append(" classDef highCC fill:#ff6b6b,stroke:#c92a2a,color:#fff") + lines.append(" classDef medCC fill:#ffd43b,stroke:#f08c00,color:#000") + lines.append(" classDef entry fill:#4dabf7,stroke:#1971c2,color:#fff") + lines.append("") + filtered_funcs = _filtered_functions( + result, + module_of, + should_skip_module, + include_examples, + ) + entry_points = _entry_points(filtered_funcs, result, is_entry_point) + modules = _group_functions_by_module(filtered_funcs, module_of) + _render_module_subgraphs( + lines, + modules, + entry_points, + short_len=35, + readable_id=readable_id, + safe_module=safe_module, + get_cc=get_cc, + sort_funcs=True, + max_funcs=40, + ) + _render_flow_edges( + lines, + filtered_funcs, + readable_id, + resolve_callee, + calls_per_function=10, + limit=200, + name_index=name_index, + ) + _render_flow_styles( + lines, + filtered_funcs, + entry_points, + readable_id, + get_cc, + high_limit=30, + med_limit=30, + entry_limit=10, + ) + write_file(output_path, lines) + + +__all__ = ["export_flow_detailed"] diff --git a/packages/code2flow/code2flow/flow_full.py b/packages/code2flow/code2flow/flow_full.py new file mode 100644 index 0000000..b895b0b --- /dev/null +++ b/packages/code2flow/code2flow/flow_full.py @@ -0,0 +1,84 @@ +"""Mermaid flow full export — flow_full.mmd debug view with all nodes.""" + +from code2graph import AnalysisResult + +from .utils import ( + readable_id, + safe_module, + resolve_callee, + write_file, + get_cc, + build_name_index, +) +from .flow_helpers import ( + _entry_points, + _filtered_functions, + _group_functions_by_module, + _render_module_subgraphs, + _render_flow_edges, + _render_flow_styles, +) +from .flow_compact import should_skip_module, is_entry_point + + +def export_flow_full( + result: AnalysisResult, output_path: str, include_examples: bool = False +) -> None: + """Export full debug view with all nodes (original flow.mmd). + + This is the original export() behavior with optional filtering. + """ + from .utils import module_of + + lines = ["flowchart TD"] + + # Build name index for O(1) callee resolution + name_index = build_name_index(result.functions) + lines.append("") + lines.append(" %% Styling definitions") + lines.append(" classDef highCC fill:#ff6b6b,stroke:#c92a2a,color:#fff") + lines.append(" classDef medCC fill:#ffd43b,stroke:#f08c00,color:#000") + lines.append(" classDef entry fill:#4dabf7,stroke:#1971c2,color:#fff") + lines.append("") + filtered_funcs = _filtered_functions( + result, + module_of, + should_skip_module, + include_examples, + ) + entry_points = _entry_points(filtered_funcs, result, is_entry_point) + modules = _group_functions_by_module(filtered_funcs, module_of) + _render_module_subgraphs( + lines, + modules, + entry_points, + short_len=35, + readable_id=readable_id, + safe_module=safe_module, + get_cc=get_cc, + sort_funcs=False, + max_funcs=None, + ) + _render_flow_edges( + lines, + filtered_funcs, + readable_id, + resolve_callee, + calls_per_function=15, + limit=None, + name_index=name_index, + ) + _render_flow_styles( + lines, + filtered_funcs, + entry_points, + readable_id, + get_cc, + high_limit=50, + med_limit=50, + entry_limit=15, + ) + write_file(output_path, lines) + + +__all__ = ["export_flow_full"] diff --git a/packages/code2flow/code2flow/flow_helpers.py b/packages/code2flow/code2flow/flow_helpers.py new file mode 100644 index 0000000..14711d3 --- /dev/null +++ b/packages/code2flow/code2flow/flow_helpers.py @@ -0,0 +1,295 @@ +"""Shared helpers for Mermaid flow rendering.""" + +# ── Node / edge helpers ─────────────────────────────────────────────────────── +# _filtered_functions, _entry_points: filter and rank functions for graph nodes. +# _group_functions_by_module: group by module for subgraph layout. +# _classify_architecture_module, _group_architecture_functions: architecture view. +# _select_key_functions: cap node count to keep diagrams readable. +# ── Rendering helpers ───────────────────────────────────────────────────────── +# _append_flow_node: emit a single Mermaid node line. +# _render_module_subgraphs: emit all subgraph blocks. +# _resolve_callee, _add_edge_if_new, _render_flow_edges: edge rendering. +# _append_entry_styles, _render_flow_styles: CSS class application. +# _render_architecture_view: top-level architecture view renderer. + +from collections import defaultdict +from typing import Any, Callable, Dict, List, Optional, Sequence, Set, Tuple + + +def _filtered_functions( + result: Any, + module_of: Callable[[str], str], + should_skip_module: Callable[[str, bool], bool], + include_examples: bool = False, +) -> Dict[str, Any]: + """Return functions after applying the flow diagram skip rules.""" + if include_examples: + return dict(result.functions) + return { + name: fi + for name, fi in result.functions.items() + if not should_skip_module(module_of(name), include_examples) + } + + +def _entry_points( + filtered_funcs: Dict[str, Any], + result: Any, + is_entry_point: Callable[[str, Any, Any], bool], +) -> List[str]: + """Collect entry points in a stable order.""" + return [ + name for name, fi in filtered_funcs.items() if is_entry_point(name, fi, result) + ] + + +def _group_functions_by_module( + funcs: Dict[str, Any], + module_of: Callable[[str], str], +) -> Dict[str, List[Tuple[str, Any]]]: + """Group functions by module for rendering.""" + modules: Dict[str, List[Tuple[str, Any]]] = defaultdict(list) + for func_name, fi in funcs.items(): + modules[module_of(func_name)].append((func_name, fi)) + return modules + + +def _classify_architecture_module(func_name: str, module: str) -> str: + """Assign a function to a compact architecture bucket.""" + if "cli" in module.lower(): + return "CLI" + if "exporter" in module.lower() or "export" in func_name.lower(): + return "Exporters" + return "Core" + + +def _group_architecture_functions( + funcs: Dict[str, Any], + module_of: Callable[[str], str], +) -> Dict[str, List[str]]: + """Group functions into the compact architecture buckets.""" + arch_modules: Dict[str, List[str]] = { + "CLI": [], + "Core": [], + "Exporters": [], + } + for func_name in funcs: + bucket = _classify_architecture_module(func_name, module_of(func_name)) + arch_modules[bucket].append(func_name) + return arch_modules + + +def _select_key_functions( + func_names: List[str], + funcs: Dict[str, Any], + entry_points: Set[str], + critical_path: Set[str], + get_cc: Callable[[Any], int], + threshold: int = 15, +) -> List[str]: + """Choose the compact set of functions to show in the architectural view.""" + key_funcs: List[str] = [] + for fn in func_names: + fi = funcs.get(fn) + if not fi: + continue + cc = get_cc(fi) + if fn in entry_points or fn in critical_path or cc >= threshold: + key_funcs.append(fn) + return key_funcs + + +def _append_flow_node( + lines: List[str], + func_name: str, + fi: Any, + short_len: int, + entry_points: Set[str], + readable_id: Callable[[str], str], + get_cc: Callable[[Any], int], + high_threshold: int = 15, + med_threshold: int = 8, +) -> None: + """Append a single Mermaid node with the appropriate CC styling.""" + sid = readable_id(func_name) + short = fi.name[:short_len] + cc = get_cc(fi) + + if func_name in entry_points: + lines.append(f' {sid}["{short}"]') + elif cc >= high_threshold: + lines.append(f" {sid}{{{{{short} CC={cc}}}}}") + elif cc >= med_threshold: + lines.append(f' {sid}("{short} CC={cc}")') + else: + lines.append(f' {sid}["{short}"]') + + +def _render_module_subgraphs( + lines: List[str], + modules: Dict[str, List[Tuple[str, Any]]], + entry_points: Sequence[str], + short_len: int, + readable_id: Callable[[str], str], + safe_module: Callable[[str], str], + get_cc: Callable[[Any], int], + sort_funcs: bool = False, + max_funcs: Optional[int] = None, + high_threshold: int = 15, + med_threshold: int = 8, +) -> None: + """Render module subgraphs for the detailed and full flow views.""" + entry_set = set(entry_points) + for module, module_funcs in sorted(modules.items()): + lines.append(f" subgraph {safe_module(module)}") + + ordered_funcs = module_funcs + if sort_funcs: + ordered_funcs = sorted( + module_funcs, key=lambda x: (-get_cc(x[1]), x[1].name) + ) + + visible_funcs = ( + ordered_funcs if max_funcs is None else ordered_funcs[:max_funcs] + ) + for func_name, fi in visible_funcs: + _append_flow_node( + lines, + func_name, + fi, + short_len, + entry_set, + readable_id, + get_cc, + high_threshold=high_threshold, + med_threshold=med_threshold, + ) + + if max_funcs is not None and len(ordered_funcs) > max_funcs: + lines.append(f' ...["+{len(ordered_funcs) - max_funcs} more"]') + lines.append(" end") + lines.append("") + + +def _resolve_callee(callee, funcs, resolve, name_index): + """Resolve callee name using index-aware or simple resolver.""" + if name_index: + return resolve(callee, funcs, name_index) + return resolve(callee, funcs) + + +def _add_edge_if_new(src, dst, seen_edges, lines, limit): + """Add a Mermaid edge if not already seen; return True if limit reached.""" + edge = (src, dst) + if edge not in seen_edges: + seen_edges.add(edge) + lines.append(f" {src} --> {dst}") + if limit is not None and len(seen_edges) >= limit: + return True + return False + + +def _render_flow_edges( + lines: List[str], + funcs: Dict[str, Any], + readable_id: Callable[[str], str], + resolve: Callable[..., Optional[str]], + calls_per_function: int = 10, + limit: Optional[int] = 200, + name_index: Optional[Dict[str, List[str]]] = None, +) -> None: + """Render cross-function call edges with optional limits.""" + seen_edges: Set[Tuple[str, str]] = set() + for func_name, fi in funcs.items(): + src = readable_id(func_name) + for callee in fi.calls[:calls_per_function]: + resolved = _resolve_callee(callee, funcs, resolve, name_index) + if resolved and resolved != func_name: + dst = readable_id(resolved) + if _add_edge_if_new(src, dst, seen_edges, lines, limit): + return + if limit is not None and len(seen_edges) >= limit: + return + + +def _append_entry_styles( + lines: List[str], + entry_points: Sequence[str], + readable_id: Callable[[str], str], + entry_limit: int = 15, +) -> None: + """Style the entry points in the Mermaid graph.""" + if entry_points: + entry_ids = [readable_id(ep) for ep in entry_points[:entry_limit]] + lines.append(f" class {','.join(entry_ids)} entry") + + +def _render_flow_styles( + lines: List[str], + funcs: Dict[str, Any], + entry_points: Sequence[str], + readable_id: Callable[[str], str], + get_cc: Callable[[Any], int], + high_threshold: int = 15, + med_threshold: int = 8, + high_limit: int = 30, + med_limit: int = 30, + entry_limit: int = 15, +) -> None: + """Render CC-based styling for the detailed and full flow views.""" + high_nodes: List[str] = [] + med_nodes: List[str] = [] + + for func_name, fi in funcs.items(): + cc = get_cc(fi) + sid = readable_id(func_name) + if cc >= high_threshold: + high_nodes.append(sid) + elif cc >= med_threshold: + med_nodes.append(sid) + + if high_nodes: + lines.append(f" class {','.join(high_nodes[:high_limit])} highCC") + if med_nodes: + lines.append(f" class {','.join(med_nodes[:med_limit])} medCC") + _append_entry_styles(lines, entry_points, readable_id, entry_limit) + + +def _render_architecture_view( + lines: List[str], + filtered_funcs: Dict[str, Any], + entry_points: Sequence[str], + critical_path: Set[str], + module_of: Callable[[str], str], + readable_id: Callable[[str], str], + get_cc: Callable[[Any], int], +) -> None: + """Render the compact architecture view used by export_flow_compact.""" + arch_modules = _group_architecture_functions(filtered_funcs, module_of) + entry_set = set(entry_points) + for arch_name, func_names in arch_modules.items(): + if not func_names: + continue + + lines.append(f" subgraph {arch_name}") + key_funcs = _select_key_functions( + func_names, + filtered_funcs, + entry_set, + critical_path, + get_cc, + threshold=15, + ) + for func_name in key_funcs[:15]: + fi = filtered_funcs.get(func_name) + if fi: + _append_flow_node( + lines, func_name, fi, 30, entry_set, readable_id, get_cc + ) + + if len(key_funcs) > 15: + lines.append(f' ...["+{len(key_funcs) - 15} more"]') + lines.append(" end") + lines.append("") + + _append_entry_styles(lines, entry_points, readable_id, entry_limit=10) diff --git a/packages/code2flow/code2flow/utils.py b/packages/code2flow/code2flow/utils.py new file mode 100644 index 0000000..36a51f9 --- /dev/null +++ b/packages/code2flow/code2flow/utils.py @@ -0,0 +1,101 @@ +"""Mermaid exporter utilities — identifiers, module extraction, file writing.""" + +import re +from collections import defaultdict +from pathlib import Path +from typing import Dict, List, Optional + +from code2graph import FunctionInfo + + +def readable_id(name: str) -> str: + """Create human-readable Mermaid-safe unique node ID.""" + return _sanitize_identifier(name, prefix="N") + + +def safe_module(name: str) -> str: + """Create safe subgraph name.""" + return _sanitize_identifier(name, prefix="M") + + +def _sanitize_identifier(name: str, prefix: str) -> str: + """Convert an arbitrary string into a Mermaid-safe identifier.""" + safe = (name or "").replace(".", "__") + safe = re.sub(r"[^A-Za-z0-9_]+", "_", safe) + if not safe: + return prefix + if not safe[0].isalpha(): + safe = f"{prefix}_{safe}" + return safe + + +def module_of(func_name: str) -> str: + """Extract module from qualified name. + + Returns up to 2 levels (e.g. 'code2llm.core', 'code2llm.exporters') + so that subpackage-level cross-edges are visible in compact_flow. + """ + parts = func_name.split(".") + if len(parts) >= 3: + return ".".join(parts[:2]) + if len(parts) == 2: + return parts[0] + return parts[0] if parts else "unknown" + + +def build_name_index(funcs: dict) -> Dict[str, List[str]]: + """Build index mapping simple names to qualified names for O(1) lookup.""" + index: Dict[str, List[str]] = defaultdict(list) + for qn in funcs: + simple_name = qn.split(".")[-1] + index[simple_name].append(qn) + return index + + +def resolve_callee( + callee: str, funcs: dict, name_index: Optional[Dict[str, List[str]]] = None +) -> Optional[str]: + """Resolve callee to a known qualified name. + + Args: + callee: The callee name to resolve + funcs: Dictionary of all functions + name_index: Optional pre-built index from build_name_index() for O(1) lookup + """ + if callee in funcs: + return callee + if name_index is not None: + candidates = name_index.get(callee, []) + else: + # Fallback to slow path if index not provided + candidates = [qn for qn in funcs if qn.endswith(f".{callee}")] + if len(candidates) == 1: + return candidates[0] + return None + + +def write_file(path: str, lines: list) -> Path: + """Write lines to file.""" + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + with open(p, "w", encoding="utf-8") as f: + f.write("\n".join(lines) + "\n") + return p + + +def get_cc(fi: FunctionInfo) -> int: + """Extract cyclomatic complexity from FunctionInfo.""" + if isinstance(fi.complexity, dict): + return fi.complexity.get("cyclomatic_complexity", 0) + return fi.complexity or 0 + + +__all__ = [ + "readable_id", + "safe_module", + "module_of", + "build_name_index", + "resolve_callee", + "write_file", + "get_cc", +] diff --git a/packages/code2flow/code2flow/validation.py b/packages/code2flow/code2flow/validation.py new file mode 100644 index 0000000..0e9f74b --- /dev/null +++ b/packages/code2flow/code2flow/validation.py @@ -0,0 +1,138 @@ +"""Mermaid file validation — check for syntax errors and structural issues.""" + +from pathlib import Path +from typing import List, Union + + +def validate_mermaid_file(mmd_path: Union[str, Path]) -> List[str]: + """Validate Mermaid file and return list of errors.""" + mmd_path = Path(mmd_path) + if not mmd_path.exists(): + return [f"File not found: {mmd_path}"] + + try: + content = mmd_path.read_text(encoding="utf-8") + lines = content.strip().split("\n") + errors = [] + + # Check for proper graph declaration + if not lines or not any( + line.strip().startswith(("graph", "flowchart")) for line in lines + ): + errors.append( + "Missing graph declaration (should start with 'graph' or 'flowchart')" + ) + + _check_bracket_balance(lines, errors) + _check_node_ids(lines, errors) + + return errors + + except Exception as e: + return [f"Error reading file: {e}"] + + +def _strip_label_segments(s: str) -> str: + """Remove label segments that frequently contain Mermaid syntax chars.""" + import re + + s = re.sub(r"\|[^|]*\|", "||", s) + s = re.sub(r"\[\"[^\"]*\"\]", "[]", s) + s = re.sub(r"\(\"[^\"]*\"\)", "()", s) + s = re.sub(r"\{\"[^\"]*\"\}", "{}", s) + s = re.sub(r"\[/[^\]]*?/\]", "[]", s) + s = re.sub(r"\(/[^)]*?/\)", "()", s) + return s + + +def _is_balanced_node_line(line: str) -> bool: + """Check if a line has balanced brackets — likely a node definition.""" + return ( + ("[" in line and "]" in line) + or ("(" in line and ")" in line) + or ("{" in line and "}" in line) + ) + + +def _check_bracket_balance(lines: List[str], errors: List[str]) -> None: + """Check for unmatched brackets/parentheses outside label segments.""" + bracket_stack = [] + paren_stack = [] + + for line_num, line in enumerate(lines, 1): + line = line.strip() + if not line or line.startswith("%%"): + continue + if _is_balanced_node_line(line): + continue + check_line = _strip_label_segments(line) + _scan_brackets(check_line, line_num, bracket_stack, paren_stack, errors) + + for expected, line_num in bracket_stack: + errors.append(f"Line {line_num}: Unclosed '[' (missing '{expected}')") + for expected, line_num in paren_stack: + errors.append(f"Line {line_num}: Unclosed '(' (missing '{expected}')") + + +def _scan_brackets( + text: str, line_num: int, bracket_stack: list, paren_stack: list, errors: List[str] +) -> None: + """Process bracket/paren chars in a single line.""" + for char in text: + if char == "[": + bracket_stack.append(("]", line_num)) + elif char == "]": + if not bracket_stack or bracket_stack[-1][0] != "]": + errors.append(f"Line {line_num}: Unmatched ']'") + else: + bracket_stack.pop() + elif char == "(": + paren_stack.append((")", line_num)) + elif char == ")": + if not paren_stack or paren_stack[-1][0] != ")": + errors.append(f"Line {line_num}: Unmatched ')'") + else: + paren_stack.pop() + + +def _check_single_node_line( + line: str, line_num: int, node_pattern, errors: List[str] +) -> None: + """Validate a single mermaid node line and append error if invalid.""" + import re + if not any(char in line for char in ["[", "(", "{"]): + return + if node_pattern.match(line): + return + match = re.match(r"^\s*([A-Za-z0-9_]+)", line) + if match: + node_id = match.group(1) + if not re.match(r"^[A-Z]\d+$|^[Ff]\d+_\w+$", node_id): + errors.append( + f"Line {line_num}: Invalid node ID '{node_id}' (should be like 'N1' or 'F123_name')" + ) + + +def _check_node_ids(lines: List[str], errors: List[str]) -> None: + """Check for invalid node IDs.""" + import re + node_pattern = re.compile(r'^\s*([A-Z]\d+|[Ff]\d+_\w+)\s*["\'\[\{]') + for line_num, line in enumerate(lines, 1): + line = line.strip() + if not line or line.startswith("%%"): + continue + if line.startswith("subgraph ") or line == "end": + continue + if _is_balanced_node_line(line): + continue + _check_single_node_line(line, line_num, node_pattern, errors) + + +__all__ = [ + "validate_mermaid_file", + "_strip_label_segments", + "_is_balanced_node_line", + "_check_bracket_balance", + "_scan_brackets", + "_check_node_ids", +] diff --git a/packages/code2flow/pyproject.toml b/packages/code2flow/pyproject.toml new file mode 100644 index 0000000..465ff75 --- /dev/null +++ b/packages/code2flow/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "code2flow" +version = "0.1.0" +description = "Mermaid flow diagrams, call graph visualization, and architecture diagrams for code analysis" +readme = "README.md" +requires-python = ">=3.9" +license = {text = "Apache-2.0"} +authors = [ + {name = "Tom Sapletta", email = "tom@sapletta.com"} +] +keywords = ["mermaid", "flowchart", "call-graph", "diagram", "code-visualization"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Topic :: Software Development :: Libraries :: Python Modules", +] +dependencies = [ + "code2graph>=0.1.0", +] + +[tool.setuptools.packages.find] +where = ["."] +include = ["code2flow*"] diff --git a/packages/code2flow/tests/test_code2flow.py b/packages/code2flow/tests/test_code2flow.py new file mode 100644 index 0000000..ea6bed8 --- /dev/null +++ b/packages/code2flow/tests/test_code2flow.py @@ -0,0 +1,47 @@ +"""Unit tests for code2flow exporters and validation.""" +from code2graph import AnalysisResult, FunctionInfo +from code2flow import ( + export_flow_compact, + export_calls, + validate_mermaid_file, + fix_mermaid_file, +) + + +def test_export_flow_compact(tmp_path): + result = AnalysisResult() + result.functions = { + "app.main": FunctionInfo(name="main", qualified_name="app.main", file="app.py", line=1, calls=["app.worker"]), + "app.worker": FunctionInfo(name="worker", qualified_name="app.worker", file="app.py", line=10, called_by=["app.main"]), + } + out_file = tmp_path / "compact_flow.mmd" + export_flow_compact(result, str(out_file)) + + assert out_file.exists() + content = out_file.read_text() + assert "graph TD" in content or "flowchart" in content or "subgraph" in content + + +def test_export_calls(tmp_path): + result = AnalysisResult() + result.functions = { + "app.main": FunctionInfo(name="main", qualified_name="app.main", file="app.py", line=1, calls=["app.worker"]), + "app.worker": FunctionInfo(name="worker", qualified_name="app.worker", file="app.py", line=10, called_by=["app.main"]), + } + out_file = tmp_path / "calls.mmd" + export_calls(result, str(out_file)) + + assert out_file.exists() + content = out_file.read_text() + assert "flowchart" in content or "graph" in content + + +def test_validate_and_fix_mermaid(tmp_path): + mmd_file = tmp_path / "test.mmd" + mmd_file.write_text("graph TD\n A -->|bad_label(| B\n") + errors = validate_mermaid_file(str(mmd_file)) + assert isinstance(errors, list) + + fixed = fix_mermaid_file(str(mmd_file)) + assert fixed is True + assert "bad_label" in mmd_file.read_text() diff --git a/pyproject.toml b/pyproject.toml index a892b00..3607011 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,6 +74,9 @@ graph = [ toon = [ "code2toon>=0.1.0", ] +diagrams = [ + "code2flow>=0.1.0", +] dev = [ "pytest>=6.2", "pytest-cov>=2.12", @@ -85,6 +88,7 @@ dev = [ "code2llm-rust>=0.1.0", "code2graph>=0.1.0", "code2toon>=0.1.0", + "code2flow>=0.1.0", ] [project.scripts]