use stream_packet and stream_sim from litex

liteeth/software/etherbone.py

@@ -1,4 +1,5 @@
-import copy
+from copy import deepcopy
+
 from liteeth.common import *
 
 
@@ -254,7 +255,7 @@ class EtherbonePacket(Packet):
         while len(payload) != 0:
             record = EtherboneRecord(payload)
             record.decode()
-            records.append(copy.deepcopy(record))
+            records.append(deepcopy(record))
             payload = record
         return records
 

test/arp_tb.py

@@ -2,12 +2,12 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core.mac import LiteEthMAC
 from liteeth.core.arp import LiteEthARP
 
-from test.common import *
 from test.model import phy, mac, arp
 
 ip_address = 0x12345678

test/common.py

@@ -1,184 +0,0 @@
-import random
-from copy import deepcopy
-
-from litex.gen import *
-from litex.soc.interconnect.stream import Sink, Source
-
-from liteeth.common import *
-
-
-def print_with_prefix(s, prefix=""):
-    if not isinstance(s, str):
-        s = s.__repr__()
-    s = s.split("\n")
-    for l in s:
-        print(prefix + l)
-
-
-def seed_to_data(seed, random=True):
-    if random:
-        return (seed * 0x31415979 + 1) & 0xffffffff
-    else:
-        return seed
-
-
-def split_bytes(v, n, endianness="big"):
-    r = []
-    r_bytes = v.to_bytes(n, byteorder=endianness)
-    for byte in r_bytes:
-        r.append(int(byte))
-    return r
-
-
-def merge_bytes(b, endianness="big"):
-    return int.from_bytes(bytes(b), endianness)
-
-
-def get_field_data(field, datas):
-    v = merge_bytes(datas[field.byte:field.byte+math.ceil(field.width/8)])
-    return (v >> field.offset) & (2**field.width-1)
-
-
-def comp(p1, p2):
-    r = True
-    for x, y in zip(p1, p2):
-        if x != y:
-            r = False
-    return r
-
-
-def check(p1, p2):
-    p1 = deepcopy(p1)
-    p2 = deepcopy(p2)
-    if isinstance(p1, int):
-        return 0, 1, int(p1 != p2)
-    else:
-        if len(p1) >= len(p2):
-            ref, res = p1, p2
-        else:
-            ref, res = p2, p1
-        shift = 0
-        while((ref[0] != res[0]) and (len(res) > 1)):
-            res.pop(0)
-            shift += 1
-        length = min(len(ref), len(res))
-        errors = 0
-        for i in range(length):
-            if ref.pop(0) != res.pop(0):
-                errors += 1
-        return shift, length, errors
-
-
-def randn(max_n):
-    return random.randint(0, max_n-1)
-
-
-class Packet(list):
-    def __init__(self, init=[]):
-        self.ongoing = False
-        self.done = False
-        for data in init:
-            self.append(data)
-
-
-class PacketStreamer(Module):
-    def __init__(self, description, last_be=None):
-        self.source = Source(description)
-        self.last_be = last_be
-
-        # # #
-
-        self.packets = []
-        self.packet = Packet()
-        self.packet.done = True
-
-    def send(self, packet):
-        packet = deepcopy(packet)
-        self.packets.append(packet)
-        return packet
-
-    def send_blocking(self, packet):
-        packet = self.send(packet)
-        while not packet.done:
-            yield
-
-    def do_simulation(self, selfp):
-        if len(self.packets) and self.packet.done:
-            self.packet = self.packets.pop(0)
-        if not self.packet.ongoing and not self.packet.done:
-            selfp.source.stb = 1
-            if self.source.description.packetized:
-                selfp.source.sop = 1
-            selfp.source.data = self.packet.pop(0)
-            self.packet.ongoing = True
-        elif selfp.source.stb == 1 and selfp.source.ack == 1:
-            if self.source.description.packetized:
-                selfp.source.sop = 0
-                if len(self.packet) == 1:
-                    selfp.source.eop = 1
-                    if self.last_be is not None:
-                        selfp.source.last_be = self.last_be
-                else:
-                    selfp.source.eop = 0
-                    if self.last_be is not None:
-                        selfp.source.last_be = 0
-            if len(self.packet) > 0:
-                selfp.source.stb = 1
-                selfp.source.data = self.packet.pop(0)
-            else:
-                self.packet.done = True
-                selfp.source.stb = 0
-
-
-class PacketLogger(Module):
-    def __init__(self, description):
-        self.sink = Sink(description)
-
-        # # #
-
-        self.packet = Packet()
-
-    def receive(self):
-        self.packet.done = False
-        while not self.packet.done:
-            yield
-
-    def do_simulation(self, selfp):
-        selfp.sink.ack = 1
-        if selfp.sink.stb:
-            if self.sink.description.packetized:
-                if selfp.sink.sop:
-                    self.packet = Packet()
-                    self.packet.append(selfp.sink.data)
-                else:
-                    self.packet.append(selfp.sink.data)
-                if selfp.sink.eop:
-                    self.packet.done = True
-            else:
-                self.packet.append(selfp.sink.data)
-
-
-class AckRandomizer(Module):
-    def __init__(self, description, level=0):
-        self.level = level
-
-        self.sink = Sink(description)
-        self.source = Source(description)
-
-        self.run = Signal()
-
-        self.comb += \
-            If(self.run,
-                Record.connect(self.sink, self.source)
-            ).Else(
-                self.source.stb.eq(0),
-                self.sink.ack.eq(0),
-            )
-
-    def do_simulation(self, selfp):
-        n = randn(100)
-        if n < self.level:
-            selfp.run = 0
-        else:
-            selfp.run = 1
-

test/etherbone_tb.py

@@ -2,12 +2,12 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core import LiteEthUDPIPCore
 from liteeth.frontend.etherbone import LiteEthEtherbone
 
-from test.common import *
 from test.model import phy, mac, arp, ip, udp, etherbone
 
 ip_address = 0x12345678

test/icmp_tb.py

@@ -2,11 +2,11 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core import LiteEthIPCore
 
-from test.common import *
 from test.model.dumps import *
 from test.model.mac import *
 from test.model.ip import *

test/ip_tb.py

@@ -2,11 +2,11 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core import LiteEthIPCore
 
-from test.common import *
 from test.model import phy, mac, arp, ip
 
 ip_address = 0x12345678

test/mac_core_tb.py

@@ -2,11 +2,11 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core.mac.core import LiteEthMACCore
 
-from test.common import *
 from test.model import phy, mac
 
 

test/mac_wishbone_tb.py

@@ -2,11 +2,11 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core.mac import LiteEthMAC
 
-from test.common import *
 from test.model import phy, mac
 
 

test/model/arp.py

@@ -1,8 +1,9 @@
 import math
 
+from litex.soc.interconnect.stream_sim import *
+
 from liteeth.common import *
 
-from test.common import *
 from test.model import mac
 
 

test/model/etherbone.py

@@ -1,10 +1,11 @@
 import math
 import copy
 
+from litex.soc.interconnect.stream_sim import *
+
 from liteeth.common import *
 from liteeth.software.etherbone import *
 
-from test.common import *
 from test.model import udp
 
 
@@ -74,7 +75,7 @@ if __name__ == "__main__":
 
     # Packet
     packet = EtherbonePacket()
-    packet.records = [copy.deepcopy(record) for i in range(8)]
+    packet.records = [deepcopy(record) for i in range(8)]
     packet.nr = 0
     packet.pr = 0
     packet.pf = 0

test/model/icmp.py

@@ -1,8 +1,9 @@
 import math
 
+from litex.soc.interconnect.stream_sim import *
+
 from liteeth.common import *
 
-from test.common import *
 from test.model import ip
 
 

test/model/ip.py

@@ -1,8 +1,9 @@
 import math
 
+from litex.soc.interconnect.stream_sim import *
+
 from liteeth.common import *
 
-from test.common import *
 from test.model import mac
 
 

test/model/mac.py

@@ -1,9 +1,9 @@
 import math
 import binascii
 
-from liteeth.common import *
+from litex.soc.interconnect.stream_sim import *
 
-from test.common import *
+from liteeth.common import *
 
 
 def print_mac(s):

test/model/phy.py

@@ -1,6 +1,6 @@
-from liteeth.common import *
+from litex.soc.interconnect.stream_sim import *
 
-from test.common import *
+from liteeth.common import *
 
 
 def print_phy(s):

test/model/udp.py

@@ -1,8 +1,9 @@
 import math
 
+from litex.soc.interconnect.stream_sim import *
+
 from liteeth.common import *
 
-from test.common import *
 from test.model import ip
 
 

test/udp_tb.py

@@ -2,11 +2,11 @@ from litex.gen import *
 from litex.gen.sim.generic import run_simulation
 
 from litex.soc.interconnect import wishbone
+from litex.soc.interconnect.stream_sim import *
 
 from liteeth.common import *
 from liteeth.core import LiteEthUDPIPCore
 
-from test.common import *
 from test.model import phy, mac, arp, ip, udp
 
 ip_address = 0x12345678