diff --git a/designs/full_new_reflectometer/tests/full_reflectometer_test/Makefile b/designs/full_new_reflectometer/tests/full_reflectometer_test/Makefile new file mode 100644 index 0000000..bf66feb --- /dev/null +++ b/designs/full_new_reflectometer/tests/full_reflectometer_test/Makefile @@ -0,0 +1,67 @@ +TOPLEVEL_LANG = verilog +SIM ?= questa +WAVES = 1 +WLF_FILE := $(SIM_BUILD)/waves.wlf + +PWD := $(shell pwd) + + +RTL_DIR = $(PWD)/../../src +RTL_ACCUM_DIR = $(PWD)/../../src/accum/src +RTL_GENERATOR_DIR = $(PWD)/../../src/generator/src +RTL_SAMPLER_DIR = $(PWD)/../../src/sampler/src +RTL_CLK_DIR = $(PWD)/../../src/clk_ctrl_wiz/clk_wiz_0 +LIBS_DIR = $(PWD)/../../../../external/rtl_libs + + +VERILOG_SOURCES += $(LIBS_DIR)/axi/rtl/axi_pkg.sv +VERILOG_SOURCES += $(RTL_DIR)/dma_reg_pkg.sv +VERILOG_SOURCES += $(LIBS_DIR)/axi/rtl/axi_if.sv +VERILOG_SOURCES += $(LIBS_DIR)/axi/rtl/axis_if.sv +VERILOG_SOURCES += $(LIBS_DIR)/axi/rtl/axis_if_to_flat.sv +VERILOG_SOURCES += $(LIBS_DIR)/axi/rtl/axi4l_flat_to_if.sv +VERILOG_SOURCES += $(LIBS_DIR)/axi/axi_reg/axi4l_reg_map.sv +VERILOG_SOURCES += $(LIBS_DIR)/external/verilog-axi/rtl/axi_dma_rd.v +VERILOG_SOURCES += $(LIBS_DIR)/external/verilog-axi/rtl/axi_dma_wr.v +VERILOG_SOURCES += $(LIBS_DIR)/external/verilog-axi/rtl/axi_dma.v +VERILOG_SOURCES += $(LIBS_DIR)/external/verilog-axi/rtl/axi_ram.v +VERILOG_SOURCES += $(RTL_DIR)/axi_ram_wrapper.sv +VERILOG_SOURCES += $(RTL_DIR)/controller.sv +VERILOG_SOURCES += $(RTL_DIR)/dma_controller.sv +VERILOG_SOURCES += $(RTL_DIR)/shaper_axis_desc.sv +VERILOG_SOURCES += $(RTL_DIR)/shaper_axis_status.sv +VERILOG_SOURCES += $(RTL_DIR)/controller_wrapper_axil.sv +VERILOG_SOURCES += $(RTL_DIR)/axi4l_reg_map_controller_pkg.sv +VERILOG_SOURCES += $(RTL_DIR)/axis_defaults_helper.sv +VERILOG_SOURCES += $(RTL_DIR)/axi4l_reg_map_controller.sv +VERILOG_SOURCES += $(RTL_DIR)/axi_dma_wrapper_if.sv +VERILOG_SOURCES += $(RTL_DIR)/wrapper_controller_dma.sv +VERILOG_SOURCES += $(RTL_ACCUM_DIR)/adder.sv +VERILOG_SOURCES += $(RTL_ACCUM_DIR)/out_axis_fifo.sv +VERILOG_SOURCES += $(RTL_ACCUM_DIR)/accum.sv +VERILOG_SOURCES += $(RTL_ACCUM_DIR)/accum_top.sv +VERILOG_SOURCES += $(RTL_GENERATOR_DIR)/generator.sv +VERILOG_SOURCES += $(RTL_SAMPLER_DIR)/sampler.sv +VERILOG_SOURCES += $(RTL_CLK_DIR)/clk_wiz_0_clk_wiz.v +VERILOG_SOURCES += $(RTL_CLK_DIR)/clk_wiz_0.v + +VERILOG_SOURCES += $(PWD)/adc_model.sv +VERILOG_SOURCES += $(PWD)/dac_model.sv +VERILOG_SOURCES += $(PWD)/reflectometer_ip.sv +VERILOG_SOURCES += $(PWD)/reflectometer_and_dma_wrapper.sv +VERILOG_SOURCES += $(PWD)/tb_full_reflectometer.sv +VERILOG_SOURCES += /mnt/c/Xilinx/Vivado/2021.2/data/verilog/src/glbl.v + +TOPLEVEL = tb_full_reflectometer +MODULE = full_reflectometer_test + + +ifeq ($(SIM),questa) + SIM_ARGS += -L xpm -L unisim + SIM_ARGS += work.glbl + SIM_ARGS += -wlf $(WLF_FILE) + COMPILE_ARGS += +acc +endif + + +include $(shell cocotb-config --makefiles)/Makefile.sim diff --git a/designs/full_new_reflectometer/tests/full_reflectometer_test/drivers.py b/designs/full_new_reflectometer/tests/full_reflectometer_test/drivers.py new file mode 100644 index 0000000..70f3796 --- /dev/null +++ b/designs/full_new_reflectometer/tests/full_reflectometer_test/drivers.py @@ -0,0 +1,191 @@ +import cocotb +from cocotb.clock import Clock +from cocotb.triggers import RisingEdge +from cocotbext.axi import AxiLiteBus, AxiLiteMaster +from cocotbext.axi import AxiStreamBus, AxiStreamSink, AxiStreamSource, AxiStreamFrame + +from reg_map import * + +def reg_addr(reg_index: int) -> int: + # AXI-Lite uses byte addresses, 32-bit registers are spaced by 4 bytes. + return reg_index * 4 + + +def u32(value: int) -> bytes: + return int(value & 0xFFFFFFFF).to_bytes(4, "little") + +class Drivers: + def __init__(self, dut): + self.dut = dut + + cocotb.start_soon(Clock(dut.ctrl_clk, 10, unit="ns").start()) + + self.axil = AxiLiteMaster( + AxiLiteBus.from_prefix(dut, "s_axil"), + dut.ctrl_clk, + dut.rst) + + + # self.axis_source = AxiStreamSource( + # AxiStreamBus.from_prefix(dut, "s_axis_write_data"), + # dut.ctrl_clk, + # dut.rst) + + self.axis_sink = AxiStreamSink( + AxiStreamBus.from_prefix(dut, "m_axis_read_data"), + dut.ctrl_clk, + dut.rst) + + async def reset(self): + self.dut.rst.value = 1 + for _ in range(20): + await RisingEdge(self.dut.ctrl_clk) + self.dut.rst.value = 0 + for _ in range(20): + await RisingEdge(self.dut.ctrl_clk) + + async def wait_locked(self, timeout_cycles=1000): + for _ in range(timeout_cycles): + if int(self.dut.locked.value) == 1: + print("MMCM locked") + return + + await RisingEdge(self.dut.ctrl_clk) + + raise TimeoutError( f"Timeout waiting for locked == 1. " f"Current locked = {self.dut.locked.value}") + + async def write_reg(self, reg_index: int, value: int): + await self.axil.write(reg_addr(reg_index), u32(value)) + + async def read_reg(self, reg_index: int) -> int: + resp = await self.axil.read(reg_addr(reg_index), 4) + return int.from_bytes(bytes(resp.data), "little") + + async def pulse_control(self, mask): + await self.write_reg( REG_CONTROL, mask ) + + # Reflectometer Driver + + async def configure_reflectometer ( + self, + pulse_width, + pulse_period, + pulse_num, + pulse_height, + adc_period, + window_size, + timeout_cycles): + await self.write_reg(REG_DAC_WIDTH, pulse_width) + await self.write_reg(REG_DAC_PERIOD, pulse_period) + await self.write_reg(REG_DAC_PULSE_NUM, pulse_num) + await self.write_reg(REG_DAC_PULSE_HEIGHT, pulse_height) + await self.write_reg(REG_ADC_PERIOD, adc_period) + await self.write_reg(REG_WINDOW_SIZE, window_size) + + await self.pulse_control(CTRL_CFG_BUS_VALID) + + control = self.dut.dut.reflectometer_top_inst.controller_wrapper_axil_inst.controller + + for cycle in range(timeout_cycles): + dac_wait = int(control.cfg_wait_dac_ack.value) + adc_wait = int(control.cfg_wait_adc_ack.value) + + if dac_wait == 0 and adc_wait == 0: + print(f"Configuration done after {cycle} ctrl_clk cycles") + return + await RisingEdge(self.dut.ctrl_clk) + + async def send_start(self): + await self.pulse_control(CTRL_START) + + async def soft_reset(self): + await self.pulse_control(CTRL_RST_SOFT) + + async def get_status(self): + status = await self.read_reg(REG_STATUS) + return { + "busy": bool(status & STATUS_BUSY), + "processing_done": bool(status & STATUS_PROCESSING_DONE), + "desc_read_busy": bool(status & STATUS_DESC_READ_BUSY), + "desc_write_busy": bool(status & STATUS_DESC_WRITE_BUSY), + "status_read_busy": bool(status & STATUS_STATUS_READ_BUSY), + "status_write_busy": bool(status & STATUS_STATUS_WRITE_BUSY), + "desc_read_hs": bool(status & STATUS_DESC_READ_HS), + "desc_write_hs": bool(status & STATUS_DESC_WRITE_HS), + "status_read_hs": bool(status & STATUS_STATUS_READ_HS), + "status_write_hs": bool(status & STATUS_STATUS_WRITE_HS) + } + + async def wait_status(self, field, value=True, timeout_cycles=1000): + + for _ in range(timeout_cycles): + status = await self.get_status() + + if status[field] == value: + return + + await RisingEdge(self.dut.ctrl_clk) + + status = await self.get_status() + + raise TimeoutError( f"Timeout waiting for status.{field} == {value}. " f"Current status = {status}") + + async def wait_processing_done(self, timeout_cycles=1000): + await self.wait_status("processing_done", True, timeout_cycles) + + + async def wait_finish(self, timeout_cycles=1000): + await self.wait_status("busy", False, timeout_cycles) + + # DMA Driver + + async def send_desc_write(self, addr, length_tag): + + await self.write_reg(REG_DESC_WRITE_ADDR, addr) + await self.write_reg(REG_DESC_WRITE_LEN_AND_TAG, length_tag) + await self.pulse_control(CTRL_SEND_DESC_WRITE) + + async def send_desc_read(self, addr, length, config): + + await self.write_reg(REG_DESC_READ_ADDR, addr) + await self.write_reg(REG_DESC_READ_LEN, length) + await self.write_reg(REG_DESC_READ_CONFIG, config) + await self.pulse_control( CTRL_SEND_DESC_READ) + + async def take_status_write(self, status_write_len, status_write_config): + await self.write_reg(REG_CONTROL, CTRL_TAKE_STATUS_WRITE) + + + assert await self.read_reg(REG_STATUS_WRITE_CONFIG) == status_write_config + assert await self.read_reg(REG_STATUS_WRITE_LEN) == status_write_len + + async def take_status_read(self, status_read): + await self.pulse_control( CTRL_TAKE_STATUS_READ) + for _ in range(10): + await RisingEdge(self.dut.ctrl_clk) + + assert await self.read_reg(REG_READ_STATUS) == status_read + + async def wait_dma_write_done(self, timeout_cycles=1000): + await self.wait_status("desc_write_busy", False, timeout_cycles) + + async def wait_dma_read_done(self, timeout_cycles=1000): + await self.wait_status("desc_read_busy", False, timeout_cycles) + + async def wait_status_read_handshake(self, timeout_cycles=1000): + await self.wait_status("status_read_hs", True, timeout_cycles) + + async def wait_status_write_handshake(self, timeout_cycles=1000): + await self.wait_status("status_write_hs", True, timeout_cycles) + + # AxiStream Driver + + # async def send_axis_data(self, data: bytes): + + # await self.axis_source.send(AxiStreamFrame(data) ) + + async def receive_axis_data(self): + + frame = await self.axis_sink.recv() + return bytes(frame) + diff --git a/designs/full_new_reflectometer/tests/full_reflectometer_test/full_reflectometer_test.py b/designs/full_new_reflectometer/tests/full_reflectometer_test/full_reflectometer_test.py new file mode 100644 index 0000000..7ac3d11 --- /dev/null +++ b/designs/full_new_reflectometer/tests/full_reflectometer_test/full_reflectometer_test.py @@ -0,0 +1,175 @@ +import cocotb +from cocotb.triggers import RisingEdge + +from drivers import Drivers +from reference_model import Reference_model +from scoreboard import Scoreboard + +from reg_map import * + +class TB: + + def __init__( + self, + dut, + pulse_width, + pulse_period, + pulse_num, + pulse_height, + adc_period, + window_size + ): + self.dut = dut + + self.driver = Drivers(dut) + + self.reference = Reference_model( + dut=dut, + + pulse_width=pulse_width, + pulse_period=pulse_period, + pulse_num=pulse_num, + pulse_height=pulse_height, + adc_period=adc_period, + window_size=window_size, + + DAC_DATA_WIDTH=DAC_DATA_WIDTH, + ADC_DATA_WIDTH=ADC_DATA_WIDTH, + PACK_FACTOR=PACK_FACTOR, + PROCESS_MODE=PROCESS_MODE, + ZERO_LEVEL=ZERO_LEVEL, + ACCUM_WIDTH=ACCUM_WIDTH, + N_MAX=N_MAX, + PACKET_SIZE=PACKET_SIZE, + RD_FIFO_WIDTH=RD_FIFO_WIDTH, + ) + + self.scoreboard = Scoreboard() + + +SEQ_NUM = 1 +WINDOW_SIZE = 1 + +PULSE_WIDTH = 10 +PULSE_PERIOD = 20 +PULSE_HEIGHT = 15000 +ADC_PERIOD = 20 + +@cocotb.test() +async def rest_init(dut): + tb = TB(dut, + pulse_width=PULSE_WIDTH, + pulse_period=PULSE_PERIOD, + pulse_num=SEQ_NUM, + pulse_height=PULSE_HEIGHT, + adc_period=ADC_PERIOD, + window_size=WINDOW_SIZE) + await tb.driver.reset() + +@cocotb.test() +async def base_full_test(dut): + tb = TB(dut, + pulse_width=PULSE_WIDTH, + pulse_period=PULSE_PERIOD, + pulse_num=SEQ_NUM, + pulse_height=PULSE_HEIGHT, + adc_period=ADC_PERIOD, + window_size=WINDOW_SIZE + ) + await tb.driver.reset() + await tb.driver.wait_locked() + await tb.driver.soft_reset() + + RESULT_ADDR = 0x1000 + + await tb.driver.configure_reflectometer( + pulse_width = PULSE_WIDTH, + pulse_period=PULSE_PERIOD, + pulse_num=SEQ_NUM, + pulse_height=PULSE_HEIGHT, + adc_period=ADC_PERIOD, + window_size=WINDOW_SIZE, + timeout_cycles=1000 + ) + + samples = tb.reference.gen_input_samples() + + SMP_NUM = len(samples[0]) + # SMP_NUM = ADC_PERIOD + + expected = tb.reference.calculate_expected( samples, WINDOW_SIZE, ACCUM_WIDTH) + + print("") + print("========================================") + print("ACCUMULATOR RANDOM TEST") + print("========================================") + print(f"seq_num = {SEQ_NUM}") + print(f"smp_num = {SMP_NUM}") + print(f"window_size = {WINDOW_SIZE}") + print(f"data_width = {ADC_DATA_WIDTH}") + print(f"accum_width = {ACCUM_WIDTH}") + print(f"expected words = {len(expected)}") + + RESULT_WORDS = len(expected) + RESULT_BYTES = RESULT_WORDS * 4 + + + print("==============================") + print("ACCUM TEST") + print("words =", RESULT_WORDS) + print("bytes =", RESULT_BYTES) + print("==============================") + + for _ in range(10): + await RisingEdge(dut.ctrl_clk) + + await tb.driver.send_start() + + for _ in range(4): + await RisingEdge(dut.ctrl_clk) + + await tb.driver.wait_processing_done(timeout_cycles=100000) + + await tb.driver.send_desc_write( addr=RESULT_ADDR, length_tag=RESULT_BYTES ) + await tb.driver.wait_dma_write_done(timeout_cycles=5000) + + for _ in range(100): + await RisingEdge(dut.ctrl_clk) + + await tb.driver.take_status_write(status_write_len=RESULT_BYTES, status_write_config=0) + + await tb.driver.send_desc_read(addr=RESULT_ADDR, length=RESULT_BYTES, config=0x0000_0001 ) + + received_data = await tb.driver.receive_axis_data() + + await tb.driver.wait_dma_read_done(timeout_cycles=5000) + + + await tb.driver.take_status_read(status_read=0x1) + + + received = tb.scoreboard.bytes_to_words(received_data, RD_FIFO_WIDTH) + + print("") + print("Expected:") + for i, value in enumerate(expected): + print(f" [{i}] = 0x{value:08X}") + + print("") + print("Received:") + for i, value in enumerate(received): + print(f" [{i}] = 0x{value:08X}") + + tb.scoreboard.check_results( + expected=expected, + received=received + ) + + print("") + print("========================================") + print("ACCUMULATOR RANDOM TEST PASSED") + print("========================================") + + + + \ No newline at end of file diff --git a/designs/full_new_reflectometer/tests/full_reflectometer_test/reference_model.py b/designs/full_new_reflectometer/tests/full_reflectometer_test/reference_model.py new file mode 100644 index 0000000..fe9cefe --- /dev/null +++ b/designs/full_new_reflectometer/tests/full_reflectometer_test/reference_model.py @@ -0,0 +1,253 @@ +from reg_map import * + +class Reference_model: + def __init__( + self, + dut, + pulse_width, + pulse_period, + pulse_num, + pulse_height, + adc_period, + window_size, + DAC_DATA_WIDTH, + ADC_DATA_WIDTH, + PACK_FACTOR, + PROCESS_MODE, + ZERO_LEVEL, + ACCUM_WIDTH, + N_MAX, + PACKET_SIZE, + RD_FIFO_WIDTH + ): + + self.dut = dut + + # configuration + self.pulse_width = pulse_width + self.pulse_period = pulse_period + self.pulse_num = pulse_num + self.pulse_height = pulse_height + self.adc_sample_num = adc_period + self.window_size = window_size + + # parameters + self.DAC_DATA_WIDTH = DAC_DATA_WIDTH + self.ADC_DATA_WIDTH = ADC_DATA_WIDTH + self.PACK_FACTOR = PACK_FACTOR + self.PROCESS_MODE = PROCESS_MODE + self.ZERO_LEVEL = ZERO_LEVEL + self.ACCUM_WIDTH = ACCUM_WIDTH + self.N_MAX = N_MAX + self.PACKET_SIZE = PACKET_SIZE + self.RD_FIFO_WIDTH = RD_FIFO_WIDTH + + # intermediate data + self.samples = [] + self.expected = [] + + + def gen_input_samples(self): + + self.samples = [] + + # ----------------------------- + # DAC: 14 bit + # ----------------------------- + DAC_ZERO = self.ZERO_LEVEL + DAC_MAX = (1 << self.DAC_DATA_WIDTH) - 1 + + DAC_RANGE = 5.0 + + DAC_STEP = ( + 2.0 * DAC_RANGE + ) / DAC_MAX + + # ----------------------------- + # ADC: 12 bit + # ----------------------------- + ADC_ZERO = 1 << (self.ADC_DATA_WIDTH - 1) + ADC_MAX = (1 << self.ADC_DATA_WIDTH) - 1 + + ADC_RANGE = 1.0 + ADC_GAIN = 0.2 + GROUND_BIAS = 0.0 + + ADC_STEP = ( + 2.0 * ADC_RANGE + ) / ADC_MAX + + # ----------------------------- + # Generate every pulse sequence + # ----------------------------- + for _ in range(self.pulse_num): + + pulse_samples = [] + + for sample_idx in range(self.adc_sample_num): + + # ========================================== + # 1. Generator produces a 14-bit DAC code + # ========================================== + if sample_idx < self.pulse_width: + dac_code = self.pulse_height + else: + dac_code = DAC_ZERO + + # Limit to actual DAC width + dac_code = max(0, min(dac_code, DAC_MAX)) + + # ========================================== + # 2. 14-bit DAC code -> analog voltage + # ========================================== + voltage = ( + (dac_code - DAC_ZERO) + * DAC_STEP + ) + + # ========================================== + # 3. Analog path -> ADC input voltage + # ========================================== + voltage = ( + (voltage - GROUND_BIAS) + * ADC_GAIN + ) + + # ========================================== + # 4. Analog voltage -> 12-bit ADC code + # ========================================== + if voltage <= -ADC_RANGE: + + adc_code = 0 + + elif voltage >= ADC_RANGE: + + adc_code = ADC_MAX + + else: + + adc_code = int( + round( + voltage / ADC_STEP + + ADC_ZERO + ) + ) + + # Make absolutely sure that the result + # is a valid 12-bit value. + adc_code = max( + 0, + min(adc_code, ADC_MAX) + ) + + # ========================================== + # 5. ADC out-of-range processing + # ========================================== + out_of_range = ( + abs(voltage) >= ADC_RANGE + ) + + if self.PROCESS_MODE: + + msb = ( + adc_code + >> (self.ADC_DATA_WIDTH - 1) + ) & 1 + + if out_of_range: + + if msb: + sample = ADC_MAX + else: + sample = 0 + + else: + + sample = ( + (((~msb) & 1) + << (self.ADC_DATA_WIDTH - 1)) + | + ( + adc_code + & ( + (1 << (self.ADC_DATA_WIDTH - 1)) + - 1 + ) + ) + ) + + else: + + if out_of_range: + + if adc_code & ADC_ZERO: + sample = ADC_MAX + else: + sample = 0 + + else: + + sample = adc_code + + # ========================================== + # Final sample is ALWAYS 12-bit ADC data + # ========================================== + sample &= ADC_MAX + + pulse_samples.append(sample) + + self.samples.append(pulse_samples) + + return self.samples + + + + + def calculate_expected( + self, + samples, + window_size: int, + accum_width: int): + if window_size <= 0: + raise ValueError( f"window_size must be > 0, got {window_size}") + + if not samples: + raise ValueError("samples must not be empty") + + seq_num = len(samples) + smp_num = len(samples[0]) + + if smp_num == 0: + raise ValueError("samples must not contain empty sequences") + + for seq_idx, seq_samples in enumerate(samples): + if len(seq_samples) != smp_num: + raise ValueError(f"Sequence {seq_idx} has {len(seq_samples)} samples, " f"expected {smp_num}" ) + + if smp_num % window_size != 0: + raise ValueError(f"smp_num ({smp_num}) must be divisible " f"by window_size ({window_size})") + + exp_word_count = smp_num // window_size + accum_mask = (1 << accum_width) - 1 + + expected = [] + + for word_idx in range(exp_word_count): + local_sum = 0 + + for seq_idx in range(seq_num): + for k in range(window_size): + sample_idx = word_idx * window_size + k + local_sum += samples[seq_idx][sample_idx] + + expected.append(local_sum & accum_mask) + + return expected + + + def run(self): + self.gen_input_samples() + + self.expected = self.calculate_expected( self.samples, self.window_size, self.ACCUM_WIDTH) + + return self.expected \ No newline at end of file diff --git a/designs/full_new_reflectometer/tests/full_reflectometer_test/reg_map.py b/designs/full_new_reflectometer/tests/full_reflectometer_test/reg_map.py new file mode 100644 index 0000000..349a119 --- /dev/null +++ b/designs/full_new_reflectometer/tests/full_reflectometer_test/reg_map.py @@ -0,0 +1,54 @@ +# Register indexes from axi4l_reg_map_controller_pkg.sv +REG_CONTROL = 0 +REG_STATUS = 1 +REG_DAC_WIDTH = 2 +REG_DAC_PERIOD = 3 +REG_DAC_PULSE_NUM = 4 +REG_DAC_PULSE_HEIGHT = 5 +REG_ADC_PERIOD = 6 +REG_WINDOW_SIZE = 7 +REG_ERROR = 8 +REG_DESC_READ_ADDR = 9 +REG_DESC_READ_LEN = 10 +REG_DESC_READ_CONFIG = 11 +REG_READ_STATUS = 12 +REG_DESC_WRITE_ADDR = 13 +REG_DESC_WRITE_LEN_AND_TAG = 14 +REG_STATUS_WRITE_LEN = 15 +REG_STATUS_WRITE_CONFIG = 16 + + +# REG_CONTROL pulse bits +CTRL_START = 1 << 0 +CTRL_RST_SOFT = 1 << 1 +CTRL_CFG_BUS_VALID = 1 << 2 +CTRL_SEND_DESC_READ = 1 << 3 +CTRL_SEND_DESC_WRITE = 1 << 4 +CTRL_TAKE_STATUS_READ = 1 << 5 +CTRL_TAKE_STATUS_WRITE = 1 << 6 + +# REG_STATUS bits +STATUS_BUSY = 1 << 0 +STATUS_PROCESSING_DONE = 1 << 1 + +STATUS_DESC_READ_BUSY = 1 << 2 +STATUS_DESC_WRITE_BUSY = 1 << 3 +STATUS_STATUS_READ_BUSY = 1 << 4 +STATUS_STATUS_WRITE_BUSY = 1 << 5 + +STATUS_DESC_READ_HS = 1 << 6 +STATUS_DESC_WRITE_HS = 1 << 7 +STATUS_STATUS_READ_HS = 1 << 8 +STATUS_STATUS_WRITE_HS = 1 << 9 + +# PARAMETERS for accumulator reference model + +DAC_DATA_WIDTH = 14 +ADC_DATA_WIDTH = 12 +PACK_FACTOR = 1 +PROCESS_MODE = 0 +ZERO_LEVEL = 8192 +ACCUM_WIDTH = 32 +N_MAX = 4096 +PACKET_SIZE = 1024 +RD_FIFO_WIDTH = 32 \ No newline at end of file diff --git a/designs/full_new_reflectometer/tests/full_reflectometer_test/scoreboard.py b/designs/full_new_reflectometer/tests/full_reflectometer_test/scoreboard.py new file mode 100644 index 0000000..e73bd8f --- /dev/null +++ b/designs/full_new_reflectometer/tests/full_reflectometer_test/scoreboard.py @@ -0,0 +1,40 @@ +from reg_map import * + +class Scoreboard: + def __init__(self): + + self.test_passed = False + + + def bytes_to_words(self, data: bytes, word_width: int = 32): + word_bytes = word_width // 8 + + if len(data) % word_bytes != 0: + raise ValueError(f"Data length {len(data)} is not divisible " f"by word size {word_bytes}" ) + + words = [] + + for i in range(0, len(data), word_bytes): + word = int.from_bytes(data[i:i + word_bytes], byteorder="little" ) + words.append(word) + + return words + + + def check_results(self, expected, received): + + assert len(received) == len(expected), ( + f"Number of words mismatch: " + f"expected={len(expected)}, " + f"received={len(received)}" ) + + for i, (exp, rec) in enumerate(zip(expected, received)): + assert rec == exp, ( + f"Payload mismatch at index {i}: " + f"expected=0x{exp:08X}, " + f"received=0x{rec:08X}" ) + self.test_passed = True + + print( f"Payload check passed: " f"{len(expected)} words") + + return True