FPGA-Neural/hardware/v1/tools/netasm/assembler.py

"""
netasm assembler -- turns a parsed DenseNet/GraphNet (see parser.py)
into exact byte layouts (descriptor tables, edge blocks) and the SPI
command sequence to load them, per the data formats in the project
spec (§4). Runs entirely on the host; nothing here executes on the
FPGA (§10).

Compile-time validation performed here (spec §9's whole point: catch
these BEFORE the runtime load-time guard in rtl/graph_engine.v ever
sees them):
  - graph: src_id < out_id, src_id/out_id < N_TOTAL, an OUTPUT neuron
    is never used as another neuron's source, every symbolic/literal
    CONN reference resolves to a real signal id.
  - graph: a neuron's padded connection count (see PARALLEL padding
    below) must fit the hardware's build-time MAX_CONN.
  - dense: every layer's real input count must be a PARALLEL
    multiple (the same runtime convention neuron_parallel.v/
    neuron_memory.v already require of n_inputs_real).
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from typing import Dict, List, Optional

try:
    from . import frames as F
    from .parser import Conn, DenseNet, GraphNet, Neuron
except ImportError:
    # See cli.py's matching fallback: allows this module to load when
    # imported as a bare top-level module too, not only as part of
    # the `tools.netasm` package.
    import frames as F
    from parser import Conn, DenseNet, GraphNet, Neuron


class NetasmError(Exception):
    pass


def _act_code(name: str) -> int:
    return F.ACT_RELU if name.lower() == "relu" else F.ACT_NONE


def _pad_to_parallel(n_conn: int, parallel: int) -> int:
    # At least one full PARALLEL group even for n_conn == 0 -- a
    # zero-group neuron would forward n_inputs_real=0 into
    # neuron_parallel, which hangs (see rtl/neuron_parallel.v's
    # GROUPS==0 failure mode, and rtl/graph_engine.v's matching
    # load-time guard). Padding edges are (src_id=0, weight=0);
    # src_id=0 is always a valid, already-computed signal for any
    # neuron with id > 0 (every real network has n_in >= 1 inputs at
    # id 0), so this never trips the src_id < out_id guard.
    return max(parallel, math.ceil(max(n_conn, 1) / parallel) * parallel)


# ================================================================
# GRAPH (Type #2)
# ================================================================


@dataclass
class GraphLayout:
    n_in: int
    num_neurons: int
    n_out: int
    n_total: int
    id_of: Dict[str, int]
    order: List[str]  # neuron names in ascending-id order
    table_base: int
    x_base: int
    out_base: int
    edges_base: Dict[str, int]
    descriptor_bytes: bytes
    edge_bytes: Dict[str, bytes]
    n_conn_real: Dict[str, int]
    n_conn_padded: Dict[str, int]
    out_ids: List[str]  # neuron names, in out_base byte order
    frames: List[F.Frame] = field(default_factory=list)


def _resolve_src(
    src_token: str, id_of: Dict[str, int], n_in: int, line: int
) -> int:
    if src_token in id_of:
        return id_of[src_token]
    try:
        v = int(src_token, 0)
    except ValueError:
        raise NetasmError(f"line {line}: unknown source '{src_token}'")
    return v


def assemble_graph(
    net: GraphNet,
    parallel: int,
    max_conn: int = 32,
    n_total: int = 4096,
    table_base: int = 0x000000,
    edges_base: int = 0x010000,
    x_base: int = 0x000000,
    out_base: int = 0x020000,
) -> GraphLayout:

    if parallel <= 0 or (parallel & (parallel - 1)) != 0:
        raise NetasmError(f"PARALLEL must be a positive power of two, got {parallel}")

    n_in = net.n_inputs
    declared_names = [n.name for n in net.neurons]
    neuron_by_name = {n.name: n for n in net.neurons}

    for out_name in net.outputs:
        if out_name not in neuron_by_name:
            raise NetasmError(f"OUTPUT '{out_name}' refers to an undeclared neuron")
    if len(set(net.outputs)) != len(net.outputs):
        raise NetasmError("duplicate name in OUTPUT list")

    # A neuron used as ANY other neuron's source can never be an
    # output sink (spec §4.4's invariant, enforced here at compile
    # time rather than left to the runtime guard).
    referenced_as_source = set()
    for n in net.neurons:
        for c in n.conns:
            if c.src in neuron_by_name:
                referenced_as_source.add(c.src)
    for out_name in net.outputs:
        if out_name in referenced_as_source:
            raise NetasmError(
                f"OUTPUT '{out_name}' is used as a source by another neuron -- "
                "output ids must be pure sinks (spec §4.4)"
            )

    output_set = set(net.outputs)
    non_output_order = [n for n in declared_names if n not in output_set]
    order = non_output_order + list(net.outputs)

    id_of: Dict[str, int] = {}
    for i, name in enumerate(order):
        id_of[name] = n_in + i

    num_neurons = len(order)
    n_total_used = n_in + num_neurons
    if n_total_used > n_total:
        raise NetasmError(
            f"network needs {n_total_used} signal ids (N_in={n_in} + "
            f"{num_neurons} neurons), exceeds N_TOTAL={n_total}"
        )

    edges_base_of: Dict[str, int] = {}
    edge_bytes: Dict[str, bytes] = {}
    n_conn_real: Dict[str, int] = {}
    n_conn_padded: Dict[str, int] = {}
    cursor = edges_base

    for name in order:
        neuron = neuron_by_name[name]
        out_id = id_of[name]
        n_conn = len(neuron.conns)
        padded = _pad_to_parallel(n_conn, parallel)
        if padded > max_conn:
            raise NetasmError(
                f"neuron '{name}' (line {neuron.line}): {n_conn} connection(s) pad "
                f"to {padded} at PARALLEL={parallel}, exceeds MAX_CONN={max_conn}"
            )

        edges_base_of[name] = cursor
        buf = bytearray()

        for c in neuron.conns:
            src_id = _resolve_src(c.src, id_of, n_in, c.line)
            if src_id >= n_total:
                raise NetasmError(
                    f"line {c.line}: src id {src_id} >= N_TOTAL={n_total}"
                )
            if src_id >= out_id:
                raise NetasmError(
                    f"line {c.line}: neuron '{name}' (id {out_id}) connects from "
                    f"src id {src_id}, which is not a strictly earlier signal "
                    "(src_id must be < out_id, §7)"
                )
            buf += src_id.to_bytes(2, "big")
            buf.append(c.weight & 0xFF)
            buf.append(0x00)  # reserved

        for _ in range(padded - n_conn):
            buf += (0).to_bytes(2, "big")  # src_id = 0 (always valid, weight 0)
            buf.append(0x00)  # weight = 0
            buf.append(0x00)  # reserved

        edge_bytes[name] = bytes(buf)
        n_conn_real[name] = n_conn
        n_conn_padded[name] = padded
        cursor += len(buf)

    descriptor = bytearray()
    for name in order:
        neuron = neuron_by_name[name]
        out_id = id_of[name]
        n_conn = n_conn_real[name]
        descriptor += edges_base_of[name].to_bytes(3, "big")
        descriptor += n_conn.to_bytes(2, "big")
        descriptor += out_id.to_bytes(2, "big")
        descriptor.append(_act_code(neuron.activation))
        descriptor.append(neuron.bias & 0xFF)
        descriptor += b"\x00\x00"  # reserved

    n_out = len(net.outputs)

    layout = GraphLayout(
        n_in=n_in,
        num_neurons=num_neurons,
        n_out=n_out,
        n_total=n_total_used,
        id_of=id_of,
        order=order,
        table_base=table_base,
        x_base=x_base,
        out_base=out_base,
        edges_base=edges_base_of,
        descriptor_bytes=bytes(descriptor),
        edge_bytes=edge_bytes,
        n_conn_real=n_conn_real,
        n_conn_padded=n_conn_padded,
        out_ids=list(net.outputs),
    )

    fr: List[F.Frame] = []
    fr.append(F.write_ram(table_base, layout.descriptor_bytes))
    for name in order:
        fr.append(F.write_ram(edges_base_of[name], edge_bytes[name]))
    fr.append(F.set_net_type(F.NET_TYPE_GRAPH))
    fr.append(F.set_base(F.SEL_X_BASE, x_base))
    fr.append(F.set_base(F.SEL_TABLE_BASE, table_base))
    fr.append(F.set_base(F.SEL_BUF_A_BASE, out_base))
    fr.append(F.set_base(F.SEL_N_INPUTS, n_in))
    fr.append(F.set_base(F.SEL_NUM_NEURONS_GRAPH, num_neurons))
    fr.append(F.set_base(F.SEL_N_OUT, n_out))
    fr.append(F.run_network(0))
    layout.frames = fr

    return layout


def dump_graph_debug(layout: GraphLayout) -> str:
    lines = []
    lines.append("=== netasm graph debug dump ===")
    lines.append(f"N_in={layout.n_in} num_neurons={layout.num_neurons} "
                 f"n_out={layout.n_out} N_TOTAL_used={layout.n_total}")
    lines.append(f"table_base=0x{layout.table_base:06x} x_base=0x{layout.x_base:06x} "
                 f"out_base=0x{layout.out_base:06x}")
    lines.append("")
    lines.append("id assignment (ascending):")
    for name in layout.order:
        marker = " <- OUTPUT" if name in layout.out_ids else ""
        lines.append(
            f"  id={layout.id_of[name]:4d}  {name:16s} n_conn={layout.n_conn_real[name]} "
            f"padded={layout.n_conn_padded[name]} edges@0x{layout.edges_base[name]:06x}{marker}"
        )
    lines.append("")
    lines.append(f"descriptor table ({len(layout.descriptor_bytes)} bytes):")
    lines.append("  " + layout.descriptor_bytes.hex(" "))
    lines.append("")
    for name in layout.order:
        lines.append(f"edges for {name} ({len(layout.edge_bytes[name])} bytes):")
        lines.append("  " + layout.edge_bytes[name].hex(" "))
    lines.append("")
    lines.append("SPI load sequence:")
    lines.append(F.frames_as_hex(layout.frames))
    return "\n".join(lines)


# ================================================================
# DENSE (Type #1)
#
# The grammar (spec §9) only declares layer SIZES/activations, not
# weight VALUES -- those come from a trained model and are loaded by
# the host separately (existing WRITE_RAM flow, unchanged from
# before this tool existed). netasm's job for dense is therefore
# layout + descriptor table + load/run command generation: it
# allocates address ranges for each layer's weight matrix and bias
# vector, validates PARALLEL alignment, and reports exactly where
# the host must WRITE_RAM the real weight/bias content before
# RUN_NETWORK.
# ================================================================


@dataclass
class DenseLayerLayout:
    n_inputs_real: int
    n_neurons_real: int
    activation: str
    w_base: int
    bias_addr: int


@dataclass
class DenseLayout:
    n_inputs: int
    layers: List[DenseLayerLayout]
    table_base: int
    x_base: int
    buf_a_base: int
    buf_b_base: int
    descriptor_bytes: bytes
    frames: List[F.Frame] = field(default_factory=list)


def assemble_dense(
    net: DenseNet,
    parallel: int,
    table_base: int = 0x000000,
    weights_base: int = 0x010000,
    x_base: int = 0x000000,
    buf_a_base: int = 0x020000,
    buf_b_base: int = 0x021000,
) -> DenseLayout:

    if parallel <= 0 or (parallel & (parallel - 1)) != 0:
        raise NetasmError(f"PARALLEL must be a positive power of two, got {parallel}")

    layers: List[DenseLayerLayout] = []
    cursor = weights_base
    prev_n = net.n_inputs

    for layer in net.layers:
        if prev_n % parallel != 0:
            raise NetasmError(
                f"line {layer.line}: layer's real input count {prev_n} is not a "
                f"multiple of PARALLEL={parallel} (neuron_parallel.v/"
                "neuron_memory.v both require this at runtime)"
            )
        w_base = cursor
        cursor += prev_n * layer.n_neurons
        bias_addr = cursor
        cursor += layer.n_neurons
        layers.append(
            DenseLayerLayout(
                n_inputs_real=prev_n,
                n_neurons_real=layer.n_neurons,
                activation=layer.activation,
                w_base=w_base,
                bias_addr=bias_addr,
            )
        )
        prev_n = layer.n_neurons

    descriptor = bytearray()
    for l in layers:
        descriptor += l.w_base.to_bytes(3, "big")
        descriptor += l.bias_addr.to_bytes(3, "big")
        descriptor.append(_act_code(l.activation))
        descriptor += l.n_inputs_real.to_bytes(2, "big")
        descriptor += l.n_neurons_real.to_bytes(2, "big")

    layout = DenseLayout(
        n_inputs=net.n_inputs,
        layers=layers,
        table_base=table_base,
        x_base=x_base,
        buf_a_base=buf_a_base,
        buf_b_base=buf_b_base,
        descriptor_bytes=bytes(descriptor),
    )

    fr: List[F.Frame] = []
    fr.append(F.write_ram(table_base, layout.descriptor_bytes))
    fr.append(F.set_net_type(F.NET_TYPE_DENSE))
    fr.append(F.set_base(F.SEL_X_BASE, x_base))
    fr.append(F.set_base(F.SEL_TABLE_BASE, table_base))
    fr.append(F.set_base(F.SEL_BUF_A_BASE, buf_a_base))
    fr.append(F.set_base(F.SEL_BUF_B_BASE, buf_b_base))
    fr.append(F.run_network(len(net.layers)))
    layout.frames = fr

    return layout


def dump_dense_debug(layout: DenseLayout) -> str:
    lines = []
    lines.append("=== netasm dense debug dump ===")
    lines.append(f"N_inputs={layout.n_inputs} layers={len(layout.layers)}")
    lines.append(f"table_base=0x{layout.table_base:06x} x_base=0x{layout.x_base:06x} "
                 f"buf_a_base=0x{layout.buf_a_base:06x} buf_b_base=0x{layout.buf_b_base:06x}")
    lines.append("")
    lines.append("NOTE: weight/bias VALUES are not part of this program -- the")
    lines.append("host must WRITE_RAM the real trained weights/bias at the")
    lines.append("addresses below before RUN_NETWORK.")
    lines.append("")
    for i, l in enumerate(layout.layers):
        lines.append(
            f"  layer {i}: n_inputs_real={l.n_inputs_real} n_neurons_real={l.n_neurons_real} "
            f"activation={l.activation}"
        )
        lines.append(
            f"    w_base=0x{l.w_base:06x} ({l.n_inputs_real * l.n_neurons_real} bytes) "
            f"bias_addr=0x{l.bias_addr:06x} ({l.n_neurons_real} bytes)"
        )
    lines.append("")
    lines.append(f"descriptor table ({len(layout.descriptor_bytes)} bytes):")
    lines.append("  " + layout.descriptor_bytes.hex(" "))
    lines.append("")
    lines.append("SPI load sequence:")
    lines.append(F.frames_as_hex(layout.frames))
    return "\n".join(lines)