Skip to content

gds_viz

Public API — all visualization functions.

Visualization utilities for GDS specifications.

spec_to_mermaid(spec, *, group_by=None, show_entities=True, show_wires=True, theme=None)

Generate a Mermaid flowchart from a GDSSpec.

Renders an architecture-level view with blocks grouped by role or tag, entity cylinders, and dependency wires.

Parameters:

Name Type Description Default
spec GDSSpec

The GDS specification to visualize.

required
group_by str | None

Tag key to group blocks by. None groups by GDS role.

None
show_entities bool

If True, render entity cylinders with state variables.

True
show_wires bool

If True, render dependency edges from wirings.

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/architecture.py
def spec_to_mermaid(
    spec: GDSSpec,
    *,
    group_by: str | None = None,
    show_entities: bool = True,
    show_wires: bool = True,
    theme: MermaidTheme | None = None,
) -> str:
    """Generate a Mermaid flowchart from a GDSSpec.

    Renders an architecture-level view with blocks grouped by role or tag,
    entity cylinders, and dependency wires.

    Args:
        spec: The GDS specification to visualize.
        group_by: Tag key to group blocks by. None groups by GDS role.
        show_entities: If True, render entity cylinders with state variables.
        show_wires: If True, render dependency edges from wirings.
        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 TD"]
    query = SpecQuery(spec)

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

    # Render grouped blocks
    if group_by is not None:
        sg_styles = _render_tag_groups(lines, spec, group_by)
    else:
        sg_styles = _render_role_groups(lines, query, spec)

    # Entity cylinders
    if show_entities:
        _render_entities(lines, spec, query)

    # Dependency wires
    if show_wires:
        _render_wires(lines, spec, query)

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

    return "\n".join(lines)

canonical_to_mermaid(canonical, *, show_updates=True, show_parameters=True, theme=None)

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)

block_to_mermaid(block, *, theme=None)

Generate a Mermaid flowchart from a Block composition tree.

This is a convenience wrapper that flattens the block and creates a minimal diagram showing the composition structure.

Parameters:

Name Type Description Default
block Block

The root block (atomic or composite).

required
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.

Example
from gds.blocks.roles import BoundaryAction, Policy, Mechanism
from gds.types.interface import Interface, port
from gds_viz import block_to_mermaid

observe = BoundaryAction(
    name="Observe",
    interface=Interface(forward_out=(port("Signal"),))
)
decide = Policy(
    name="Decide",
    interface=Interface(
        forward_in=(port("Signal"),),
        forward_out=(port("Action"),)
    )
)
update = Mechanism(
    name="Update",
    interface=Interface(forward_in=(port("Action"),)),
    updates=[("Entity", "state")]
)

pipeline = observe >> decide >> update
print(block_to_mermaid(pipeline))
Source code in packages/gds-viz/gds_viz/mermaid.py
def block_to_mermaid(block: Block, *, theme: MermaidTheme | None = None) -> str:
    """Generate a Mermaid flowchart from a Block composition tree.

    This is a convenience wrapper that flattens the block and creates
    a minimal diagram showing the composition structure.

    Args:
        block: The root block (atomic or composite).
        theme: Mermaid theme — one of 'default', 'neutral', 'dark', 'forest',
               'base'. None uses the default ('neutral').

    Returns:
        Mermaid flowchart diagram as a string.

    Example:
        ```python
        from gds.blocks.roles import BoundaryAction, Policy, Mechanism
        from gds.types.interface import Interface, port
        from gds_viz import block_to_mermaid

        observe = BoundaryAction(
            name="Observe",
            interface=Interface(forward_out=(port("Signal"),))
        )
        decide = Policy(
            name="Decide",
            interface=Interface(
                forward_in=(port("Signal"),),
                forward_out=(port("Action"),)
            )
        )
        update = Mechanism(
            name="Update",
            interface=Interface(forward_in=(port("Action"),)),
            updates=[("Entity", "state")]
        )

        pipeline = observe >> decide >> update
        print(block_to_mermaid(pipeline))
        ```
    """
    from gds.compiler.compile import compile_system

    # Compile with default settings
    system = compile_system(name=block.name, root=block)
    return system_to_mermaid(system, show_hierarchy=False, theme=theme)

system_to_mermaid(system, show_hierarchy=False, *, theme=None)

Generate a Mermaid flowchart from a SystemIR.

Parameters:

Name Type Description Default
system SystemIR

The compiled system to visualize.

required
show_hierarchy bool

If True, uses the hierarchy tree to organize subgraphs. If False, renders a flat graph of all blocks.

False
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.

Example
from examples.sir_epidemic.model import build_system
from gds_viz import system_to_mermaid

system = build_system()
mermaid = system_to_mermaid(system)
print(mermaid)
Source code in packages/gds-viz/gds_viz/mermaid.py
def system_to_mermaid(
    system: SystemIR,
    show_hierarchy: bool = False,
    *,
    theme: MermaidTheme | None = None,
) -> str:
    """Generate a Mermaid flowchart from a SystemIR.

    Args:
        system: The compiled system to visualize.
        show_hierarchy: If True, uses the hierarchy tree to organize subgraphs.
                       If False, renders a flat graph of all blocks.
        theme: Mermaid theme — one of 'default', 'neutral', 'dark', 'forest',
               'base'. None uses the default ('neutral').

    Returns:
        Mermaid flowchart diagram as a string.

    Example:
        ```python
        from examples.sir_epidemic.model import build_system
        from gds_viz import system_to_mermaid

        system = build_system()
        mermaid = system_to_mermaid(system)
        print(mermaid)
        ```
    """
    lines = [theme_directive(theme), "flowchart TD"]

    # Class definitions for role-based styling
    lines.extend(classdefs_for_roles(theme))

    if show_hierarchy and system.hierarchy:
        lines.append(_hierarchy_to_mermaid(system.hierarchy, indent=1))
    else:
        # Flat block diagram with role-based classes
        block_shapes = _get_block_shapes(system)
        block_roles = _get_block_roles(system)
        for block in system.blocks:
            shape_open, shape_close = block_shapes.get(block.name, ("[", "]"))
            safe_name = sanitize_id(block.name)
            role = block_roles.get(block.name, "generic")
            lines.append(
                f"    {safe_name}{shape_open}{block.name}{shape_close}:::{role}"
            )

    # Add wirings
    for wiring in system.wirings:
        src = sanitize_id(wiring.source)
        tgt = sanitize_id(wiring.target)
        label = wiring.label

        if wiring.is_temporal:
            # Temporal loop: dashed line with arrow back
            lines.append(f"    {src} -.{label}..-> {tgt}")
        elif wiring.is_feedback:
            # Feedback: thick arrow
            lines.append(f"    {src} =={label}==> {tgt}")
        elif wiring.direction == FlowDirection.CONTRAVARIANT:
            # Contravariant: backward arrow
            lines.append(f"    {tgt} <--{label}--- {src}")
        else:
            # Covariant forward: normal arrow
            lines.append(f"    {src} --{label}--> {tgt}")

    return "\n".join(lines)

params_to_mermaid(spec, *, theme=None)

Generate a parameter influence diagram from a GDSSpec.

Shows Θ parameters → blocks that use them → entities they update. Only includes blocks that reference at least one parameter, and entities reachable from those blocks via the update map.

Parameters:

Name Type Description Default
spec GDSSpec

The GDS specification.

required
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/traceability.py
def params_to_mermaid(spec: GDSSpec, *, theme: MermaidTheme | None = None) -> str:
    """Generate a parameter influence diagram from a GDSSpec.

    Shows Θ parameters → blocks that use them → entities they update.
    Only includes blocks that reference at least one parameter, and
    entities reachable from those blocks via the update map.

    Args:
        spec: The GDS specification.
        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"]
    query = SpecQuery(spec)

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

    param_to_blocks = query.param_to_blocks()
    entity_update_map = query.entity_update_map()

    # Collect which params and blocks are actually connected
    active_params = {p for p, blocks in param_to_blocks.items() if blocks}
    if not active_params:
        lines.append("    no_params[No parameters defined]:::empty")
        return "\n".join(lines)

    # Parameter nodes (hexagons)
    for pname in sorted(active_params):
        pid = _param_id(pname)
        lines.append(f'    {pid}{{{{"{pname}"}}}}:::param')

    # Block nodes — only those referenced by parameters
    param_blocks: set[str] = set()
    for blocks in param_to_blocks.values():
        param_blocks.update(blocks)

    for bname in sorted(param_blocks):
        bid = sanitize_id(bname)
        lines.append(f"    {bid}[{bname}]")

    # Entity nodes — only those updated by param-connected blocks
    # Build reverse map: mechanism -> [(entity, var)]
    mech_to_updates: dict[str, list[tuple[str, str]]] = {}
    for ename, var_map in entity_update_map.items():
        for vname, mechs in var_map.items():
            for mname in mechs:
                mech_to_updates.setdefault(mname, []).append((ename, vname))

    active_entities: set[str] = set()
    for bname in param_blocks:
        if bname in mech_to_updates:
            for ename, _ in mech_to_updates[bname]:
                active_entities.add(ename)

    # Also include entities reachable via dependency chain from param blocks
    dep_graph = query.dependency_graph()
    visited: set[str] = set()
    frontier = list(param_blocks)
    while frontier:
        current = frontier.pop()
        if current in visited:
            continue
        visited.add(current)
        if current in mech_to_updates:
            for ename, _ in mech_to_updates[current]:
                active_entities.add(ename)
        for target in dep_graph.get(current, set()):
            frontier.append(target)

    for ename in sorted(active_entities):
        entity = spec.entities[ename]
        var_parts = []
        for vname, var in entity.variables.items():
            var_parts.append(var.symbol if var.symbol else vname)
        var_str = ", ".join(var_parts)
        eid = _entity_id(ename)
        lines.append(f'    {eid}[("{ename}<br/>{var_str}")]:::entity')

    # Edges: param -> block
    for pname in sorted(active_params):
        pid = _param_id(pname)
        for bname in param_to_blocks[pname]:
            bid = sanitize_id(bname)
            lines.append(f"    {pid} -.-> {bid}")

    # Edges: block -> entity (for blocks in the param-reachable set)
    seen_edges: set[tuple[str, str]] = set()
    for bname in visited:
        if bname in mech_to_updates:
            bid = sanitize_id(bname)
            for ename, _vname in mech_to_updates[bname]:
                eid = _entity_id(ename)
                if (bid, eid) not in seen_edges:
                    seen_edges.add((bid, eid))
                    lines.append(f"    {bid} -.-> {eid}")

    # Edges: block -> block (dependency flow within param-reachable set)
    for source in sorted(visited):
        for target in sorted(dep_graph.get(source, set())):
            if target in visited:
                sid = sanitize_id(source)
                tid = sanitize_id(target)
                lines.append(f"    {sid} --> {tid}")

    return "\n".join(lines)

trace_to_mermaid(spec, entity, variable, *, theme=None)

Generate a traceability diagram for a single entity variable.

Shows every block that can transitively affect the variable, the parameters feeding those blocks, and the causal chain.

Parameters:

Name Type Description Default
spec GDSSpec

The GDS specification.

required
entity str

Entity name (e.g. "Susceptible").

required
variable str

Variable name (e.g. "count").

required
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/traceability.py
def trace_to_mermaid(
    spec: GDSSpec,
    entity: str,
    variable: str,
    *,
    theme: MermaidTheme | None = None,
) -> str:
    """Generate a traceability diagram for a single entity variable.

    Shows every block that can transitively affect the variable,
    the parameters feeding those blocks, and the causal chain.

    Args:
        spec: The GDS specification.
        entity: Entity name (e.g. "Susceptible").
        variable: Variable name (e.g. "count").
        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 RL"]
    query = SpecQuery(spec)

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

    affecting = query.blocks_affecting(entity, variable)
    if not affecting:
        lines.append(f"    target[{entity}.{variable}]:::target")
        lines.append("    none[No affecting blocks]:::empty")
        return "\n".join(lines)

    # Target node
    ent = spec.entities[entity]
    var = ent.variables[variable]
    symbol = var.symbol if var.symbol else variable
    lines.append(f'    target(["{entity}.{variable} ({symbol})"]):::target')

    # Block nodes
    for bname in affecting:
        bid = sanitize_id(bname)
        lines.append(f"    {bid}[{bname}]")

    # Parameter nodes for affecting blocks
    block_to_params = query.block_to_params()
    active_params: set[str] = set()
    for bname in affecting:
        for pname in block_to_params.get(bname, []):
            active_params.add(pname)

    for pname in sorted(active_params):
        pid = _param_id(pname)
        lines.append(f'    {pid}{{{{"{pname}"}}}}:::param')

    # Edges: mechanism -> target
    entity_update_map = query.entity_update_map()
    direct_mechs = entity_update_map.get(entity, {}).get(variable, [])
    for mname in direct_mechs:
        mid = sanitize_id(mname)
        lines.append(f"    {mid} ==> target")

    # Edges: block -> block (dependency within affecting set)
    dep_graph = query.dependency_graph()
    for source in affecting:
        sid = sanitize_id(source)
        for target in dep_graph.get(source, set()):
            if target in affecting:
                tid = sanitize_id(target)
                lines.append(f"    {sid} --> {tid}")

    # Edges: param -> block
    for bname in affecting:
        bid = sanitize_id(bname)
        for pname in block_to_params.get(bname, []):
            pid = _param_id(pname)
            lines.append(f"    {pid} -.-> {bid}")

    return "\n".join(lines)