eth: Add GMII-1000BASE-X encode/decode modules

Signed-off-by: Alex Forencich <alex@alexforencich.com>
This commit is contained in:
Alex Forencich
2026-06-12 10:20:15 -07:00
parent 6e1e9c905f
commit 507143fbcf
8 changed files with 931 additions and 0 deletions

View File

@@ -0,0 +1,205 @@
// SPDX-License-Identifier: CERN-OHL-S-2.0
/*
Copyright (c) 2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
*/
`resetall
`timescale 1ns / 1ps
`default_nettype none
/*
* 1000BASE-X decoder
*/
module taxi_gmii_basex_dec #
(
parameter DATA_W = 16,
parameter CTRL_W = (DATA_W/8),
parameter logic GBX_IF_EN = 1'b0
)
(
input wire logic clk,
input wire logic rst,
/*
* 1000BASE-X encoded input
*/
input wire logic [DATA_W-1:0] encoded_rx_data,
input wire logic [CTRL_W-1:0] encoded_rx_data_k,
input wire logic encoded_rx_data_valid = 1'b1,
/*
* GMII interface
*/
output wire logic [DATA_W-1:0] gmii_rxd,
output wire logic [CTRL_W-1:0] gmii_rx_dv,
output wire logic [CTRL_W-1:0] gmii_rx_er,
output wire logic gmii_rx_valid,
/*
* Status
*/
output wire logic stat_rx_err_bad_block,
output wire logic stat_rx_err_framing
);
// check configuration
if (DATA_W != CTRL_W*8)
$fatal(0, "Error: DATA_W must equal CTRL_W*8 (instance %m)");
if (2**$clog2(CTRL_W) != CTRL_W)
$fatal(0, "Error: CTRL_W must be even power of two (instance %m)");
typedef enum logic [7:0] {
ETH_PRE = 8'h55,
ETH_SFD = 8'hD5
} eth_pre_t;
function [7:0] D(input [4:0] edcba, input [2:0] hgf);
D = {hgf, edcba};
endfunction
function [7:0] K(input [4:0] edcba, input [2:0] hgf);
K = {hgf, edcba};
endfunction
localparam logic [15:0] CTRL_C1 = {D(21,5), K(28,5)};
localparam logic [15:0] CTRL_C2 = {D(2,2), K(28,5)};
localparam logic [15:0] CTRL_I1 = {D(5,6), K(28,5)};
localparam logic [15:0] CTRL_I2 = {D(16,2), K(28,5)};
localparam logic [7:0] CTRL_R = K(23,7);
localparam logic [7:0] CTRL_S = K(27,7);
localparam logic [7:0] CTRL_T = K(29,7);
localparam logic [7:0] CTRL_V = K(30,7);
localparam logic [15:0] CTRL_L1 = {D(6,5), K(28,5)};
localparam logic [15:0] CTRL_L2 = {D(26,4), K(28,5)};
logic frame_reg = 1'b0, frame_next;
logic frame_cyc;
logic odd_reg = 1'b0, odd_next;
logic odd_cyc;
logic [DATA_W-1:0] gmii_rxd_reg = '0, gmii_rxd_next;
logic [CTRL_W-1:0] gmii_rx_dv_reg = '0, gmii_rx_dv_next;
logic [CTRL_W-1:0] gmii_rx_er_reg = '0, gmii_rx_er_next;
logic gmii_rx_valid_reg = '0, gmii_rx_valid_next;
logic stat_rx_err_bad_block_reg = '0, stat_rx_err_bad_block_next;
logic stat_rx_err_framing_reg = '0, stat_rx_err_framing_next;
assign gmii_rxd = gmii_rxd_reg;
assign gmii_rx_dv = gmii_rx_dv_reg;
assign gmii_rx_er = gmii_rx_er_reg;
assign gmii_rx_valid = gmii_rx_valid_reg;
assign stat_rx_err_bad_block = stat_rx_err_bad_block_reg;
assign stat_rx_err_framing = stat_rx_err_framing_reg;
always_comb begin
frame_next = frame_reg;
odd_next = odd_reg;
gmii_rxd_next = '0;
gmii_rx_dv_next = '0;
gmii_rx_er_next = '0;
gmii_rx_valid_next = 1'b0;
stat_rx_err_bad_block_next = 1'b0;
stat_rx_err_framing_next = 1'b0;
frame_cyc = frame_reg;
odd_cyc = odd_reg;
if (encoded_rx_data_valid) begin
// loop over bytes
for (integer seg = 0; seg < CTRL_W; seg = seg + 1) begin
if (CTRL_W > 1) begin
odd_cyc = 1'(seg & 1);
end
if (encoded_rx_data_k[seg]) begin
// Kx.y
if (encoded_rx_data[seg*8 +: 8] == K(28,5)) begin
// K28.5
odd_cyc = 1'b0; // sync
frame_cyc = 1'b0;
stat_rx_err_framing_next = frame_cyc;
end else if (encoded_rx_data[seg*8 +: 8] == CTRL_T) begin
// terminate
frame_cyc = 1'b0;
end else if (encoded_rx_data[seg*8 +: 8] == CTRL_R) begin
// carrier extend
frame_cyc = 1'b0;
gmii_rx_er_next[seg] = 1'b1;
stat_rx_err_framing_next = frame_cyc;
end else if (encoded_rx_data[seg*8 +: 8] == CTRL_V) begin
// error
gmii_rxd_next[seg*8 +: 8] = encoded_rx_data[seg*8 +: 8];
gmii_rx_dv_next[seg] = frame_cyc;
gmii_rx_er_next[seg] = 1'b1;
end else if (encoded_rx_data[seg*8 +: 8] == CTRL_S && odd_cyc == 0) begin
// start
frame_cyc = 1'b1;
gmii_rxd_next[seg*8 +: 8] = ETH_PRE;
gmii_rx_dv_next[seg] = 1'b1;
gmii_rx_er_next[seg] = 1'b0;
stat_rx_err_framing_next = frame_cyc;
end else begin
// unknown control character
frame_cyc = 1'b0;
stat_rx_err_bad_block_next = 1'b1;
stat_rx_err_framing_next = frame_cyc;
end
end else begin
// Dx.y
if (frame_cyc) begin
// frame data
gmii_rxd_next[seg*8 +: 8] = encoded_rx_data[seg*8 +: 8];
gmii_rx_dv_next[seg] = 1'b1;
gmii_rx_er_next[seg] = 1'b0;
end
end
odd_cyc = !odd_cyc;
end
frame_next = frame_cyc;
odd_next = odd_cyc;
gmii_rx_valid_next = 1'b1;
end
end
always_ff @(posedge clk) begin
frame_reg = frame_next;
odd_reg = odd_next;
gmii_rxd_reg = gmii_rxd_next;
gmii_rx_dv_reg = gmii_rx_dv_next;
gmii_rx_er_reg = gmii_rx_er_next;
gmii_rx_valid_reg = gmii_rx_valid_next;
stat_rx_err_bad_block_reg = stat_rx_err_bad_block_next;
stat_rx_err_framing_reg = stat_rx_err_framing_next;
if (rst) begin
frame_reg = 1'b0;
odd_reg = 1'b0;
gmii_rxd_reg = '0;
gmii_rx_dv_reg = '0;
gmii_rx_er_reg = '0;
gmii_rx_valid_reg = 1'b0;
stat_rx_err_bad_block_reg = 1'b0;
stat_rx_err_framing_reg = 1'b0;
end
end
endmodule
`resetall

View File

@@ -0,0 +1,274 @@
// SPDX-License-Identifier: CERN-OHL-S-2.0
/*
Copyright (c) 2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
*/
`resetall
`timescale 1ns / 1ps
`default_nettype none
/*
* 1000BASE-X encoder
*/
module taxi_gmii_basex_enc #
(
parameter DATA_W = 16,
parameter CTRL_W = (DATA_W/8),
parameter logic GBX_IF_EN = 1'b0,
parameter GBX_CNT = 1
)
(
input wire logic clk,
input wire logic rst,
/*
* GMII interface
*/
input wire logic [DATA_W-1:0] gmii_txd,
input wire logic [CTRL_W-1:0] gmii_tx_en,
input wire logic [CTRL_W-1:0] gmii_tx_er,
input wire logic gmii_tx_valid = 1'b1,
input wire logic [GBX_CNT-1:0] tx_gbx_sync_in = '0,
/*
* 1000BASE-X encoded interface
*/
output wire logic [DATA_W-1:0] encoded_tx_data,
output wire logic [CTRL_W-1:0] encoded_tx_data_k,
output wire logic [CTRL_W-1:0] encoded_tx_data_dm,
output wire logic [CTRL_W-1:0] encoded_tx_data_dv,
output wire logic encoded_tx_data_valid,
output wire logic [GBX_CNT-1:0] tx_gbx_sync_out
);
// check configuration
if (DATA_W != CTRL_W*8)
$fatal(0, "Error: DATA_W must equal CTRL_W*8 (instance %m)");
if (2**$clog2(CTRL_W) != CTRL_W)
$fatal(0, "Error: CTRL_W must be even power of two (instance %m)");
typedef enum logic [7:0] {
ETH_PRE = 8'h55,
ETH_SFD = 8'hD5
} eth_pre_t;
function [7:0] D(input [4:0] edcba, input [2:0] hgf);
D = {hgf, edcba};
endfunction
function [7:0] K(input [4:0] edcba, input [2:0] hgf);
K = {hgf, edcba};
endfunction
localparam logic [15:0] CTRL_C1 = {D(21,5), K(28,5)};
localparam logic [15:0] CTRL_C2 = {D(2,2), K(28,5)};
localparam logic [15:0] CTRL_I1 = {D(5,6), K(28,5)};
localparam logic [15:0] CTRL_I2 = {D(16,2), K(28,5)};
localparam logic [7:0] CTRL_R = K(23,7);
localparam logic [7:0] CTRL_S = K(27,7);
localparam logic [7:0] CTRL_T = K(29,7);
localparam logic [7:0] CTRL_V = K(30,7);
localparam logic [15:0] CTRL_L1 = {D(6,5), K(28,5)};
localparam logic [15:0] CTRL_L2 = {D(26,4), K(28,5)};
function logic rd_flip_3b4b(input [2:0] hgf);
case (hgf)
3'b000: rd_flip_3b4b = 1'b1;
3'b001: rd_flip_3b4b = 1'b0;
3'b010: rd_flip_3b4b = 1'b0;
3'b011: rd_flip_3b4b = 1'b0;
3'b100: rd_flip_3b4b = 1'b1;
3'b101: rd_flip_3b4b = 1'b0;
3'b110: rd_flip_3b4b = 1'b0;
3'b111: rd_flip_3b4b = 1'b1;
endcase
endfunction
function logic rd_flip_5b6b(input [4:0] edcba, input k);
case (edcba)
5'b00000: rd_flip_5b6b = 1'b1;
5'b00001: rd_flip_5b6b = 1'b1;
5'b00010: rd_flip_5b6b = 1'b1;
5'b00011: rd_flip_5b6b = 1'b0;
5'b00100: rd_flip_5b6b = 1'b1;
5'b00101: rd_flip_5b6b = 1'b0;
5'b00110: rd_flip_5b6b = 1'b0;
5'b00111: rd_flip_5b6b = 1'b0;
5'b01000: rd_flip_5b6b = 1'b1;
5'b01001: rd_flip_5b6b = 1'b0;
5'b01010: rd_flip_5b6b = 1'b0;
5'b01011: rd_flip_5b6b = 1'b0;
5'b01100: rd_flip_5b6b = 1'b0;
5'b01101: rd_flip_5b6b = 1'b0;
5'b01110: rd_flip_5b6b = 1'b0;
5'b01111: rd_flip_5b6b = 1'b1;
5'b10000: rd_flip_5b6b = 1'b1;
5'b10001: rd_flip_5b6b = 1'b0;
5'b10010: rd_flip_5b6b = 1'b0;
5'b10011: rd_flip_5b6b = 1'b0;
5'b10100: rd_flip_5b6b = 1'b0;
5'b10101: rd_flip_5b6b = 1'b0;
5'b10110: rd_flip_5b6b = 1'b0;
5'b10111: rd_flip_5b6b = 1'b1;
5'b11000: rd_flip_5b6b = 1'b1;
5'b11001: rd_flip_5b6b = 1'b0;
5'b11010: rd_flip_5b6b = 1'b0;
5'b11011: rd_flip_5b6b = 1'b1;
5'b11100: rd_flip_5b6b = k; // K28
5'b11101: rd_flip_5b6b = 1'b1;
5'b11110: rd_flip_5b6b = 1'b1;
5'b11111: rd_flip_5b6b = 1'b1;
endcase
endfunction
function logic rd_flip_8b10b(input [7:0] d, input k);
rd_flip_8b10b = rd_flip_5b6b(d[4:0], k) ^ rd_flip_3b4b(d[7:5]);
endfunction
logic frame_reg = 1'b0, frame_next;
logic frame_cyc;
logic odd_reg = 1'b0, odd_next;
logic odd_cyc;
logic cext_reg = 1'b0, cext_next;
logic cext_cyc;
logic rd_reg = 1'b0, rd_next;
logic rd_cyc;
logic [DATA_W-1:0] encoded_tx_data_reg = '0, encoded_tx_data_next;
logic [CTRL_W-1:0] encoded_tx_data_k_reg = '0, encoded_tx_data_k_next;
logic [CTRL_W-1:0] encoded_tx_data_dm_reg = '0, encoded_tx_data_dm_next;
logic [CTRL_W-1:0] encoded_tx_data_dv_reg = '0, encoded_tx_data_dv_next;
logic encoded_tx_data_valid_reg = '0, encoded_tx_data_valid_next;
logic [GBX_CNT-1:0] tx_gbx_sync_reg = '0, tx_gbx_sync_next;
assign encoded_tx_data = encoded_tx_data_reg;
assign encoded_tx_data_k = encoded_tx_data_k_reg;
assign encoded_tx_data_dm = encoded_tx_data_dm_reg;
assign encoded_tx_data_dv = encoded_tx_data_dv_reg;
assign encoded_tx_data_valid = encoded_tx_data_valid_reg;
assign tx_gbx_sync_out = tx_gbx_sync_reg;
always_comb begin
frame_next = frame_reg;
odd_next = odd_reg;
cext_next = cext_reg;
rd_next = rd_reg;
encoded_tx_data_next = '0;
encoded_tx_data_k_next = '0;
encoded_tx_data_dm_next = '0;
encoded_tx_data_dv_next = '0;
encoded_tx_data_valid_next = '0;
frame_cyc = frame_reg;
odd_cyc = odd_reg;
cext_cyc = cext_reg;
rd_cyc = rd_reg;
tx_gbx_sync_next = tx_gbx_sync_in;
if (gmii_tx_valid) begin
// loop over bytes
for (integer seg = 0; seg < CTRL_W; seg = seg + 1) begin
if (CTRL_W > 1) begin
odd_cyc = 1'(seg & 1);
end
if (frame_cyc) begin
if (gmii_tx_en[seg]) begin
if (gmii_tx_er[seg]) begin
// propagate error
encoded_tx_data_next[seg*8 +: 8] = CTRL_V;
encoded_tx_data_k_next[seg] = 1'b1;
end else begin
// data
encoded_tx_data_next[seg*8 +: 8] = gmii_txd[seg*8 +: 8];
encoded_tx_data_k_next[seg] = 1'b0;
end
end else begin
// end of frame
frame_cyc = 1'b0;
encoded_tx_data_next[seg*8 +: 8] = CTRL_T;
encoded_tx_data_k_next[seg] = 1'b1;
cext_cyc = 1'b1;
end
end else begin
if (gmii_tx_en[seg] && odd_cyc == 0) begin
// start of frame
frame_cyc = 1'b1;
encoded_tx_data_next[seg*8 +: 8] = CTRL_S;
encoded_tx_data_k_next[seg] = 1'b1;
end else begin
if (cext_cyc) begin
// carrier extend
encoded_tx_data_next[seg*8 +: 8] = CTRL_R;
encoded_tx_data_k_next[seg] = 1'b1;
if (odd_cyc) begin
cext_cyc = 1'b0;
end
end else if (!odd_cyc) begin
// K28.5
encoded_tx_data_next[seg*8 +: 8] = K(28,5);
encoded_tx_data_k_next[seg] = 1'b1;
encoded_tx_data_dm_next[seg] = 1'b1;
encoded_tx_data_dv_next[seg] = rd_cyc;
end else begin
// idle
encoded_tx_data_next[seg*8 +: 8] = rd_cyc ? D(16,2) : D(5,6);
encoded_tx_data_k_next[seg] = 1'b0;
encoded_tx_data_dm_next[seg] = 1'b1;
encoded_tx_data_dv_next[seg] = rd_cyc;
end
end
end
odd_cyc = !odd_cyc;
rd_cyc = rd_cyc ^ rd_flip_8b10b(encoded_tx_data_next[seg*8 +: 8], encoded_tx_data_k_next[seg]);
end
frame_next = frame_cyc;
odd_next = odd_cyc;
cext_next = cext_cyc;
rd_next = rd_cyc;
encoded_tx_data_valid_next = 1'b1;
end
end
always_ff @(posedge clk) begin
frame_reg <= frame_next;
odd_reg <= odd_next;
cext_reg <= cext_next;
rd_reg <= rd_next;
encoded_tx_data_reg <= encoded_tx_data_next;
encoded_tx_data_k_reg <= encoded_tx_data_k_next;
encoded_tx_data_dm_reg <= encoded_tx_data_dm_next;
encoded_tx_data_dv_reg <= encoded_tx_data_dv_next;
encoded_tx_data_valid_reg <= encoded_tx_data_valid_next;
tx_gbx_sync_reg <= tx_gbx_sync_next;
if (rst) begin
frame_reg <= 1'b0;
odd_reg <= 1'b0;
cext_reg <= 1'b0;
rd_reg <= 1'b0;
encoded_tx_data_reg <= '0;
encoded_tx_data_k_reg <= '0;
encoded_tx_data_dm_reg <= '0;
encoded_tx_data_dv_reg <= '0;
encoded_tx_data_valid_reg <= 1'b0;
tx_gbx_sync_reg <= '0;
end
end
endmodule
`resetall

View File

@@ -0,0 +1,50 @@
# SPDX-License-Identifier: CERN-OHL-S-2.0
#
# Copyright (c) 2026 FPGA Ninja, LLC
#
# Authors:
# - Alex Forencich
TOPLEVEL_LANG = verilog
SIM ?= verilator
WAVES ?= 0
COCOTB_HDL_TIMEUNIT = 1ns
COCOTB_HDL_TIMEPRECISION = 1ps
RTL_DIR = ../../rtl
LIB_DIR = ../../lib
TAXI_SRC_DIR = $(LIB_DIR)/taxi/src
DUT = taxi_gmii_basex_dec
COCOTB_TEST_MODULES = test_$(DUT)
COCOTB_TOPLEVEL = $(DUT)
MODULE = $(COCOTB_TEST_MODULES)
TOPLEVEL = $(COCOTB_TOPLEVEL)
VERILOG_SOURCES += $(RTL_DIR)/$(DUT).sv
# handle file list files
process_f_file = $(call process_f_files,$(addprefix $(dir $1),$(shell cat $1)))
process_f_files = $(foreach f,$1,$(if $(filter %.f,$f),$(call process_f_file,$f),$f))
uniq_base = $(if $1,$(call uniq_base,$(foreach f,$1,$(if $(filter-out $(notdir $(lastword $1)),$(notdir $f)),$f,))) $(lastword $1))
VERILOG_SOURCES := $(call uniq_base,$(call process_f_files,$(VERILOG_SOURCES)))
# module parameters
export PARAM_DATA_W := 16
export PARAM_GBX_IF_EN := 0
ifeq ($(SIM), icarus)
PLUSARGS += -fst
COMPILE_ARGS += $(foreach v,$(filter PARAM_%,$(.VARIABLES)),-P $(COCOTB_TOPLEVEL).$(subst PARAM_,,$(v))=$($(v)))
else ifeq ($(SIM), verilator)
COMPILE_ARGS += $(foreach v,$(filter PARAM_%,$(.VARIABLES)),-G$(subst PARAM_,,$(v))=$($(v)))
ifeq ($(WAVES), 1)
COMPILE_ARGS += --trace-fst
VERILATOR_TRACE = 1
endif
endif
include $(shell cocotb-config --makefiles)/Makefile.sim

View File

@@ -0,0 +1 @@
../basex.py

View File

@@ -0,0 +1,174 @@
#!/usr/bin/env python
# SPDX-License-Identifier: CERN-OHL-S-2.0
"""
Copyright (c) 2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
"""
import itertools
import logging
import os
import sys
import cocotb_test.simulator
import pytest
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from cocotb.regression import TestFactory
from cocotbext.eth import GmiiFrame, GmiiSink
try:
from basex import BaseXSerdesSource
except ImportError:
# attempt import from current directory
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
try:
from basex import BaseXSerdesSource
finally:
del sys.path[0]
class TB:
def __init__(self, dut, gbx_cfg=None):
self.dut = dut
self.log = logging.getLogger("cocotb.tb")
self.log.setLevel(logging.DEBUG)
if gbx_cfg:
self.clk_period = 16
else:
self.clk_period = 16
cocotb.start_soon(Clock(dut.clk, self.clk_period, units="ns").start())
self.source = BaseXSerdesSource(
data=dut.encoded_rx_data,
data_k=dut.encoded_rx_data_k,
data_valid=dut.encoded_rx_data_valid,
clock=dut.clk,
enc_8b10b=False,
gbx_cfg=gbx_cfg
)
self.sink = GmiiSink(dut.gmii_rxd, dut.gmii_rx_er, dut.gmii_rx_dv,
dut.clk, dut.rst)
async def reset(self):
self.dut.rst.setimmediatevalue(0)
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst.value = 1
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst.value = 0
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
async def run_test(dut, gbx_cfg=None, payload_lengths=None, payload_data=None, ifg=12, pre_len=8):
tb = TB(dut, gbx_cfg)
tb.source.ifg = ifg
await tb.reset()
test_frames = [payload_data(x) for x in payload_lengths()]
for test_data in test_frames:
test_frame = GmiiFrame.from_payload(test_data)
test_frame.data = test_frame.data[8-pre_len:]
await tb.source.send(test_frame)
for test_data in test_frames:
rx_frame = await tb.sink.recv()
assert rx_frame.get_payload() == test_data
assert rx_frame.check_fcs()
assert rx_frame.error is None
assert tb.sink.empty()
for k in range(10):
await RisingEdge(dut.clk)
def size_list():
return list(range(60, 128)) + [512, 1514, 9214] + [60]*10 + [i for i in range(64, 73) for k in range(8)]
def incrementing_payload(length):
return bytearray(itertools.islice(itertools.cycle(range(256)), length))
if getattr(cocotb, 'top', None) is not None:
factory = TestFactory(run_test)
factory.add_option("payload_lengths", [size_list])
factory.add_option("payload_data", [incrementing_payload])
factory.add_option("ifg", list(range(0, 13)))
factory.add_option("pre_len", [8, 7])
factory.generate_tests()
# cocotb-test
tests_dir = os.path.abspath(os.path.dirname(__file__))
rtl_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'rtl'))
lib_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'lib'))
taxi_src_dir = os.path.abspath(os.path.join(lib_dir, 'taxi', 'src'))
def process_f_files(files):
lst = {}
for f in files:
if f[-2:].lower() == '.f':
with open(f, 'r') as fp:
l = fp.read().split()
for f in process_f_files([os.path.join(os.path.dirname(f), x) for x in l]):
lst[os.path.basename(f)] = f
else:
lst[os.path.basename(f)] = f
return list(lst.values())
@pytest.mark.parametrize("data_w", [8, 16])
def test_taxi_gmii_basex_dec(request, data_w):
dut = "taxi_gmii_basex_dec"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = dut
verilog_sources = [
os.path.join(rtl_dir, f"{dut}.sv"),
]
verilog_sources = process_f_files(verilog_sources)
parameters = {}
parameters['DATA_W'] = data_w
parameters['GBX_IF_EN'] = 0
extra_env = {f'PARAM_{k}': str(v) for k, v in parameters.items()}
sim_build = os.path.join(tests_dir, "sim_build",
request.node.name.replace('[', '-').replace(']', ''))
cocotb_test.simulator.run(
simulator="verilator",
python_search=[tests_dir],
verilog_sources=verilog_sources,
toplevel=toplevel,
module=module,
parameters=parameters,
sim_build=sim_build,
extra_env=extra_env,
)

View File

@@ -0,0 +1,50 @@
# SPDX-License-Identifier: CERN-OHL-S-2.0
#
# Copyright (c) 2026 FPGA Ninja, LLC
#
# Authors:
# - Alex Forencich
TOPLEVEL_LANG = verilog
SIM ?= verilator
WAVES ?= 0
COCOTB_HDL_TIMEUNIT = 1ns
COCOTB_HDL_TIMEPRECISION = 1ps
RTL_DIR = ../../rtl
LIB_DIR = ../../lib
TAXI_SRC_DIR = $(LIB_DIR)/taxi/src
DUT = taxi_gmii_basex_enc
COCOTB_TEST_MODULES = test_$(DUT)
COCOTB_TOPLEVEL = $(DUT)
MODULE = $(COCOTB_TEST_MODULES)
TOPLEVEL = $(COCOTB_TOPLEVEL)
VERILOG_SOURCES += $(RTL_DIR)/$(DUT).sv
# handle file list files
process_f_file = $(call process_f_files,$(addprefix $(dir $1),$(shell cat $1)))
process_f_files = $(foreach f,$1,$(if $(filter %.f,$f),$(call process_f_file,$f),$f))
uniq_base = $(if $1,$(call uniq_base,$(foreach f,$1,$(if $(filter-out $(notdir $(lastword $1)),$(notdir $f)),$f,))) $(lastword $1))
VERILOG_SOURCES := $(call uniq_base,$(call process_f_files,$(VERILOG_SOURCES)))
# module parameters
export PARAM_DATA_W := 16
export PARAM_GBX_IF_EN := 0
ifeq ($(SIM), icarus)
PLUSARGS += -fst
COMPILE_ARGS += $(foreach v,$(filter PARAM_%,$(.VARIABLES)),-P $(COCOTB_TOPLEVEL).$(subst PARAM_,,$(v))=$($(v)))
else ifeq ($(SIM), verilator)
COMPILE_ARGS += $(foreach v,$(filter PARAM_%,$(.VARIABLES)),-G$(subst PARAM_,,$(v))=$($(v)))
ifeq ($(WAVES), 1)
COMPILE_ARGS += --trace-fst
VERILATOR_TRACE = 1
endif
endif
include $(shell cocotb-config --makefiles)/Makefile.sim

View File

@@ -0,0 +1 @@
../basex.py

View File

@@ -0,0 +1,176 @@
#!/usr/bin/env python
# SPDX-License-Identifier: CERN-OHL-S-2.0
"""
Copyright (c) 2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
"""
import itertools
import logging
import os
import sys
import cocotb_test.simulator
import pytest
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from cocotb.regression import TestFactory
from cocotbext.eth import GmiiFrame, GmiiSource
try:
from basex import BaseXSerdesSink
except ImportError:
# attempt import from current directory
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
try:
from basex import BaseXSerdesSink
finally:
del sys.path[0]
class TB:
def __init__(self, dut, gbx_cfg=None):
self.dut = dut
self.log = logging.getLogger("cocotb.tb")
self.log.setLevel(logging.DEBUG)
if gbx_cfg:
self.clk_period = 16
else:
self.clk_period = 16
cocotb.start_soon(Clock(dut.clk, self.clk_period, units="ns").start())
self.source = GmiiSource(dut.gmii_txd, dut.gmii_tx_er, dut.gmii_tx_en,
dut.clk, dut.rst)
self.sink = BaseXSerdesSink(
data=dut.encoded_tx_data,
data_k=dut.encoded_tx_data_k,
data_valid=dut.encoded_tx_data_valid,
# gbx_req_sync=dut.tx_gbx_req_sync,
# gbx_req_stall=dut.tx_gbx_req_stall,
# gbx_sync=dut.tx_gbx_sync,
clock=dut.clk,
dec_8b10b=False,
gbx_cfg=gbx_cfg
)
async def reset(self):
self.dut.rst.setimmediatevalue(0)
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst.value = 1
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst.value = 0
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
async def run_test(dut, gbx_cfg=None, payload_lengths=None, payload_data=None, ifg=12, pre_len=8):
tb = TB(dut, gbx_cfg)
tb.source.ifg = ifg
await tb.reset()
test_frames = [payload_data(x) for x in payload_lengths()]
for test_data in test_frames:
test_frame = GmiiFrame.from_payload(test_data)
test_frame.data = test_frame.data[8-pre_len:]
await tb.source.send(test_frame)
for test_data in test_frames:
rx_frame = await tb.sink.recv()
assert rx_frame.get_payload() == test_data
assert rx_frame.check_fcs()
assert rx_frame.error is None
assert tb.sink.empty()
for k in range(10):
await RisingEdge(dut.clk)
def size_list():
return list(range(60, 128)) + [512, 1514, 9214] + [60]*10 + [i for i in range(64, 73) for k in range(8)]
def incrementing_payload(length):
return bytearray(itertools.islice(itertools.cycle(range(256)), length))
if getattr(cocotb, 'top', None) is not None:
factory = TestFactory(run_test)
factory.add_option("payload_lengths", [size_list])
factory.add_option("payload_data", [incrementing_payload])
factory.add_option("ifg", list(range(0, 13)))
factory.add_option("pre_len", [8, 7])
factory.generate_tests()
# cocotb-test
tests_dir = os.path.abspath(os.path.dirname(__file__))
rtl_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'rtl'))
lib_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'lib'))
taxi_src_dir = os.path.abspath(os.path.join(lib_dir, 'taxi', 'src'))
def process_f_files(files):
lst = {}
for f in files:
if f[-2:].lower() == '.f':
with open(f, 'r') as fp:
l = fp.read().split()
for f in process_f_files([os.path.join(os.path.dirname(f), x) for x in l]):
lst[os.path.basename(f)] = f
else:
lst[os.path.basename(f)] = f
return list(lst.values())
@pytest.mark.parametrize("data_w", [8, 16])
def test_taxi_gmii_basex_enc(request, data_w):
dut = "taxi_gmii_basex_enc"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = dut
verilog_sources = [
os.path.join(rtl_dir, f"{dut}.sv"),
]
verilog_sources = process_f_files(verilog_sources)
parameters = {}
parameters['DATA_W'] = data_w
parameters['GBX_IF_EN'] = 0
extra_env = {f'PARAM_{k}': str(v) for k, v in parameters.items()}
sim_build = os.path.join(tests_dir, "sim_build",
request.node.name.replace('[', '-').replace(']', ''))
cocotb_test.simulator.run(
simulator="verilator",
python_search=[tests_dir],
verilog_sources=verilog_sources,
toplevel=toplevel,
module=module,
parameters=parameters,
sim_build=sim_build,
extra_env=extra_env,
)