NIR Bridge API

sc_neurocore.nir_bridge

NIR integration for SC-NeuroCore.

Provides bidirectional conversion between NIR graphs and SC-NeuroCore networks.

>>> import nir
>>> from sc_neurocore.nir_bridge import from_nir
>>> graph = nir.read("model.nir")
>>> network = from_nir(graph, dt=1.0)
>>> network.run(inputs, steps=100)

from_nir(source, dt=1.0, reset_mode='reset')

Convert a NIR graph to an executable SC-NeuroCore network.

Parameters

source : nir.NIRGraph or str or Path NIR graph object, or path to a .nir file. dt : float Timestep for leaky integrator dynamics. reset_mode : str Spike reset mechanism: "reset" (v = v_reset, NIR spec default) or "subtract" (v = v - v_threshold, used by snnTorch).

Returns

SCNetwork Executable network with topologically sorted forward pass.

Source code in src/sc_neurocore/nir_bridge/parser.py

def from_nir(source, dt: float = 1.0, reset_mode: str = "reset") -> SCNetwork:
    """Convert a NIR graph to an executable SC-NeuroCore network.

    Parameters
    ----------
    source : nir.NIRGraph or str or Path
        NIR graph object, or path to a .nir file.
    dt : float
        Timestep for leaky integrator dynamics.
    reset_mode : str
        Spike reset mechanism: "reset" (v = v_reset, NIR spec default)
        or "subtract" (v = v - v_threshold, used by snnTorch).

    Returns
    -------
    SCNetwork
        Executable network with topologically sorted forward pass.
    """
    if isinstance(source, (str, Path)):
        graph = nir.read(str(source))
    elif isinstance(source, nir.NIRGraph):
        graph = source
    else:
        raise TypeError(f"Expected NIRGraph or path, got {type(source)}")

    return _parse_graph(graph, dt=dt, reset_mode=reset_mode)

to_nir(network, path=None)

Export an SC-NeuroCore SCNetwork to NIR format.

Parameters

network : SCNetwork The network to export. path : str or Path, optional If provided, write the NIR graph to this file.

Returns

nir.NIRGraph

Source code in src/sc_neurocore/nir_bridge/export.py

def to_nir(network, path: str | Path | None = None) -> nir.NIRGraph:
    """Export an SC-NeuroCore SCNetwork to NIR format.

    Parameters
    ----------
    network : SCNetwork
        The network to export.
    path : str or Path, optional
        If provided, write the NIR graph to this file.

    Returns
    -------
    nir.NIRGraph
    """
    from .parser import SCMultiPortSubgraphNode, SCNetwork, SCSubgraphNode, _UnitDelayNode

    if not isinstance(network, SCNetwork):
        raise TypeError(f"Expected SCNetwork, got {type(network)}")

    # Ensure topo_order has been computed (triggers delay insertion)
    _ = network.topo_order

    nodes = {}
    edges = list(network.edges)

    for name, node in network.nodes.items():
        # Skip internal delay nodes — reconstruct as direct recurrent edges
        if isinstance(node, _UnitDelayNode):
            continue
        # Recursively export subgraphs
        if isinstance(node, (SCSubgraphNode, SCMultiPortSubgraphNode)):
            nodes[name] = to_nir(node.network)
            continue
        nir_node = _node_to_nir(name, node)
        if nir_node is None:
            raise ValueError(f"Cannot export node {name!r} of type {type(node).__name__} to NIR")
        nodes[name] = nir_node

    # Replace delay edges with original recurrent edges
    clean_edges = []
    for src, dst in edges:
        if src.startswith("_delay_") and src in network._recurrent_map:
            # Restore original back edge: recurrent_source -> dst
            original_src = network._recurrent_map[src]
            clean_edges.append((original_src, dst))
        elif dst.startswith("_delay_"):
            # Skip the edge feeding INTO the delay node (it's implicit)
            continue
        else:
            clean_edges.append((src, dst))

    graph = nir.NIRGraph(nodes=nodes, edges=clean_edges)

    if path is not None:
        nir.write(str(path), graph)

    return graph

Parser

sc_neurocore.nir_bridge.parser

SCNetwork dataclass

Executable network parsed from a NIR graph.

Nodes are stored by name. Edges define the forward pass order. Calling run() feeds input through the graph for the given number of timesteps and returns the output node's accumulated result.

Recurrent edges (cycles) are automatically handled by inserting unit-delay nodes that feed from the previous timestep.

Source code in src/sc_neurocore/nir_bridge/parser.py

@dataclass
class SCNetwork:
    """Executable network parsed from a NIR graph.

    Nodes are stored by name. Edges define the forward pass order.
    Calling ``run()`` feeds input through the graph for the given
    number of timesteps and returns the output node's accumulated result.

    Recurrent edges (cycles) are automatically handled by inserting
    unit-delay nodes that feed from the previous timestep.
    """

    nodes: dict[str, Any] = field(default_factory=dict)
    edges: list[tuple[str, str]] = field(default_factory=list)
    input_nodes: list[str] = field(default_factory=list)
    output_nodes: list[str] = field(default_factory=list)
    _topo_order: list[str] | None = None
    # Maps delay_node_name → source_node_name for recurrent connections
    _recurrent_map: dict[str, str] = field(default_factory=dict)

    def _find_back_edges(self) -> list[tuple[str, str]]:
        """DFS-based back-edge detection."""
        WHITE, GRAY, BLACK = 0, 1, 2
        color: dict[str, int] = {n: WHITE for n in self.nodes}
        adj: dict[str, list[str]] = {n: [] for n in self.nodes}
        for src, dst in self.edges:
            adj[src].append(dst)

        back_edges: list[tuple[str, str]] = []

        def dfs(u: str) -> None:
            color[u] = GRAY
            for v in adj[u]:
                if v not in color:
                    continue
                if color[v] == GRAY:
                    back_edges.append((u, v))
                elif color[v] == WHITE:
                    dfs(v)
            color[u] = BLACK

        for n in self.nodes:
            if color[n] == WHITE:
                dfs(n)
        return back_edges

    def _break_cycles(self) -> None:
        """Replace back edges with unit-delay source nodes."""
        back_edges = self._find_back_edges()
        if not back_edges:
            return

        for src, dst in back_edges:
            delay_name = f"_delay_{src}_to_{dst}"
            self.edges.remove((src, dst))
            # Delay node is a DAG source (no incoming edges) — feeds dst
            self.nodes[delay_name] = _UnitDelayNode(name=delay_name)
            self.edges.append((delay_name, dst))
            self._recurrent_map[delay_name] = src

    def _topological_sort(self) -> list[str]:
        """Kahn's algorithm with automatic cycle breaking via delay nodes."""
        self._break_cycles()

        adj: dict[str, list[str]] = {n: [] for n in self.nodes}
        in_deg: dict[str, int] = {n: 0 for n in self.nodes}
        for src, dst in self.edges:
            adj[src].append(dst)
            in_deg[dst] = in_deg.get(dst, 0) + 1

        queue = [n for n, d in in_deg.items() if d == 0]
        order = []
        while queue:
            node = queue.pop(0)
            order.append(node)
            for nxt in adj[node]:
                in_deg[nxt] -= 1
                if in_deg[nxt] == 0:
                    queue.append(nxt)

        if len(order) != len(self.nodes):
            raise ValueError("NIR graph contains a cycle that cannot be broken by delay insertion")
        return order

    @property
    def topo_order(self) -> list[str]:
        if self._topo_order is None:
            self._topo_order = self._topological_sort()
        return self._topo_order

    def step(self, inputs: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
        """Execute one timestep through the graph.

        Parameters
        ----------
        inputs : dict mapping input node name → input array

        Returns
        -------
        dict mapping output node name → output array
        """
        values: dict[str, np.ndarray] = {}

        for name in self.topo_order:
            node = self.nodes[name]

            if name in self.input_nodes:
                x = inputs.get(name, np.array([0.0]))
                values[name] = node.forward(x)
            elif isinstance(node, _UnitDelayNode):
                # Delay nodes are sources — forward() returns buffered value
                values[name] = node.forward(np.array([0.0]))
            else:
                predecessors = [src for src, dst in self.edges if dst == name]
                if len(predecessors) == 1:
                    x = values[predecessors[0]]
                elif len(predecessors) > 1:
                    x = sum(values[p] for p in predecessors)
                else:
                    x = np.array([0.0])
                values[name] = node.forward(x)

        # Update delay buffers with this timestep's source values
        for delay_name, src_name in self._recurrent_map.items():
            if src_name in values:
                self.nodes[delay_name].update_buffer(values[src_name])

        return {name: values[name] for name in self.output_nodes if name in values}

    def run(self, inputs: dict[str, np.ndarray], steps: int = 100) -> dict[str, list[np.ndarray]]:
        """Run the network for multiple timesteps.

        Parameters
        ----------
        inputs : dict mapping input node name → input array (constant across steps)
        steps : number of timesteps

        Returns
        -------
        dict mapping output node name → list of output arrays per timestep
        """
        results: dict[str, list[np.ndarray]] = {n: [] for n in self.output_nodes}
        for _ in range(steps):
            out = self.step(inputs)
            for name, val in out.items():
                results[name].append(val.copy())
        return results

    def reset(self):
        """Reset all stateful nodes."""
        for node in self.nodes.values():
            if hasattr(node, "reset"):
                node.reset()

    def summary(self) -> str:
        """Human-readable network summary."""
        lines = [f"SCNetwork: {len(self.nodes)} nodes, {len(self.edges)} edges"]
        for name in self.topo_order:
            node = self.nodes[name]
            lines.append(f"  {name}: {type(node).__name__}")
        if self._recurrent_map:
            lines.append(f"  recurrent: {list(self._recurrent_map.values())}")
        lines.append(f"  inputs: {self.input_nodes}")
        lines.append(f"  outputs: {self.output_nodes}")
        return "\n".join(lines)

step(inputs)

Execute one timestep through the graph.

Parameters

inputs : dict mapping input node name → input array

Returns

dict mapping output node name → output array

Source code in src/sc_neurocore/nir_bridge/parser.py

def step(self, inputs: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
    """Execute one timestep through the graph.

    Parameters
    ----------
    inputs : dict mapping input node name → input array

    Returns
    -------
    dict mapping output node name → output array
    """
    values: dict[str, np.ndarray] = {}

    for name in self.topo_order:
        node = self.nodes[name]

        if name in self.input_nodes:
            x = inputs.get(name, np.array([0.0]))
            values[name] = node.forward(x)
        elif isinstance(node, _UnitDelayNode):
            # Delay nodes are sources — forward() returns buffered value
            values[name] = node.forward(np.array([0.0]))
        else:
            predecessors = [src for src, dst in self.edges if dst == name]
            if len(predecessors) == 1:
                x = values[predecessors[0]]
            elif len(predecessors) > 1:
                x = sum(values[p] for p in predecessors)
            else:
                x = np.array([0.0])
            values[name] = node.forward(x)

    # Update delay buffers with this timestep's source values
    for delay_name, src_name in self._recurrent_map.items():
        if src_name in values:
            self.nodes[delay_name].update_buffer(values[src_name])

    return {name: values[name] for name in self.output_nodes if name in values}

run(inputs, steps=100)

Run the network for multiple timesteps.

Parameters

inputs : dict mapping input node name → input array (constant across steps) steps : number of timesteps

Returns

dict mapping output node name → list of output arrays per timestep

Source code in src/sc_neurocore/nir_bridge/parser.py

def run(self, inputs: dict[str, np.ndarray], steps: int = 100) -> dict[str, list[np.ndarray]]:
    """Run the network for multiple timesteps.

    Parameters
    ----------
    inputs : dict mapping input node name → input array (constant across steps)
    steps : number of timesteps

    Returns
    -------
    dict mapping output node name → list of output arrays per timestep
    """
    results: dict[str, list[np.ndarray]] = {n: [] for n in self.output_nodes}
    for _ in range(steps):
        out = self.step(inputs)
        for name, val in out.items():
            results[name].append(val.copy())
    return results

reset()

Reset all stateful nodes.

Source code in src/sc_neurocore/nir_bridge/parser.py

def reset(self):
    """Reset all stateful nodes."""
    for node in self.nodes.values():
        if hasattr(node, "reset"):
            node.reset()

summary()

Human-readable network summary.

Source code in src/sc_neurocore/nir_bridge/parser.py

def summary(self) -> str:
    """Human-readable network summary."""
    lines = [f"SCNetwork: {len(self.nodes)} nodes, {len(self.edges)} edges"]
    for name in self.topo_order:
        node = self.nodes[name]
        lines.append(f"  {name}: {type(node).__name__}")
    if self._recurrent_map:
        lines.append(f"  recurrent: {list(self._recurrent_map.values())}")
    lines.append(f"  inputs: {self.input_nodes}")
    lines.append(f"  outputs: {self.output_nodes}")
    return "\n".join(lines)

SCSubgraphNode dataclass

Executable wrapper for a nested NIR subgraph (single I/O port).

Source code in src/sc_neurocore/nir_bridge/parser.py

@dataclass
class SCSubgraphNode:
    """Executable wrapper for a nested NIR subgraph (single I/O port)."""

    name: str
    network: SCNetwork

    def __post_init__(self):
        if len(self.network.input_nodes) != 1 or len(self.network.output_nodes) != 1:
            raise ValueError("Nested NIRGraph nodes must expose exactly one input and one output")

    def forward(self, x: np.ndarray) -> np.ndarray:
        outputs = self.network.step({self.network.input_nodes[0]: np.atleast_1d(np.asarray(x))})
        return outputs[self.network.output_nodes[0]]

    def reset(self):
        self.network.reset()

SCMultiPortSubgraphNode dataclass

Executable wrapper for a nested NIR subgraph with multiple I/O ports.

Supports modular architectures where subgraphs expose multiple named inputs and outputs (e.g., encoder-decoder, skip connections).

Source code in src/sc_neurocore/nir_bridge/parser.py

@dataclass
class SCMultiPortSubgraphNode:
    """Executable wrapper for a nested NIR subgraph with multiple I/O ports.

    Supports modular architectures where subgraphs expose multiple named
    inputs and outputs (e.g., encoder-decoder, skip connections).
    """

    name: str
    network: SCNetwork

    def __post_init__(self):
        if not self.network.input_nodes or not self.network.output_nodes:
            raise ValueError("Multi-port subgraph must have at least one input and one output")

    @property
    def input_ports(self) -> list[str]:
        return self.network.input_nodes

    @property
    def output_ports(self) -> list[str]:
        return self.network.output_nodes

    def forward(self, x: np.ndarray) -> np.ndarray:
        """Single-input convenience: feeds x to first input, returns first output."""
        inputs = {self.network.input_nodes[0]: np.atleast_1d(np.asarray(x))}
        outputs = self.network.step(inputs)
        return outputs[self.network.output_nodes[0]]

    def forward_multi(self, inputs: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
        """Multi-port forward: provide named inputs, get named outputs."""
        return self.network.step(inputs)

    def reset(self):
        self.network.reset()

forward(x)

Single-input convenience: feeds x to first input, returns first output.

Source code in src/sc_neurocore/nir_bridge/parser.py

def forward(self, x: np.ndarray) -> np.ndarray:
    """Single-input convenience: feeds x to first input, returns first output."""
    inputs = {self.network.input_nodes[0]: np.atleast_1d(np.asarray(x))}
    outputs = self.network.step(inputs)
    return outputs[self.network.output_nodes[0]]

forward_multi(inputs)

Multi-port forward: provide named inputs, get named outputs.

Source code in src/sc_neurocore/nir_bridge/parser.py

def forward_multi(self, inputs: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
    """Multi-port forward: provide named inputs, get named outputs."""
    return self.network.step(inputs)

from_nir(source, dt=1.0, reset_mode='reset')

Convert a NIR graph to an executable SC-NeuroCore network.

Parameters

source : nir.NIRGraph or str or Path NIR graph object, or path to a .nir file. dt : float Timestep for leaky integrator dynamics. reset_mode : str Spike reset mechanism: "reset" (v = v_reset, NIR spec default) or "subtract" (v = v - v_threshold, used by snnTorch).

Returns

SCNetwork Executable network with topologically sorted forward pass.

Source code in src/sc_neurocore/nir_bridge/parser.py

def from_nir(source, dt: float = 1.0, reset_mode: str = "reset") -> SCNetwork:
    """Convert a NIR graph to an executable SC-NeuroCore network.

    Parameters
    ----------
    source : nir.NIRGraph or str or Path
        NIR graph object, or path to a .nir file.
    dt : float
        Timestep for leaky integrator dynamics.
    reset_mode : str
        Spike reset mechanism: "reset" (v = v_reset, NIR spec default)
        or "subtract" (v = v - v_threshold, used by snnTorch).

    Returns
    -------
    SCNetwork
        Executable network with topologically sorted forward pass.
    """
    if isinstance(source, (str, Path)):
        graph = nir.read(str(source))
    elif isinstance(source, nir.NIRGraph):
        graph = source
    else:
        raise TypeError(f"Expected NIRGraph or path, got {type(source)}")

    return _parse_graph(graph, dt=dt, reset_mode=reset_mode)

Recurrent Edge Handling

Graphs with cycles (feedback connections) are automatically handled by inserting unit-delay nodes on back edges. The delay node buffers the previous timestep's value, breaking algebraic loops while preserving temporal dynamics. See _UnitDelayNode.

Multi-Port Subgraphs

Nested NIR graphs with multiple inputs/outputs use SCMultiPortSubgraphNode, which exposes forward_multi(inputs_dict) → outputs_dict for named I/O ports.

Node Map

sc_neurocore.nir_bridge.node_map

NODE_MAP = {nir.Input: lambda name, node, **kw: SCInputNode(name=name, shape=(tuple((int(x)) for x in (next(iter(node.input_type.values())).flatten())) if node.input_type else ())), nir.Output: lambda name, node, **kw: SCOutputNode(name=name, shape=(tuple((int(x)) for x in (next(iter(node.output_type.values())).flatten())) if node.output_type else ())), nir.LIF: lambda name, node, **kw: SCLIFNode.from_nir(name, node, dt=(kw.get('dt', 1.0)), reset_mode=(kw.get('reset_mode', 'reset'))), nir.IF: lambda name, node, **kw: SCIFNode.from_nir(name, node, dt=(kw.get('dt', 1.0)), reset_mode=(kw.get('reset_mode', 'reset'))), nir.LI: lambda name, node, **kw: SCLINode.from_nir(name, node, dt=(kw.get('dt', 1.0))), nir.I: lambda name, node, **kw: SCIntegratorNode.from_nir(name, node, dt=(kw.get('dt', 1.0))), nir.Affine: lambda name, node, **kw: SCAffineNode.from_nir(name, node), nir.Linear: lambda name, node, **kw: SCLinearNode.from_nir(name, node), nir.Scale: lambda name, node, **kw: SCScaleNode.from_nir(name, node), nir.Threshold: lambda name, node, **kw: SCThresholdNode.from_nir(name, node), nir.Flatten: lambda name, node, **kw: SCFlattenNode.from_nir(name, node), nir.Delay: lambda name, node, **kw: SCDelayNode.from_nir(name, node, dt=(kw.get('dt', 1.0))), nir.CubaLIF: lambda name, node, **kw: SCCubaLIFNode.from_nir(name, node, dt=(kw.get('dt', 1.0)), reset_mode=(kw.get('reset_mode', 'reset'))), nir.CubaLI: lambda name, node, **kw: SCCubaLINode.from_nir(name, node, dt=(kw.get('dt', 1.0))), nir.SumPool2d: lambda name, node, **kw: SCSumPool2dNode.from_nir(name, node), nir.AvgPool2d: lambda name, node, **kw: SCAvgPool2dNode.from_nir(name, node), nir.Conv1d: lambda name, node, **kw: SCConv1dNode.from_nir(name, node), nir.Conv2d: lambda name, node, **kw: SCConv2dNode.from_nir(name, node)} module-attribute

SCLIFNode dataclass

LIF neuron mapped from NIR LIF primitive.

NIR LIF: taudv/dt = (v_leak - v) + RI, spike when v > v_threshold Euler: v += ((v_leak - v) + R*I) * dt/tau

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCLIFNode:
    """LIF neuron mapped from NIR LIF primitive.

    NIR LIF: tau*dv/dt = (v_leak - v) + R*I, spike when v > v_threshold
    Euler: v += ((v_leak - v) + R*I) * dt/tau
    """

    name: str
    n_neurons: int
    tau: np.ndarray
    r: np.ndarray
    v_leak: np.ndarray
    v_threshold: np.ndarray
    v_reset: np.ndarray
    v: np.ndarray | None = None
    dt: float = 1.0
    reset_mode: str = "reset"

    @classmethod
    def from_nir(
        cls,
        name: str,
        node: nir.LIF,
        dt: float = 1.0,
        reset_mode: str = "reset",
    ) -> SCLIFNode:
        tau = np.atleast_1d(node.tau).flatten()
        r = np.atleast_1d(node.r).flatten()
        v_leak = np.atleast_1d(node.v_leak).flatten()
        v_threshold = np.atleast_1d(node.v_threshold).flatten()
        v_reset = (
            np.atleast_1d(node.v_reset).flatten()
            if node.v_reset is not None
            else np.zeros_like(v_threshold)
        )
        return cls(
            name=name,
            n_neurons=len(tau),
            tau=tau,
            r=r,
            v_leak=v_leak,
            v_threshold=v_threshold,
            v_reset=v_reset,
            dt=dt,
            reset_mode=reset_mode,
        )

    def __post_init__(self):
        if self.v is None:
            self.v = self.v_leak.copy()

    def _broadcast_to(self, size: int):
        self.n_neurons = size
        for attr in ("tau", "r", "v_leak", "v_threshold", "v_reset"):
            arr = getattr(self, attr)
            if len(arr) == 1 and size > 1:
                setattr(self, attr, np.broadcast_to(arr, (size,)).copy())
        self.v = np.broadcast_to(self.v, (size,)).copy()

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        if self.n_neurons == 1 and len(x) > 1:
            self._broadcast_to(len(x))
        x = x[: self.n_neurons]
        dv = (self.v_leak - self.v + self.r * x) * (self.dt / self.tau)
        self.v += dv
        spikes = (self.v > self.v_threshold).astype(np.float64)
        if self.reset_mode == "subtract":
            self.v = np.where(spikes > 0, self.v - self.v_threshold, self.v)
        else:
            self.v = np.where(spikes > 0, self.v_reset, self.v)
        return spikes

    def reset(self):
        self.v = self.v_leak.copy()

SCIFNode dataclass

IF neuron — integrator with threshold, no leak.

NIR IF: dv/dt = RI, spike when v > v_threshold Euler: v += RI*dt

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCIFNode:
    """IF neuron — integrator with threshold, no leak.

    NIR IF: dv/dt = R*I, spike when v > v_threshold
    Euler: v += R*I*dt
    """

    name: str
    n_neurons: int
    r: np.ndarray
    v_threshold: np.ndarray
    v_reset: np.ndarray
    v: np.ndarray | None = None
    dt: float = 1.0
    reset_mode: str = "reset"

    @classmethod
    def from_nir(
        cls,
        name: str,
        node: nir.IF,
        dt: float = 1.0,
        reset_mode: str = "reset",
    ) -> SCIFNode:
        r = np.atleast_1d(node.r).flatten()
        v_threshold = np.atleast_1d(node.v_threshold).flatten()
        v_reset = (
            np.atleast_1d(node.v_reset).flatten() if node.v_reset is not None else np.zeros_like(r)
        )
        return cls(
            name=name,
            n_neurons=len(r),
            r=r,
            v_threshold=v_threshold,
            v_reset=v_reset,
            dt=dt,
            reset_mode=reset_mode,
        )

    def __post_init__(self):
        if self.v is None:
            self.v = np.zeros(self.n_neurons)

    def _broadcast_to(self, size: int):
        self.n_neurons = size
        for attr in ("r", "v_threshold", "v_reset"):
            arr = getattr(self, attr)
            if len(arr) == 1 and size > 1:
                setattr(self, attr, np.broadcast_to(arr, (size,)).copy())
        self.v = np.zeros(size)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        if self.n_neurons == 1 and len(x) > 1:
            self._broadcast_to(len(x))
        x = x[: self.n_neurons]
        self.v += self.r * x * self.dt
        spikes = (self.v > self.v_threshold).astype(np.float64)
        if self.reset_mode == "subtract":
            self.v = np.where(spikes > 0, self.v - self.v_threshold, self.v)
        else:
            self.v = np.where(spikes > 0, self.v_reset, self.v)
        return spikes

    def reset(self):
        self.v = np.zeros(self.n_neurons)

SCLINode dataclass

Leaky integrator — LIF without threshold.

NIR LI: taudv/dt = (v_leak - v) + RI

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCLINode:
    """Leaky integrator — LIF without threshold.

    NIR LI: tau*dv/dt = (v_leak - v) + R*I
    """

    name: str
    n_neurons: int
    tau: np.ndarray
    r: np.ndarray
    v_leak: np.ndarray
    v: np.ndarray | None = None
    dt: float = 1.0

    @classmethod
    def from_nir(cls, name: str, node: nir.LI, dt: float = 1.0) -> SCLINode:
        tau = np.atleast_1d(node.tau).flatten()
        r = np.atleast_1d(node.r).flatten()
        v_leak = np.atleast_1d(node.v_leak).flatten()
        return cls(name=name, n_neurons=len(tau), tau=tau, r=r, v_leak=v_leak, dt=dt)

    def __post_init__(self):
        if self.v is None:
            self.v = self.v_leak.copy()

    def _broadcast_to(self, size: int):
        self.n_neurons = size
        for attr in ("tau", "r", "v_leak"):
            arr = getattr(self, attr)
            if len(arr) == 1 and size > 1:
                setattr(self, attr, np.broadcast_to(arr, (size,)).copy())
        self.v = np.broadcast_to(self.v, (size,)).copy()

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        if self.n_neurons == 1 and len(x) > 1:
            self._broadcast_to(len(x))
        x = x[: self.n_neurons]
        dv = (self.v_leak - self.v + self.r * x) * (self.dt / self.tau)
        self.v += dv
        return self.v.copy()

    def reset(self):
        self.v = self.v_leak.copy()

SCIntegratorNode dataclass

Pure integrator: dv/dt = RI (no leak, no threshold). Euler: v += RI*dt

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCIntegratorNode:
    """Pure integrator: dv/dt = R*I (no leak, no threshold). Euler: v += R*I*dt"""

    name: str
    r: np.ndarray
    v: np.ndarray | None = None
    dt: float = 1.0

    @classmethod
    def from_nir(cls, name: str, node: nir.I, dt: float = 1.0) -> SCIntegratorNode:
        r = np.atleast_1d(node.r).flatten()
        return cls(name=name, r=r, dt=dt)

    def __post_init__(self):
        if self.v is None:
            self.v = np.zeros_like(self.r)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()[: len(self.r)]
        self.v += self.r * x * self.dt
        return self.v.copy()

    def reset(self):
        self.v = np.zeros_like(self.r)

SCAffineNode dataclass

Dense linear transform with bias: y = Wx + b

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCAffineNode:
    """Dense linear transform with bias: y = Wx + b"""

    name: str
    weight: np.ndarray
    bias: np.ndarray

    @classmethod
    def from_nir(cls, name: str, node: nir.Affine) -> SCAffineNode:
        return cls(name=name, weight=node.weight, bias=node.bias)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        return self.weight @ x + self.bias

SCLinearNode dataclass

Matrix multiply without bias: y = Wx

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCLinearNode:
    """Matrix multiply without bias: y = Wx"""

    name: str
    weight: np.ndarray

    @classmethod
    def from_nir(cls, name: str, node: nir.Linear) -> SCLinearNode:
        return cls(name=name, weight=node.weight)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        return self.weight @ x

SCScaleNode dataclass

Element-wise scaling: y = s * x

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCScaleNode:
    """Element-wise scaling: y = s * x"""

    name: str
    scale: np.ndarray

    @classmethod
    def from_nir(cls, name: str, node: nir.Scale) -> SCScaleNode:
        return cls(name=name, scale=node.scale)

    def forward(self, x: np.ndarray) -> np.ndarray:
        return self.scale * x

SCThresholdNode dataclass

Spike threshold: y = 1 if x >= threshold else 0

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCThresholdNode:
    """Spike threshold: y = 1 if x >= threshold else 0"""

    name: str
    threshold: np.ndarray

    @classmethod
    def from_nir(cls, name: str, node: nir.Threshold) -> SCThresholdNode:
        return cls(name=name, threshold=node.threshold)

    def forward(self, x: np.ndarray) -> np.ndarray:
        return (x >= self.threshold).astype(np.float64)

SCFlattenNode dataclass

Reshape tensor — flatten dimensions.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCFlattenNode:
    """Reshape tensor — flatten dimensions."""

    name: str
    start_dim: int
    end_dim: int

    @classmethod
    def from_nir(cls, name: str, node: nir.Flatten) -> SCFlattenNode:
        return cls(name=name, start_dim=node.start_dim, end_dim=node.end_dim)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.asarray(x)
        if x.ndim == 0:
            if self.start_dim not in (0, -1) or self.end_dim not in (0, -1):
                raise ValueError(
                    f"Invalid flatten dims {self.start_dim}:{self.end_dim} for shape {x.shape}"
                )
            return x.reshape(1)

        start = self.start_dim if self.start_dim >= 0 else x.ndim + self.start_dim
        end = self.end_dim if self.end_dim >= 0 else x.ndim + self.end_dim
        if not 0 <= start < x.ndim or not 0 <= end < x.ndim or start > end:
            raise ValueError(
                f"Invalid flatten dims {self.start_dim}:{self.end_dim} for shape {x.shape}"
            )
        if start == end:
            return x.copy()

        merged = int(np.prod(x.shape[start : end + 1], dtype=np.int64))
        new_shape = x.shape[:start] + (merged,) + x.shape[end + 1 :]
        return x.reshape(new_shape)

SCInputNode dataclass

Graph entry point — passes input through unchanged.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCInputNode:
    """Graph entry point — passes input through unchanged."""

    name: str
    shape: tuple

    def forward(self, x: np.ndarray) -> np.ndarray:
        return x

SCOutputNode dataclass

Graph exit point — collects output.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCOutputNode:
    """Graph exit point — collects output."""

    name: str
    shape: tuple
    last_output: np.ndarray | None = None

    def forward(self, x: np.ndarray) -> np.ndarray:
        self.last_output = x
        return x

SCDelayNode dataclass

Temporal delay: output = input(t - delay).

NIR Delay: I(t - tau). Implemented as a circular buffer per element. Delay values are rounded to integer timesteps.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCDelayNode:
    """Temporal delay: output = input(t - delay).

    NIR Delay: I(t - tau). Implemented as a circular buffer per element.
    Delay values are rounded to integer timesteps.
    """

    name: str
    delay_steps: np.ndarray
    delay_time: np.ndarray | None = None  # original physical time for lossless export
    _buffers: list[list[np.ndarray]] | None = None

    @classmethod
    def from_nir(cls, name: str, node: nir.Delay, dt: float = 1.0) -> SCDelayNode:
        delay = np.atleast_1d(node.delay).flatten()
        steps = np.round(delay / dt).astype(int)
        return cls(name=name, delay_steps=steps, delay_time=delay.copy())

    def __post_init__(self):
        if self._buffers is None:
            self._buffers = [[np.zeros(1) for _ in range(int(d))] for d in self.delay_steps]

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        out = np.zeros(len(self.delay_steps))
        for i, buf in enumerate(self._buffers):
            xi = float(x[i]) if i < len(x) else 0.0
            if len(buf) == 0:
                out[i] = xi  # zero-delay passthrough
            else:
                out[i] = buf[0][0]
                buf.append(np.array([xi]))
                buf.pop(0)
        return out

    def reset(self):
        self._buffers = [[np.zeros(1) for _ in range(int(d))] for d in self.delay_steps]

SCCubaLIFNode dataclass

Current-based LIF with synaptic filter.

tau_syn * dI_syn/dt = -I_syn + w_in * I

tau_mem * dv/dt = (v_leak - v) + R * I_syn spike when v > v_threshold, reset to v_reset

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCCubaLIFNode:
    """Current-based LIF with synaptic filter.

    NIR CubaLIF: tau_syn * dI_syn/dt = -I_syn + w_in * I
                 tau_mem * dv/dt = (v_leak - v) + R * I_syn
                 spike when v > v_threshold, reset to v_reset
    """

    name: str
    n_neurons: int
    tau_syn: np.ndarray
    tau_mem: np.ndarray
    r: np.ndarray
    v_leak: np.ndarray
    v_threshold: np.ndarray
    v_reset: np.ndarray
    w_in: np.ndarray
    v: np.ndarray | None = None
    i_syn: np.ndarray | None = None
    dt: float = 1.0
    reset_mode: str = "reset"

    @classmethod
    def from_nir(
        cls,
        name: str,
        node: nir.CubaLIF,
        dt: float = 1.0,
        reset_mode: str = "reset",
    ) -> SCCubaLIFNode:
        tau_syn = np.atleast_1d(node.tau_syn).flatten()
        tau_mem = np.atleast_1d(node.tau_mem).flatten()
        r = np.atleast_1d(node.r).flatten()
        v_leak = np.atleast_1d(node.v_leak).flatten()
        v_threshold = np.atleast_1d(node.v_threshold).flatten()
        v_reset = (
            np.atleast_1d(node.v_reset).flatten()
            if node.v_reset is not None
            else np.zeros_like(v_threshold)
        )
        w_in = np.atleast_1d(node.w_in).flatten()
        return cls(
            name=name,
            n_neurons=len(tau_mem),
            tau_syn=tau_syn,
            tau_mem=tau_mem,
            r=r,
            v_leak=v_leak,
            v_threshold=v_threshold,
            v_reset=v_reset,
            w_in=w_in,
            dt=dt,
            reset_mode=reset_mode,
        )

    def __post_init__(self):
        if self.v is None:
            self.v = self.v_leak.copy()
        if self.i_syn is None:
            self.i_syn = np.zeros(self.n_neurons)

    def _broadcast_to(self, size: int):
        """Broadcast scalar params to match actual input size."""
        self.n_neurons = size
        for attr in ("tau_syn", "tau_mem", "r", "v_leak", "v_threshold", "v_reset", "w_in"):
            arr = getattr(self, attr)
            if len(arr) == 1 and size > 1:
                setattr(self, attr, np.broadcast_to(arr, (size,)).copy())
        self.v = np.broadcast_to(self.v, (size,)).copy()
        self.i_syn = np.zeros(size)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        if self.n_neurons == 1 and len(x) > 1:
            self._broadcast_to(len(x))
        x = x[: self.n_neurons]
        di = (-self.i_syn + self.w_in * x) * (self.dt / self.tau_syn)
        self.i_syn += di
        dv = (self.v_leak - self.v + self.r * self.i_syn) * (self.dt / self.tau_mem)
        self.v += dv
        spikes = (self.v > self.v_threshold).astype(np.float64)
        if self.reset_mode == "subtract":
            self.v = np.where(spikes > 0, self.v - self.v_threshold, self.v)
        else:
            self.v = np.where(spikes > 0, self.v_reset, self.v)
        return spikes

    def reset(self):
        self.v = self.v_leak.copy()
        self.i_syn = np.zeros(self.n_neurons)

SCCubaLINode dataclass

Current-based leaky integrator (CubaLIF without threshold).

tau_syn * dI_syn/dt = -I_syn + w_in * I

tau_mem * dv/dt = (v_leak - v) + R * I_syn

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCCubaLINode:
    """Current-based leaky integrator (CubaLIF without threshold).

    NIR CubaLI: tau_syn * dI_syn/dt = -I_syn + w_in * I
                tau_mem * dv/dt = (v_leak - v) + R * I_syn
    """

    name: str
    n_neurons: int
    tau_syn: np.ndarray
    tau_mem: np.ndarray
    r: np.ndarray
    v_leak: np.ndarray
    w_in: np.ndarray
    v: np.ndarray | None = None
    i_syn: np.ndarray | None = None
    dt: float = 1.0

    @classmethod
    def from_nir(cls, name: str, node: nir.CubaLI, dt: float = 1.0) -> SCCubaLINode:
        tau_syn = np.atleast_1d(node.tau_syn).flatten()
        tau_mem = np.atleast_1d(node.tau_mem).flatten()
        r = np.atleast_1d(node.r).flatten()
        v_leak = np.atleast_1d(node.v_leak).flatten()
        w_in = np.atleast_1d(node.w_in).flatten()
        return cls(
            name=name,
            n_neurons=len(tau_mem),
            tau_syn=tau_syn,
            tau_mem=tau_mem,
            r=r,
            v_leak=v_leak,
            w_in=w_in,
            dt=dt,
        )

    def __post_init__(self):
        if self.v is None:
            self.v = self.v_leak.copy()
        if self.i_syn is None:
            self.i_syn = np.zeros(self.n_neurons)

    def _broadcast_to(self, size: int):
        self.n_neurons = size
        for attr in ("tau_syn", "tau_mem", "r", "v_leak", "w_in"):
            arr = getattr(self, attr)
            if len(arr) == 1 and size > 1:
                setattr(self, attr, np.broadcast_to(arr, (size,)).copy())
        self.v = np.broadcast_to(self.v, (size,)).copy()
        self.i_syn = np.zeros(size)

    def forward(self, x: np.ndarray) -> np.ndarray:
        x = np.atleast_1d(x).flatten()
        if self.n_neurons == 1 and len(x) > 1:
            self._broadcast_to(len(x))
        x = x[: self.n_neurons]
        di = (-self.i_syn + self.w_in * x) * (self.dt / self.tau_syn)
        self.i_syn += di
        dv = (self.v_leak - self.v + self.r * self.i_syn) * (self.dt / self.tau_mem)
        self.v += dv
        return self.v.copy()

    def reset(self):
        self.v = self.v_leak.copy()
        self.i_syn = np.zeros(self.n_neurons)

SCConv1dNode dataclass

1D convolution: y = conv1d(x, weight) + bias.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCConv1dNode:
    """1D convolution: y = conv1d(x, weight) + bias."""

    name: str
    weight: np.ndarray
    bias: np.ndarray
    stride: int
    padding: int
    dilation: int
    groups: int
    input_shape: int | None = None

    @classmethod
    def from_nir(cls, name: str, node: nir.Conv1d) -> SCConv1dNode:
        if isinstance(node.padding, str):
            raise NotImplementedError(
                f"String padding '{node.padding}' not supported; use integer padding"
            )
        padding = int(node.padding)
        return cls(
            name=name,
            weight=node.weight,
            bias=node.bias if node.bias is not None else np.zeros(node.weight.shape[0]),
            stride=node.stride,
            padding=padding,
            dilation=node.dilation,
            groups=node.groups,
            input_shape=getattr(node, "input_shape", None),
        )

    def forward(self, x: np.ndarray) -> np.ndarray:
        # x: (C_in, L) or (L,)
        if x.ndim == 1:
            x = x[np.newaxis, :]
        c_out, c_in_per_group, k = self.weight.shape
        c_in, length = x.shape
        if self.padding > 0:
            x = np.pad(x, ((0, 0), (self.padding, self.padding)), mode="constant")
            length = x.shape[1]
        out_len = (length - self.dilation * (k - 1) - 1) // self.stride + 1
        out = np.zeros((c_out, out_len))
        for o in range(c_out):
            g = o // (c_out // self.groups)
            c_start = g * c_in_per_group
            for l in range(out_len):
                val = 0.0
                for ci in range(c_in_per_group):
                    for ki in range(k):
                        idx = l * self.stride + ki * self.dilation
                        if 0 <= idx < x.shape[1]:
                            val += self.weight[o, ci, ki] * x[c_start + ci, idx]
                out[o, l] = val + self.bias[o]
        return out.squeeze()

SCConv2dNode dataclass

2D convolution: y = conv2d(x, weight) + bias.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCConv2dNode:
    """2D convolution: y = conv2d(x, weight) + bias."""

    name: str
    weight: np.ndarray
    bias: np.ndarray
    stride: tuple[int, int]
    padding: tuple[int, int]
    dilation: tuple[int, int]
    groups: int
    input_shape: tuple[int, int] | None = None

    @classmethod
    def from_nir(cls, name: str, node: nir.Conv2d) -> SCConv2dNode:
        stride = node.stride if isinstance(node.stride, tuple) else (node.stride, node.stride)
        padding = node.padding if isinstance(node.padding, tuple) else (node.padding, node.padding)
        if isinstance(padding[0], str):
            raise NotImplementedError(
                f"String padding '{padding[0]}' not supported; use integer padding"
            )
        dilation = (
            node.dilation if isinstance(node.dilation, tuple) else (node.dilation, node.dilation)
        )
        return cls(
            name=name,
            weight=node.weight,
            bias=node.bias if node.bias is not None else np.zeros(node.weight.shape[0]),
            stride=stride,
            padding=padding,
            dilation=dilation,
            groups=node.groups,
            input_shape=getattr(node, "input_shape", None),
        )

    def forward(self, x: np.ndarray) -> np.ndarray:
        # x: (C_in, H, W) or (H, W)
        if x.ndim == 2:
            x = x[np.newaxis, :, :]
        c_out, c_in_per_group, kh, kw = self.weight.shape
        c_in, h, w = x.shape
        ph, pw = self.padding
        if ph > 0 or pw > 0:
            x = np.pad(x, ((0, 0), (ph, ph), (pw, pw)), mode="constant")
            h, w = x.shape[1], x.shape[2]
        sh, sw = self.stride
        dh, dw = self.dilation
        oh = (h - dh * (kh - 1) - 1) // sh + 1
        ow = (w - dw * (kw - 1) - 1) // sw + 1
        out = np.zeros((c_out, oh, ow))
        for o in range(c_out):
            g = o // (c_out // self.groups)
            c_start = g * c_in_per_group
            for i in range(oh):
                for j in range(ow):
                    val = 0.0
                    for ci in range(c_in_per_group):
                        for ki in range(kh):
                            for kj in range(kw):
                                ii = i * sh + ki * dh
                                jj = j * sw + kj * dw
                                if 0 <= ii < h and 0 <= jj < w:
                                    val += self.weight[o, ci, ki, kj] * x[c_start + ci, ii, jj]
                    out[o, i, j] = val + self.bias[o]
        return out.squeeze()

SCSumPool2dNode dataclass

2D sum pooling: sum over spatial kernel windows.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCSumPool2dNode:
    """2D sum pooling: sum over spatial kernel windows."""

    name: str
    kernel_size: tuple[int, int]
    stride: tuple[int, int]
    padding: tuple[int, int]

    @classmethod
    def from_nir(cls, name: str, node: nir.SumPool2d) -> SCSumPool2dNode:
        ks = tuple(int(x) for x in np.atleast_1d(node.kernel_size).flatten()[:2])
        st = tuple(int(x) for x in np.atleast_1d(node.stride).flatten()[:2])
        pad = tuple(int(x) for x in np.atleast_1d(node.padding).flatten()[:2])
        if len(ks) == 1:
            ks = (ks[0], ks[0])
        if len(st) == 1:
            st = (st[0], st[0])
        if len(pad) == 1:
            pad = (pad[0], pad[0])
        return cls(name=name, kernel_size=ks, stride=st, padding=pad)

    def forward(self, x: np.ndarray) -> np.ndarray:
        if x.ndim < 2:
            return x
        # Expect (C, H, W) or (H, W)
        if x.ndim == 2:
            x = x[np.newaxis, :, :]
        c, h, w = x.shape
        ph, pw = self.padding
        if ph > 0 or pw > 0:
            x = np.pad(x, ((0, 0), (ph, ph), (pw, pw)), mode="constant")
            h, w = x.shape[1], x.shape[2]
        kh, kw = self.kernel_size
        sh, sw = self.stride
        oh = (h - kh) // sh + 1
        ow = (w - kw) // sw + 1
        out = np.zeros((c, oh, ow))
        for i in range(oh):
            for j in range(ow):
                out[:, i, j] = x[:, i * sh : i * sh + kh, j * sw : j * sw + kw].sum(axis=(1, 2))
        return out.squeeze()

SCAvgPool2dNode dataclass

2D average pooling: SumPool / kernel_area.

Source code in src/sc_neurocore/nir_bridge/node_map.py

@dataclass
class SCAvgPool2dNode:
    """2D average pooling: SumPool / kernel_area."""

    name: str
    kernel_size: tuple[int, int]
    stride: tuple[int, int]
    padding: tuple[int, int]

    @classmethod
    def from_nir(cls, name: str, node: nir.AvgPool2d) -> SCAvgPool2dNode:
        ks = tuple(int(x) for x in np.atleast_1d(node.kernel_size).flatten()[:2])
        st = tuple(int(x) for x in np.atleast_1d(node.stride).flatten()[:2])
        pad = tuple(int(x) for x in np.atleast_1d(node.padding).flatten()[:2])
        if len(ks) == 1:
            ks = (ks[0], ks[0])
        if len(st) == 1:
            st = (st[0], st[0])
        if len(pad) == 1:
            pad = (pad[0], pad[0])
        return cls(name=name, kernel_size=ks, stride=st, padding=pad)

    def forward(self, x: np.ndarray) -> np.ndarray:
        sum_node = SCSumPool2dNode(
            name=self.name + "_sum",
            kernel_size=self.kernel_size,
            stride=self.stride,
            padding=self.padding,
        )
        summed = sum_node.forward(x)
        area = self.kernel_size[0] * self.kernel_size[1]
        return summed / area

map_node(name, node, **kwargs)

Convert a single NIR node to its SC-NeuroCore equivalent.

Source code in src/sc_neurocore/nir_bridge/node_map.py

def map_node(name: str, node: nir.NIRNode, **kwargs) -> Any:
    """Convert a single NIR node to its SC-NeuroCore equivalent."""
    factory = NODE_MAP.get(type(node))
    if factory is None:
        raise NotImplementedError(
            f"NIR node type {type(node).__name__} not yet supported (node: {name!r})"
        )
    return factory(name, node, **kwargs)

Export

sc_neurocore.nir_bridge.export

to_nir(network, path=None)

Export an SC-NeuroCore SCNetwork to NIR format.

Parameters

network : SCNetwork The network to export. path : str or Path, optional If provided, write the NIR graph to this file.

Returns

nir.NIRGraph

Source code in src/sc_neurocore/nir_bridge/export.py

def to_nir(network, path: str | Path | None = None) -> nir.NIRGraph:
    """Export an SC-NeuroCore SCNetwork to NIR format.

    Parameters
    ----------
    network : SCNetwork
        The network to export.
    path : str or Path, optional
        If provided, write the NIR graph to this file.

    Returns
    -------
    nir.NIRGraph
    """
    from .parser import SCMultiPortSubgraphNode, SCNetwork, SCSubgraphNode, _UnitDelayNode

    if not isinstance(network, SCNetwork):
        raise TypeError(f"Expected SCNetwork, got {type(network)}")

    # Ensure topo_order has been computed (triggers delay insertion)
    _ = network.topo_order

    nodes = {}
    edges = list(network.edges)

    for name, node in network.nodes.items():
        # Skip internal delay nodes — reconstruct as direct recurrent edges
        if isinstance(node, _UnitDelayNode):
            continue
        # Recursively export subgraphs
        if isinstance(node, (SCSubgraphNode, SCMultiPortSubgraphNode)):
            nodes[name] = to_nir(node.network)
            continue
        nir_node = _node_to_nir(name, node)
        if nir_node is None:
            raise ValueError(f"Cannot export node {name!r} of type {type(node).__name__} to NIR")
        nodes[name] = nir_node

    # Replace delay edges with original recurrent edges
    clean_edges = []
    for src, dst in edges:
        if src.startswith("_delay_") and src in network._recurrent_map:
            # Restore original back edge: recurrent_source -> dst
            original_src = network._recurrent_map[src]
            clean_edges.append((original_src, dst))
        elif dst.startswith("_delay_"):
            # Skip the edge feeding INTO the delay node (it's implicit)
            continue
        else:
            clean_edges.append((src, dst))

    graph = nir.NIRGraph(nodes=nodes, edges=clean_edges)

    if path is not None:
        nir.write(str(path), graph)

    return graph