USB3: Implement compliance mode, de-emphasis, and tune TX amplitude on ECP5

luna/gateware/interface/pipe.py

@@ -8,6 +8,7 @@
 
 from amaranth import *
 from amaranth.lib.cdc  import FFSynchronizer, ResetSynchronizer
+from amaranth.lib.enum import IntEnum
 from amaranth.lib.fifo import AsyncFIFOBuffered
 
 
@@ -89,8 +90,8 @@ class PIPEInterface(Elaboratable):
     tx_compliance : Signal(), input
         If asserted, sets the running disparity to negative for the first symbol on the transmit
         data bus. This signal is implemented only for PHYs that can operate in the PCI Express mode.
-    tx_ones_zeroes : Signal(), input
-        If asserted, the PHY transmits an alternating sequence of 50-250 ones and 50-250 zeroes
+    tx_ones_zeros : Signal(), input
+        If asserted, the PHY transmits an alternating sequence of 50-250 ones and 50-250 zeros
         instead of the data on the transmit data bus. This signal is implemented only for PHYs
         that can operate in the SuperSpeed USB mode.
     rx_polarity : Signal(), input
@@ -156,13 +157,13 @@ def __init__(self, *, width):
         self.elas_buf_mode  = Signal(1)
         self.rate           = Signal(1)
         self.power_down     = Signal(2)
-        self.tx_deemph      = Signal(2)
+        self.tx_deemph      = Signal(TXDeemphMode)
         self.tx_margin      = Signal(3)
         self.tx_swing       = Signal(1)
         self.tx_detrx_lpbk  = Signal()
         self.tx_elec_idle   = Signal()
         self.tx_compliance  = Signal()
-        self.tx_ones_zeroes = Signal()
+        self.tx_ones_zeros  = Signal()
         self.rx_polarity    = Signal()
         self.rx_eq_training = Signal()
         self.rx_termination = Signal()
@@ -252,20 +253,20 @@ def elaborate(self, platform):
         geared_tx_data          = Signal.like(self.tx_data)
         geared_tx_datak         = Signal.like(self.tx_datak)
         geared_tx_compliance    = Signal.like(self.tx_compliance)
-        geared_tx_ones_zeroes   = Signal.like(self.tx_ones_zeroes)
+        geared_tx_ones_zeros    = Signal.like(self.tx_ones_zeros)
         mac_tx_bus_signals = Cat(
             self.tx_data,
             self.tx_datak,
             # These control signals are additional inputs to the 8b10b encoder; and must be
             # exactly synchronized to the transmit data bus.
             self.tx_compliance,
-            self.tx_ones_zeroes,
+            self.tx_ones_zeros,
         )
         phy_tx_bus_signals = Cat(
             geared_tx_data,
             geared_tx_datak,
             geared_tx_compliance,
-            geared_tx_ones_zeroes,
+            geared_tx_ones_zeros,
         )
 
         m.d.comb += [
@@ -275,8 +276,8 @@ def elaborate(self, platform):
         # TxCompliance affects only the first transmitted symbol; keep that property after gearing.
         with m.If(gear_index == 0):
             m.d.comb += phy.tx_compliance   .eq(geared_tx_compliance)
-        # TxOnesZeroes replaces all symbols on the transmit data bus.
-        m.d.comb += phy.tx_ones_zeroes  .eq(geared_tx_ones_zeroes)
+        # TxOnesZeros replaces all symbols on the transmit data bus.
+        m.d.comb += phy.tx_ones_zeros  .eq(geared_tx_ones_zeros)
 
         m.submodules.tx_fifo = tx_fifo = AsyncFIFOBuffered(
             width=len(mac_tx_bus_signals),
@@ -561,3 +562,8 @@ def elaborate(self, platform):
         return m
 
 
+class TXDeemphMode(IntEnum, shape=2):
+    DEEMPH_6DB      = 0b00
+    DEEMPH_3P5DB    = 0b01
+    DEEMPH_NONE     = 0b10
+    DEEMPH_RESERVED = 0b11

luna/gateware/interface/serdes_phy/ecp5.py

@@ -17,7 +17,7 @@
 from amaranth.lib.cdc import FFSynchronizer
 
 from .lfps         import LFPSSquareWaveGenerator, LFPSSquareWaveDetector
-from ..pipe        import PIPEInterface
+from ..pipe        import PIPEInterface, TXDeemphMode
 
 
 class ECP5SerDesPLLConfiguration:
@@ -113,8 +113,10 @@ def __init__(self, serdes, sci):
         #
         # I/O port
         #
+        self.enc_bypass  = Signal()
         self.loopback    = Signal()
         self.rx_polarity = Signal()
+        self.tx_deemph   = Signal(TXDeemphMode)
         self.tx_idle     = Signal()
         self.tx_polarity = Signal()
         self.rx_termination = Signal()
@@ -189,6 +191,67 @@ def elaborate(self, platform):
                     sci.dat_w[6].eq(self.tx_idle),  # pcie_ei_en
                 ]
 
+                with m.If(~first & sci.done):
+                    m.d.pipe += first.eq(1)
+                    m.d.comb += sci.we.eq(0)
+                    m.next = "READ-CH_03"
+
+            with m.State("READ-CH_03"):
+                m.d.pipe += first.eq(0)
+                m.d.comb += [
+                    sci.chan_sel.eq(1),
+                    sci.re.eq(1),
+                    sci.adr.eq(0x03),
+                ]
+
+                with m.If(~first & sci.done):
+                    m.d.pipe += first.eq(1)
+                    m.d.comb += sci.re.eq(0)
+                    m.d.pipe += data.eq(sci.dat_r)
+                    m.next = "WRITE-CH_03"
+
+            with m.State("WRITE-CH_03"):
+                m.d.pipe += first.eq(0)
+                m.d.comb += [
+                    sci.chan_sel.eq(1),
+                    sci.we.eq(1),
+                    sci.adr.eq(0x03),
+                    sci.dat_w.eq(data),
+                    sci.dat_w[3].eq(self.enc_bypass),  # enc_bypass
+                ]
+
+                with m.If(~first & sci.done):
+                    m.d.pipe += first.eq(1)
+                    m.d.comb += sci.we.eq(0)
+                    m.next = "READ-CH_12"
+
+            with m.State("READ-CH_12"):
+                m.d.pipe += first.eq(0)
+                m.d.comb += [
+                    sci.chan_sel.eq(1),
+                    sci.re.eq(1),
+                    sci.adr.eq(0x12),
+                ]
+
+                with m.If(~first & sci.done):
+                    m.d.comb += sci.re.eq(0)
+                    m.d.pipe += first.eq(1)
+                    m.d.pipe += data.eq(sci.dat_r)
+                    m.next = "WRITE-CH_12"
+
+            with m.State("WRITE-CH_12"):
+                m.d.pipe += first.eq(0)
+                # If de-emphasis is enabled, drive one slice with Post Data,
+                # else drive it with Main Data.
+                slice2_sel = Mux(self.tx_deemph == TXDeemphMode.DEEMPH_3P5DB, 0b11, 0b01)
+                m.d.comb += [
+                    sci.chan_sel.eq(1),
+                    sci.we.eq(1),
+                    sci.adr.eq(0x12),
+                    sci.dat_w.eq(data),
+                    sci.dat_w[4:6].eq(slice2_sel),
+                ]
+
                 with m.If(~first & sci.done):
                     m.d.pipe += first.eq(1)
                     m.d.comb += sci.we.eq(0)
@@ -220,6 +283,7 @@ def elaborate(self, platform):
                     m.d.comb += sci.dat_w[0:6].eq(0b110010) # lb_ctl
 
                 with m.If(~first & sci.done):
+                    m.d.pipe += first.eq(1)
                     m.d.comb += sci.we.eq(0)
                     m.next = "READ-CH_17"
 
@@ -695,6 +759,8 @@ def __init__(self, pll_config, tx_pads, rx_pads, dual=0, channel=0):
         self.rx_datak       = Signal(self._io_words)
 
         # TX controls
+        self.tx_deemph      = Signal(TXDeemphMode)
+        self.tx_ones_zeros  = Signal()
         self.tx_polarity    = Signal()
         self.tx_elec_idle   = Signal()
         self.tx_gpio_en     = Signal()
@@ -763,6 +829,8 @@ def elaborate(self, platform):
         m.submodules.sci = sci = ECP5SerDesConfigInterface(self)
         m.submodules.sci_trans = sci_trans = ECP5SerDesRegisterTranslator(self, sci)
         m.d.comb += [
+            sci_trans.enc_bypass    .eq(self.tx_ones_zeros),
+            sci_trans.tx_deemph     .eq(self.tx_deemph),
             sci_trans.tx_polarity   .eq(self.tx_polarity),
             sci_trans.rx_polarity   .eq(self.rx_polarity),
             sci_trans.rx_termination.eq(self.rx_termination),
@@ -1006,18 +1074,27 @@ def elaborate(self, platform):
             p_CHX_TXAMPLITUDE       = "0d1000", # 1000 mV
 
             # CHX TX — equalization
-            p_CHX_TDRV_SLICE0_CUR   = "0b011",  # 400 uA
+            p_CHX_TDRV_SLICE0_CUR   = "0b001",  # 200 uA
             p_CHX_TDRV_SLICE0_SEL   = "0b01",   # main data
+
             p_CHX_TDRV_SLICE1_CUR   = "0b000",  # 100 uA
             p_CHX_TDRV_SLICE1_SEL   = "0b00",   # power down
-            p_CHX_TDRV_SLICE2_CUR   = "0b11",   # 3200 uA
+
+            # This slice will be switched to post data by SCI when de-emphasis is enabled.
+            p_CHX_TDRV_SLICE2_CUR   = "0b01",   # 1600 uA
             p_CHX_TDRV_SLICE2_SEL   = "0b01",   # main data
-            p_CHX_TDRV_SLICE3_CUR   = "0b10",   # 2400 uA
+
+            p_CHX_TDRV_SLICE3_CUR   = "0b11",   # 3200 uA
             p_CHX_TDRV_SLICE3_SEL   = "0b01",   # main data
-            p_CHX_TDRV_SLICE4_CUR   = "0b00",   # 800 uA
-            p_CHX_TDRV_SLICE4_SEL   = "0b00",   # power down
-            p_CHX_TDRV_SLICE5_CUR   = "0b00",   # 800 uA
-            p_CHX_TDRV_SLICE5_SEL   = "0b00",   # power down
+
+            p_CHX_TDRV_SLICE4_CUR   = "0b11",   # 3200 uA
+            p_CHX_TDRV_SLICE4_SEL   = "0b01",   # main data
+
+            p_CHX_TDRV_SLICE5_CUR   = "0b01",   # 1600 uA
+            p_CHX_TDRV_SLICE5_SEL   = "0b01",   # main data
+
+            p_CHX_TDRV_PRE_EN       = "0b0",
+            p_CHX_TDRV_POST_EN      = "0b1",
 
             # CHX TX — clocking
             p_CHX_FF_TX_F_CLK_DIS   = "0b0",    # enable  DIV/1 output clock
@@ -1084,6 +1161,21 @@ def elaborate(self, platform):
             tx_bus[20]          .eq(self.tx_datak[1]),
         ]
 
+        # If `tx_ones_zeros` is set, override the TX data.
+        tx_pattern = Signal.like(tx_bus)
+        counter = Signal(range(9))
+        m.d.pipe += [
+            counter.eq(counter + 1),
+            # When bypassing the 8b10b encoder, the bus expands to 10-bit.
+            # However, we're still in PCIe mode and need to avoid setting `pcie_ei_en`.
+            # (table 7.3 in the user guide is not very clear on this case)
+            tx_pattern[0 :10].eq(Mux(counter[-1], -1, 0)),
+            tx_pattern[12:22].eq(Mux(counter[-1], -1, 0)),
+        ]
+        with m.If(self.tx_ones_zeros):
+            m.d.comb += tx_bus.eq(tx_pattern)
+
+
         # The SerDes is providing us with two RxStatus words, one per byte; but we emit only one
         # for the entire word. Combine the status conditions together according to their priorities.
         for rx_status_code in 0b011, 0b010, 0b001, 0b111, 0b110, 0b101, 0b100:
@@ -1124,8 +1216,9 @@ class ECP5SerDesPIPE(PIPEInterface, Elaboratable):
     tx_detrx_lpbk :
     tx_elec_idle :
         Transmit control signals. Loopback and receiver detection are not implemented.
+    tx_ones_zeros :
+        Transmit 50-250 ones and 50-250 zeros. This implementation transmits 160 of each.
     tx_compliance :
-    tx_ones_zeroes :
     rx_eq_training :
         These inputs are not implemented.
     power_present :
@@ -1157,6 +1250,7 @@ def elaborate(self, platform):
             pll_config  = pll_config,
             tx_pads     = self._tx_pads,
             rx_pads     = self._rx_pads,
+            dual        = self._dual,
             channel     = self._channel,
         )
 
@@ -1187,7 +1281,9 @@ def elaborate(self, platform):
             serdes.reset            .eq(self.reset),
             self.pclk               .eq(serdes.pclk),
 
+            serdes.tx_deemph        .eq(self.tx_deemph),
             serdes.tx_elec_idle     .eq(self.tx_elec_idle),
+            serdes.tx_ones_zeros    .eq(self.tx_ones_zeros),
             serdes.rx_polarity      .eq(self.rx_polarity),
             serdes.rx_termination   .eq(self.rx_termination),
             lfps_generator.generate .eq(self.tx_detrx_lpbk & self.tx_elec_idle),

luna/gateware/interface/serdes_phy/xc7_gtp.py

@@ -959,7 +959,7 @@ class XC7GTPSerDesPIPE(PIPEInterface, Elaboratable):
     tx_elec_idle :
         Transmit control signals. Loopback and receiver detection are not implemented.
     tx_compliance :
-    tx_ones_zeroes :
+    tx_ones_zeros :
         These inputs are not implemented.
     power_present :
         This output is not implemented. External logic may drive it if necessary.

luna/gateware/interface/serdes_phy/xc7_gtx.py

@@ -1017,7 +1017,7 @@ class XC7GTXSerDesPIPE(PIPEInterface, Elaboratable):
     tx_elec_idle :
         Transmit control signals. Loopback and receiver detection are not implemented.
     tx_compliance :
-    tx_ones_zeroes :
+    tx_ones_zeros :
         These inputs are not implemented.
     power_present :
         This output is not implemented. External logic may drive it if necessary.

luna/gateware/usb/usb3/link/compliance.py

@@ -0,0 +1,74 @@
+#
+# This file is part of LUNA.
+#
+# Copyright (c) 2026 Great Scott Gadgets <info@greatscottgadgets.com>
+#
+# SPDX-License-Identifier: BSD-3-Clause
+""" Compliance pattern generation gateware.
+"""
+
+from collections import namedtuple
+
+from amaranth import *
+
+from ..physical.coding import NamedSymbol, D, K, COM, IDL
+from ...stream         import USBRawSuperSpeedStream
+
+Pattern = namedtuple(
+    'Pattern',
+    ['value', 'ctrl', 'scrambling', 'lfps', 'deemphasis', 'oneszeros'],
+    defaults=[0, 0, False, False, True, False],
+)
+
+PATTERNS = [
+    Pattern(IDL.value, IDL.ctrl, scrambling=True), # CP0: D0.0 scrambled
+    Pattern(D(10, 2)),                             # CP1: Nyquist frequency
+    Pattern(D(24, 3)),                             # CP2: Nyquist/2
+    Pattern(COM.value, COM.ctrl),                  # CP3: COM pattern
+    Pattern(lfps=True),                            # CP4: LFPS
+    Pattern(K(27, 7), ctrl=1),                     # CP5: K27.7 with de-emphasis
+    Pattern(K(27, 7), ctrl=1, deemphasis=False),   # CP6: K27.7 without de-emphasis
+    Pattern(oneszeros=True),                       # CP7: 50-250 ones & 50-250 zeros with de-emphasis
+    Pattern(oneszeros=True, deemphasis=False),     # CP8: 50-250 ones & 50-250 zeros without de-emphasis
+]
+
+
+class CompliancePatternEmitter(Elaboratable):
+    """ TODO
+    """
+    def __init__(self):
+
+        #
+        # I/O port
+        #
+        self.source            = USBRawSuperSpeedStream()
+
+        self.enable            = Signal()
+        self.enable_scrambling = Signal()
+        self.disable_deemph    = Signal()
+        self.tx_ones_zeros     = Signal()
+
+
+
+    def elaborate(self, platform):
+        m = Module()
+
+        # TODO: hook up LFPS.Ping to switch patterns
+        pmod = platform.request("pmod", 0).i
+        current_pattern = pmod[0:4]
+
+        with m.Switch(current_pattern):
+            for i, pattern in enumerate(PATTERNS):
+                with m.Case(i):
+                    with m.If(self.enable):
+                        m.d.comb += [
+                            self.source.valid     .eq(1),
+                            self.source.data      .eq(C(pattern.value, 8).replicate(4)),
+                            self.source.ctrl      .eq(C(pattern.ctrl, 1).replicate(4)),
+                            self.enable_scrambling.eq(pattern.scrambling),
+                            self.disable_deemph   .eq(~pattern.deemphasis),
+                            self.tx_ones_zeros    .eq(pattern.oneszeros),
+                        ]
+
+        return m
+