phy: add debounced link_up/connect/disconnect events

litesata/gen.py

@@ -83,8 +83,11 @@ def get_sata_ios():
 
 def get_ctrl_ios():
     return [
-        ("enable", 0, Pins(1)),
-        ("ready",  0, Pins(1)),
+        ("enable",     0, Pins(1)),
+        ("ready",      0, Pins(1)),
+        ("link_up",    0, Pins(1)), # Debounced link status.
+        ("connect",    0, Pins(1)), # Single-cycle pulse on link up.
+        ("disconnect", 0, Pins(1)), # Single-cycle pulse on (debounced) link down.
     ]
 
 def get_irq_ios():
@@ -165,6 +168,9 @@ def __init__(self, platform, core_config):
             with_csr   = bus_standard is not None)
         self.comb += self.sata_phy.enable.eq(platform.request("enable"))
         self.comb += platform.request("ready").eq(self.sata_phy.ready)
+        self.comb += platform.request("link_up").eq(self.sata_phy.link_up)
+        self.comb += platform.request("connect").eq(self.sata_phy.connect_irq)
+        self.comb += platform.request("disconnect").eq(self.sata_phy.disconnect_irq)
 
         if bus_standard is None:
             # Native core: SATA Core + Crossbar only -----------------------------------------------

litesata/phy/__init__.py

@@ -31,7 +31,7 @@ class LiteSATAPHY(LiteXModule):
     """
     def __init__(self, device, pads, gen, clk_freq, refclk=None, data_width=16,
                  qpll=None, gt_type="GTY", use_gtgrefclk=True, with_csr=True,
-                 dual=0, channel=0):
+                 dual=0, channel=0, link_down_timeout=100e-3):
         self.pads   = pads
         self.gen    = gen
         self.refclk = refclk
@@ -118,6 +118,13 @@ def __init__(self, device, pads, gen, clk_freq, refclk=None, data_width=16,
             self.comb += self.phy.serdes.init.rst.eq(~self.enable)
         self.comb += self.ready.eq(self.phy.ready & self.ctrl.ready)
 
+        # Link events.
+        # ------------
+        self.link_monitor   = LiteSATAPHYLinkMonitor(self.ready, down_cycles=int(link_down_timeout*clk_freq))
+        self.link_up        = self.link_monitor.link_up
+        self.connect_irq    = self.link_monitor.connect
+        self.disconnect_irq = self.link_monitor.disconnect
+
         # CSRs.
         # -----
         if with_csr:
@@ -142,6 +149,10 @@ def  add_csr(self):
                     ("``0b0``", "Ctrl/OOB not initialized."),
                     ("``0b1``", "Ctrl/OOB initialized and ready.")
             ]),
+            CSRField("link_up", size=1, values=[
+                    ("``0b0``", "Link down (debounced)."),
+                    ("``0b1``", "Link up.")
+            ]),
         ])
 
         self.comb += self.enable.eq(self._enable.storage)
@@ -156,3 +167,4 @@ def  add_csr(self):
             self.comb += self._status.fields.tx_ready.eq(self.phy.ready)
             self.comb += self._status.fields.rx_ready.eq(self.phy.ready)
         self.comb += self._status.fields.ctrl_ready.eq(self.ctrl.ready)
+        self.comb += self._status.fields.link_up.eq(self.link_up)

litesata/phy/ctrl.py

@@ -187,3 +187,34 @@ def __init__(self, trx, crg, clk_freq):
     def us(self, t):
         clk_period_us = 1e6/self.clk_freq
         return ceil(t/clk_period_us)
+
+# LiteSATAPHYLinkMonitor ---------------------------------------------------------------------------
+
+class LiteSATAPHYLinkMonitor(LiteXModule):
+    """Debounced link up/down events derived from the PHY ready signal.
+
+    link_up rises as soon as ready is asserted (ready already includes the controller's stability
+    time) and falls only once ready has stayed low for down_cycles, so the short re-initializations
+    performed by the controller (misalignment, idle retries) do not generate events while a device
+    unplug does. connect/disconnect are single-cycle pulses (EventSourcePulse compatible).
+    """
+    def __init__(self, ready, down_cycles):
+        self.link_up    = Signal()
+        self.connect    = Signal()
+        self.disconnect = Signal()
+
+        # # #
+
+        self.timer = timer = WaitTimer(down_cycles)
+        link_up_d = Signal()
+        self.comb += timer.wait.eq(~ready)
+        self.sync += [
+            If(ready,
+                self.link_up.eq(1)
+            ).Elif(timer.done,
+                self.link_up.eq(0)
+            ),
+            link_up_d.eq(self.link_up),
+            self.connect.eq(self.link_up & ~link_up_d),
+            self.disconnect.eq(~self.link_up & link_up_d),
+        ]

test/test_gen.py

@@ -107,6 +107,8 @@ def test_examples(self):
                 verilog, files = generate_example(name)
                 ports = re.findall(r"^\s*(?:input|output)\s+wire\s*(?:\[[^\]]*\])?\s*(\w+),?$", verilog, re.M)
                 self.assertIn(primitive, verilog)
+                for pad in ["enable", "ready", "link_up", "connect", "disconnect"]:
+                    self.assertIn(pad, ports)
                 self.assertIn("litesata_core.v", files)
                 self.assertIn("csr.csv", files)
                 # Native port: the sink data pad must drive the crossbar (it was left unconnected).

test/test_phy_ctrl.py

@@ -11,7 +11,7 @@
 from migen.sim import run_simulation
 
 from litesata.common import primitives
-from litesata.phy.ctrl import LiteSATAPHYCtrl
+from litesata.phy.ctrl import LiteSATAPHYCtrl, LiteSATAPHYLinkMonitor
 from litesata.phy.ecp5sataphy import ECP5LiteSATAPHYCtrl
 
 
@@ -116,5 +116,47 @@ def test_early_align_latch_is_opt_in(self):
         self.assertEqual(result["data"], primitives["ALIGN"])
 
 
+class TestPHYLinkMonitor(unittest.TestCase):
+    def test_link_events(self):
+        class DUT(Module):
+            def __init__(self):
+                self.ready = Signal()
+                self.submodules.monitor = LiteSATAPHYLinkMonitor(self.ready, down_cycles=16)
+
+        dut    = DUT()
+        events = []
+
+        def generator():
+            def drive(ready, cycles):
+                yield dut.ready.eq(ready)
+                for _ in range(cycles):
+                    yield
+                    if (yield dut.monitor.connect):
+                        events.append("connect")
+                    if (yield dut.monitor.disconnect):
+                        events.append("disconnect")
+
+            yield from drive(0, 4)
+            self.assertEqual(events, [])
+            # Ready: immediate link up + one connect pulse.
+            yield from drive(1, 4)
+            self.assertEqual((yield dut.monitor.link_up), 1)
+            self.assertEqual(events, ["connect"])
+            # Short drop (re-initialization): no event.
+            yield from drive(0, 8)
+            yield from drive(1, 4)
+            self.assertEqual((yield dut.monitor.link_up), 1)
+            self.assertEqual(events, ["connect"])
+            # Long drop (unplug): one disconnect pulse after the timeout.
+            yield from drive(0, 32)
+            self.assertEqual((yield dut.monitor.link_up), 0)
+            self.assertEqual(events, ["connect", "disconnect"])
+            # Back: one connect pulse.
+            yield from drive(1, 4)
+            self.assertEqual(events, ["connect", "disconnect", "connect"])
+
+        run_simulation(dut, generator())
+
+
 if __name__ == "__main__":
     unittest.main()