Add FCS insertion, removal, and verification to GmiiFrame and XgmiiFrame

cocotbext/eth/gmii.py

@@ -23,6 +23,8 @@ THE SOFTWARE.
 """
 
 import logging
+import struct
+import zlib
 from collections import deque
 
 import cocotb
@@ -48,9 +50,11 @@ class GmiiFrame(object):
             self.error = error
 
     @classmethod
-    def from_payload(cls, payload):
+    def from_payload(cls, payload, add_fcs=True):
         data = bytearray(ETH_PREAMBLE)
         data.extend(payload)
+        if add_fcs:
+            data.extend(struct.pack('<L', zlib.crc32(payload)))
         return cls(data)
 
     def get_preamble_len(self):
@@ -59,8 +63,17 @@ class GmiiFrame(object):
     def get_preamble(self):
         return self.data[0:self.get_preamble_len()]
 
-    def get_payload(self):
-        return self.data[self.get_preamble_len():]
+    def get_payload(self, strip_fcs=True):
+        if strip_fcs:
+            return self.data[self.get_preamble_len():-4]
+        else:
+            return self.data[self.get_preamble_len():]
+
+    def get_fcs(self):
+        return self.data[-4:]
+
+    def check_fcs(self):
+        return self.get_fcs() == struct.pack('<L', zlib.crc32(self.get_payload(strip_fcs=True)))
 
     def normalize(self):
         n = len(self.data)

cocotbext/eth/xgmii.py

@@ -23,6 +23,8 @@ THE SOFTWARE.
 """
 
 import logging
+import struct
+import zlib
 from collections import deque
 
 import cocotb
@@ -50,9 +52,11 @@ class XgmiiFrame(object):
             self.ctrl = ctrl
 
     @classmethod
-    def from_payload(cls, payload):
+    def from_payload(cls, payload, add_fcs=True):
         data = bytearray(ETH_PREAMBLE)
         data.extend(payload)
+        if add_fcs:
+            data.extend(struct.pack('<L', zlib.crc32(payload)))
         return cls(data)
 
     def get_preamble_len(self):
@@ -61,8 +65,17 @@ class XgmiiFrame(object):
     def get_preamble(self):
         return self.data[0:self.get_preamble_len()]
 
-    def get_payload(self):
-        return self.data[self.get_preamble_len():]
+    def get_payload(self, strip_fcs=True):
+        if strip_fcs:
+            return self.data[self.get_preamble_len():-4]
+        else:
+            return self.data[self.get_preamble_len():]
+
+    def get_fcs(self):
+        return self.data[-4:]
+
+    def check_fcs(self):
+        return self.get_fcs() == struct.pack('<L', zlib.crc32(self.get_payload(strip_fcs=True)))
 
     def normalize(self):
         n = len(self.data)

tests/gmii/test_gmii.py

@@ -119,7 +119,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
 
 
 def size_list():
-    return list(range(64, 128)) + [512, 1514, 9214] + [64]*10
+    return list(range(60, 128)) + [512, 1514, 9214] + [60]*10
 
 
 def incrementing_payload(length):

tests/rgmii/test_rgmii.py

@@ -108,6 +108,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
         rx_frame = 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()
@@ -117,7 +118,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
 
 
 def size_list():
-    return list(range(64, 128)) + [512, 1514, 9214] + [64]*10
+    return list(range(60, 128)) + [512, 1514, 9214] + [60]*10
 
 
 def incrementing_payload(length):

tests/xgmii/test_xgmii.py

@@ -110,6 +110,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
         rx_frame = tb.sink.recv()
 
         assert rx_frame.get_payload() == test_data
+        assert rx_frame.check_fcs()
         assert rx_frame.ctrl is None
 
     assert tb.sink.empty()
@@ -132,7 +133,7 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True,
     if enable_gen is not None:
         tb.set_enable_generator(enable_gen())
 
-    for length in range(64, 96):
+    for length in range(60, 92):
 
         await tb.reset()
 
@@ -148,6 +149,7 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True,
             rx_frame = tb.sink.recv()
 
             assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
             assert rx_frame.ctrl is None
 
             start_lane.append(rx_frame.rx_start_lane)
@@ -168,7 +170,7 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True,
                 lane = 4
 
             start_lane_ref.append(lane)
-            lane = (lane + len(test_data)+ifg) % byte_width
+            lane = (lane + len(test_data)+4+ifg) % byte_width
 
             if enable_dic:
                 offset = lane % 4
@@ -195,7 +197,7 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True,
 
 
 def size_list():
-    return list(range(64, 128)) + [512, 1514, 9214] + [64]*10 + [65]*10 + [66]*10 + [67]*10
+    return list(range(60, 128)) + [512, 1514, 9214] + [60]*10 + [61]*10 + [62]*10 + [63]*10
 
 
 def incrementing_payload(length):