tfg.py

#!/usr/bin/python3
from migen.fhdl.module import Module
from migen.fhdl.bitcontainer import *
from migen.fhdl.structure import Replicate
from migen import *
from migen.genlib.fsm import FSM, NextState, NextValue
from litex.soc.interconnect import stream

from testframe import testFrameDescriptor

class TestFrameGenerator(Module):
    def __init__(self, maxlen=65536, data_width=256):
        self.start = Signal()
        self.length = Signal(max=maxlen)
        self.busy = Signal()
        self.testFrame = testFrame = stream.Endpoint(
            testFrameDescriptor(data_width),
            name="tf_src")
        # # #
        beats = Signal(max=maxlen, reset_less=True)

        fsm = FSM(reset_state="IDLE")
        fsm.act("IDLE",
            If(self.start,
                NextState("RUN"),
                NextValue(self.busy, 1),
                NextValue(beats, 0),
            )
        )
        fsm.act("RUN",
            testFrame.valid.eq(1),
            testFrame.data.eq(Replicate(beats, data_width//len(beats))),
            testFrame.last.eq(beats == self.length),
            If(testFrame.ready == 1,
                If(beats == self.length,
                    NextState("DONE"),
                    NextValue(self.busy, 0)
                ).Else(
                    NextValue(beats, beats + 1)
                )
            )
        )
        fsm.act("DONE",
            self.testFrame.valid.eq(0),
            NextState("IDLE")
        )
        self.submodules.fsm = fsm

def tfg_test(dut, timeout=1000):
    yield dut.start.eq(0)
    yield dut.length.eq(10)
    yield dut.testFrame.ready.eq(0)
    yield
    yield dut.start.eq(1)
    yield 
    yield dut.start.eq(0)
    yield dut.testFrame.ready.eq(1)
    for i in range(timeout):
        yield
        busy = yield dut.busy
        print(busy)
    yield
    yield
    yield

if __name__ == "__main__":
    dut=TestFrameGenerator()
    run_simulation(dut, tfg_test(dut), vcd_name="tfg.vcd")