morgan
/
artiq-zynq
forked from M-Labs/artiq-zynq
1
0
Fork 0
Code Pull Requests Activity

sim: add frame sim

This commit is contained in:
morgan 2024-11-11 11:22:15 +08:00
parent c5d16668a0
commit e79dac7da4
1 changed files with 157 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
sim_frame.py Normal file
View File

@ -0,0 +1,157 @@
from migen import *
from misoc.interconnect import stream
from sim_pipeline import *
from sim_generator import CXPCRC32Inserter
from src.gateware.cxp_pipeline import *
class Frame(Module):
def __init__(self):
# to construct correct crc and ack/stb signal
self.submodules.buffer = buffer = stream.SyncFIFO(word_layout, 32)
self.submodules.crc_inserter = crc_inserter = CXPCRC32Inserter()
self.submodules.dchar_decoder = dchar_decoder = Duplicated_Char_Decoder()
self.submodules.stream_pipe = stream_pipe = Stream_Pipeline()
pipeline = [buffer, crc_inserter, dchar_decoder, stream_pipe]
for s, d in zip(pipeline, pipeline[1:]):
self.comb += s.source.connect(d.sink)
self.sink = pipeline[0].sink
self.source = pipeline[-1].source
# no backpressure for sim
self.sync += self.source.ack.eq(1)
dut = Frame()
def check_case(packet=[]):
print("=================TEST========================")
sink = dut.sink
stream_pipe = dut.stream_pipe
for i, p in enumerate(packet):
yield sink.data.eq(p["data"])
yield sink.k.eq(p["k"])
yield sink.stb.eq(1)
if "eop" in p:
yield sink.eop.eq(1)
else:
yield sink.eop.eq(0)
# check cycle result
yield
# source = dut.dchar_decoder.source
source = dut.stream_pipe.frame_extractor.sink
print(
f"\nCYCLE#{i} : source char = {yield source.data:#X} k = {yield source.k:#X} stb = {yield source.stb} ack = {yield source.ack} eop = {yield source.eop}"
# f" source dchar = {yield source.dchar:#X} dchar_k = {yield source.dchar_k:#X}"
# f"\nCYCLE#{i} : stream id = {yield decoder.stream_id:#X} pak_tag = {yield decoder.pak_tag:#X}"
# f" stream_pak_size = {yield decoder.stream_pak_size:#X}"
)
# crc = downconns[1].generator.crc_inserter.crc
# crc = dut.double_buffer.crc
# print(
# f"CYCLE#{i} : crc error = {yield crc.error:#X} crc value = {yield crc.value:#X}"
# f" crc data = {yield crc.data:#X} engine next = {yield crc.engine.next:#X} ce = {yield crc.ce}"
# )
# extra clk cycles
cyc = i + 1
for i in range(cyc, cyc + 50):
yield sink.data.eq(0)
yield sink.k.eq(0)
yield sink.stb.eq(0)
yield sink.eop.eq(0)
yield
print(
f"\nCYCLE#{i} : source char = {yield source.data:#X} k = {yield source.k:#X} stb = {yield source.stb} ack = {yield source.ack} eop = {yield source.eop}"
# f" source dchar = {yield source.dchar:#X} dchar_k = {yield source.dchar_k:#X}"
# f"\nCYCLE#{i} : stream id = {yield decoder.stream_id:#X} pak_tag = {yield decoder.pak_tag:#X}"
# f" stream_pak_size = {yield decoder.stream_pak_size:#X}"
)
metadata = dut.stream_pipe.frame_extractor.metadata
img_header_layout = [
"stream_id",
"source_tag",
"x_size",
"x_offset",
"y_size",
"y_offset",
"l_size", # number of data words per image line
"pixel_format",
"tap_geo",
"flag",
]
for name in img_header_layout:
print(f"{name} = {yield getattr(metadata, name):#04X} ", end="")
print()
assert True
def testbench():
# stream_id = 0x01
streams = [
[
{"data": 0x11111111, "k": Replicate(0, 4)},
{"data": 0xB105F00D, "k": Replicate(0, 4)},
],
[
{"data": 0x22222222, "k": Replicate(0, 4)},
{"data": 0xC001BEA0, "k": Replicate(0, 4)},
],
[
{"data": Replicate(KCode["stream_marker"], 4), "k": Replicate(1, 4)},
{"data": Replicate(C(0x01,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x02,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x03,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x04,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x05,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x06,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x07,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x08,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x09,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x10,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x11,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x12,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x13,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x14,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x15,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x16,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x17,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x18,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x19,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x20,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x21,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x22,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x23,char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(0x24,char_width), 4), "k": Replicate(0, 4)},
],
]
packet = []
for i, s in enumerate(streams):
s[-1]["eop"] = 0
packet += [
{"data": Replicate(C(i % 2, char_width), 4), "k": Replicate(0, 4)},
{"data": Replicate(C(i, char_width), 4), "k": Replicate(0, 4)},
{
"data": Replicate(C(len(s) >> 8 & 0xFF, char_width), 4),
"k": Replicate(0, 4),
},
{"data": Replicate(C(len(s) & 0xFF, char_width), 4), "k": Replicate(0, 4)},
*s,
]
yield from check_case(packet)
run_simulation(dut, testbench(), vcd_name="sim-cxp.vcd")