serdes: move USB3SerDesModel to sim.py

sim.py

@@ -12,7 +12,7 @@ from litex.build.sim.config import SimConfig
 from litex.soc.integration.soc_core import *
 from litex.soc.integration.builder import *
 
-from usb3_pipe import USB3SerDesModel
+from usb3_pipe.serdes import *
 from usb3_pipe import USB3PIPE
 
 # IOs ----------------------------------------------------------------------------------------------
@@ -38,6 +38,87 @@ class Platform(SimPlatform):
     def do_finalize(self, fragment):
         pass
 
+# Simulation Serializer/Deserializer Model ---------------------------------------------------------
+
+class USB3SerDesModel(Module):
+    def __init__(self, rx_word_shift=0):
+        self.sink   = stream.Endpoint([("data", 32), ("ctrl", 4)])
+        self.source = stream.Endpoint([("data", 32), ("ctrl", 4)])
+        self.tx     = stream.Endpoint([("data", 20)])
+        self.rx     = stream.Endpoint([("data", 20)])
+
+        self.enable = Signal(reset=1) # i
+        self.ready  = Signal()        # o
+
+        self.tx_polarity = Signal()   # i
+        self.tx_idle     = Signal()   # i
+        self.tx_pattern  = Signal(20) # i
+
+        self.rx_polarity = Signal()   # i
+        self.rx_idle     = Signal()   # o
+        self.rx_align    = Signal()   # i
+
+        # # #
+
+        tx_datapath = SerdesTXDatapath()
+        rx_datapath = SerdesRXDatapath()
+        self.submodules += tx_datapath, rx_datapath
+        self.comb += [
+            self.sink.connect(tx_datapath.sink),
+            rx_datapath.word_aligner.enable.eq(self.rx_align),
+            rx_datapath.source.connect(self.source)
+        ]
+
+        encoder  = Encoder(2, True)
+        decoders = [Decoder(True) for _ in range(2)]
+        self.submodules += encoder, decoders
+        self.comb += tx_datapath.source.ready.eq(1)
+        self.comb += rx_datapath.sink.valid.eq(1)
+        for i in range(2):
+            self.comb += [
+                encoder.k[i].eq(tx_datapath.source.ctrl[i]),
+                encoder.d[i].eq(tx_datapath.source.data[8*i:8*(i+1)]),
+                rx_datapath.sink.ctrl[i].eq(decoders[i].k),
+                rx_datapath.sink.data[8*i:8*(i+1)].eq(decoders[i].d),
+            ]
+
+        tx_data    = Signal(20)
+        rx_data    = Signal(20)
+        rx_data_sr = Signal(40)
+        self.comb += [
+            If(self.tx_pattern != 0,
+                tx_data.eq(self.tx_pattern)
+            ).Else(
+                tx_data.eq(Cat(*[encoder.output[i] for i in range(2)])),
+            ),
+            If(self.tx_polarity,
+                self.tx.data.eq(~tx_data)
+            ).Else(
+                self.tx.data.eq(tx_data)
+            )
+        ]
+        self.comb += [
+            If(self.rx_polarity,
+                rx_data.eq(~self.rx.data)
+            ).Else(
+                rx_data.eq(self.rx.data)
+            )
+        ]
+        self.sync += rx_data_sr.eq(Cat(rx_data, rx_data_sr))
+        for i in range(2):
+            self.comb += decoders[i].input.eq(rx_data_sr[10*(rx_word_shift+i):10*(rx_word_shift+i+1)])
+
+        # Ready when enabled
+        self.comb += self.ready.eq(self.enable)
+
+    def connect(self, serdes):
+        self.comb += [
+            self.tx.connect(serdes.rx),
+            serdes.tx.connect(self.rx),
+            self.rx_idle.eq(serdes.tx_idle),
+            serdes.rx_idle.eq(self.tx_idle),
+        ]
+
 # USB3PIPESim --------------------------------------------------------------------------------------
 
 class USB3PIPESim(SoCMini):

usb3_pipe/__init__.py

@@ -1,5 +1,4 @@
 from usb3_pipe.serdes import K7USB3SerDes
 from usb3_pipe.serdes import A7USB3SerDes
 from usb3_pipe.serdes import ECP5USB3SerDes
-from usb3_pipe.serdes import USB3SerDesModel
 from usb3_pipe.core import USB3PIPE

usb3_pipe/serdes.py

@@ -434,84 +434,3 @@ class ECP5USB3SerDes(Module):
         # FIXME: Add keep and false path?
         platform.add_period_constraint(serdes.txoutclk, 1e9/serdes.tx_clk_freq)
         platform.add_period_constraint(serdes.rxoutclk, 1e9/serdes.rx_clk_freq)
-
-# Simulation Serializer/Deserializer Model ---------------------------------------------------------
-
-class USB3SerDesModel(Module):
-    def __init__(self, rx_word_shift=0):
-        self.sink   = stream.Endpoint([("data", 32), ("ctrl", 4)])
-        self.source = stream.Endpoint([("data", 32), ("ctrl", 4)])
-        self.tx     = stream.Endpoint([("data", 20)])
-        self.rx     = stream.Endpoint([("data", 20)])
-
-        self.enable = Signal(reset=1) # i
-        self.ready  = Signal()        # o
-
-        self.tx_polarity = Signal()   # i
-        self.tx_idle     = Signal()   # i
-        self.tx_pattern  = Signal(20) # i
-
-        self.rx_polarity = Signal()   # i
-        self.rx_idle     = Signal()   # o
-        self.rx_align    = Signal()   # i # not used
-
-        # # #
-
-        tx_datapath = SerdesTXDatapath()
-        rx_datapath = SerdesRXDatapath()
-        self.submodules += tx_datapath, rx_datapath
-        self.comb += [
-            self.sink.connect(tx_datapath.sink),
-            rx_datapath.word_aligner.enable.eq(self.rx_align),
-            rx_datapath.source.connect(self.source)
-        ]
-
-        encoder  = Encoder(2, True)
-        decoders = [Decoder(True) for _ in range(2)]
-        self.submodules += encoder, decoders
-        self.comb += tx_datapath.source.ready.eq(1)
-        self.comb += rx_datapath.sink.valid.eq(1)
-        for i in range(2):
-            self.comb += [
-                encoder.k[i].eq(tx_datapath.source.ctrl[i]),
-                encoder.d[i].eq(tx_datapath.source.data[8*i:8*(i+1)]),
-                rx_datapath.sink.ctrl[i].eq(decoders[i].k),
-                rx_datapath.sink.data[8*i:8*(i+1)].eq(decoders[i].d),
-            ]
-
-        tx_data    = Signal(20)
-        rx_data    = Signal(20)
-        rx_data_sr = Signal(40)
-        self.comb += [
-            If(self.tx_pattern != 0,
-                tx_data.eq(self.tx_pattern)
-            ).Else(
-                tx_data.eq(Cat(*[encoder.output[i] for i in range(2)])),
-            ),
-            If(self.tx_polarity,
-                self.tx.data.eq(~tx_data)
-            ).Else(
-                self.tx.data.eq(tx_data)
-            )
-        ]
-        self.comb += [
-            If(self.rx_polarity,
-                rx_data.eq(~self.rx.data)
-            ).Else(
-                rx_data.eq(self.rx.data)
-            )
-        ]
-        self.sync += rx_data_sr.eq(Cat(rx_data, rx_data_sr))
-        for i in range(2):
-            self.comb += decoders[i].input.eq(rx_data_sr[10*(rx_word_shift+i):10*(rx_word_shift+i+1)])
-
-        # Ready when enabled
-        self.comb += self.ready.eq(self.enable)
-
-    def connect(self, serdes):
-        self.comb += [
-            self.tx.connect(serdes.rx),
-            serdes.tx.connect(self.rx),
-            self.rx_idle.eq(serdes.tx_idle),
-            serdes.rx_idle.eq(self.tx_idle),
-        ]