gds_viz.traceability

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)