quivers.analysis

Static analysis and inspection utilities derive chain shapes, initialization recipes, saturation diagnostics, scope paths, and plate graphs without rewriting a compiled model. The package-level exports cover the training diagnostics; the plate and scope modules support the REPL and can also be used by downstream tooling.

analysis

Algebra-guided training tooling.

This package inspects a compiled QVR program and returns chain metadata, initialization recipes, and source-keyed diagnostics without rewriting the program. It accepts either a quivers.dsl.ast_nodes.Module AST or its runtime quivers.continuous.programs.MonadicProgram.

The pieces:

• ChainShape (quivers.analysis.chain_shape): walks a program's let / latent steps and tags each one with its source location, governing algebra, and intermediate dimensionality. This metadata supplies the initialization and saturation analyses.
• recommend_init (quivers.analysis.init_spec): given a program, produces a per-latent InitSpec from each algebra's initialization recipe. Pair with apply_init_spec to materialize the initial values onto the program's learnable parameters.
• saturation_warnings (quivers.analysis.saturation): given a program, returns source-keyed warnings about latents that, under the recommended init, would saturate the surrounding algebra's value range.

See Analysis Pipelines: Fitting and Diagnostics for the user-facing workflow.

StepKind module-attribute

StepKind = Literal['latent', 'observe', 'marginalize', 'let']

The kind tag of a StepShape: a stochastic bind (latent, observe, marginalize) or a deterministic let.

NodeKind module-attribute

NodeKind = Literal['latent', 'observed', 'marginalized', 'deterministic']

The kind tag of a PlateNode, which selects how a renderer draws it.

SCOPE_SEPARATOR module-attribute

SCOPE_SEPARATOR = '::'

The separator between the segments of a scoped path such as "prog::x".

TOP_LEVEL_KINDS module-attribute

TOP_LEVEL_KINDS: tuple[ScopeKind, ...] = ('object', 'space', 'morphism', 'rule', 'program', 'deduction', 'signature', 'encoder', 'decoder', 'loss', 'bundle', 'contraction', 'composition', 'category')

The top-level container kinds, those reachable directly from the compiler's public accessors, in the order ReplSession.browse() walks them for output.

ScopeKind module-attribute

ScopeKind = Literal['object', 'space', 'morphism', 'rule', 'program', 'deduction', 'signature', 'encoder', 'decoder', 'loss', 'bundle', 'contraction', 'composition', 'category', 'param', 'sample-site', 'observe-site', 'let-site', 'marginalize-site', 'score-site', 'return-site', 'atom', 'deduction-rule', 'lexicon-entry', 'sort', 'constructor', 'binder', 'vertex-kind', 'edge-kind', 'op-rule', 'init-rule', 'message-rule', 'update-rule', 'var-init', 'decoder-head', 'composition-entry', 'bundle-member']

The kind tag carried by every ScopedRef, used for pretty-printing, click-target classification, and the :browse section grouping.

ChainShape didactic-model

ChainShape(*, algebra_name: str = '', object_cardinalities: Mapping[str, int] = field(default_factory=dict, opaque=True), steps: tuple[StepShape, ...] = ())

Bases: Model

Sequence of StepShape records for a program.

ATTRIBUTE	DESCRIPTION
algebra_name	Name of the program's governing algebra (the top-level algebra <name> declaration). TYPE: str
object_cardinalities	Cardinality of every object X : N declaration preceding the program. Used to resolve plate sizes referenced by : T annotations on latents / observes. TYPE: Mapping[str, int]
steps	Per-step metadata in source order. TYPE: tuple[StepShape, ...]

stochastic_depth property

stochastic_depth: int

Number of stochastic-bind steps (latent, observe, or marginalize).

algebra property

algebra: Algebra | None

Resolve algebra_name against the algebra registry. Returns None if the name is unknown (e.g. a user-defined inline composition rule not registered).

latents

latents() -> tuple[StepShape, ...]

All latent (sample-mode) steps.

Source code in src/quivers/analysis/chain_shape.py

def latents(self) -> tuple[StepShape, ...]:
    """All ``latent`` (sample-mode) steps."""
    return tuple(s for s in self.steps if s.kind == "latent")

observes

observes() -> tuple[StepShape, ...]

All observe (score-mode) steps.

Source code in src/quivers/analysis/chain_shape.py

def observes(self) -> tuple[StepShape, ...]:
    """All ``observe`` (score-mode) steps."""
    return tuple(s for s in self.steps if s.kind == "observe")

from_module classmethod

from_module(module: Module) -> 'ChainShape'

Build a ChainShape from a compiled Module AST.

Walks the module's statement list, captures the algebra name from any top-level CompositionDecl, captures every ObjectDecl's numeric cardinality, then walks the unique ProgramDecl's steps in source order. MarginalizeStep bodies are walked recursively; their inner steps are recorded after the enclosing MarginalizeStep.

Source code in src/quivers/analysis/chain_shape.py

@classmethod
def from_module(cls, module: Module) -> "ChainShape":
    """Build a `ChainShape` from a compiled
    `Module` AST.

    Walks the module's statement list, captures the algebra
    name from any top-level `CompositionDecl`, captures
    every `ObjectDecl`'s numeric cardinality, then walks
    the unique `ProgramDecl`'s steps in source order.
    `MarginalizeStep` bodies are walked recursively;
    their inner steps are recorded after the enclosing
    `MarginalizeStep`.
    """
    algebra_name = "product_fuzzy"
    cardinalities: dict[str, int] = {}
    program: ProgramDecl | None = None
    for stmt in module.statements:
        if isinstance(stmt, CompositionDecl):
            algebra_name = stmt.name
        elif isinstance(stmt, ObjectDecl):
            cardinality = _type_decl_cardinality(stmt)
            if cardinality is not None:
                for decl_name in stmt.names:
                    cardinalities[decl_name] = cardinality
        elif isinstance(stmt, ProgramDecl) and program is None:
            program = stmt

    steps: list[StepShape] = []
    depth = 0

    def record(
        name: str,
        kind: StepKind,
        line: int,
        col: int,
        intermediate: int | None,
    ) -> None:
        nonlocal depth
        depth += 1
        steps.append(
            StepShape(
                name=name,
                kind=kind,
                source_line=line,
                source_col=col,
                depth=depth,
                algebra_name=algebra_name,
                intermediate_size=intermediate,
            )
        )

    def walk(program_steps: tuple[ProgramStep, ...]) -> None:
        for step in program_steps:
            if isinstance(step, SampleStep):
                record(
                    step.vars[0] if step.vars else "",
                    "latent",
                    step.line,
                    step.col,
                    _index_size(step.index, cardinalities),
                )
            elif isinstance(step, ObserveStep):
                record(
                    step.vars[0],
                    "observe",
                    step.line,
                    step.col,
                    _index_size(step.index, cardinalities),
                )
            elif isinstance(step, MarginalizeStep):
                record(
                    step.var,
                    "marginalize",
                    step.line,
                    step.col,
                    _index_size(step.index, cardinalities),
                )
                walk(step.scope)
            elif isinstance(step, LetStep):
                steps.append(
                    StepShape(
                        name=step.name,
                        kind="let",
                        source_line=step.line,
                        source_col=step.col,
                        depth=depth,
                        algebra_name=algebra_name,
                        intermediate_size=None,
                    )
                )

    if program is not None:
        walk(program.draws)

    return cls(
        algebra_name=algebra_name,
        object_cardinalities=dict(cardinalities),
        steps=tuple(steps),
    )

StepShape didactic-model

StepShape(*, name: str, kind: StepKind, source_line: int = 0, source_col: int = 0, depth: int = 0, algebra_name: str = '', intermediate_size: int | None = None)

Bases: Model

Per-step metadata derived by ChainShape.

ATTRIBUTE	DESCRIPTION
name	Bound variable name (the LHS of the step). TYPE: str
kind	One of "latent" / "observe" / "marginalize" (stochastic-bind steps) or "let" (deterministic). TYPE: str
source_line, source_col	Source position where the step was declared. TYPE: int
depth	The 1-indexed position of the step in the chain of stochastic binds (latent / observe / marginalize). Let steps inherit the depth of the most recent stochastic predecessor; a let before any stochastic step has depth 0. TYPE: int
algebra_name	Name of the program's governing algebra. TYPE: str
intermediate_size	Cardinality of the bound coordinate's intermediate axis, when statically inferable from the AST. None if not. TYPE: int | None

InitSpec didactic-model

InitSpec(*, distribution: Literal['normal', 'uniform', 'constant'] = 'normal', mean: float = 0.0, std: float = 1.0, lower: float = float('-inf'), upper: float = float('inf'), rationale: str = '')

Bases: Model

Parameters of a one-dimensional initial distribution.

ATTRIBUTE	DESCRIPTION
distribution	One of "normal" / "uniform" / "constant". TYPE: str
mean	Location parameter for "normal" / "constant"; the midpoint (lower + upper) / 2 for "uniform". TYPE: float
std	Scale parameter for "normal"; the half-width (upper - lower) / 2 for "uniform"; 0.0 for "constant". TYPE: float
lower, upper	Bounds for "uniform"; for "normal" / "constant" they record the algebra's value-space bounds if any (-inf / inf otherwise). TYPE: float
rationale	One-line explanation of the recipe, suitable for surfacing in a warning or in qvr check output. TYPE: str

Edge didactic-model

Edge(*, src: str, dst: str)

Bases: Model

One directed dependency edge in the plate diagram.

Plate didactic-model

Plate(*, name: str, cardinality: int | None = None, parent: str | None = None)

Bases: Model

One plate (indexing axis) of the diagram.

ATTRIBUTE	DESCRIPTION
name	The plate object's name (Doc, Topic, Word). Multiple plates can share a name when the same axis is indexed at multiple levels in the program; the extractor de-duplicates them. TYPE: str
cardinality	The plate's size if statically inferable from the module's object NAME : FinSet N declarations; None otherwise. TYPE: int | None
parent	Name of an enclosing plate, or None when this plate is at the program's outermost level. Set when a marginalize block nests inside another marginalize. TYPE: str | None

PlateGraph didactic-model

PlateGraph(*, program_name: str, domain: str = '', codomain: str = '', nodes: tuple[PlateNode, ...] = (), plates: tuple[Plate, ...] = (), edges: tuple[Edge, ...] = ())

Bases: Model

Structural plate diagram of a single QVR program.

PlateNode didactic-model

PlateNode(*, name: str, kind: NodeKind, family: str | None = None, plates: tuple[str, ...] = (), args: tuple[str, ...] = (), scope_path: str = '')

Bases: Model

One random variable / deterministic intermediate.

ATTRIBUTE	DESCRIPTION
name	The variable's bound name in the program (theta, z, w, mu). TYPE: str
kind	"latent" for unobserved samples, "observed" for observe sites (clamped at runtime), "marginalized" for sample sites that are integrated out by an enclosing marginalize block, "deterministic" for let bindings whose value is a closed-form function of upstream nodes (no stochastic draw). TYPE: NodeKind
family	The distribution family name ("Normal", "Dirichlet", "Categorical") for stochastic nodes. None for let nodes. TYPE: str | None
plates	The plates this node sits inside, ordered from outermost to innermost. Empty tuple for scalar nodes. TYPE: tuple[str, ...]
args	The variable names this node's family / let-expression references. The graph's edges are derived from these. TYPE: tuple[str, ...]
scope_path	The ::-separated scope path to this binding (e.g. lda::theta, lda::z::w). Used by renderers that link plate-diagram clicks back to :info lookups. TYPE: str

SaturationWarning didactic-model

SaturationWarning(*, name: str, source_line: int = 0, source_col: int = 0, algebra_name: str = '', depth: int = 0, intermediate_size: int | None = None, init_spec: InitSpec | None = None)

Bases: Model

One saturation-risk diagnosis at a specific source location.

ATTRIBUTE	DESCRIPTION
name	Latent variable name. TYPE: str
source_line, source_col	Position of the latent declaration in the source. TYPE: int
algebra_name	Governing algebra at the latent's location. TYPE: str
depth	Chain depth of the latent (number of stochastic-bind steps from the program entry, 1-indexed). TYPE: int
intermediate_size	Inferred plate / shared-axis size. TYPE: int | None
init_spec	The recommended algebra-specific initialization for this latent. TYPE: InitSpec

message

message() -> str

One-line human-readable diagnosis suitable for surfacing at qvr check / Compiler warning level.

Source code in src/quivers/analysis/saturation.py

def message(self) -> str:
    """One-line human-readable diagnosis suitable for surfacing
    at ``qvr check`` / Compiler warning level."""
    loc = f"line {self.source_line}" if self.source_line else "(unknown source)"
    spec = self.init_spec
    if spec is None:
        return (
            f"{self.name!r} at {loc}: saturation risk under "
            f"{self.algebra_name} at depth {self.depth}; no init "
            "recipe registered for this algebra"
        )
    return (
        f"{self.name!r} at {loc}: {spec.rationale}. Consider declaring "
        f"the latent with the recommended init "
        f"({spec.distribution}, mean={spec.mean:.4g}, std={spec.std:.4g})."
    )

ScopedRef didactic-model

ScopedRef(*, name: str, kind: ScopeKind, path: str, parent_kind: ScopeKind | None = None, node: object = field(opaque=True))

Bases: Model

One resolved binding.

ATTRIBUTE	DESCRIPTION
name	Last path segment. "lda" for lda, "w" for lda::z::w. TYPE: str
kind	Categorical tag indicating which declaration / step kind this ref points at. TYPE: ScopeKind
path	The full ::-separated path from the module root. TYPE: str
parent_kind	None for a top-level binding; otherwise the enclosing container's kind (useful for context-aware rendering: a sample-site belongs to a program or a marginalize-site). TYPE: ScopeKind | None

apply_init_spec

apply_init_spec(parameter: Parameter | Tensor, spec: InitSpec) -> None

Materialize an InitSpec onto a learnable tensor.

Delegates to torch.nn.init for normal / uniform and to a plain fill_ for constant. The tensor's shape is preserved; only its entries are overwritten in-place.

PARAMETER	DESCRIPTION
parameter	The tensor to overwrite in place. TYPE: Parameter or Tensor
spec	The recipe to draw the new entries from. TYPE: InitSpec

Source code in src/quivers/analysis/init_spec.py

def apply_init_spec(parameter: nn.Parameter | torch.Tensor, spec: InitSpec) -> None:
    """Materialize an `InitSpec` onto a learnable tensor.

    Delegates to ``torch.nn.init`` for ``normal`` / ``uniform`` and
    to a plain ``fill_`` for ``constant``. The tensor's shape is
    preserved; only its entries are overwritten in-place.

    Parameters
    ----------
    parameter : nn.Parameter or torch.Tensor
        The tensor to overwrite in place.
    spec : InitSpec
        The recipe to draw the new entries from.
    """
    with torch.no_grad():
        if spec.distribution == "normal":
            nn.init.normal_(parameter, mean=spec.mean, std=spec.std)
        elif spec.distribution == "uniform":
            nn.init.uniform_(parameter, a=spec.lower, b=spec.upper)
        else:
            parameter.fill_(spec.mean)

recommend_init

recommend_init(module: Module) -> dict[str, InitSpec]

Per-latent algebra-specific initialization recipe.

Walks the program's ChainShape, looks up the governing algebra, and returns a mapping from latent variable name to the InitSpec that places the chain at the algebra's mid-saturation point at depth-and-size.

PARAMETER	DESCRIPTION
module	The parsed QVR module whose program is analysed. TYPE: Module

RETURNS	DESCRIPTION
dict[str, InitSpec]	The recipe for each latent, keyed by variable name; empty when the module declares no governing algebra.

Source code in src/quivers/analysis/init_spec.py

def recommend_init(module: Module) -> dict[str, InitSpec]:
    """Per-latent algebra-specific initialization recipe.

    Walks the program's `ChainShape`, looks up the governing
    algebra, and returns a mapping from latent variable name to the
    `InitSpec` that places the chain at the algebra's
    mid-saturation point at depth-and-size.

    Parameters
    ----------
    module : Module
        The parsed QVR module whose program is analysed.

    Returns
    -------
    dict[str, InitSpec]
        The recipe for each latent, keyed by variable name; empty when
        the module declares no governing algebra.
    """
    shape = ChainShape.from_module(module)
    algebra = shape.algebra
    if algebra is None:
        return {}
    out: dict[str, InitSpec] = {}
    for step in shape.steps:
        if step.kind != "latent":
            continue
        size = step.intermediate_size or 1
        out[step.name] = _algebra_init_spec(algebra, step.depth, size)
    return out

build_plate_graph

build_plate_graph(compiler, program_name: str) -> PlateGraph | None

Build the :class:PlateGraph for program_name.

Returns None if program_name isn't a registered program on compiler. Walks the program's draws and any nested marginalize body recursively. Marginalized sample sites (the variable bound by the marginalize header) get kind="marginalized" so renderers can shade them differently from a plain "latent".

PARAMETER	DESCRIPTION
compiler	A quivers.dsl.Compiler that has compiled the module holding the program. TYPE: Compiler
program_name	The name of the program to diagram. TYPE: str

RETURNS	DESCRIPTION
PlateGraph or None	The program's plate graph, or None when no program of that name is registered.

Source code in src/quivers/analysis/plate_graph.py

def build_plate_graph(  # type: ignore[no-untyped-def]
    compiler,
    program_name: str,
) -> PlateGraph | None:
    """Build the :class:`PlateGraph` for ``program_name``.

    Returns ``None`` if ``program_name`` isn't a registered program
    on ``compiler``. Walks the program's ``draws`` and any nested
    ``marginalize`` body recursively. Marginalized sample sites
    (the variable bound by the marginalize header) get
    ``kind="marginalized"`` so renderers can shade them
    differently from a plain ``"latent"``.

    Parameters
    ----------
    compiler : Compiler
        A `quivers.dsl.Compiler` that has compiled the module holding
        the program.
    program_name : str
        The name of the program to diagram.

    Returns
    -------
    PlateGraph or None
        The program's plate graph, or ``None`` when no program of that
        name is registered.
    """
    # Locate the program declaration. Parametric programs live on
    # ``compiler.programs`` (the template registry); non-parametric
    # programs compile straight to a morphism and are accessible
    # only through the source module's statement list. Check both.
    programs = getattr(compiler, "programs", {}) or {}
    decl = programs.get(program_name)
    if not isinstance(decl, ProgramDecl):
        module = getattr(compiler, "_module", None)
        for stmt in getattr(module, "statements", ()) or ():
            if (
                isinstance(stmt, ProgramDecl)
                and getattr(stmt, "name", None) == program_name
            ):
                decl = stmt
                break
    if not isinstance(decl, ProgramDecl):
        return None

    nodes: list[PlateNode] = []
    plate_names: dict[str, str | None] = {}  # plate -> parent plate
    edges: list[Edge] = []

    def _walk(  # type: ignore[no-untyped-def]
        steps: tuple[object, ...],
        parent_plates: tuple[str, ...],
        path_prefix: str,
    ) -> None:
        for step in steps:
            if isinstance(step, SampleStep):
                name = step.vars[0] if step.vars else "?"
                plates = _plates_for_step(step, parent_plates)
                for p in plates:
                    if p not in plate_names:
                        plate_names[p] = parent_plates[-1] if parent_plates else None
                args = _step_args(step)
                nodes.append(
                    PlateNode(
                        name=name,
                        kind="latent",
                        family=step.morphism,
                        plates=plates,
                        args=args,
                        scope_path=f"{path_prefix}::{name}",
                    )
                )
                for a in args:
                    edges.append(Edge(src=a, dst=name))
            elif isinstance(step, ObserveStep):
                name = step.vars[0]
                plates = _plates_for_step(step, parent_plates)
                for p in plates:
                    if p not in plate_names:
                        plate_names[p] = parent_plates[-1] if parent_plates else None
                args = _step_args(step)
                nodes.append(
                    PlateNode(
                        name=name,
                        kind="observed",
                        family=step.morphism,
                        plates=plates,
                        args=args,
                        scope_path=f"{path_prefix}::{name}",
                    )
                )
                for a in args:
                    edges.append(Edge(src=a, dst=name))
            elif isinstance(step, MarginalizeStep):
                name = step.var
                plates = _plates_for_step(step, parent_plates)
                for p in plates:
                    if p not in plate_names:
                        plate_names[p] = parent_plates[-1] if parent_plates else None
                args = _step_args(step)
                nodes.append(
                    PlateNode(
                        name=name,
                        kind="marginalized",
                        family=step.morphism,
                        plates=plates,
                        args=args,
                        scope_path=f"{path_prefix}::{name}",
                    )
                )
                for a in args:
                    edges.append(Edge(src=a, dst=name))
                # Recurse into the marginalize body. Inner plates
                # inherit ``parent_plates + [over]`` rather than
                # the marginalize variable's own index plate,
                # because the marginalized variable is integrated
                # out and inner observations don't lie on its axis.
                inner_prefix = f"{path_prefix}::{name}"
                inner_plates = _child_plates_for_marginalize(step, parent_plates)
                _walk(step.scope, inner_plates, inner_prefix)
            elif isinstance(step, LetStep):
                name = step.name
                args = _let_value_args(step.value)
                nodes.append(
                    PlateNode(
                        name=name,
                        kind="deterministic",
                        family=None,
                        plates=parent_plates,
                        args=args,
                        scope_path=f"{path_prefix}::{name}",
                    )
                )
                for a in args:
                    edges.append(Edge(src=a, dst=name))
            elif isinstance(step, ReturnStep):
                # Return doesn't introduce a node; downstream
                # consumers see the return vars via the program
                # decl.
                continue

    _walk(decl.draws, (), program_name)

    plates_out: list[Plate] = []
    for p, parent in plate_names.items():
        plates_out.append(
            Plate(
                name=p,
                cardinality=_cardinality_lookup(compiler, p),
                parent=parent,
            )
        )

    domain = getattr(decl.domain, "name", "") if decl.domain is not None else ""
    codomain = getattr(decl.codomain, "name", "") if decl.codomain is not None else ""
    return PlateGraph(
        program_name=program_name,
        domain=domain,
        codomain=codomain,
        nodes=tuple(nodes),
        plates=tuple(plates_out),
        edges=tuple(edges),
    )

render_daft

render_daft(graph: PlateGraph) -> str

Return a Python script that uses daft to render the plate diagram. The script defines a build_pgm() function returning a daft.PGM instance, ready for the user to render() and save.

PARAMETER	DESCRIPTION
graph	The plate graph to draw. TYPE: PlateGraph

RETURNS	DESCRIPTION
str	Python source defining build_pgm().

Source code in src/quivers/analysis/plate_render.py

def render_daft(graph: PlateGraph) -> str:
    """Return a Python script that uses ``daft`` to render the
    plate diagram. The script defines a ``build_pgm()`` function
    returning a ``daft.PGM`` instance, ready for the user to
    ``render()`` and save.

    Parameters
    ----------
    graph : PlateGraph
        The plate graph to draw.

    Returns
    -------
    str
        Python source defining ``build_pgm()``.
    """
    # Topological depth -> column position.
    depth: dict[str, int] = {n.name: 0 for n in graph.nodes}
    changed = True
    while changed:
        changed = False
        for e in graph.edges:
            new_d = depth.get(e.src, 0) + 1
            if new_d > depth.get(e.dst, 0):
                depth[e.dst] = new_d
                changed = True

    by_col: dict[int, list[PlateNode]] = {}
    for n in graph.nodes:
        by_col.setdefault(depth.get(n.name, 0), []).append(n)
    positions: dict[str, tuple[float, float]] = {}
    for col, members in by_col.items():
        for row, n in enumerate(members):
            positions[n.name] = (col + 1.0, -row + len(members) / 2.0)

    width = max(depth.values(), default=0) + 2
    height = max(len(m) for m in by_col.values()) + 1 if by_col else 1

    lines: list[str] = [
        "import daft",
        "",
        "",
        "def build_pgm():",
        f'    """Build the plate diagram for {graph.program_name}."""',
        f"    pgm = daft.PGM(shape=[{width}, {height}])",
    ]
    for n in graph.nodes:
        x, y = positions[n.name]
        observed = "True" if n.kind == "observed" else "False"
        fixed = "True" if n.kind == "deterministic" else "False"
        lines.append(
            f'    pgm.add_node("{n.name}", r"${n.name}$", {x:.2f}, {y:.2f}, '
            f"observed={observed}, fixed={fixed})"
        )
    for e in graph.edges:
        if any(n.name == e.src for n in graph.nodes):
            lines.append(f'    pgm.add_edge("{e.src}", "{e.dst}")')

    # Plates: bounding box per plate stack, derived from node positions.
    plate_stacks = sorted({tuple(n.plates) for n in graph.nodes}, key=len)
    for stack in plate_stacks:
        if not stack:
            continue
        members = [n for n in graph.nodes if n.plates == stack]
        if not members:
            continue
        xs = [positions[m.name][0] for m in members]
        ys = [positions[m.name][1] for m in members]
        x0 = min(xs) - 0.5
        y0 = min(ys) - 0.5
        w = max(xs) - x0 + 0.5
        h = max(ys) - y0 + 0.5
        label = " \\times ".join(stack)
        lines.append(
            f"    pgm.add_plate([{x0:.2f}, {y0:.2f}, {w:.2f}, {h:.2f}], "
            f'label=r"${label}$")'
        )

    lines.append("    return pgm")
    lines.append("")
    lines.append("")
    lines.append('if __name__ == "__main__":')
    lines.append("    pgm = build_pgm()")
    lines.append("    pgm.render()")
    lines.append('    pgm.savefig("plate.png")')
    return "\n".join(lines)

render_dot

render_dot(graph: PlateGraph) -> str

Return Graphviz DOT source. Plates become cluster_* subgraphs; nodes carry shape attributes based on kind.

PARAMETER	DESCRIPTION
graph	The plate graph to draw. TYPE: PlateGraph

RETURNS	DESCRIPTION
str	Graphviz DOT source.

Source code in src/quivers/analysis/plate_render.py

def render_dot(graph: PlateGraph) -> str:
    """Return Graphviz DOT source. Plates become ``cluster_*``
    subgraphs; nodes carry shape attributes based on kind.

    Parameters
    ----------
    graph : PlateGraph
        The plate graph to draw.

    Returns
    -------
    str
        Graphviz DOT source.
    """
    shape_for: dict[str, str] = {
        "latent": "circle",
        "observed": "doublecircle",
        "marginalized": "octagon",
        "deterministic": "box",
    }
    fillcolor_for: dict[str, str] = {
        "latent": "white",
        "observed": "lightgray",
        "marginalized": "white",
        "deterministic": "white",
    }
    lines: list[str] = [
        f"digraph {_safe_id(graph.program_name)} {{",
        "  graph [rankdir=TB, fontsize=10]",
        "  node  [fontsize=10, style=filled]",
        "  edge  [fontsize=9]",
    ]

    by_plate: dict[tuple[str, ...], list[PlateNode]] = {}
    for n in graph.nodes:
        by_plate.setdefault(n.plates, []).append(n)

    def _node_id(n: PlateNode) -> str:
        return _safe_id(n.scope_path or n.name)

    def _emit_node(n: PlateNode, indent: str) -> str:
        nid = _node_id(n)
        label = n.name
        if n.family:
            label = f"{n.name}\\n{n.family}"
        shape = shape_for.get(n.kind, "box")
        fc = fillcolor_for.get(n.kind, "white")
        if n.kind == "marginalized":
            return f'{indent}{nid} [label="{label}", shape={shape}, fillcolor={fc}, style="filled,dashed"]'
        return f'{indent}{nid} [label="{label}", shape={shape}, fillcolor={fc}]'

    # Cluster per plate stack.
    plate_stacks = sorted({tuple(n.plates) for n in graph.nodes}, key=len)
    for stack in plate_stacks:
        if not stack:
            for n in by_plate.get(stack, ()):
                lines.append(_emit_node(n, "  "))
            continue
        cid = _safe_id("_x_".join(stack))
        label = " x ".join(stack)
        lines.append(f"  subgraph cluster_{cid} {{")
        lines.append(f'    label="{label}";')
        lines.append('    style="rounded,dashed";')
        for n in by_plate.get(stack, ()):
            lines.append(_emit_node(n, "    "))
        lines.append("  }")

    for e in graph.edges:
        src = _safe_id(e.src)
        dst_node = next((n for n in graph.nodes if n.name == e.dst), None)
        dst = _node_id(dst_node) if dst_node else _safe_id(e.dst)
        lines.append(f"  {src} -> {dst}")
    lines.append("}")
    return "\n".join(lines)

render_mermaid

render_mermaid(graph: PlateGraph) -> str

Return Mermaid graph TD source for graph.

Each plate becomes a subgraph cluster. Each node is drawn as a circle (( )) if latent, a double-circle ((( ))) if observed, an octagon ({{ }}) if marginalized, and a hexagon (([ ])) if deterministic. Edges follow the graph's edges field.

Mermaid lacks a native plate / nested-plate primitive; the subgraph blocks approximate it well enough that GitHub-rendered markdown and mermaid.live both display the structure.

PARAMETER	DESCRIPTION
graph	The plate graph to draw. TYPE: PlateGraph

RETURNS	DESCRIPTION
str	Mermaid graph TD source.

Source code in src/quivers/analysis/plate_render.py

def render_mermaid(graph: PlateGraph) -> str:
    """Return Mermaid ``graph TD`` source for ``graph``.

    Each plate becomes a ``subgraph`` cluster. Each node is
    drawn as a circle (``( )``) if latent, a double-circle
    (``(( ))``) if observed, an octagon (``{{ }}``) if
    marginalized, and a hexagon (``([ ])``) if deterministic.
    Edges follow the graph's ``edges`` field.

    Mermaid lacks a native plate / nested-plate primitive; the
    subgraph blocks approximate it well enough that
    GitHub-rendered markdown and mermaid.live both display the
    structure.

    Parameters
    ----------
    graph : PlateGraph
        The plate graph to draw.

    Returns
    -------
    str
        Mermaid ``graph TD`` source.
    """
    lines: list[str] = []
    lines.append("graph TD")
    # Node declarations, grouped by their innermost plate.
    by_plate: dict[tuple[str, ...], list[PlateNode]] = {}
    for n in graph.nodes:
        by_plate.setdefault(n.plates, []).append(n)

    def _node_decl(n: PlateNode) -> str:
        nid = _safe_id(n.scope_path or n.name)
        label = n.name
        if n.family:
            label = f"{n.name}<br/>{n.family}"
        if n.kind == "latent":
            return f"    {nid}(({label}))"
        if n.kind == "observed":
            return f"    {nid}(((({label}))))"
        if n.kind == "marginalized":
            return f"    {nid}{{{{ {label} }}}}"
        return f"    {nid}([{label}])"

    # Emit plate clusters in nesting order (longest plate stacks
    # last so inner plates appear inside outer ones in the source).
    plate_stacks = sorted({tuple(n.plates) for n in graph.nodes}, key=len)
    seen_cluster_keys: set[tuple[str, ...]] = set()
    for stack in plate_stacks:
        if not stack:
            # Outer scope: emit naked.
            for n in by_plate.get(stack, ()):
                lines.append(_node_decl(n))
            continue
        if stack in seen_cluster_keys:
            continue
        seen_cluster_keys.add(stack)
        label = " x ".join(stack)
        cid = _safe_id("_x_".join(stack))
        lines.append(f'    subgraph {cid} ["{label}"]')
        for n in by_plate.get(stack, ()):
            lines.append(_node_decl(n))
        lines.append("    end")

    # Edges.
    for e in graph.edges:
        # Some edges reference variables that aren't in the
        # graph's nodes (e.g. external program inputs). Emit them
        # anyway; Mermaid creates an implicit node.
        src = _safe_id(e.src)
        dst_node = next((n for n in graph.nodes if n.name == e.dst), None)
        dst = _safe_id(dst_node.scope_path) if dst_node else _safe_id(e.dst)
        lines.append(f"    {src} --> {dst}")

    return "\n".join(lines)

render_table

render_table(graph: PlateGraph)

Return a rich.table.Table describing graph.

The caller is responsible for printing it (Console.print for the plain prompt; RichLog.write for the Textual TUI).

PARAMETER	DESCRIPTION
graph	The plate graph to tabulate. TYPE: PlateGraph

RETURNS	DESCRIPTION
Table	One row per node, with its kind, family, plates, and parents.

Source code in src/quivers/analysis/plate_render.py

def render_table(graph: PlateGraph):  # type: ignore[no-untyped-def]
    """Return a `rich.table.Table` describing ``graph``.

    The caller is responsible for printing it (`Console.print` for
    the plain prompt; `RichLog.write` for the Textual TUI).

    Parameters
    ----------
    graph : PlateGraph
        The plate graph to tabulate.

    Returns
    -------
    rich.table.Table
        One row per node, with its kind, family, plates, and parents.
    """
    from rich.table import Table
    from rich.text import Text

    title = Text()
    title.append("plate graph of ", style="dim")
    title.append(graph.program_name, style="bold")
    if graph.domain or graph.codomain:
        title.append(f"  ({graph.domain} -> {graph.codomain})", style="dim")

    table = Table(title=title, show_header=True, expand=False)
    table.add_column("#", justify="right", style="dim")
    table.add_column("kind", justify="left")
    table.add_column("variable", justify="left")
    table.add_column("family", justify="left")
    table.add_column("plates", justify="left")
    table.add_column("parents", justify="left")

    parents_by_dst: dict[str, list[str]] = {}
    for e in graph.edges:
        parents_by_dst.setdefault(e.dst, []).append(e.src)

    for i, node in enumerate(graph.nodes, start=1):
        badge = _KIND_BADGE.get(node.kind, "·")
        kind_cell = Text(f"{badge} {node.kind}", style=_KIND_STYLE.get(node.kind, ""))
        plates = " ⨯ ".join(node.plates) if node.plates else "—"
        parents = ", ".join(parents_by_dst.get(node.name, ())) or "—"
        family = node.family or "—"
        table.add_row(
            str(i),
            kind_cell,
            node.name,
            family,
            plates,
            parents,
        )
    return table

render_table_plain

render_table_plain(graph: PlateGraph) -> str

Plain-text rendering of the table, for non-Rich callers.

Used by :plate from qvr repl --plain and from the Jupyter kernel.

PARAMETER	DESCRIPTION
graph	The plate graph to tabulate. TYPE: PlateGraph

RETURNS	DESCRIPTION
str	The table as aligned plain text.

Source code in src/quivers/analysis/plate_render.py

def render_table_plain(graph: PlateGraph) -> str:
    """Plain-text rendering of the table, for non-Rich callers.

    Used by `:plate` from `qvr repl --plain` and from the Jupyter
    kernel.

    Parameters
    ----------
    graph : PlateGraph
        The plate graph to tabulate.

    Returns
    -------
    str
        The table as aligned plain text.
    """
    parents_by_dst: dict[str, list[str]] = {}
    for e in graph.edges:
        parents_by_dst.setdefault(e.dst, []).append(e.src)

    cols = ("#", "kind", "variable", "family", "plates", "parents")
    rows: list[tuple[str, ...]] = []
    for i, node in enumerate(graph.nodes, start=1):
        plates = " x ".join(node.plates) if node.plates else "-"
        parents = ", ".join(parents_by_dst.get(node.name, ())) or "-"
        family = node.family or "-"
        rows.append(
            (
                str(i),
                f"{_KIND_BADGE.get(node.kind, '.')} {node.kind}",
                node.name,
                family,
                plates,
                parents,
            )
        )

    # Compute column widths
    widths = [len(c) for c in cols]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))

    def _fmt(cells: tuple[str, ...]) -> str:
        return "  ".join(c.ljust(widths[i]) for i, c in enumerate(cells))

    out_lines = [
        f"plate graph of {graph.program_name}"
        + (
            f"  ({graph.domain} -> {graph.codomain})"
            if graph.domain or graph.codomain
            else ""
        ),
        "",
        _fmt(cols),
        "  ".join("-" * w for w in widths),
    ]
    out_lines.extend(_fmt(row) for row in rows)

    if graph.plates:
        out_lines.append("")
        out_lines.append("plates:")
        for p in graph.plates:
            card = f" [{p.cardinality}]" if p.cardinality else ""
            parent = f" parent={p.parent}" if p.parent else ""
            out_lines.append(f"  {p.name}{card}{parent}")

    return "\n".join(out_lines)

render_tikz

render_tikz(graph: PlateGraph) -> str

Return a LaTeX snippet using tikz and the bayesnet package conventions. Computes positions by laying nodes out in columns (one per depth level) and rows (one per plate stack); good enough for diagrams up to ~20 nodes.

PARAMETER	DESCRIPTION
graph	The plate graph to draw. TYPE: PlateGraph

RETURNS	DESCRIPTION
str	A LaTeX tikzpicture snippet.

Source code in src/quivers/analysis/plate_render.py

def render_tikz(graph: PlateGraph) -> str:
    """Return a LaTeX snippet using ``tikz`` and the ``bayesnet``
    package conventions. Computes positions by laying nodes out
    in columns (one per depth level) and rows (one per plate
    stack); good enough for diagrams up to ~20 nodes.

    Parameters
    ----------
    graph : PlateGraph
        The plate graph to draw.

    Returns
    -------
    str
        A LaTeX ``tikzpicture`` snippet.
    """
    # Topological depth = longest path of edges into a node.
    depth: dict[str, int] = {}
    for n in graph.nodes:
        depth[n.name] = 0
    changed = True
    while changed:
        changed = False
        for e in graph.edges:
            new_d = depth.get(e.src, 0) + 1
            if new_d > depth.get(e.dst, 0):
                depth[e.dst] = new_d
                changed = True

    # Group nodes by (depth, plate_stack)
    positions: dict[str, tuple[float, float]] = {}
    by_col: dict[int, list[PlateNode]] = {}
    for n in graph.nodes:
        by_col.setdefault(depth.get(n.name, 0), []).append(n)
    for col, members in by_col.items():
        for row, n in enumerate(members):
            positions[n.name] = (col * 2.0, -row * 1.5)

    lines: list[str] = [
        "% Plate diagram of " + graph.program_name,
        "% Requires \\usepackage{tikz} and \\usetikzlibrary{bayesnet}",
        "\\begin{tikzpicture}",
    ]
    for n in graph.nodes:
        x, y = positions[n.name]
        nid = _safe_id(n.scope_path or n.name)
        label = n.name
        if n.kind == "observed":
            lines.append(f"  \\node[obs] ({nid}) at ({x:.1f},{y:.1f}) {{${label}$}};")
        elif n.kind == "marginalized":
            lines.append(
                f"  \\node[latent, dashed] ({nid}) at ({x:.1f},{y:.1f}) {{${label}$}};"
            )
        elif n.kind == "deterministic":
            lines.append(f"  \\node[det] ({nid}) at ({x:.1f},{y:.1f}) {{${label}$}};")
        else:
            lines.append(
                f"  \\node[latent] ({nid}) at ({x:.1f},{y:.1f}) {{${label}$}};"
            )

    for e in graph.edges:
        src_node = next((n for n in graph.nodes if n.name == e.src), None)
        if src_node is None:
            continue
        dst_node = next((n for n in graph.nodes if n.name == e.dst), None)
        if dst_node is None:
            continue
        src = _safe_id(src_node.scope_path or src_node.name)
        dst = _safe_id(dst_node.scope_path or dst_node.name)
        lines.append(f"  \\edge {{{src}}} {{{dst}}};")

    # Plates: each plate becomes a \plate around its members.
    plate_stacks = sorted({tuple(n.plates) for n in graph.nodes}, key=len)
    plate_id = 0
    for stack in plate_stacks:
        if not stack:
            continue
        members = [n for n in graph.nodes if n.plates == stack]
        if not members:
            continue
        nodes = " ".join(_safe_id(n.scope_path or n.name) for n in members)
        label = " \\times ".join(stack)
        lines.append(f"  \\plate {{p{plate_id}}} {{({nodes})}} {{${label}$}};")
        plate_id += 1

    lines.append("\\end{tikzpicture}")
    return "\n".join(lines)

saturation_warnings

saturation_warnings(module: Module) -> tuple[SaturationWarning, ...]

Per-latent saturation diagnoses.

Returns a warning for every latent step in the program at depth ≥ 2 or with intermediate size > 1 under an algebra whose algebra-specific recipe differs materially from a default Normal(0, 1) init. The threshold for "differs materially" is the absolute distance between the recipe's mean and 0 (Normal default) plus the relative distance between its std and 1; we surface a warning when either pulls the recommendation more than 20% away from the default.

Suppressing warnings: declare the latent with an explicit init via the DSL surface (when available) or apply the recommended init programmatically via quivers.analysis.init_spec.apply_init_spec.

PARAMETER	DESCRIPTION
module	The parsed QVR module whose program is analysed. TYPE: Module

RETURNS	DESCRIPTION
tuple[SaturationWarning, ...]	One warning per latent whose recommended init departs from the default; empty when the module declares no governing algebra.

Source code in src/quivers/analysis/saturation.py

def saturation_warnings(module: Module) -> tuple[SaturationWarning, ...]:
    """Per-latent saturation diagnoses.

    Returns a warning for every ``latent`` step in the program at
    depth ≥ 2 or with intermediate size > 1 under an algebra whose
    algebra-specific recipe differs materially from a default
    ``Normal(0, 1)`` init. The threshold for "differs materially"
    is the absolute distance between the recipe's mean and 0
    (Normal default) plus the relative distance between its std
    and 1; we surface a warning when either pulls the
    recommendation more than 20% away from the default.

    Suppressing warnings: declare the latent with an explicit init
    via the DSL surface (when available) or apply the recommended
    init programmatically via
    [`quivers.analysis.init_spec.apply_init_spec`][quivers.analysis.init_spec.apply_init_spec].

    Parameters
    ----------
    module : Module
        The parsed QVR module whose program is analysed.

    Returns
    -------
    tuple[SaturationWarning, ...]
        One warning per latent whose recommended init departs from the
        default; empty when the module declares no governing algebra.
    """
    shape = ChainShape.from_module(module)
    algebra = shape.algebra
    if algebra is None:
        return ()
    out: list[SaturationWarning] = []
    for step in shape.steps:
        if step.kind != "latent":
            continue
        size = step.intermediate_size or 1
        if step.depth < 2 and size <= 1:
            continue
        spec = _algebra_init_spec(algebra, step.depth, size)
        if not _spec_differs_from_default(spec):
            continue
        out.append(
            SaturationWarning(
                name=step.name,
                source_line=step.source_line,
                source_col=step.source_col,
                algebra_name=step.algebra_name,
                depth=step.depth,
                intermediate_size=step.intermediate_size,
                init_spec=spec,
            )
        )
    return tuple(out)

find_all_references

find_all_references(compiler, name: str) -> list[ScopedRef]

Find every binding in compiler's module whose final path segment equals name.

Returns refs sorted by depth (top-level first) then by path lexicographic order. Used by :where NAME to surface every scope a bare name appears in.

PARAMETER	DESCRIPTION
compiler	A quivers.dsl.Compiler whose module has been compiled. TYPE: Compiler
name	The bare name to search for. TYPE: str

RETURNS	DESCRIPTION
list[ScopedRef]	Every binding whose final path segment is name.

Source code in src/quivers/analysis/scope.py

def find_all_references(compiler, name: str) -> list[ScopedRef]:  # type: ignore[no-untyped-def]
    """Find every binding in ``compiler``'s module whose final
    path segment equals ``name``.

    Returns refs sorted by depth (top-level first) then by path
    lexicographic order. Used by ``:where NAME`` to surface every
    scope a bare name appears in.

    Parameters
    ----------
    compiler : Compiler
        A `quivers.dsl.Compiler` whose module has been compiled.
    name : str
        The bare name to search for.

    Returns
    -------
    list[ScopedRef]
        Every binding whose final path segment is ``name``.
    """
    hits: list[ScopedRef] = []
    # Walk every top-level binding and recursively visit scopes.
    # Bounded by the AST size; QVR programs are small.
    seen: set[str] = set()
    for attr, kind in (
        ("objects", "object"),
        ("spaces", "space"),
        ("morphisms", "morphism"),
        ("rules", "rule"),
        ("programs", "program"),
        ("deductions", "deduction"),
        ("signatures", "signature"),
        ("encoders", "encoder"),
        ("decoders", "decoder"),
        ("losses", "loss"),
        ("bundles", "bundle"),
        ("contractions", "contraction"),
    ):
        mapping = getattr(compiler, attr, None) or {}
        for n, node in mapping.items():
            top = ScopedRef(name=n, kind=kind, path=n, parent_kind=None, node=node)
            _collect_named(top, name, hits, seen)
    hits.sort(key=lambda r: (r.path.count(SCOPE_SEPARATOR), r.path))
    return hits

resolve_scoped_path

resolve_scoped_path(compiler, path: str) -> ScopedRef | None

Resolve a ::-separated path against compiler's elaborated module. Returns None if any segment fails to resolve.

PARAMETER	DESCRIPTION
compiler	A quivers.dsl.compiler.Compiler instance whose compile_env() has run.
path	A non-empty ::-separated string. TYPE: str

RETURNS	DESCRIPTION
ScopedRef | None	The terminal binding, or None if any segment doesn't exist in its parent's scope.

Source code in src/quivers/analysis/scope.py

def resolve_scoped_path(compiler, path: str) -> ScopedRef | None:  # type: ignore[no-untyped-def]
    """Resolve a ``::``-separated path against ``compiler``'s
    elaborated module. Returns ``None`` if any segment fails to
    resolve.

    Parameters
    ----------
    compiler
        A ``quivers.dsl.compiler.Compiler`` instance whose
        ``compile_env()`` has run.
    path
        A non-empty ``::``-separated string.

    Returns
    -------
    ScopedRef | None
        The terminal binding, or ``None`` if any segment doesn't
        exist in its parent's scope.
    """
    segments = split_path(path)
    if not segments or any(not s for s in segments):
        return None
    current = _top_level_lookup(compiler, segments[0])
    if current is None:
        return None
    for seg in segments[1:]:
        children = scope_children(current)
        if seg not in children:
            return None
        current = children[seg]
    return current

scope_children

scope_children(ref: ScopedRef) -> Mapping[str, ScopedRef]

Return the named children of ref's scope, or an empty mapping if ref.kind introduces no further bindings.

Idempotent + pure; callers can cache. Reads only the AST fields exposed on each declaration shape, so adding a new declaration kind requires registering one entry in _SCOPE_DISPATCH.

PARAMETER	DESCRIPTION
ref	The binding whose scope is listed. TYPE: ScopedRef

RETURNS	DESCRIPTION
Mapping[str, ScopedRef]	The child bindings, keyed by name.

Source code in src/quivers/analysis/scope.py

def scope_children(ref: ScopedRef) -> Mapping[str, ScopedRef]:
    """Return the named children of ``ref``'s scope, or an empty
    mapping if ``ref.kind`` introduces no further bindings.

    Idempotent + pure; callers can cache. Reads only the AST
    fields exposed on each declaration shape, so adding a new
    declaration kind requires registering one entry in
    ``_SCOPE_DISPATCH``.

    Parameters
    ----------
    ref : ScopedRef
        The binding whose scope is listed.

    Returns
    -------
    Mapping[str, ScopedRef]
        The child bindings, keyed by name.
    """
    builder = _SCOPE_DISPATCH.get(ref.kind)
    if builder is None:
        return {}
    return builder(ref)  # type: ignore[operator]

split_path

split_path(path: str) -> list[str]

Split "a::b::c" into ["a", "b", "c"]. Leading or trailing separators yield empty segments, which the resolver rejects.

PARAMETER	DESCRIPTION
path	A SCOPE_SEPARATOR-separated path. TYPE: str

RETURNS	DESCRIPTION
list[str]	The path's segments, in order.

Source code in src/quivers/analysis/scope.py

def split_path(path: str) -> list[str]:
    """Split ``"a::b::c"`` into ``["a", "b", "c"]``. Leading or
    trailing separators yield empty segments, which the resolver
    rejects.

    Parameters
    ----------
    path : str
        A `SCOPE_SEPARATOR`-separated path.

    Returns
    -------
    list[str]
        The path's segments, in order.
    """
    return path.split(SCOPE_SEPARATOR)