Array Transfer Interface

ArrayTransferIF is a logical interface that carries a variable-length array of typed elements between simulation components. The element type can be any DataSchema subclass — a scalar field (Float32, U8), a composite DataList, or a DataArray.

ArrayTransferIF generalises SchemaTransferIF: a single-element array transfer with count=1 is equivalent to a schema transfer, and the implementations share the same PhysicalTransport layer.

Application layer:   Component.write(elements)    rx_proc(elements) / get(count)
                           │                                  │
Logical layer:   ArrayTransferIFMaster              ArrayTransferIFSlave
                           │                                  │
Transport layer:     PhysicalTransport       (StreamTransport | …)
                           │                                  │
Physical layer:    StreamIFMaster                    StreamIFSlave

---

Classes

Class	Role
ArrayTransferIFMaster	Serializes an element list → one word burst → transport
ArrayTransferIFSlave	Receives a word burst → deserializes → delivers via rx_proc or get(count)
ArrayTransferIF	Optional logical container; validates endpoint types, element_type, and bitwidth

PhysicalTransport and StreamTransport are shared with SchemaTransferIF; see the Schema Transfer Interface page.

---

A minimal simulation

Two raw SimObjs sending one variable-length array master→slave. Each transfer endpoint owns an internal stream_ep; bind those over a StreamIF to wire the physical link. No HwComponent. (The yield from / run_proc mechanics are in Process generators.)

from dataclasses import dataclass

import numpy as np

from waveflow.hw.clock import Clock
from waveflow.hw.dataschema import FloatField
from waveflow.hw.interface import StreamIF
from waveflow.hw.schema_transfer_interface import (
    ArrayTransferIFMaster, ArrayTransferIFSlave,
)
from waveflow.simulation.simobj import ProcessGen, SimObj
from waveflow.simulation.simulation import Simulation

Float32 = FloatField.specialize(bitwidth=32)


@dataclass
class Producer(SimObj):
    """Holds the array master; sends one variable-length burst of samples."""

    master: ArrayTransferIFMaster | None = None

    def run_proc(self) -> ProcessGen[None]:
        samples = np.array([1.0, -2.5, 3.14], dtype=np.float32)
        yield from self.master.write(samples)     # one packet, TLAST on the last element


@dataclass
class Consumer(SimObj):
    """Holds the array slave; rx_proc fires once with the whole received array."""

    def __post_init__(self) -> None:
        super().__post_init__()
        self.received: np.ndarray | None = None
        self.slave = ArrayTransferIFSlave(
            sim=self.sim, element_type=Float32, bitwidth=32, rx_proc=self.on_samples,
        )

    def on_samples(self, elements: np.ndarray) -> ProcessGen[None]:
        self.received = elements                  # np.ndarray[float32]
        yield self.timeout(0)


sim = Simulation()
clk = Clock(freq=1e9)

producer = Producer(
    name="producer", sim=sim,
    master=ArrayTransferIFMaster(sim=sim, element_type=Float32, bitwidth=32),
)
consumer = Consumer(name="consumer", sim=sim)

# Each transfer endpoint owns an internal stream_ep; bind those to wire the physical link.
stream_if = StreamIF(sim=sim, clk=clk)
stream_if.bind("master", producer.master.stream_ep)
stream_if.bind("slave", consumer.slave.stream_ep)

sim.run_sim()
print("consumer received:", consumer.received.tolist())

The producer writes the three-element array as one burst; the slave infers the element count from the burst length and delivers the whole np.ndarray[float32] to rx_proc, so consumer.received is [1.0, -2.5, 3.14]. See SimObj for the base object and lifecycle; push- vs pull-mode receive is detailed below.

---

ArrayTransferIFMaster

Parameter	Type	Default	Meaning
element_type	type[DataSchema]	—	Schema class for each element
bitwidth	int	32	Word width for serialization

The master creates an internal StreamIFMaster exposed as .stream_ep — bind it over a StreamIF.

master = ArrayTransferIFMaster(
    sim=sim,
    element_type=Float32,
    bitwidth=32,
)

Usage — from inside a run_proc:

def run_proc(self) -> ProcessGen[None]:
    samples = np.array([1.0, -2.5, 3.14], dtype=np.float32)
    yield from self.arr_master.write(samples)

write(elements) serializes each element with element_type.nwords_per_inst(bitwidth) words, concatenates them into one burst, and forwards the burst to the transport as a single AXI-Stream packet (TLAST asserted on the last word).

Numpy fast path — when element_type is a scalar FloatField or IntField and elements is a np.ndarray, the array is converted to words in a single vectorized operation (no per-element schema instance allocation):

yield from self.arr_master.write(np.array([1.0, -2.5, 3.14], dtype=np.float32))  # Float32
yield from self.arr_master.write(np.array([10, 20, 30], dtype=np.uint8))          # U8

elements may also contain schema instances or raw Python values — anything that element_type(value) accepts (slow path):

yield from self.arr_master.write([Float32(1.0), Float32(-2.5)])  # schema instances
yield from self.arr_master.write([1.0, -2.5, 3.14])              # raw values → Float32

---

ArrayTransferIFSlave

Parameter	Type	Default	Meaning
element_type	type[DataSchema]	—	Schema class for each element
bitwidth	int	32	Word width for deserialization
rx_proc	Callable[[list], ProcessGen[None]] \| None	None	Push-mode callback
pull_mode	bool	False	When True, enables get(count) and disables the push callback

The slave creates an internal StreamIFSlave exposed as .stream_ep — bind it over a StreamIF. Two receive modes are available.

Push mode (default)

Set rx_proc to a callback and call pre_sim(). When a burst arrives, the element count is inferred from the burst length divided by element_type.nwords_per_inst(bitwidth), and rx_proc(elements) is called with the result.

For scalar FloatField / IntField element types, elements is a np.ndarray of the element’s native dtype (e.g. np.float32). For composite element types it is a list of deserialized instances.

def on_samples(self, elements: np.ndarray) -> ProcessGen[None]:
    # elements is np.ndarray[float32] — no per-element boxing
    process(elements)
    yield self.env.timeout(0)

slave = ArrayTransferIFSlave(
    sim=sim,
    element_type=Float32,
    bitwidth=32,
    rx_proc=self.on_samples,
)
# bind slave.stream_ep over a StreamIF; run_sim() then calls pre_sim() to install the callback

Pull mode

Set pull_mode=True (and do not set rx_proc). The owning component drives sequencing by calling get(count=n) directly. An exact element count is required; the burst length is validated against count * element_type.nwords_per_inst(bitwidth).

slave = ArrayTransferIFSlave(
    sim=sim,
    element_type=Float32,
    bitwidth=32,
    pull_mode=True,
)

# Inside the component's run_proc:
samples = yield from self.samp_slave.get(count=nsamp)
# samples is np.ndarray[float32] — use directly in numpy operations
values = samples.astype(float)

TLAST validation — get(count) raises RuntimeError if the burst does not match the expected length:

Condition	Message
actual_words < count * nwords_per_elem	TLAST early: expected N words …
actual_words > count * nwords_per_elem	Missing TLAST: expected N words …

This matches the error contract in the generated C++ utilities (TLAST_EARLY_SAMP_IN, NO_TLAST_SAMP_IN).

Lifecycle

In push mode, pre_sim() installs _on_words_received as the transport’s receive callback. When using Simulation.run_sim() this is called automatically. When driving env.run() directly, call slave.pre_sim() manually.

In pull mode, pre_sim() is a no-op and the physical slave’s run_proc exits immediately, leaving the data buffer available for get().

---

ArrayTransferIF

ArrayTransferIF is an optional Interface container that enforces consistency when binding endpoints:

from waveflow.hw.schema_transfer_interface import ArrayTransferIF

iface = ArrayTransferIF(sim=sim)
iface.bind("master", master_ep)
iface.bind("slave",  slave_ep)

Binding raises:

• TypeError if the wrong endpoint class is used for a side
• ValueError if master and slave have different element_type or bitwidth

Like SchemaTransferIF, this container only validates the logical pair — it does not wire the physical link. You still bind each endpoint’s .stream_ep over a StreamIF to connect them.

---

Example: push mode

The master sends a burst; the slave infers element count from burst length.

from dataclasses import dataclass

import numpy as np
from waveflow.hw.clock import Clock
from waveflow.hw.dataschema import FloatField
from waveflow.hw.interface import StreamIF
from waveflow.hw.schema_transfer_interface import (
    ArrayTransferIFMaster, ArrayTransferIFSlave,
)
from waveflow.simulation.simulation import Simulation
from waveflow.simulation.simobj import ProcessGen, SimObj

Float32 = FloatField.specialize(bitwidth=32)

sim = Simulation()
clk = Clock(freq=1e9)

received: list[np.ndarray] = []

def on_samples(elements: np.ndarray) -> ProcessGen[None]:
    received.append(elements)   # np.ndarray[float32]
    yield sim.env.timeout(0)

arr_master = ArrayTransferIFMaster(sim=sim, element_type=Float32, bitwidth=32)
arr_slave  = ArrayTransferIFSlave(sim=sim, element_type=Float32, bitwidth=32, rx_proc=on_samples)

# Each transfer endpoint owns an internal stream_ep; bind those over a StreamIF.
stream_if = StreamIF(sim=sim, clk=clk)
stream_if.bind("master", arr_master.stream_ep)
stream_if.bind("slave",  arr_slave.stream_ep)


@dataclass
class Tx(SimObj):
    def run_proc(self) -> ProcessGen[None]:
        yield from arr_master.write(np.array([1.0, 2.0, 3.0], dtype=np.float32))


Tx(name="tx", sim=sim)
sim.run_sim()   # calls arr_slave.pre_sim() to install the receive callback
# received[0] == np.array([1.0, 2.0, 3.0], dtype=np.float32)

---

Example: pull mode

In pull mode the consumer drives sequencing: it calls get(count=n) for an exact element count (TLAST-validated) instead of registering a callback.

from dataclasses import dataclass

import numpy as np
from waveflow.hw.clock import Clock
from waveflow.hw.dataschema import FloatField
from waveflow.hw.interface import StreamIF
from waveflow.hw.schema_transfer_interface import (
    ArrayTransferIFMaster, ArrayTransferIFSlave,
)
from waveflow.simulation.simulation import Simulation
from waveflow.simulation.simobj import ProcessGen, SimObj

Float32 = FloatField.specialize(bitwidth=32)


@dataclass
class Producer(SimObj):
    master: ArrayTransferIFMaster | None = None

    def run_proc(self) -> ProcessGen[None]:
        yield from self.master.write(np.array([1.0, 2.0, 3.0], dtype=np.float32))


@dataclass
class Consumer(SimObj):
    slave: ArrayTransferIFSlave | None = None

    def __post_init__(self) -> None:
        super().__post_init__()
        self.samples: np.ndarray | None = None

    def run_proc(self) -> ProcessGen[None]:
        self.samples = yield from self.slave.get(count=3)   # exact count; TLAST-validated


sim = Simulation()
clk = Clock(freq=1e9)

producer = Producer(
    name="producer", sim=sim,
    master=ArrayTransferIFMaster(sim=sim, element_type=Float32, bitwidth=32),
)
consumer = Consumer(
    name="consumer", sim=sim,
    slave=ArrayTransferIFSlave(sim=sim, element_type=Float32, bitwidth=32, pull_mode=True),
)

stream_if = StreamIF(sim=sim, clk=clk)
stream_if.bind("master", producer.master.stream_ep)
stream_if.bind("slave",  consumer.slave.stream_ep)

sim.run_sim()
# consumer.samples == np.array([1.0, 2.0, 3.0], dtype=np.float32)

Header + payload on one physical stream. Each transfer endpoint owns its own stream_ep, so two transfer slaves cannot share a single stream. For the “typed header first, then pull nsamp samples from the same stream” pattern (the poly accelerator), use a single raw StreamIFSlave and successive get() calls — cmd = yield from s_in.get(PolyCmdHdr) then samp = yield from s_in.get_pipelined(Float32, count=cmd.nsamp) — as examples/stream_inband/poly.py does.

---

Synthesis mapping

The synthesizable side — the generated <element>_array_utils calls a kernel uses to move this transfer over an m_axi / stream port (write_axi4_stream_lane / read_axi4_stream_lane, or read_array_slice for a resident array) — is documented, with the Python→C++ mapping table, in Custom Hooks: kernel transfer reference.

---

Wire footprint

element_type	Words per element	Words per count-element transfer
Float32 (32-bit, word_bw=32)	1	count
U8 (8-bit, word_bw=32)	1	count
S16 (16-bit, word_bw=32)	1	count
SomeDataList	SomeDataList.nwords_per_inst(word_bw)	count × nwords_per_inst

Elements are packed tightly; no padding is inserted between elements.

---

Quick reference

from waveflow.hw.schema_transfer_interface import (
    ArrayTransferIFMaster,
    ArrayTransferIFSlave,
    ArrayTransferIF,
    StreamTransport,
)

Operation	Code
Create master	ArrayTransferIFMaster(sim=sim, element_type=T, bitwidth=32)
Create slave (push)	ArrayTransferIFSlave(sim=sim, element_type=T, bitwidth=32, rx_proc=fn)
Create slave (pull)	ArrayTransferIFSlave(sim=sim, element_type=T, bitwidth=32, pull_mode=True)
Wire the physical link	stream_if.bind("master", master.stream_ep); stream_if.bind("slave", slave.stream_ep)
Transmit (numpy fast path)	yield from master.write(np.array(..., dtype=np.float32))
Transmit (schema / raw values)	yield from master.write([v1, v2, …])
Receive (push, manual setup)	slave.pre_sim() before env.run()
Receive (pull)	arr = yield from slave.get(count=n) — np.ndarray for scalar types
Words per element	element_type.nwords_per_inst(bitwidth)

See also: Schema Transfer Interface for the shared PhysicalTransport abstraction.