gds_viz.canonical

Generate a Mermaid flowchart from a CanonicalGDS projection.

Renders the formal GDS decomposition: X_t -> U -> g -> f -> X_{t+1} with optional parameter space (Theta) and update map labels.

Parameters:

Name	Type	Description	Default
canonical	CanonicalGDS	The canonical GDS projection to visualize.	required
show_updates	bool	If True, label mechanism->X edges with entity.variable.	True
show_parameters	bool	If True, show the Theta node when parameters exist.	True
theme	MermaidTheme | None	Mermaid theme — one of 'default', 'neutral', 'dark', 'forest', 'base'. None uses the default ('neutral').	None

Returns:

Type	Description
str	Mermaid flowchart diagram as a string.

Source code in packages/gds-viz/gds_viz/canonical.py

def canonical_to_mermaid(
    canonical: CanonicalGDS,
    *,
    show_updates: bool = True,
    show_parameters: bool = True,
    theme: MermaidTheme | None = None,
) -> str:
    """Generate a Mermaid flowchart from a CanonicalGDS projection.

    Renders the formal GDS decomposition: X_t -> U -> g -> f -> X_{t+1}
    with optional parameter space (Theta) and update map labels.

    Args:
        canonical: The canonical GDS projection to visualize.
        show_updates: If True, label mechanism->X edges with entity.variable.
        show_parameters: If True, show the Theta node when parameters exist.
        theme: Mermaid theme — one of 'default', 'neutral', 'dark', 'forest',
               'base'. None uses the default ('neutral').

    Returns:
        Mermaid flowchart diagram as a string.
    """
    lines = [theme_directive(theme), "flowchart LR"]

    # Class definitions
    lines.extend(classdefs_for_all(theme))

    # State variable listing for X_t / X_{t+1}
    # Use entity.var format to disambiguate variables with the same name
    var_names = [v for _, v in canonical.state_variables]
    has_dupes = len(var_names) != len(set(var_names))
    if has_dupes:
        var_list = ", ".join(f"{e}.{v}" for e, v in canonical.state_variables)
    else:
        var_list = ", ".join(var_names)
    if var_list:
        x_label = f"X_t<br/>{var_list}"
        x_next_label = f"X_{{t+1}}<br/>{var_list}"
    else:
        x_label = "X_t"
        x_next_label = "X_{t+1}"

    lines.append(f'    X_t(["{x_label}"]):::state')
    lines.append(f'    X_next(["{x_next_label}"]):::state')

    # Parameter node (Theta)
    if show_parameters and canonical.has_parameters:
        param_names = ", ".join(canonical.parameter_schema.names())
        lines.append(f'    Theta{{{{"\u0398<br/>{param_names}"}}}}:::param')

    # Role subgraphs — only render non-empty ones
    rendered_sgs: dict[str, str] = {}
    for sg_id, label, blocks, role in [
        ("U", "Boundary (U)", canonical.boundary_blocks, "boundary"),
        ("g", "Policy (g)", canonical.policy_blocks, "policy"),
        ("f", "Mechanism (f)", canonical.mechanism_blocks, "mechanism"),
        ("ctrl", "Control", canonical.control_blocks, "control"),
    ]:
        if blocks:
            _render_subgraph(lines, sg_id, label, blocks, role)
            rendered_sgs[sg_id] = role

    # Edges between layers
    _render_flow_edges(lines, canonical)

    # Update edges: mechanism -> X_{t+1}
    _render_update_edges(lines, canonical, show_updates)

    # Control feedback edges
    if canonical.control_blocks:
        for cname in canonical.control_blocks:
            cid = sanitize_id(cname)
            # f -> ctrl (dashed)
            lines.append(f"    f -.-> {cid}")
            # ctrl -> g (dashed)
            lines.append(f"    {cid} -.-> g")

    # Parameter edges
    if show_parameters and canonical.has_parameters:
        if canonical.policy_blocks:
            lines.append("    Theta -.-> g")
        if canonical.mechanism_blocks:
            lines.append("    Theta -.-> f")

    # Subgraph background styling
    lines.extend(subgraph_style_lines(rendered_sgs, theme))

    return "\n".join(lines)