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

Skip using bus for more convenient support of non-standard signal names

Browse Source
This commit is contained in:
Alex Forencich
2020-11-25 17:34:19 -08:00
parent ceabdabb11
commit c4ed6073d6
6 changed files with 103 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
cocotbext/eth/gmii.py
View File

@@ -22,14 +22,13 @@ THE SOFTWARE.
"""
import logging
from collections import deque
import cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Timer, First, Event
from cocotb.bus import Bus
from cocotb.log import SimLog
from cocotb.utils import get_sim_time
from collections import deque
from .version import __version__
from .constants import EthPre, ETH_PREAMBLE
@@ -95,17 +94,15 @@ class GmiiFrame(object):
class GmiiSource(object):
_signals = ["d"]
_optional_signals = ["er", "en", "dv"]
def __init__(self, entity, name, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
self.entity = entity
def __init__(self, data, er, dv, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{data._path}")
self.data = data
self.er = er
self.dv = dv
self.clock = clock
self.reset = reset
self.enable = enable
self.mii_select = mii_select
self.bus = Bus(self.entity, name, self._signals, optional_signals=self._optional_signals, **kwargs)
self.log.info("GMII source")
self.log.info("cocotbext-eth version %s", __version__)
@@ -127,18 +124,13 @@ class GmiiSource(object):
self.reset = reset
assert len(self.bus.d) == 8
self.bus.d.setimmediatevalue(0)
if self.bus.er is not None:
assert len(self.bus.er) == 1
self.bus.er.setimmediatevalue(0)
if self.bus.en is not None:
assert len(self.bus.en) == 1
self.bus.en.setimmediatevalue(0)
self.bus.dv = self.bus.en
if self.bus.dv is not None:
assert len(self.bus.dv) == 1
self.bus.dv.setimmediatevalue(0)
assert len(self.data) == 8
self.data.setimmediatevalue(0)
if self.er is not None:
assert len(self.er) == 1
self.er.setimmediatevalue(0)
assert len(self.dv) == 1
self.dv.setimmediatevalue(0)
cocotb.fork(self._run())
@@ -174,10 +166,10 @@ class GmiiSource(object):
frame = None
ifg_cnt = 0
self.active = False
self.bus.d <= 0
if self.bus.er is not None:
self.bus.er <= 0
self.bus.en <= 0
self.data <= 0
if self.er is not None:
self.er <= 0
self.dv <= 0
continue
await RisingEdge(self.clock)
@@ -209,35 +201,33 @@ class GmiiSource(object):
self.active = True
if frame is not None:
self.bus.d <= frame.data.pop(0)
if self.bus.er is not None:
self.bus.er <= frame.error.pop(0)
self.bus.en <= 1
self.data <= frame.data.pop(0)
if self.er is not None:
self.er <= frame.error.pop(0)
self.dv <= 1
if not frame.data:
ifg_cnt = max(self.ifg, 1)
frame = None
else:
self.bus.d <= 0
if self.bus.er is not None:
self.bus.er <= 0
self.bus.en <= 0
self.data <= 0
if self.er is not None:
self.er <= 0
self.dv <= 0
self.active = False
class GmiiSink(object):
_signals = ["d"]
_optional_signals = ["er", "en", "dv"]
def __init__(self, entity, name, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
self.entity = entity
def __init__(self, data, er, dv, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{data._path}")
self.data = data
self.er = er
self.dv = dv
self.clock = clock
self.reset = reset
self.enable = enable
self.mii_select = mii_select
self.bus = Bus(self.entity, name, self._signals, optional_signals=self._optional_signals, **kwargs)
self.log.info("GMII sink")
self.log.info("cocotbext-eth version %s", __version__)
@@ -258,14 +248,11 @@ class GmiiSink(object):
self.reset = reset
assert len(self.bus.d) == 8
if self.bus.er is not None:
assert len(self.bus.er) == 1
if self.bus.en is not None:
assert len(self.bus.en) == 1
self.bus.dv = self.bus.en
if self.bus.dv is not None:
assert len(self.bus.dv) == 1
assert len(self.data) == 8
if self.er is not None:
assert len(self.er) == 1
if self.dv is not None:
assert len(self.dv) == 1
cocotb.fork(self._run())
@@ -309,9 +296,9 @@ class GmiiSink(object):
continue
if self.enable is None or self.enable.value:
d_val = self.bus.d.value.integer
dv_val = self.bus.dv.value.integer
er_val = 0 if self.bus.er is None else self.bus.er.value.integer
d_val = self.data.value.integer
dv_val = self.dv.value.integer
er_val = 0 if self.er is None else self.er.value.integer
if frame is None:
if dv_val:

61
cocotbext/eth/rgmii.py
View File

@@ -22,14 +22,13 @@ THE SOFTWARE.
"""
import logging
from collections import deque
import cocotb
from cocotb.triggers import RisingEdge, FallingEdge, ReadOnly, Timer, First, Event
from cocotb.bus import Bus
from cocotb.log import SimLog
from cocotb.utils import get_sim_time
from collections import deque
from .version import __version__
from .gmii import GmiiFrame
from .constants import EthPre
@@ -37,17 +36,14 @@ from .constants import EthPre
class RgmiiSource(object):
_signals = ["d", "ctl"]
_optional_signals = []
def __init__(self, entity, name, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
self.entity = entity
def __init__(self, data, ctrl, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{data._path}")
self.data = data
self.ctrl = ctrl
self.clock = clock
self.reset = reset
self.enable = enable
self.mii_select = mii_select
self.bus = Bus(self.entity, name, self._signals, optional_signals=self._optional_signals, **kwargs)
self.log.info("RGMII source")
self.log.info("cocotbext-eth version %s", __version__)
@@ -69,10 +65,10 @@ class RgmiiSource(object):
self.reset = reset
assert len(self.bus.d) == 4
self.bus.d.setimmediatevalue(0)
assert len(self.bus.ctl) == 1
self.bus.ctl.setimmediatevalue(0)
assert len(self.data) == 4
self.data.setimmediatevalue(0)
assert len(self.ctrl) == 1
self.ctrl.setimmediatevalue(0)
cocotb.fork(self._run())
@@ -111,16 +107,16 @@ class RgmiiSource(object):
frame = None
ifg_cnt = 0
self.active = False
self.bus.d <= 0
self.bus.ctl <= 0
self.data <= 0
self.ctrl <= 0
continue
await RisingEdge(self.clock)
if self.mii_select is None or not self.mii_select.value:
# send high nibble after rising edge, leading in to falling edge
self.bus.d <= d >> 4
self.bus.ctl <= en ^ er
self.data <= d >> 4
self.ctrl <= en ^ er
if self.enable is None or self.enable.value:
if ifg_cnt > 0:
@@ -165,23 +161,20 @@ class RgmiiSource(object):
await FallingEdge(self.clock)
# send low nibble after falling edge, leading in to rising edge
self.bus.d <= d & 0x0F
self.bus.ctl <= en
self.data <= d & 0x0F
self.ctrl <= en
class RgmiiSink(object):
_signals = ["d", "ctl"]
_optional_signals = []
def __init__(self, entity, name, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
self.entity = entity
def __init__(self, data, ctrl, clock, reset=None, enable=None, mii_select=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{data._path}")
self.data = data
self.ctrl = ctrl
self.clock = clock
self.reset = reset
self.enable = enable
self.mii_select = mii_select
self.bus = Bus(self.entity, name, self._signals, optional_signals=self._optional_signals, **kwargs)
self.log.info("RGMII sink")
self.log.info("cocotbext-eth version %s", __version__)
@@ -202,8 +195,8 @@ class RgmiiSink(object):
self.reset = reset
assert len(self.bus.d) == 4
assert len(self.bus.ctl) == 1
assert len(self.data) == 4
assert len(self.ctrl) == 1
cocotb.fork(self._run())
@@ -250,8 +243,8 @@ class RgmiiSink(object):
continue
# capture high nibble after rising edge, leading in to falling edge
d_val |= self.bus.d.value.integer << 4
er_val = dv_val ^ self.bus.ctl.value.integer
d_val |= self.data.value.integer << 4
er_val = dv_val ^ self.ctrl.value.integer
if self.enable is None or self.enable.value:
@@ -304,8 +297,8 @@ class RgmiiSink(object):
await ReadOnly()
# capture low nibble after falling edge, leading in to rising edge
d_val = self.bus.d.value.integer
dv_val = self.bus.ctl.value.integer
d_val = self.data.value.integer
dv_val = self.ctrl.value.integer
await RisingEdge(self.clock)

57
cocotbext/eth/xgmii.py
View File

@@ -22,14 +22,13 @@ THE SOFTWARE.
"""
import logging
from collections import deque
import cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Timer, First, Event
from cocotb.bus import Bus
from cocotb.log import SimLog
from cocotb.utils import get_sim_time
from collections import deque
from .version import __version__
from .constants import EthPre, ETH_PREAMBLE, XgmiiCtrl
@@ -98,16 +97,13 @@ class XgmiiFrame(object):
class XgmiiSource(object):
_signals = ["d", "c"]
_optional_signals = []
def __init__(self, entity, name, clock, reset=None, enable=None, *args, **kwargs):
self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
self.entity = entity
def __init__(self, data, ctrl, clock, reset=None, enable=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{data._path}")
self.data = data
self.ctrl = ctrl
self.clock = clock
self.reset = reset
self.enable = enable
self.bus = Bus(self.entity, name, self._signals, optional_signals=self._optional_signals, **kwargs)
self.log.info("XGMII source")
self.log.info("cocotbext-eth version %s", __version__)
@@ -126,8 +122,8 @@ class XgmiiSource(object):
self.queue_occupancy_bytes = 0
self.queue_occupancy_frames = 0
self.width = len(self.bus.d)
self.byte_width = len(self.bus.c)
self.width = len(self.data)
self.byte_width = len(self.ctrl)
self.reset = reset
@@ -140,8 +136,8 @@ class XgmiiSource(object):
self.idle_d |= XgmiiCtrl.IDLE << k*8
self.idle_c |= 1 << k
self.bus.d.setimmediatevalue(0)
self.bus.c.setimmediatevalue(0)
self.data.setimmediatevalue(0)
self.ctrl.setimmediatevalue(0)
cocotb.fork(self._run())
@@ -179,8 +175,8 @@ class XgmiiSource(object):
ifg_cnt = 0
deficit_idle_cnt = 0
self.active = False
self.bus.d <= 0
self.bus.c <= 0
self.data <= 0
self.ctrl <= 0
continue
await RisingEdge(self.clock)
@@ -246,26 +242,23 @@ class XgmiiSource(object):
d_val |= XgmiiCtrl.IDLE << k*8
c_val |= 1 << k
self.bus.d <= d_val
self.bus.c <= c_val
self.data <= d_val
self.ctrl <= c_val
else:
self.bus.d <= self.idle_d
self.bus.c <= self.idle_c
self.data <= self.idle_d
self.ctrl <= self.idle_c
self.active = False
class XgmiiSink(object):
_signals = ["d", "c"]
_optional_signals = []
def __init__(self, entity, name, clock, reset=None, enable=None, *args, **kwargs):
self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
self.entity = entity
def __init__(self, data, ctrl, clock, reset=None, enable=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{data._path}")
self.data = data
self.ctrl = ctrl
self.clock = clock
self.reset = reset
self.enable = enable
self.bus = Bus(self.entity, name, self._signals, optional_signals=self._optional_signals, **kwargs)
self.log.info("XGMII sink")
self.log.info("cocotbext-eth version %s", __version__)
@@ -281,8 +274,8 @@ class XgmiiSink(object):
self.queue_occupancy_bytes = 0
self.queue_occupancy_frames = 0
self.width = len(self.bus.d)
self.byte_width = len(self.bus.c)
self.width = len(self.data)
self.byte_width = len(self.ctrl)
self.reset = reset
@@ -331,8 +324,8 @@ class XgmiiSink(object):
if self.enable is None or self.enable.value:
for offset in range(self.byte_width):
d_val = (self.bus.d.value.integer >> (offset*8)) & 0xff
c_val = (self.bus.c.value.integer >> offset) & 1
d_val = (self.data.value.integer >> (offset*8)) & 0xff
c_val = (self.ctrl.value.integer >> offset) & 1
if frame is None:
if c_val and d_val == XgmiiCtrl.START:

9
tests/gmii/test_gmii.py
View File

@@ -30,7 +30,6 @@ import os
import cocotb_test.simulator
import cocotb
from cocotb.log import SimLog
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from cocotb.regression import TestFactory
@@ -42,7 +41,7 @@ class TB(object):
def __init__(self, dut):
self.dut = dut
self.log = SimLog("cocotb.tb")
self.log = logging.getLogger("cocotb.tb")
self.log.setLevel(logging.DEBUG)
self._enable_generator = None
@@ -50,8 +49,10 @@ class TB(object):
cocotb.fork(Clock(dut.clk, 2, units="ns").start())
self.source = GmiiSource(dut, "gmii", dut.clk, dut.rst, dut.gmii_clk_en, dut.gmii_mii_sel)
self.sink = GmiiSink(dut, "gmii", dut.clk, dut.rst, dut.gmii_clk_en, dut.gmii_mii_sel)
self.source = GmiiSource(dut.gmii_d, dut.gmii_er, dut.gmii_en,
dut.clk, dut.rst, dut.gmii_clk_en, dut.gmii_mii_sel)
self.sink = GmiiSink(dut.gmii_d, dut.gmii_er, dut.gmii_en,
dut.clk, dut.rst, dut.gmii_clk_en, dut.gmii_mii_sel)
dut.gmii_clk_en.setimmediatevalue(1)
dut.gmii_mii_sel.setimmediatevalue(0)

7
tests/rgmii/test_rgmii.py
View File

@@ -30,7 +30,6 @@ import os
import cocotb_test.simulator
import cocotb
from cocotb.log import SimLog
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from cocotb.regression import TestFactory
@@ -42,7 +41,7 @@ class TB(object):
def __init__(self, dut):
self.dut = dut
self.log = SimLog("cocotb.tb")
self.log = logging.getLogger("cocotb.tb")
self.log.setLevel(logging.DEBUG)
self._enable_generator = None
@@ -50,8 +49,8 @@ class TB(object):
cocotb.fork(Clock(dut.clk, 2, units="ns").start())
self.source = RgmiiSource(dut, "rgmii", dut.clk, dut.rst, dut.rgmii_clk_en, dut.rgmii_mii_sel)
self.sink = RgmiiSink(dut, "rgmii", dut.clk, dut.rst, dut.rgmii_clk_en, dut.rgmii_mii_sel)
self.source = RgmiiSource(dut.rgmii_d, dut.rgmii_ctl, dut.clk, dut.rst, dut.rgmii_clk_en, dut.rgmii_mii_sel)
self.sink = RgmiiSink(dut.rgmii_d, dut.rgmii_ctl, dut.clk, dut.rst, dut.rgmii_clk_en, dut.rgmii_mii_sel)
dut.rgmii_clk_en.setimmediatevalue(1)
dut.rgmii_mii_sel.setimmediatevalue(0)

7
tests/xgmii/test_xgmii.py
View File

@@ -31,7 +31,6 @@ import cocotb_test.simulator
import pytest
import cocotb
from cocotb.log import SimLog
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from cocotb.regression import TestFactory
@@ -43,7 +42,7 @@ class TB(object):
def __init__(self, dut):
self.dut = dut
self.log = SimLog("cocotb.tb")
self.log = logging.getLogger("cocotb.tb")
self.log.setLevel(logging.DEBUG)
self._enable_generator = None
@@ -51,8 +50,8 @@ class TB(object):
cocotb.fork(Clock(dut.clk, 2, units="ns").start())
self.source = XgmiiSource(dut, "xgmii", dut.clk, dut.rst, dut.xgmii_clk_en)
self.sink = XgmiiSink(dut, "xgmii", dut.clk, dut.rst, dut.xgmii_clk_en)
self.source = XgmiiSource(dut.xgmii_d, dut.xgmii_c, dut.clk, dut.rst, dut.xgmii_clk_en)
self.sink = XgmiiSink(dut.xgmii_d, dut.xgmii_c, dut.clk, dut.rst, dut.xgmii_clk_en)
dut.xgmii_clk_en.setimmediatevalue(1)