pyfdn-system: one JSON document format for every pyFDN backend (design + schema sketch)

HISTORY.rst

@@ -2,6 +2,22 @@
 History
 =======
 
+Unreleased
+----------
+
+* New ``pyFDN.system`` module (draft, not yet exported from ``pyFDN``): the
+  ``pyfdn-system`` v1 JSON document, which holds either an unchanged
+  ``pyfdn-fdn-build`` or a node ``graph`` for systems a build cannot express.
+  Includes validation with path-named errors, ``build_to_graph`` /
+  ``graph_to_build`` and ``save_fdn_system`` / ``load_fdn_system``.
+  ``save_fdn_system`` stamps ``metadata.created`` and
+  ``metadata.pyfdn_version`` when they are missing. See
+  ``docs/design/system-format.md``.
+* New ``pyFDN.auxiliary.graph_plot.plot_system_graph`` draws the signal flow of
+  a system document, a ``GraphNode`` tree or an ``FDNBuild`` without building
+  a FLAMO model. ``plot_flamo_graph`` now uses the same renderer; its output is
+  unchanged.
+
 0.5.0 (2026-09-28)
 ------------------
 

src/pyFDN/__init__.py

@@ -520,11 +520,15 @@
         mimo_rir_eigenvalues_per_frequency,
         model_response,
         param,
-        params,
         train_fdn,
         trainable_from_build,
         trainable_from_preset,
     )
+
+    # ``params`` is also the name of the submodule it lives in; importing it
+    # from ``.train`` lets mypy resolve it to the module under some import
+    # orders ("already defined" no-redef).
+    from .train.params import params
     from .translate.flamo_to_pr import (
         FlamoDecompositionForPR,
         flamo_decompose_for_pr,

src/pyFDN/auxiliary/flamo_graph.py

@@ -430,107 +430,6 @@ def _hook_sos(name: str) -> np.ndarray | None:
     )
 
 
-# ---------------------------------------------------------------------------
-# Matplotlib renderer
-# ---------------------------------------------------------------------------
-
-# Layout constants (abstract units; figure is scaled at the end)
-_LEAF_H = 0.9  # leaf box height
-_GAP_X = 0.6  # horizontal wire length between series elements
-_GAP_Y = 0.5  # vertical gap between parallel branches / fF and fB
-_NODE_R = 0.13  # radius of sum (+) and pickoff nodes
-_PAD = 0.3  # padding inside recursion container
-_STUB = 0.35  # split/merge wire stub length in Parallel
-
-_WIRE = "0.25"
-_LOOP_EDGE = "#b05a4a"
-
-# Leaf box colors by module category: (facecolor, edgecolor)
-_BLOCK_COLORS = {
-    "transform": ("#efefef", "#7a7a7a"),  # FFT / iFFT
-    "gain": ("#dbe9f6", "#3d6d9e"),  # gains
-    "matrix": ("#e8e1f2", "#7b5ea7"),  # mixing / feedback matrices
-    "delay": ("#fdf1d6", "#c08a2d"),  # delays
-    "filter": ("#e1efe3", "#4f8a5e"),  # filters / EQs
-    "other": ("#f5f5f5", "#8c8c8c"),
-}
-
-
-def _block_category(mod_type: str) -> str:
-    t = mod_type.lower()
-    if "fft" in t:
-        return "transform"
-    if "delay" in t:
-        return "delay"
-    if any(k in t for k in ("filter", "biquad", "svf", "geq", "peq", "eq", "sos")):
-        return "filter"
-    if any(k in t for k in ("matrix", "scattering", "householder", "hadamard")):
-        return "matrix"
-    if "gain" in t:
-        return "gain"
-    return "other"
-
-
-def _leaf_label(node: dict[str, Any]) -> str:
-    mod = node.get("module")
-    mod_type = type(mod).__name__ if mod is not None else "?"
-    name = node.get("name", "")
-    if name and name != mod_type and not name.isdigit():
-        return f"{name}\n{mod_type}"
-    return mod_type
-
-
-def _measure(node: dict[str, Any]) -> tuple[float, float, float]:
-    """
-    Bottom-up size pass. Returns (w, h, ay) where ay is the distance from the
-    top of the bounding box to the signal-flow axis. Cached in node["_size"].
-    """
-    ntype = node.get("type", "Leaf")
-
-    if ntype == "Shell":
-        parts = []
-        if node.get("input_layer") is not None:
-            parts.append(node["input_layer"])
-        parts.extend(node.get("children") or [])
-        if node.get("output_layer") is not None:
-            parts.append(node["output_layer"])
-        node["_parts"] = parts
-        sizes = [_measure(ch) for ch in parts]
-        w = sum(s[0] for s in sizes) + _GAP_X * max(len(parts) - 1, 0)
-        ay = max((s[2] for s in sizes), default=_LEAF_H / 2)
-        below = max((s[1] - s[2] for s in sizes), default=_LEAF_H / 2)
-        size = (w, ay + below, ay)
-
-    elif ntype == "Series":
-        sizes = [_measure(ch) for ch in node.get("children") or []]
-        w = sum(s[0] for s in sizes) + _GAP_X * max(len(sizes) - 1, 0)
-        ay = max((s[2] for s in sizes), default=_LEAF_H / 2)
-        below = max((s[1] - s[2] for s in sizes), default=_LEAF_H / 2)
-        size = (w, ay + below, ay)
-
-    elif ntype == "Parallel":
-        sizes = [_measure(ch) for ch in node.get("children") or []]
-        w = max((s[0] for s in sizes), default=1.0) + 2 * (_STUB + _NODE_R * 2)
-        h = sum(s[1] for s in sizes) + _GAP_Y * max(len(sizes) - 1, 0)
-        size = (w, h, h / 2)
-
-    elif ntype == "Recursion":
-        wf, hf, ayf = _measure(node["fF"]) if node.get("fF") else (1.0, _LEAF_H, 0.45)
-        wb, hb, _ = _measure(node["fB"]) if node.get("fB") else (1.0, _LEAF_H, 0.45)
-        inner_w = max(wf, wb)
-        w = _PAD + 2 * _NODE_R + _GAP_X + inner_w + _GAP_X + 2 * _NODE_R + _PAD
-        h = _PAD + hf + _GAP_Y + hb + _PAD
-        size = (w, h, _PAD + ayf)
-
-    else:  # Leaf
-        label = _leaf_label(node)
-        wmax = max(len(line) for line in label.split("\n"))
-        size = (max(1.1, 0.115 * wmax + 0.35), _LEAF_H, _LEAF_H / 2)
-
-    node["_size"] = size
-    return size
-
-
 def plot_flamo_graph(
     model: Any,
     *,
@@ -544,7 +443,9 @@ def plot_flamo_graph(
 
     Signal flows left to right; only the feedback path of a Recursion flows
     right to left, drawn below the forward path with a loop back to a sum
-    node at the forward path's input.
+    node at the forward path's input. To draw a saved ``pyfdn-system``
+    document without building a FLAMO model, use
+    :func:`pyFDN.auxiliary.graph_plot.plot_system_graph`.
 
     Parameters
     ----------
@@ -563,219 +464,7 @@ def plot_flamo_graph(
     -------
     ax : matplotlib Axes
     """
-    import matplotlib.pyplot as plt
-    from matplotlib.patches import Circle, FancyArrowPatch, FancyBboxPatch
+    from pyFDN.auxiliary.graph_plot import plot_node_tree
 
     root = flamo_model_to_nodes(model, name=name, include_shell_io=True)
-    total_w, total_h, root_ay = _measure(root)
-
-    if ax is None:
-        fig_w = (total_w + 2.0) * scale
-        fig_h = (total_h + 0.8) * scale
-        _, ax = plt.subplots(figsize=(fig_w, fig_h))
-
-    Point = tuple[float, float]
-    Ports = tuple[Point, Point]
-
-    # y grows downward in layout; flip the axis at the end.
-    def route(
-        pts: list[Point], *, arrow: bool = True, color: str = _WIRE, ls: str = "-"
-    ) -> None:
-        xs = [p[0] for p in pts]
-        ys = [p[1] for p in pts]
-        if len(pts) > 2:
-            ax.plot(xs[:-1], ys[:-1], color=color, lw=1.2, ls=ls, zorder=1)
-        if arrow:
-            patch = FancyArrowPatch(
-                pts[-2],
-                pts[-1],
-                arrowstyle="-|>",
-                mutation_scale=11,
-                color=color,
-                lw=1.2,
-                linestyle=ls,
-                shrinkA=0,
-                shrinkB=0,
-                zorder=1,
-            )
-            ax.add_patch(patch)
-        else:
-            ax.plot(xs[-2:], ys[-2:], color=color, lw=1.2, ls=ls, zorder=1)
-
-    def sum_node(x: float, y: float, color: str = _WIRE) -> None:
-        ax.add_patch(
-            Circle(
-                (x, y), _NODE_R, facecolor="white", edgecolor=color, lw=1.2, zorder=3
-            )
-        )
-        ax.text(
-            x,
-            y,
-            "+",
-            ha="center",
-            va="center_baseline",
-            fontsize=fontsize,
-            color=color,
-            zorder=4,
-        )
-
-    def pickoff(x: float, y: float, color: str = _WIRE) -> None:
-        ax.add_patch(Circle((x, y), 0.045, facecolor=color, edgecolor=color, zorder=3))
-
-    def xpos(x_left: float, w_total: float, off: float, w_el: float, d: int) -> float:
-        """Left edge of an element at offset `off` from the input side."""
-        return x_left + off if d > 0 else x_left + w_total - off - w_el
-
-    def draw(node: dict[str, Any], x: float, y_axis: float, d: int) -> Ports:
-        """Draw node with left edge x and flow axis y_axis, direction d (+1/-1).
-        Returns (p_in, p_out) port coordinates."""
-        ntype = node.get("type", "Leaf")
-        w, h, ay = node["_size"]
-        y_top = y_axis - ay
-
-        if ntype in ("Shell", "Series"):
-            children = (
-                node["_parts"] if ntype == "Shell" else node.get("children") or []
-            )
-            if not children:
-                p = ((x, y_axis), (x + w, y_axis))
-                return p if d > 0 else (p[1], p[0])
-            off = 0.0
-            ports = []
-            prev_out = None
-            for ch in children:
-                cw, _, _ = ch["_size"]
-                cx = xpos(x, w, off, cw, d)
-                p_in, p_out = draw(ch, cx, y_axis, d)
-                if prev_out is not None:
-                    route([prev_out, p_in])
-                ports.append((p_in, p_out))
-                prev_out = p_out
-                off += cw + _GAP_X
-            return ports[0][0], ports[-1][1]
-
-        if ntype == "Parallel":
-            stub = _STUB + 2 * _NODE_R
-            x_in = x if d > 0 else x + w
-            x_merge = x + w - _NODE_R if d > 0 else x + _NODE_R
-            sum_node(x_merge, y_axis)
-            pickoff(x_in, y_axis)
-            cy = y_top
-            for ch in node.get("children") or []:
-                cw, chh, cay = ch["_size"]
-                cx = xpos(x, w, stub + (max(0.0, w - 2 * stub - cw)) / 2, cw, d)
-                p_in, p_out = draw(ch, cx, cy + cay, d)
-                route([(x_in, y_axis), (x_in, p_in[1]), p_in])
-                route(
-                    [
-                        p_out,
-                        (x_merge, p_out[1]),
-                        (x_merge, y_axis - _NODE_R * _sign(y_axis - p_out[1])),
-                    ]
-                )
-                cy += chh + _GAP_Y
-            p = ((x_in, y_axis), (x_merge + _NODE_R * d, y_axis))
-            return p
-
-        if ntype == "Recursion":
-            # dashed container
-            ax.add_patch(
-                FancyBboxPatch(
-                    (x, y_top),
-                    w,
-                    h,
-                    boxstyle="round,pad=0.06",
-                    facecolor="none",
-                    edgecolor=_LOOP_EDGE,
-                    lw=1.0,
-                    ls=(0, (4, 3)),
-                    zorder=0,
-                )
-            )
-            nname = node.get("name", "")
-            if nname and not nname.isdigit():
-                ax.text(
-                    x + 0.12,
-                    y_top + 0.06,
-                    nname,
-                    ha="left",
-                    va="top",
-                    fontsize=fontsize - 1.5,
-                    color=_LOOP_EDGE,
-                    style="italic",
-                    zorder=3,
-                )
-            fF, fB = node.get("fF"), node.get("fB")
-            if fF is None:  # placeholder; matches the _measure() default size
-                fF = {
-                    "type": "Leaf",
-                    "name": "fF",
-                    "module": None,
-                    "_size": (1.0, _LEAF_H, _LEAF_H / 2),
-                }
-            wf, hf, _ = fF["_size"]
-            wb, hb, _ = fB["_size"] if fB else (0, 0, 0)
-            inner_w = max(wf, wb)
-            x_sum = xpos(x, w, _PAD + _NODE_R, 0, d)
-            x_pick = xpos(x, w, w - _PAD - _NODE_R, 0, d)
-            sum_node(x_sum, y_axis, color=_LOOP_EDGE)
-            pickoff(x_pick, y_axis)
-            fx = xpos(x, w, _PAD + 2 * _NODE_R + _GAP_X + (inner_w - wf) / 2, wf, d)
-            f_in, f_out = draw(fF, fx, y_axis, d)
-            route([(x_sum + _NODE_R * d, y_axis), f_in])
-            route([f_out, (x_pick, y_axis)], arrow=False)
-            if fB is not None:
-                yb_axis = y_top + _PAD + hf + _GAP_Y + fB["_size"][2]
-                bx = xpos(x, w, _PAD + 2 * _NODE_R + _GAP_X + (inner_w - wb) / 2, wb, d)
-                b_in, b_out = draw(fB, bx, yb_axis, -d)
-                route([(x_pick, y_axis), (x_pick, b_in[1]), b_in], color=_LOOP_EDGE)
-                route(
-                    [b_out, (x_sum, b_out[1]), (x_sum, y_axis + _NODE_R)],
-                    color=_LOOP_EDGE,
-                )
-            return (x_sum - _NODE_R * d, y_axis), (x_pick, y_axis)
-
-        # Leaf
-        mod = node.get("module")
-        face, edge = _BLOCK_COLORS[_block_category(type(mod).__name__)]
-        ax.add_patch(
-            FancyBboxPatch(
-                (x, y_top),
-                w,
-                h,
-                boxstyle="round,pad=0.02",
-                facecolor=face,
-                edgecolor=edge,
-                lw=1.2,
-                zorder=2,
-            )
-        )
-        ax.text(
-            x + w / 2,
-            y_axis,
-            _leaf_label(node),
-            ha="center",
-            va="center",
-            fontsize=fontsize,
-            zorder=3,
-            linespacing=1.3,
-        )
-        p = ((x, y_axis), (x + w, y_axis))
-        return p if d > 0 else (p[1], p[0])
-
-    def _sign(v: float) -> float:
-        return 1.0 if v >= 0 else -1.0
-
-    x0 = 1.0
-    p_in, p_out = draw(root, x0, root_ay, +1)
-    # model input / output stubs
-    route([(x0 - 0.8, p_in[1]), p_in])
-    route([p_out, (p_out[0] + 0.8, p_out[1])])
-    ax.text(x0 - 0.85, p_in[1], "in", ha="right", va="center", fontsize=fontsize)
-    ax.text(p_out[0] + 0.85, p_out[1], "out", ha="left", va="center", fontsize=fontsize)
-
-    ax.set_xlim(x0 - 1.6, x0 + total_w + 1.6)
-    ax.set_ylim(total_h + 0.4, -0.4)  # inverted: layout y grows downward
-    ax.set_aspect("equal")
-    ax.axis("off")
-    return ax
+    return plot_node_tree(root, ax=ax, scale=scale, fontsize=fontsize)