third-party/cocotbext-eth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Clean up and lint removal

Browse Source
This commit is contained in:
Alex Forencich
2020-11-25 00:51:09 -08:00
parent a6d01c5b73
commit 3d9f20a844
8 changed files with 75 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
cocotbext/eth/__init__.py
View File

@@ -26,4 +26,3 @@ from .version import __version__
from .gmii import GmiiFrame, GmiiSource, GmiiSink
from .xgmii import XgmiiFrame, XgmiiSource, XgmiiSink

14
cocotbext/eth/constants.py
View File

@@ -24,12 +24,17 @@ THE SOFTWARE.
import enum
# Ethernet frame
class EthPre(enum.IntEnum):
PRE = 0x55
SFD = 0xD5
ETH_PREAMBLE = b'\x55\x55\x55\x55\x55\x55\x55\xd5'
# XGMII control characters
class XgmiiCtrl(enum.IntEnum):
IDLE = 0x07
LPI = 0x06
@@ -45,6 +50,8 @@ class XgmiiCtrl(enum.IntEnum):
RES5 = 0xf7
SIG_OS = 0x5c
# BASE-R control characters
class BaseRCtrl(enum.IntEnum):
IDLE = 0x00
LPI = 0x06
@@ -56,14 +63,20 @@ class BaseRCtrl(enum.IntEnum):
RES_4 = 0x66
RES_5 = 0x78
# BASE-R O codes
class BaseRO(enum.IntEnum):
SEQ_OS = 0x0
SIG_OS = 0xf
# BASE-R sync header
class BaseRSync(enum.IntEnum):
DATA = 0b10
CTRL = 0b01
# BASE-R block type field
class BaseRBlockType(enum.IntEnum):
CTRL = 0x1e # C7 C6 C5 C4 C3 C2 C1 C0 BT
OS_4 = 0x2d # D7 D6 D5 O4 C3 C2 C1 C0 BT
@@ -80,4 +93,3 @@ class BaseRBlockType(enum.IntEnum):
TERM_5 = 0xd2 # C7 C6 D4 D3 D2 D1 D0 BT
TERM_6 = 0xe1 # C7 D5 D4 D3 D2 D1 D0 BT
TERM_7 = 0xff # D6 D5 D4 D3 D2 D1 D0 BT

14
cocotbext/eth/gmii.py
View File

@@ -33,6 +33,7 @@ from collections import deque
from .version import __version__
from .constants import EthPre, ETH_PREAMBLE
class GmiiFrame(object):
def __init__(self, data=None, error=None):
self.data = bytearray()
@@ -80,10 +81,10 @@ class GmiiFrame(object):
def __repr__(self):
return (
f"{type(self).__name__}(data={repr(self.data)}, " +
f"error={repr(self.error)}, " +
f"rx_sim_time={repr(self.rx_sim_time)})"
)
f"{type(self).__name__}(data={repr(self.data)}, "
f"error={repr(self.error)}, "
f"rx_sim_time={repr(self.rx_sim_time)})"
)
def __len__(self):
return len(self.data)
@@ -191,7 +192,7 @@ class GmiiSource(object):
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
self.log.info(f"TX frame: {frame}")
self.log.info("TX frame: %s", frame)
frame.normalize()
if self.mii_select is not None and self.mii_select.value:
@@ -343,7 +344,7 @@ class GmiiSink(object):
frame.error = error
frame.compact()
self.log.info(f"RX frame: {frame}")
self.log.info("RX frame: %s", frame)
self.queue_occupancy_bytes += len(frame)
self.queue_occupancy_frames += 1
@@ -362,4 +363,3 @@ class GmiiSink(object):
class GmiiMonitor(GmiiSink):
pass

23
cocotbext/eth/xgmii.py
View File

@@ -33,6 +33,7 @@ from collections import deque
from .version import __version__
from .constants import EthPre, ETH_PREAMBLE, XgmiiCtrl
class XgmiiFrame(object):
def __init__(self, data=None, ctrl=None):
self.data = bytearray()
@@ -82,11 +83,11 @@ class XgmiiFrame(object):
def __repr__(self):
return (
f"{type(self).__name__}(data={repr(self.data)}, " +
f"ctrl={repr(self.ctrl)}, " +
f"rx_sim_time={repr(self.rx_sim_time)}, " +
f"rx_start_lane={repr(self.rx_start_lane)})"
)
f"{type(self).__name__}(data={repr(self.data)}, "
f"ctrl={repr(self.ctrl)}, "
f"rx_sim_time={repr(self.rx_sim_time)}, "
f"rx_start_lane={repr(self.rx_start_lane)})"
)
def __len__(self):
return len(self.data)
@@ -200,7 +201,7 @@ class XgmiiSource(object):
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
self.log.info(f"TX frame: {frame}")
self.log.info("TX frame: %s", frame)
frame.normalize()
assert frame.data[0] == EthPre.PRE
assert frame.ctrl[0] == 0
@@ -210,7 +211,12 @@ class XgmiiSource(object):
frame.ctrl.append(1)
# offset start
if (self.byte_width > 4 and ifg_cnt > (3-deficit_idle_cnt if self.enable_dic else 0)) or self.force_offset_start:
if self.enable_dic:
min_ifg = 3 - deficit_idle_cnt
else:
min_ifg = 0
if self.byte_width > 4 and (ifg_cnt > min_ifg or self.force_offset_start):
ifg_cnt = ifg_cnt-4
frame.data = bytearray([XgmiiCtrl.IDLE]*4)+frame.data
frame.ctrl = [1]*4+frame.ctrl
@@ -343,7 +349,7 @@ class XgmiiSink(object):
frame.ctrl.append(c_val)
frame.compact()
self.log.info(f"RX frame: {frame}")
self.log.info("RX frame: %s", frame)
self.queue_occupancy_bytes += len(frame)
self.queue_occupancy_frames += 1
@@ -361,4 +367,3 @@ class XgmiiSink(object):
class XgmiiMonitor(XgmiiSink):
pass

14
setup.py
View File

@@ -11,19 +11,19 @@ with open("README.md", "r") as f:
long_description = f.read()
setup(
name = "cocotbext-eth",
name="cocotbext-eth",
author="Alex Forencich",
author_email="alex@alexforencich.com",
description="Ethernet modules for Cocotb",
long_description=long_description,
long_description_content_type='text/markdown',
url="https://github.com/alexforencich/cocotbext-eth",
download_url = 'http://github.com/alexforencich/cocotbext-eth/tarball/master',
version = version,
packages = find_namespace_packages(include=['cocotbext.*']),
install_requires = ['cocotb', 'cocotbext-axi'],
python_requires = '>=3.6',
classifiers = [
download_url='http://github.com/alexforencich/cocotbext-eth/tarball/master',
version=version,
packages=find_namespace_packages(include=['cocotbext.*']),
install_requires=['cocotb', 'cocotbext-axi'],
python_requires='>=3.6',
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",

7
tests/gmii/Makefile
View File

@@ -31,16 +31,11 @@ TOPLEVEL = $(DUT)
MODULE = $(DUT)
VERILOG_SOURCES += $(DUT).v
# module parameters
# export PARAM_DATA_WIDTH ?= 64
SIM_BUILD ?= sim_build_$(MODULE)
ifeq ($(SIM), icarus)
PLUSARGS += -fst
#COMPILE_ARGS += -P $(TOPLEVEL).DATA_WIDTH=$(PARAM_DATA_WIDTH)
ifeq ($(WAVES), 1)
VERILOG_SOURCES += iverilog_dump.v
COMPILE_ARGS += -s iverilog_dump
@@ -48,8 +43,6 @@ ifeq ($(SIM), icarus)
else ifeq ($(SIM), verilator)
COMPILE_ARGS += -Wno-SELRANGE -Wno-WIDTH
#COMPILE_ARGS += -GDATA_WIDTH=$(PARAM_DATA_WIDTH)
ifeq ($(WAVES), 1)
COMPILE_ARGS += --trace-fst
endif

23
tests/gmii/test_gmii.py
View File

@@ -28,7 +28,6 @@ import logging
import os
import cocotb_test.simulator
import pytest
import cocotb
from cocotb.log import SimLog
@@ -38,11 +37,12 @@ from cocotb.regression import TestFactory
from cocotbext.eth import GmiiFrame, GmiiSource, GmiiSink
class TB(object):
def __init__(self, dut):
self.dut = dut
self.log = SimLog(f"cocotb.tb")
self.log = SimLog("cocotb.tb")
self.log.setLevel(logging.DEBUG)
self._enable_generator = None
@@ -85,12 +85,11 @@ class TB(object):
self.dut.gmii_clk_en <= val
await RisingEdge(self.dut.clk)
async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_gen=None, mii_sel=False):
tb = TB(dut)
byte_width = tb.source.width // 8
tb.source.ifg = ifg
tb.dut.gmii_mii_sel <= mii_sel
@@ -99,7 +98,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
await tb.reset()
test_frames = [payload_data(l) for l in payload_lengths()]
test_frames = [payload_data(x) for x in payload_lengths()]
for test_data in test_frames:
test_frame = GmiiFrame.from_payload(test_data)
@@ -117,15 +116,19 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
def size_list():
return list(range(64, 128))+[512, 1514, 9214]+[64]*10
return list(range(64, 128)) + [512, 1514, 9214] + [64]*10
def incrementing_payload(length):
return bytearray(itertools.islice(itertools.cycle(range(256)), length))
def cycle_en():
return itertools.cycle([0, 0, 0, 1])
if cocotb.SIM_NAME:
factory = TestFactory(run_test)
@@ -137,9 +140,12 @@ if cocotb.SIM_NAME:
factory.generate_tests()
# cocotb-test
tests_dir = os.path.dirname(__file__)
rtl_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'rtl'))
def test_gmii(request):
dut = "test_gmii"
module = os.path.splitext(os.path.basename(__file__))[0]
@@ -151,9 +157,7 @@ def test_gmii(request):
parameters = {}
# parameters['DATA_WIDTH'] = data_width
extra_env = {f'PARAM_{k}' : str(v) for k, v in parameters.items()}
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(']', ''))
@@ -167,4 +171,3 @@ def test_gmii(request):
sim_build=sim_build,
extra_env=extra_env,
)

33
tests/xgmii/test_xgmii.py
View File

@@ -38,11 +38,12 @@ from cocotb.regression import TestFactory
from cocotbext.eth import XgmiiFrame, XgmiiSource, XgmiiSink
class TB(object):
def __init__(self, dut):
self.dut = dut
self.log = SimLog(f"cocotb.tb")
self.log = SimLog("cocotb.tb")
self.log.setLevel(logging.DEBUG)
self._enable_generator = None
@@ -84,12 +85,12 @@ class TB(object):
self.dut.xgmii_clk_en <= val
await RisingEdge(self.dut.clk)
async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_dic=True, force_offset_start=False, enable_gen=None):
async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_dic=True,
force_offset_start=False, enable_gen=None):
tb = TB(dut)
byte_width = tb.source.width // 8
tb.source.ifg = ifg
tb.source.enable_dic = enable_dic
tb.source.force_offset_start = force_offset_start
@@ -99,7 +100,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
await tb.reset()
test_frames = [payload_data(l) for l in payload_lengths()]
test_frames = [payload_data(x) for x in payload_lengths()]
for test_data in test_frames:
test_frame = XgmiiFrame.from_payload(test_data)
@@ -117,7 +118,9 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True, force_offset_start=False, enable_gen=False):
async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True,
force_offset_start=False, enable_gen=False):
tb = TB(dut)
@@ -150,8 +153,8 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True, fo
start_lane.append(rx_frame.rx_start_lane)
tb.log.info(f"length: {length}")
tb.log.info(f"start_lane: {start_lane}")
tb.log.info("length: %d", length)
tb.log.info("start_lane: %s", start_lane)
start_lane_ref = []
@@ -180,7 +183,7 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True, fo
offset += 4
lane = (lane - offset) % byte_width
tb.log.info(f"start_lane_ref: {start_lane_ref}")
tb.log.info("start_lane_ref: %s", start_lane_ref)
assert start_lane_ref == start_lane
@@ -191,15 +194,19 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True, fo
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
def size_list():
return list(range(64, 128))+[512, 1514, 9214]+[64]*10+[65]*10+[66]*10+[67]*10
return list(range(64, 128)) + [512, 1514, 9214] + [64]*10 + [65]*10 + [66]*10 + [67]*10
def incrementing_payload(length):
return bytearray(itertools.islice(itertools.cycle(range(256)), length))
def cycle_en():
return itertools.cycle([0, 0, 0, 1])
if cocotb.SIM_NAME:
factory = TestFactory(run_test)
@@ -220,9 +227,12 @@ if cocotb.SIM_NAME:
factory.generate_tests()
# cocotb-test
tests_dir = os.path.dirname(__file__)
rtl_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'rtl'))
@pytest.mark.parametrize("data_width", [32, 64])
def test_xgmii(request, data_width):
dut = "test_xgmii"
@@ -238,7 +248,7 @@ def test_xgmii(request, data_width):
parameters['DATA_WIDTH'] = data_width
parameters['CTRL_WIDTH'] = parameters['DATA_WIDTH'] // 8
extra_env = {f'PARAM_{k}' : str(v) for k, v in parameters.items()}
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(']', ''))
@@ -252,4 +262,3 @@ def test_xgmii(request, data_width):
sim_build=sim_build,
extra_env=extra_env,
)