Simulation API

Configuration

fpga_verification.sim.config

Configuration contract shared by Platform Designer and cocotb tests.

RUNTIME_CONFIG_ENV = 'FPGA_VERIFICATION_TEST_CONFIG_JSON' module-attribute

Environment variable containing the resolved test configuration JSON.

ComponentConfig

Mixin for a resolved component configuration dataclass.

A configuration has two deliberately different representations: Platform Designer receives only HDL parameters, while cocotb receives the complete, already-resolved configuration. Runtime loading is strict so a stale test cannot silently run with defaults different from the generated DUT.

Source code in src/fpga_verification/sim/config.py

class ComponentConfig:
    """Mixin for a resolved component configuration dataclass.

    A configuration has two deliberately different representations: Platform
    Designer receives only HDL parameters, while cocotb receives the complete,
    already-resolved configuration.  Runtime loading is strict so a stale test
    cannot silently run with defaults different from the generated DUT.
    """

    def to_parameters(self) -> dict[str, Any]:
        parameters = {}
        for item in fields(self):
            name = item.metadata.get("fpga_verification.hdl_parameter")
            if name:
                parameters[item.name if name is True else name] = getattr(self, item.name)
        return parameters

    def to_runtime_dict(self) -> dict[str, Any]:
        value = asdict(self)
        _validate_runtime_mapping(type(self), value)
        try:
            json.dumps(value)
        except (TypeError, ValueError) as error:
            raise TypeError("ComponentConfig runtime values must be JSON serializable") from error
        return value

    @classmethod
    def from_runtime_dict(cls, value: Mapping[str, Any]):
        if not isinstance(value, Mapping):
            raise TypeError(f"{cls.__name__} runtime configuration must be a JSON object")

        expected = {item.name for item in fields(cls)}
        actual = set(value)
        missing = expected - actual
        extra = actual - expected
        if missing or extra:
            messages = []
            if missing:
                messages.append(f"missing fields: {', '.join(sorted(missing))}")
            if extra:
                messages.append(f"unknown fields: {', '.join(sorted(extra))}")
            raise ValueError(f"Invalid {cls.__name__} runtime configuration ({'; '.join(messages)})")

        _validate_runtime_mapping(cls, value)

        try:
            return cls(**dict(value))
        except (TypeError, ValueError) as error:
            raise ValueError(f"Invalid {cls.__name__} runtime configuration values") from error

hdl_parameter(default=MISSING, *, name=None, **kwargs)

Declare a dataclass field which is also a Platform Designer parameter.

name defaults to the dataclass field name. All fields remain present in the runtime configuration; the metadata only controls to_parameters().

Source code in src/fpga_verification/sim/config.py

def hdl_parameter(default: Any = MISSING, *, name: str | None = None, **kwargs):
    """Declare a dataclass field which is also a Platform Designer parameter.

    ``name`` defaults to the dataclass field name.  All fields remain present in
    the runtime configuration; the metadata only controls ``to_parameters()``.
    """
    from dataclasses import field

    metadata = dict(kwargs.pop("metadata", {}))
    metadata["fpga_verification.hdl_parameter"] = name or True
    if default is MISSING:
        return field(metadata=metadata, **kwargs)
    return field(default=default, metadata=metadata, **kwargs)

runtime_config_environment(config)

Return the cocotb environment payload for an already-resolved config.

Source code in src/fpga_verification/sim/config.py

def runtime_config_environment(config: ComponentConfig) -> dict[str, str]:
    """Return the cocotb environment payload for an already-resolved config."""
    if not isinstance(config, ComponentConfig):
        raise TypeError("Runtime config must inherit ComponentConfig")
    return {RUNTIME_CONFIG_ENV: json.dumps(config.to_runtime_dict(), sort_keys=True)}

load_runtime_config(config_type, environ=None)

Strictly restore config_type from runner-provided JSON.

Missing variables, malformed JSON and both missing and unknown fields are errors. Tests must never fall back to constructing a default config.

Source code in src/fpga_verification/sim/config.py

def load_runtime_config(config_type, environ: Mapping[str, str] | None = None):
    """Strictly restore ``config_type`` from runner-provided JSON.

    Missing variables, malformed JSON and both missing and unknown fields are
    errors.  Tests must never fall back to constructing a default config.
    """
    if not issubclass(config_type, ComponentConfig):
        raise TypeError("config_type must inherit ComponentConfig")

    environ = os.environ if environ is None else environ
    try:
        payload = environ[RUNTIME_CONFIG_ENV]
    except KeyError as error:
        raise RuntimeError(
            f"{RUNTIME_CONFIG_ENV} is required; run the test via the FPGA verification runner"
        ) from error

    try:
        value = json.loads(payload)
    except (TypeError, json.JSONDecodeError) as error:
        raise ValueError(f"{RUNTIME_CONFIG_ENV} must contain a JSON object") from error
    return config_type.from_runtime_dict(value)

Agents

fpga_verification.sim.agents

Reusable pyuvm agents.

VIPAgent

Bases: uvm_agent

Source/sink pyuvm agent for Intel Avalon-ST Video packet streams.

Source code in src/fpga_verification/sim/agents/intel_video.py

class VIPAgent(uvm_agent):
    """Source/sink pyuvm agent for Intel Avalon-ST Video packet streams."""

    def __init__(
        self,
        name,
        parent,
        clock,
        reset,
        source_bus=None,
        sink_bus=None,
        source_fmt=None,
        sink_fmt=None,
        reset_active_level=True,
        ready_latency=0,
        ready_allowance=None,
        idle_value=0,
        randomize=False,
        is_active=uvm_active_passive_enum.UVM_ACTIVE,
        packet_logging=False,
        packet_log_level=logging.INFO,
    ):
        super().__init__(name, parent)
        if source_bus is None and sink_bus is None:
            raise ValueError("VIPAgent requires source_bus, sink_bus, or both")

        if sink_fmt is None:
            sink_fmt = source_fmt
        if source_bus is not None and source_fmt is None:
            raise ValueError("source_fmt is required when source_bus is used")
        if sink_bus is not None and sink_fmt is None:
            raise ValueError("sink_fmt is required when sink_bus is used")
        if source_fmt is not None and not isinstance(source_fmt, VideoFormat):
            raise TypeError("source_fmt must be a VideoFormat")
        if sink_fmt is not None and not isinstance(sink_fmt, VideoFormat):
            raise TypeError("sink_fmt must be a VideoFormat")

        self.clock = clock
        self.reset = reset
        self.reset_active_level = bool(reset_active_level)
        self.ready_latency = ready_latency
        self.ready_allowance = ready_allowance
        self.idle_value = idle_value
        self.source_bus = source_bus
        self.source_fmt = source_fmt
        self.sink_bus = sink_bus
        self.sink_fmt = sink_fmt
        self.randomize = randomize
        self._requested_is_active = is_active
        self.packet_logging = bool(packet_logging)
        self.packet_log_level = _normalize_log_level(packet_log_level)

        self.sequencer = None
        self.source = None
        self.source_driver = None
        self.source_monitor = None
        self.sink_monitor = None

    def build_phase(self):
        super().build_phase()
        self.is_active = self._requested_is_active

        if self.source_bus is not None:
            self.source_monitor = VIPMonitor(
                "source_monitor",
                self,
                self.source_bus,
                self.clock,
                self.source_fmt,
                self.reset,
                self.reset_active_level,
                self.ready_latency,
                self.ready_allowance,
                packet_logging=self.packet_logging and not self.active(),
                packet_log_level=self.packet_log_level,
                packet_log_action="observed",
            )

        if self.sink_bus is not None:
            self.sink_monitor = VIPMonitor(
                "sink_monitor",
                self,
                self.sink_bus,
                self.clock,
                self.sink_fmt,
                self.reset,
                self.reset_active_level,
                self.ready_latency,
                self.ready_allowance,
                drive_ready=self.active(),
                randomize=self.randomize,
                packet_logging=self.packet_logging,
                packet_log_level=self.packet_log_level,
                packet_log_action="got",
            )

        if self.active() and self.source_bus is not None:
            self.source = _make_vip_source(
                self.source_bus,
                self.clock,
                self.reset,
                self.source_fmt,
                self.reset_active_level,
                self.ready_latency,
                self.ready_allowance,
                self.idle_value,
            )
            if self.randomize:
                self.source.set_pause_generator(_random_pause_generator())
            self.sequencer = uvm_sequencer("sequencer", self)
            self.source_driver = VIPDriver(
                "source_driver",
                self,
                self.source,
                self.source_fmt,
                packet_logging=self.packet_logging,
                packet_log_level=self.packet_log_level,
            )

    def connect_phase(self):
        if self.source_driver is not None:
            self.source_driver.seq_item_port.connect(
                self.sequencer.seq_item_export
            )

    def set_packet_logging(self, enable, level=None):
        self.packet_logging = bool(enable)
        if level is not None:
            self.packet_log_level = _normalize_log_level(level)
        if self.source_driver is not None:
            self.source_driver.packet_logging = self.packet_logging
            self.source_driver.packet_log_level = self.packet_log_level
        for monitor in (self.source_monitor, self.sink_monitor):
            if monitor is not None:
                monitor.packet_log_level = self.packet_log_level
        if self.source_monitor is not None:
            self.source_monitor.packet_logging = self.packet_logging and not self.active()
        if self.sink_monitor is not None:
            self.sink_monitor.packet_logging = self.packet_logging

    def set_randomize(self, enable):
        self.randomize = bool(enable)
        pause_generator = _random_pause_generator() if self.randomize else None
        if self.source is not None:
            self.source.set_pause_generator(pause_generator)
        if self.sink_monitor is not None and self.sink_monitor.monitor is not None:
            self.sink_monitor.randomize = self.randomize
            if self.sink_monitor.drive_ready:
                pause_generator = (
                    _random_pause_generator() if self.randomize else None
                )
                self.sink_monitor.monitor.set_pause_generator(pause_generator)

    def cancel_bfms(self):
        for bfm in (
            self.source,
            self.source_monitor.monitor if self.source_monitor is not None else None,
            self.sink_monitor.monitor if self.sink_monitor is not None else None,
        ):
            if bfm is not None:
                bfm.cancel()

    def clear_bfms(self):
        for monitor in (self.source_monitor, self.sink_monitor):
            if monitor is not None:
                monitor.reset_protocol_state()
                if monitor.monitor is not None:
                    monitor.monitor.clear()
        if self.source is not None:
            self.source.clear()

    @property
    def source_analysis_port(self):
        if self.source_monitor is None:
            raise RuntimeError("VIP agent has no source monitor analysis port")
        return self.source_monitor.analysis_port

    @property
    def sink_analysis_port(self):
        if self.sink_monitor is None:
            raise RuntimeError("VIP agent has no sink monitor analysis port")
        return self.sink_monitor.analysis_port

VIPDriver

Bases: uvm_driver

Drive VIP packet sequence items onto an Avalon-ST source.

Source code in src/fpga_verification/sim/agents/intel_video.py

class VIPDriver(uvm_driver):
    """Drive VIP packet sequence items onto an Avalon-ST source."""

    def __init__(
        self,
        name,
        parent,
        source,
        fmt,
        packet_logging=False,
        packet_log_level=logging.INFO,
    ):
        super().__init__(name, parent)
        self.source = source
        self.fmt = fmt
        self.packet_logging = bool(packet_logging)
        self.packet_log_level = _normalize_log_level(packet_log_level)

    async def run_phase(self):
        while True:
            item = await self.seq_item_port.get_next_item()
            try:
                await self._send_item(item)
            finally:
                self.seq_item_port.item_done()

    async def _send_item(self, item):
        if not isinstance(item, VIPItem):
            raise TypeError("VIPDriver accepts VIPItem instances")

        tx_complete = Event()
        packet = item.to_vip_packet()

        def log_completed_frame(frame):
            self.source.log.debug("TX VIP packet: %s", frame)
            if self.packet_logging:
                _log_vip_packet(
                    self.source.log,
                    self.packet_log_level,
                    self.source._bus_label(),
                    "sent",
                    packet,
                )
            tx_complete.set()

        frame = AvalonSTFrame(
            packet.to_symbols(self.fmt.samples_per_beat),
            tx_complete=log_completed_frame,
        )
        await self.source.send(frame)
        await tx_complete.wait()

VIPItem

Bases: uvm_sequence_item

Sequence item containing one complete Intel VIP packet.

Source code in src/fpga_verification/sim/agents/intel_video.py

class VIPItem(uvm_sequence_item):
    """Sequence item containing one complete Intel VIP packet."""

    def __init__(self, packet, name="vip_item"):
        super().__init__(name)
        if not isinstance(packet, VIPPacket):
            raise TypeError("packet must be a VIPPacket")
        self.packet = packet

    def to_vip_packet(self):
        return self.packet

    @classmethod
    def from_packet(cls, packet, name="vip_item"):
        return cls(packet, name=name)

VIPMonitor

Bases: uvm_monitor

Observe VIP packets, validate stream protocol, and publish packets.

Source code in src/fpga_verification/sim/agents/intel_video.py

class VIPMonitor(uvm_monitor):
    """Observe VIP packets, validate stream protocol, and publish packets."""

    def __init__(
        self,
        name,
        parent,
        bus,
        clock,
        fmt,
        reset=None,
        reset_active_level=True,
        ready_latency=0,
        ready_allowance=None,
        drive_ready=False,
        randomize=False,
        packet_logging=False,
        packet_log_level=logging.INFO,
        packet_log_action="got",
    ):
        super().__init__(name, parent)
        if not isinstance(fmt, VideoFormat):
            raise TypeError("fmt must be a VideoFormat")
        self.bus = bus
        self.clock = clock
        self.reset = reset
        self.reset_active_level = bool(reset_active_level)
        self.ready_latency = ready_latency
        self.ready_allowance = ready_allowance
        self.fmt = fmt
        self.drive_ready = drive_ready
        self.randomize = randomize
        self.packet_logging = bool(packet_logging)
        self.packet_log_level = _normalize_log_level(packet_log_level)
        self.packet_log_action = str(packet_log_action)
        self.monitor = None
        self.protocol_checker = VIPProtocolChecker(fmt)
        self.analysis_port = uvm_analysis_port("analysis_port", self)

    def build_phase(self):
        super().build_phase()
        if self.drive_ready:
            self.monitor = _make_vip_sink(
                self.bus,
                self.clock,
                self.reset,
                self.fmt,
                self.reset_active_level,
                self.ready_latency,
                self.ready_allowance,
            )
            if self.randomize:
                self.monitor.set_pause_generator(_random_pause_generator())
        else:
            self.monitor = _make_vip_monitor(
                self.bus,
                self.clock,
                self.reset,
                self.fmt,
                self.reset_active_level,
                self.ready_latency,
                self.ready_allowance,
            )

    async def recv_packet(self):
        frame = await self.monitor.recv()
        self.monitor.log.debug("RX VIP packet: %s", frame)
        try:
            packet = vip_packet_from_symbols(
                frame.data,
                symbols_per_beat=self.fmt.samples_per_beat,
            )
            self.protocol_checker.observe(packet)
            if self.packet_logging:
                _log_vip_packet(
                    self.monitor.log,
                    self.packet_log_level,
                    self.monitor._bus_label(),
                    self.packet_log_action,
                    packet,
                )
            return packet
        except ValueError as exc:
            raise ValueError(f"{self.monitor._bus_label()}: {exc}") from exc

    async def _watch_reset(self):
        while True:
            try:
                trigger = self.reset.value_change
            except AttributeError:
                trigger = Edge(self.reset)
            await trigger
            try:
                level = bool(int(self.reset.value))
            except ValueError:
                level = self.reset_active_level
            if level == self.reset_active_level:
                self.reset_protocol_state()

    async def run_phase(self):
        reset_task = None
        if self.reset is not None:
            reset_task = cocotb.start_soon(self._watch_reset())
        try:
            while True:
                self.analysis_port.write(await self.recv_packet())
        finally:
            if reset_task is not None:
                reset_task.cancel()

    def reset_protocol_state(self):
        self.protocol_checker.reset()

VIPSequence

Bases: uvm_sequence

Sequence that sends one or more complete Intel VIP packets.

Source code in src/fpga_verification/sim/agents/intel_video.py

class VIPSequence(uvm_sequence):
    """Sequence that sends one or more complete Intel VIP packets."""

    def __init__(self, items=None, name="vip_sequence"):
        super().__init__(name)
        self.items = list(items) if items is not None else []

    @classmethod
    def from_packets(cls, packets, name="vip_sequence"):
        return cls(
            [
                VIPItem.from_packet(packet, name=f"{name}_packet_{index}")
                for index, packet in enumerate(packets)
            ],
            name=name,
        )

    async def body(self):
        for item in self.items:
            await self.send_item(item)

    async def send_item(self, item):
        await self.start_item(item)
        await self.finish_item(item)
        return item

AvalonMMMonitor

Bases: uvm_monitor

Passively observes accepted Avalon-MM read/write requests.

Source code in src/fpga_verification/sim/agents/avalon_mm.py

class AvalonMMMonitor(uvm_monitor):
    """Passively observes accepted Avalon-MM read/write requests."""

    def __init__(
        self,
        name,
        parent,
        bus,
        clock,
        reset=None,
        reset_active_level=True,
        packet_logging=False,
        packet_log_level=logging.INFO,
    ):
        super().__init__(name, parent)
        self.bus = bus
        self.clock = clock
        self.reset = reset
        self.reset_active_level = bool(reset_active_level)
        self.packet_logging = bool(packet_logging)
        self.packet_log_level = _normalize_log_level(packet_log_level)
        self.label = _bus_label(bus)
        self.log = logging.getLogger(f"cocotb.{self.label}.monitor")
        self.analysis_port = uvm_analysis_port("analysis_port", self)

    async def run_phase(self):
        while True:
            await RisingEdge(self.clock)

            if self._reset_active():
                continue

            if self._waitrequest_active():
                continue

            read = bool(_read_int(self.bus.read, "read", 0))
            write = bool(_read_int(self.bus.write, "write", 0))

            if read and write:
                raise RuntimeError(
                    f"{self.label}: Avalon-MM read and write asserted together"
                )
            if not read and not write:
                continue

            transaction = self._sample_transaction("write" if write else "read")
            self.analysis_port.write(transaction)

            if self.packet_logging:
                self._log_transaction(transaction)

    def _sample_transaction(self, kind):
        address = _read_int(self.bus.address, "address")
        byteenable = _read_int(
            self.bus.byteenable,
            "byteenable",
            _mask(_byteenable_width(self.bus)),
        )
        burstcount = _read_int(self.bus.burstcount, "burstcount", 1)
        data = (
            _read_int(self.bus.writedata, "writedata")
            if kind == "write"
            else None
        )

        return AvalonMMTransaction(
            kind=kind,
            address=address,
            data=data,
            byteenable=byteenable,
            burstcount=burstcount,
            beat_index=0,
        )

    def _waitrequest_active(self):
        return bool(_read_int(self.bus.waitrequest, "waitrequest", 0))

    def _reset_active(self):
        if self.reset is None:
            return False
        return _read_int(
            self.reset,
            "reset",
            int(self.reset_active_level),
        ) == int(self.reset_active_level)

    def _log_transaction(self, transaction):
        data = ""
        if transaction.data is not None:
            data = f" data=0x{transaction.data:X}"

        self.log.log(
            self.packet_log_level,
            "%s: observed avalon-mm %s address=0x%X%s byteenable=0x%X",
            self.label,
            transaction.kind,
            transaction.address,
            data,
            transaction.byteenable,
        )

AvalonMMAgent

Bases: uvm_agent

Avalon-MM pyuvm agent with an always-on monitor and optional master.

Source code in src/fpga_verification/sim/agents/avalon_mm.py

class AvalonMMAgent(uvm_agent):
    """Avalon-MM pyuvm agent with an always-on monitor and optional master."""

    def __init__(
        self,
        name,
        parent,
        bus,
        clock,
        reset=None,
        reset_active_level=True,
        is_active=uvm_active_passive_enum.UVM_PASSIVE,
        read_response_latency=0,
        default_byteenable=None,
        packet_logging=False,
        packet_log_level=logging.INFO,
    ):
        super().__init__(name, parent)
        self.bus = bus
        self.clock = clock
        self.reset = reset
        self.reset_active_level = bool(reset_active_level)
        self._requested_is_active = is_active
        self.read_response_latency = int(read_response_latency)
        self.default_byteenable = default_byteenable
        self.packet_logging = bool(packet_logging)
        self.packet_log_level = _normalize_log_level(packet_log_level)

        self.monitor = None
        self.master = None

    def build_phase(self):
        super().build_phase()
        self.is_active = self._requested_is_active

        self.monitor = AvalonMMMonitor(
            "monitor",
            self,
            bus=self.bus,
            clock=self.clock,
            reset=self.reset,
            reset_active_level=self.reset_active_level,
            packet_logging=self.packet_logging and not self.active(),
            packet_log_level=self.packet_log_level,
        )

        if self.active():
            self.master = AvalonMMMasterBFM(
                self.bus,
                self.clock,
                self.reset,
                read_response_latency=self.read_response_latency,
                default_byteenable=self.default_byteenable,
                packet_logging=self.packet_logging,
                packet_log_level=self.packet_log_level,
            )

    async def run_phase(self):
        if self.master is not None:
            self.master.start()

    @property
    def analysis_port(self):
        if self.monitor is None:
            raise RuntimeError("Avalon-MM agent has no monitor analysis port")
        return self.monitor.analysis_port

FrameSource

Bases: Protocol

A source with lifecycle and analysis publication compatible with VIP tests.

Source code in src/fpga_verification/sim/agents/frame_source.py

class FrameSource(Protocol):
    """A source with lifecycle and analysis publication compatible with VIP tests."""

    analysis_port: object

    async def send_frame(self, frame) -> None:
        """Publish ``FrameTransaction`` then physically drive ``frame``."""

    def stop_tasks(self) -> None:
        """Stop source-owned clocks and background drive tasks."""

send_frame(frame) async

Publish FrameTransaction then physically drive frame.

Source code in src/fpga_verification/sim/agents/frame_source.py

async def send_frame(self, frame) -> None:
    """Publish ``FrameTransaction`` then physically drive ``frame``."""

stop_tasks()

Stop source-owned clocks and background drive tasks.

Source code in src/fpga_verification/sim/agents/frame_source.py

def stop_tasks(self) -> None:
    """Stop source-owned clocks and background drive tasks."""

FrameTransaction dataclass

A frame published before its source drives physical conduit signals.

Source code in src/fpga_verification/sim/agents/frame_source.py

@dataclass(frozen=True)
class FrameTransaction:
    """A frame published before its source drives physical conduit signals."""

    frame: object
    size: FrameSize

Scoreboards

fpga_verification.sim.scoreboards

Reusable pyuvm scoreboards.

AnalysisImp

Bases: uvm_analysis_export

pyuvm analysis export forwarding writes to a Python callable.

Source code in src/fpga_verification/sim/scoreboards/analysis.py

class AnalysisImp(uvm_analysis_export):
    """pyuvm analysis export forwarding writes to a Python callable."""

    def __init__(self, name, parent, write_fn):
        super().__init__(name, parent)
        self.write_fn = write_fn

    def write(self, item):
        self.write_fn(item)

BaseVIPScoreboard

Bases: uvm_scoreboard

Ordered protocol-level comparison engine for one VIP output stream.

Subclasses implement IP semantics through process_input_packet(), process_control_transaction() and on_reset(). They create all expectations explicitly with :meth:add_expectation; this base class only enforces packet order and the selected protocol-level comparison mode.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

class BaseVIPScoreboard(uvm_scoreboard):
    """Ordered protocol-level comparison engine for one VIP output stream.

    Subclasses implement IP semantics through ``process_input_packet()``,
    ``process_control_transaction()`` and ``on_reset()``.  They create all
    expectations explicitly with :meth:`add_expectation`; this base class only
    enforces packet order and the selected protocol-level comparison mode.
    """

    user_packet_policy = UserPacketPolicy.DROP

    def __init__(
        self,
        name,
        parent,
        source_fmt=None,
        sink_fmt=None,
        *,
        clock=None,
        reset=None,
        reset_active_level=True,
        quiet_cycles=0,
    ):
        super().__init__(name, parent)
        if sink_fmt is None:
            raise ValueError("sink_fmt must be provided")
        if int(quiet_cycles) < 0:
            raise ValueError("quiet_cycles must be >= 0")

        self.data_in_export = (
            AnalysisImp("data_in_export", self, self._process_input_packet)
            if source_fmt is not None
            else None
        )
        self.data_out_export = AnalysisImp(
            "data_out_export", self, self._process_output_packet
        )
        self.control_export = AnalysisImp(
            "control_export", self, self._process_control_transaction
        )
        self.log = logging.getLogger(f"cocotb.base_vip_scoreboard.{name}")

        self.vip_input_codec = (
            IntelVIPFrameCodec(source_fmt) if source_fmt is not None else None
        )
        self.vip_output_codec = IntelVIPFrameCodec(sink_fmt)
        self.clock = clock
        self.reset_signal = reset
        self.reset_active_level = bool(reset_active_level)
        self.quiet_cycles = int(quiet_cycles)
        self._in_reset = False

        self.expected_queue = deque()
        self.last_output_size = None
        self.protocol_epoch = 0
        self.input_frames_cnt = 0
        self.output_frames_cnt = 0
        self.waited_frames_cnt = 0

        self._failure = None
        self._frame_checked = Event()
        self._queue_changed = Event()
        self._reset_task = None

    def add_expectation(self, expectation: PacketExpectation):
        """Append exactly one output packet expectation in protocol order."""
        if not isinstance(expectation, PacketExpectation):
            raise TypeError("expectation must be a PacketExpectation")
        if self._failure is not None:
            return
        self.expected_queue.append(expectation)
        self._queue_changed.set()

    def process_input_packet(self, packet: VIPPacket):
        """Map an observed input packet to zero or more output expectations."""

    def process_control_transaction(self, transaction):
        """Apply an observed Avalon-MM/control transaction to IP state."""

    def on_reset(self):
        """Reset transient custom-scoreboard/behaviour-model state."""

    def get_frame_count(self):
        return self.output_frames_cnt

    def process_frame_done(self):
        self.output_frames_cnt += 1
        self._frame_checked.set()

    def _record_failure(self, exc):
        if self._failure is None:
            self._failure = exc
        self._frame_checked.set()
        self._queue_changed.set()

    def _raise_failure(self):
        if self._failure is not None:
            raise self._failure

    def _process_input_packet(self, packet):
        if self._in_reset or self._failure is not None:
            return
        try:
            self.process_input_packet(packet)
            if isinstance(packet, VIPVideoPacket):
                self.input_frames_cnt += 1
        except Exception as exc:
            self._record_failure(exc)

    def _process_control_transaction(self, transaction):
        if self._in_reset or self._failure is not None:
            return
        try:
            self.process_control_transaction(transaction)
        except Exception as exc:
            self._record_failure(exc)

    def _process_output_packet(self, packet):
        """Match one observed output packet against the next expectation."""
        if self._in_reset or self._failure is not None:
            return
        try:
            if not self.expected_queue:
                raise AssertionError(f"Unexpected output {self._packet_name(packet)} packet")

            expectation = self.expected_queue[0]
            if type(packet) is not type(expectation.packet):
                raise AssertionError(
                    f"Unexpected output {self._packet_name(packet)} packet; "
                    f"expected {self._packet_name(expectation.packet)}"
                )

            self.expected_queue.popleft()
            self._queue_changed.set()
            self._compare_packet(packet, expectation)
            if isinstance(packet, VIPControlPacket):
                self.last_output_size = FrameSize(packet.width, packet.height)
            if isinstance(packet, VIPVideoPacket):
                self.process_frame_done()
        except Exception as exc:
            self._record_failure(exc)

    @staticmethod
    def _packet_name(packet):
        if isinstance(packet, VIPControlPacket):
            return "CONTROL"
        if isinstance(packet, VIPVideoPacket):
            return "VIDEO"
        if isinstance(packet, VIPUserPacket):
            return f"USER{packet.user_type}"
        return getattr(getattr(packet, "packet_type", None), "name", "UNKNOWN")

    def _compare_packet(self, packet, expectation):
        packet_ref = expectation.packet
        if expectation.check is CheckMode.SHAPE:
            self._compare_shape(packet, packet_ref)
            return

        if isinstance(packet, VIPControlPacket):
            self._compare_control_packet(packet, packet_ref)
        elif isinstance(packet, VIPUserPacket):
            self._compare_user_packet(packet, packet_ref, exact=True)
        elif isinstance(packet, VIPVideoPacket):
            self._compare_video_packet(packet, packet_ref, expectation.tolerance)
        else:
            raise AssertionError(f"Unsupported VIP packet type: {packet.packet_type}")

    def _compare_shape(self, packet, packet_ref):
        if isinstance(packet, VIPControlPacket):
            self._compare_control_packet(packet, packet_ref)
        elif isinstance(packet, VIPUserPacket):
            self._compare_user_packet(packet, packet_ref, exact=False)
        elif isinstance(packet, VIPVideoPacket):
            assert len(packet.payload) == len(packet_ref.payload), (
                f"expected VIDEO payload length {len(packet_ref.payload)}, "
                f"got {len(packet.payload)}"
            )
        else:
            raise AssertionError(f"Unsupported VIP packet type: {packet.packet_type}")

    @staticmethod
    def _compare_control_packet(packet, packet_ref):
        assert packet_ref.width == packet.width, (
            f"expected width {packet_ref.width}, got {packet.width}"
        )
        assert packet_ref.height == packet.height, (
            f"expected height {packet_ref.height}, got {packet.height}"
        )
        assert packet.interlacing.is_interlaced == packet_ref.interlacing.is_interlaced, (
            f"expected interlacing {packet_ref.interlacing}, "
            f"got {packet.interlacing}"
        )

    @staticmethod
    def _compare_user_packet(packet, packet_ref, *, exact):
        assert packet.user_type == packet_ref.user_type, (
            f"expected USER{packet_ref.user_type}, got USER{packet.user_type}"
        )
        if exact:
            assert packet.payload == packet_ref.payload, "USER payload mismatch"
        else:
            assert len(packet.payload) == len(packet_ref.payload), (
                f"expected USER payload length {len(packet_ref.payload)}, "
                f"got {len(packet.payload)}"
            )

    def _compare_video_packet(self, packet, packet_ref, tolerance):
        if self.last_output_size is None:
            raise AssertionError("VIDEO packet arrived before CONTROL packet")
        frame_ref = self.vip_output_codec.video_packet_to_frame(packet_ref, self.last_output_size)
        frame_out = self.vip_output_codec.video_packet_to_frame(packet, self.last_output_size)
        compare_frames(frame_out, frame_ref, tolerance=int(tolerance))

    def _reset_active(self):
        try:
            value = bool(int(self.reset_signal.value))
        except ValueError:
            # Top-level inputs can be unresolved before the testbench performs
            # its first drive (notably under Questa).  Defer classification
            # until the first signal change instead of killing the watcher.
            return False
        return value == self.reset_active_level

    def reset(self):
        """Abort pending protocol work while preserving an already-found error."""
        self._in_reset = True
        self.expected_queue.clear()
        self.last_output_size = None
        self.protocol_epoch += 1
        self._queue_changed.set()
        try:
            self.on_reset()
        except Exception as exc:
            self._record_failure(exc)

    async def _watch_reset(self):
        if self._reset_active():
            self.reset()
        while True:
            trigger = getattr(self.reset_signal, "value_change", None)
            await trigger if trigger is not None else Edge(self.reset_signal)
            if self._reset_active():
                self.reset()
            else:
                self._in_reset = False

    async def run_phase(self):
        if self.reset_signal is not None:
            self._reset_task = cocotb.start_soon(self._watch_reset())

    def check_phase(self):
        self._raise_failure()
        if self.expected_queue:
            raise AssertionError(
                f"Missing output packets: {len(self.expected_queue)} expectation(s) remain"
            )

    async def wait_frame_checked(self, timeout=1, timeout_unit="ms", *, after=None):
        """Wait until one more expected VIDEO packet has been checked."""
        target = self.waited_frames_cnt + 1 if after is None else int(after) + 1
        while self.get_frame_count() < target and self._failure is None:
            self._frame_checked.clear()
            try:
                await with_timeout(self._frame_checked.wait(), timeout, timeout_unit)
            except SimTimeoutError as exc:
                raise AssertionError(
                    f"Frame check timeout after {timeout} {timeout_unit}: "
                    f"got {self.get_frame_count()} of {target} expected frames"
                ) from exc
        self._raise_failure()
        self.waited_frames_cnt = max(self.waited_frames_cnt, target)

    async def drain(self, timeout, timeout_unit="ms", *, quiet_cycles=None):
        """Wait for all expectations, then require a quiet output window.

        An output packet during the quiet window is always a failure: by then
        there is no matching expectation left.  ``quiet_cycles=0`` disables the
        additional observation window.
        """
        cycles = self.quiet_cycles if quiet_cycles is None else int(quiet_cycles)
        if cycles < 0:
            raise ValueError("quiet_cycles must be >= 0")
        if cycles and self.clock is None:
            raise RuntimeError("drain quiet window requires scoreboard clock")

        while True:
            while self.expected_queue and self._failure is None:
                self._queue_changed.clear()
                try:
                    await with_timeout(
                        self._queue_changed.wait(),
                        timeout,
                        timeout_unit,
                    )
                except SimTimeoutError as exc:
                    raise AssertionError(
                        f"Drain timeout after {timeout} {timeout_unit}: "
                        f"{len(self.expected_queue)} expectation(s) remain"
                    ) from exc

            self._raise_failure()

            if cycles == 0:
                return

            for _ in range(cycles):
                await RisingEdge(self.clock)
                # Let monitors triggered by the same clock edge publish before
                # deciding whether the output stream remained quiet.
                await NextTimeStep()
                self._raise_failure()
                if self.expected_queue:
                    break
            else:
                return

add_expectation(expectation)

Append exactly one output packet expectation in protocol order.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

def add_expectation(self, expectation: PacketExpectation):
    """Append exactly one output packet expectation in protocol order."""
    if not isinstance(expectation, PacketExpectation):
        raise TypeError("expectation must be a PacketExpectation")
    if self._failure is not None:
        return
    self.expected_queue.append(expectation)
    self._queue_changed.set()

process_input_packet(packet)

Map an observed input packet to zero or more output expectations.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

def process_input_packet(self, packet: VIPPacket):
    """Map an observed input packet to zero or more output expectations."""

process_control_transaction(transaction)

Apply an observed Avalon-MM/control transaction to IP state.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

def process_control_transaction(self, transaction):
    """Apply an observed Avalon-MM/control transaction to IP state."""

on_reset()

Reset transient custom-scoreboard/behaviour-model state.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

def on_reset(self):
    """Reset transient custom-scoreboard/behaviour-model state."""

reset()

Abort pending protocol work while preserving an already-found error.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

def reset(self):
    """Abort pending protocol work while preserving an already-found error."""
    self._in_reset = True
    self.expected_queue.clear()
    self.last_output_size = None
    self.protocol_epoch += 1
    self._queue_changed.set()
    try:
        self.on_reset()
    except Exception as exc:
        self._record_failure(exc)

wait_frame_checked(timeout=1, timeout_unit='ms', *, after=None) async

Wait until one more expected VIDEO packet has been checked.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

async def wait_frame_checked(self, timeout=1, timeout_unit="ms", *, after=None):
    """Wait until one more expected VIDEO packet has been checked."""
    target = self.waited_frames_cnt + 1 if after is None else int(after) + 1
    while self.get_frame_count() < target and self._failure is None:
        self._frame_checked.clear()
        try:
            await with_timeout(self._frame_checked.wait(), timeout, timeout_unit)
        except SimTimeoutError as exc:
            raise AssertionError(
                f"Frame check timeout after {timeout} {timeout_unit}: "
                f"got {self.get_frame_count()} of {target} expected frames"
            ) from exc
    self._raise_failure()
    self.waited_frames_cnt = max(self.waited_frames_cnt, target)

drain(timeout, timeout_unit='ms', *, quiet_cycles=None) async

Wait for all expectations, then require a quiet output window.

An output packet during the quiet window is always a failure: by then there is no matching expectation left. quiet_cycles=0 disables the additional observation window.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

async def drain(self, timeout, timeout_unit="ms", *, quiet_cycles=None):
    """Wait for all expectations, then require a quiet output window.

    An output packet during the quiet window is always a failure: by then
    there is no matching expectation left.  ``quiet_cycles=0`` disables the
    additional observation window.
    """
    cycles = self.quiet_cycles if quiet_cycles is None else int(quiet_cycles)
    if cycles < 0:
        raise ValueError("quiet_cycles must be >= 0")
    if cycles and self.clock is None:
        raise RuntimeError("drain quiet window requires scoreboard clock")

    while True:
        while self.expected_queue and self._failure is None:
            self._queue_changed.clear()
            try:
                await with_timeout(
                    self._queue_changed.wait(),
                    timeout,
                    timeout_unit,
                )
            except SimTimeoutError as exc:
                raise AssertionError(
                    f"Drain timeout after {timeout} {timeout_unit}: "
                    f"{len(self.expected_queue)} expectation(s) remain"
                ) from exc

        self._raise_failure()

        if cycles == 0:
            return

        for _ in range(cycles):
            await RisingEdge(self.clock)
            # Let monitors triggered by the same clock edge publish before
            # deciding whether the output stream remained quiet.
            await NextTimeStep()
            self._raise_failure()
            if self.expected_queue:
                break
        else:
            return

CheckMode

Bases: str, Enum

How an expected packet is verified.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

class CheckMode(str, Enum):
    """How an expected packet is verified."""

    EXACT = "exact"
    SHAPE = "shape"

PacketExpectation dataclass

One expected output packet and its explicit comparison contract.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

@dataclass(frozen=True)
class PacketExpectation:
    """One expected output packet and its explicit comparison contract."""

    packet: VIPPacket
    check: CheckMode = CheckMode.EXACT
    reason: str = ""
    tolerance: int = 0

    def __post_init__(self):
        if not isinstance(self.packet, VIPPacket):
            raise TypeError("PacketExpectation.packet must be a VIPPacket")
        object.__setattr__(self, "check", CheckMode(self.check))
        if int(self.tolerance) < 0:
            raise ValueError("PacketExpectation.tolerance must be >= 0")

UserPacketPolicy

Bases: str, Enum

Default input-user-packet policy for custom VIP scoreboards.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

class UserPacketPolicy(str, Enum):
    """Default input-user-packet policy for custom VIP scoreboards."""

    DROP = "drop"
    PASSTHROUGH = "passthrough"

VideoPacketPolicy

Bases: str, Enum

How one input video packet maps to the output stream.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

class VideoPacketPolicy(str, Enum):
    """How one input video packet maps to the output stream."""

    EXACT = "exact"
    SHAPE = "shape"
    DROP = "drop"

VideoPacketResult dataclass

Standard behaviour-model result for one input video packet.

Source code in src/fpga_verification/sim/scoreboards/intel_video.py

@dataclass(frozen=True)
class VideoPacketResult:
    """Standard behaviour-model result for one input video packet."""

    policy: VideoPacketPolicy
    expected: object = None
    reason: str = ""
    tolerance: int = 0

    def __post_init__(self):
        object.__setattr__(self, "policy", VideoPacketPolicy(self.policy))
        object.__setattr__(self, "tolerance", int(self.tolerance))
        if self.tolerance < 0:
            raise ValueError("VideoPacketResult.tolerance must be >= 0")
        if self.policy is VideoPacketPolicy.EXACT and self.expected is None:
            raise ValueError("EXACT video result requires an expected value")
        if self.policy is VideoPacketPolicy.DROP:
            if self.expected is not None:
                raise ValueError("DROP video result cannot have an expected value")
            if self.tolerance:
                raise ValueError("DROP video result cannot have a tolerance")

    @classmethod
    def exact(cls, expected, *, tolerance=0, reason=""):
        return cls(
            VideoPacketPolicy.EXACT,
            expected=expected,
            reason=reason,
            tolerance=tolerance,
        )

    @classmethod
    def shape(cls, expected=None, *, reason=""):
        return cls(
            VideoPacketPolicy.SHAPE,
            expected=expected,
            reason=reason,
        )

    @classmethod
    def drop(cls, *, reason=""):
        return cls(VideoPacketPolicy.DROP, reason=reason)

Performance metrics

fpga_verification.sim.stream_metrics

Reusable performance metrics for packetized streaming interfaces.

PacketObservation dataclass

One packet observed at the input and output of a DUT or pipeline.

Source code in src/fpga_verification/sim/stream_metrics.py

@dataclass(frozen=True)
class PacketObservation:
    """One packet observed at the input and output of a DUT or pipeline."""

    name: str
    beats: int
    input_frame: Any
    output_frame: Any

PacketMetrics dataclass

Latency and throughput metrics for one packet.

Source code in src/fpga_verification/sim/stream_metrics.py

@dataclass(frozen=True)
class PacketMetrics:
    """Latency and throughput metrics for one packet."""

    name: str
    beats: int
    input_cycles: int
    input_efficiency: float
    input_stalls: int
    output_cycles: int
    output_efficiency: float
    output_bubbles: int
    sop_latency: int
    eop_latency: int
    end_to_end_cycles: int

    def log(self, logger):
        logger.info(
            "%s metrics:\n beats=%d,\n input=%d cycles "
            "(%.2f%%, stalls=%d),\n output=%d cycles "
            "(%.2f%%, bubbles=%d),\n SoP latency=%d,\n "
            "EoP latency=%d,\n end-to-end=%d cycles",
            self.name,
            self.beats,
            self.input_cycles,
            self.input_efficiency * 100,
            self.input_stalls,
            self.output_cycles,
            self.output_efficiency * 100,
            self.output_bubbles,
            self.sop_latency,
            self.eop_latency,
            self.end_to_end_cycles,
        )

PacketSequenceMetrics dataclass

Throughput and packet-boundary metrics for an ordered packet sequence.

Source code in src/fpga_verification/sim/stream_metrics.py

@dataclass(frozen=True)
class PacketSequenceMetrics:
    """Throughput and packet-boundary metrics for an ordered packet sequence."""

    name: str
    packets: tuple[PacketMetrics, ...]
    total_beats: int
    input_cycles: int
    input_efficiency: float
    input_stalls: int
    output_cycles: int
    output_efficiency: float
    output_bubbles: int
    input_packet_gaps: tuple[int, ...]
    output_packet_gaps: tuple[int, ...]
    input_sop_intervals: tuple[int, ...]
    output_sop_intervals: tuple[int, ...]
    boundary_overlaps: tuple[bool, ...]
    max_packets_in_flight: int

    @property
    def max_input_packet_gap(self):
        return max(self.input_packet_gaps, default=0)

    @property
    def max_output_packet_gap(self):
        return max(self.output_packet_gaps, default=0)

    @property
    def sustainable_efficiency(self):
        """Conservative sustained throughput relative to one beat per cycle."""

        return min(self.input_efficiency, self.output_efficiency)

    @property
    def required_clock_multiplier(self):
        """Clock multiplier needed to match an ideal one-beat/cycle stream."""

        return 1 / self.sustainable_efficiency

    def assert_input_packet_gap_at_most(self, maximum_cycles):
        """Require every next input SoP to be accepted within the limit."""

        violating = [
            (index, gap)
            for index, gap in enumerate(self.input_packet_gaps)
            if gap > maximum_cycles
        ]
        if violating:
            details = ", ".join(
                f"{self.packets[index].name}->{self.packets[index + 1].name}: "
                f"{gap} cycles"
                for index, gap in violating
            )
            raise AssertionError(
                f"{self.name}: input packet gap exceeds {maximum_cycles} cycles: "
                f"{details}"
            )

    def assert_all_boundaries_overlap(self):
        """Require the next packet to enter before the previous packet exits."""

        missing = [
            index
            for index, overlaps in enumerate(self.boundary_overlaps)
            if not overlaps
        ]
        if missing:
            details = ", ".join(
                f"{self.packets[index].name}->{self.packets[index + 1].name}"
                for index in missing
            )
            raise AssertionError(
                f"{self.name}: packet boundaries without pipeline overlap: {details}"
            )

    def log(self, logger):
        logger.info(
            "%s sequence metrics:\n packets=%d, beats=%d,\n "
            "input=%d cycles (%.2f%%, stalls=%d),\n "
            "output=%d cycles (%.2f%%, bubbles=%d),\n "
            "input packet gaps=%s (max=%d),\n "
            "output packet gaps=%s (max=%d),\n "
            "input SoP intervals=%s,\n output SoP intervals=%s,\n "
            "overlapping boundaries=%d/%d, max packets in flight=%d,\n "
            "sustainable efficiency=%.2f%%, required clock multiplier=%.4fx",
            self.name,
            len(self.packets),
            self.total_beats,
            self.input_cycles,
            self.input_efficiency * 100,
            self.input_stalls,
            self.output_cycles,
            self.output_efficiency * 100,
            self.output_bubbles,
            _format_samples(self.input_packet_gaps),
            self.max_input_packet_gap,
            _format_samples(self.output_packet_gaps),
            self.max_output_packet_gap,
            _format_samples(self.input_sop_intervals),
            _format_samples(self.output_sop_intervals),
            sum(self.boundary_overlaps),
            len(self.boundary_overlaps),
            self.max_packets_in_flight,
            self.sustainable_efficiency * 100,
            self.required_clock_multiplier,
        )

sustainable_efficiency property

Conservative sustained throughput relative to one beat per cycle.

required_clock_multiplier property

Clock multiplier needed to match an ideal one-beat/cycle stream.

assert_input_packet_gap_at_most(maximum_cycles)

Require every next input SoP to be accepted within the limit.

Source code in src/fpga_verification/sim/stream_metrics.py

def assert_input_packet_gap_at_most(self, maximum_cycles):
    """Require every next input SoP to be accepted within the limit."""

    violating = [
        (index, gap)
        for index, gap in enumerate(self.input_packet_gaps)
        if gap > maximum_cycles
    ]
    if violating:
        details = ", ".join(
            f"{self.packets[index].name}->{self.packets[index + 1].name}: "
            f"{gap} cycles"
            for index, gap in violating
        )
        raise AssertionError(
            f"{self.name}: input packet gap exceeds {maximum_cycles} cycles: "
            f"{details}"
        )

assert_all_boundaries_overlap()

Require the next packet to enter before the previous packet exits.

Source code in src/fpga_verification/sim/stream_metrics.py

def assert_all_boundaries_overlap(self):
    """Require the next packet to enter before the previous packet exits."""

    missing = [
        index
        for index, overlaps in enumerate(self.boundary_overlaps)
        if not overlaps
    ]
    if missing:
        details = ", ".join(
            f"{self.packets[index].name}->{self.packets[index + 1].name}"
            for index in missing
        )
        raise AssertionError(
            f"{self.name}: packet boundaries without pipeline overlap: {details}"
        )

StreamPerformanceAnalyzer

Calculate cycle-based metrics from timestamped monitor frames.

Frames must expose sim_time_start and sim_time_end attributes, as AvalonSTFrame does. The analyzer is otherwise independent of packet contents and can therefore measure a single IP or a complete pipeline.

Source code in src/fpga_verification/sim/stream_metrics.py

class StreamPerformanceAnalyzer:
    """Calculate cycle-based metrics from timestamped monitor frames.

    Frames must expose ``sim_time_start`` and ``sim_time_end`` attributes, as
    ``AvalonSTFrame`` does. The analyzer is otherwise independent of packet
    contents and can therefore measure a single IP or a complete pipeline.
    """

    def __init__(self):
        self._clock_period_steps = None

    @classmethod
    async def from_clock(cls, clock):
        """Measure the simulation clock period without requiring its frequency."""

        from cocotb.triggers import RisingEdge
        from cocotb.utils import get_sim_time

        analyzer = cls()
        await RisingEdge(clock)
        start = get_sim_time(unit="step")
        await RisingEdge(clock)
        end = get_sim_time(unit="step")
        analyzer._set_clock_period_steps(end - start)
        return analyzer

    @classmethod
    def _from_clock_period_steps(cls, clock_period_steps):
        """Construct an analyzer with a known period for unit testing."""

        analyzer = cls()
        analyzer._set_clock_period_steps(clock_period_steps)
        return analyzer

    def _set_clock_period_steps(self, clock_period_steps):
        clock_period_steps = int(clock_period_steps)
        if clock_period_steps <= 0:
            raise ValueError("clock_period_steps must be greater than zero")
        self._clock_period_steps = clock_period_steps

    def packet(self, observation):
        self._validate_observation(observation)

        input_cycles = self._frame_cycles(observation.input_frame)
        output_cycles = self._frame_cycles(observation.output_frame)

        return PacketMetrics(
            name=observation.name,
            beats=observation.beats,
            input_cycles=input_cycles,
            input_efficiency=observation.beats / input_cycles,
            input_stalls=input_cycles - observation.beats,
            output_cycles=output_cycles,
            output_efficiency=observation.beats / output_cycles,
            output_bubbles=output_cycles - observation.beats,
            sop_latency=self._cycle_delta(
                observation.input_frame.sim_time_start,
                observation.output_frame.sim_time_start,
            ),
            eop_latency=self._cycle_delta(
                observation.input_frame.sim_time_end,
                observation.output_frame.sim_time_end,
            ),
            end_to_end_cycles=self._cycle_delta(
                observation.input_frame.sim_time_start,
                observation.output_frame.sim_time_end,
            ) + 1,
        )

    def sequence(self, name, observations: Sequence[PacketObservation]):
        observations = tuple(observations)
        if not observations:
            raise ValueError("observations must contain at least one packet")

        for observation in observations:
            self._validate_observation(observation)

        packets = tuple(self.packet(observation) for observation in observations)
        total_beats = sum(packet.beats for packet in packets)

        input_cycles = self._cycle_delta(
            observations[0].input_frame.sim_time_start,
            observations[-1].input_frame.sim_time_end,
        ) + 1
        output_cycles = self._cycle_delta(
            observations[0].output_frame.sim_time_start,
            observations[-1].output_frame.sim_time_end,
        ) + 1

        input_packet_gaps = tuple(
            self._cycle_delta(
                previous.input_frame.sim_time_end,
                current.input_frame.sim_time_start,
            ) - 1
            for previous, current in _pairwise(observations)
        )
        output_packet_gaps = tuple(
            self._cycle_delta(
                previous.output_frame.sim_time_end,
                current.output_frame.sim_time_start,
            ) - 1
            for previous, current in _pairwise(observations)
        )
        input_sop_intervals = tuple(
            self._cycle_delta(
                previous.input_frame.sim_time_start,
                current.input_frame.sim_time_start,
            )
            for previous, current in _pairwise(observations)
        )
        output_sop_intervals = tuple(
            self._cycle_delta(
                previous.output_frame.sim_time_start,
                current.output_frame.sim_time_start,
            )
            for previous, current in _pairwise(observations)
        )
        boundary_overlaps = tuple(
            current.input_frame.sim_time_start <= previous.output_frame.sim_time_end
            for previous, current in _pairwise(observations)
        )

        if any(gap < 0 for gap in input_packet_gaps):
            raise ValueError("input packet observations overlap or are out of order")
        if any(gap < 0 for gap in output_packet_gaps):
            raise ValueError("output packet observations overlap or are out of order")

        return PacketSequenceMetrics(
            name=name,
            packets=packets,
            total_beats=total_beats,
            input_cycles=input_cycles,
            input_efficiency=total_beats / input_cycles,
            input_stalls=input_cycles - total_beats,
            output_cycles=output_cycles,
            output_efficiency=total_beats / output_cycles,
            output_bubbles=output_cycles - total_beats,
            input_packet_gaps=input_packet_gaps,
            output_packet_gaps=output_packet_gaps,
            input_sop_intervals=input_sop_intervals,
            output_sop_intervals=output_sop_intervals,
            boundary_overlaps=boundary_overlaps,
            max_packets_in_flight=_max_packets_in_flight(observations),
        )

    def _frame_cycles(self, frame):
        return self._cycle_delta(frame.sim_time_start, frame.sim_time_end) + 1

    def _cycle_delta(self, start, end):
        if self._clock_period_steps is None:
            raise RuntimeError(
                "StreamPerformanceAnalyzer is not calibrated; create it with "
                "'await StreamPerformanceAnalyzer.from_clock(clock)'"
            )
        delta_steps = int(end - start)
        cycles, remainder = divmod(delta_steps, self._clock_period_steps)
        if remainder:
            raise ValueError(
                f"timestamp difference {delta_steps} is not aligned to the "
                f"{self._clock_period_steps}-step clock period"
            )
        return cycles

    @staticmethod
    def _validate_observation(observation):
        if observation.beats <= 0:
            raise ValueError("packet beats must be greater than zero")

        for side, frame in (
            ("input", observation.input_frame),
            ("output", observation.output_frame),
        ):
            if frame.sim_time_start is None or frame.sim_time_end is None:
                raise ValueError(
                    f"{observation.name} {side} frame has incomplete timestamps"
                )
            if frame.sim_time_end < frame.sim_time_start:
                raise ValueError(
                    f"{observation.name} {side} frame ends before it starts"
                )

from_clock(clock) async classmethod

Measure the simulation clock period without requiring its frequency.

Source code in src/fpga_verification/sim/stream_metrics.py

@classmethod
async def from_clock(cls, clock):
    """Measure the simulation clock period without requiring its frequency."""

    from cocotb.triggers import RisingEdge
    from cocotb.utils import get_sim_time

    analyzer = cls()
    await RisingEdge(clock)
    start = get_sim_time(unit="step")
    await RisingEdge(clock)
    end = get_sim_time(unit="step")
    analyzer._set_clock_period_steps(end - start)
    return analyzer