interface/stream_utils: add the LastOnTimeout module

Author: povauboin

lambdalib/interface/stream_utils.py

@@ -11,6 +11,7 @@
     "Splitter",
     "Merger",
     "LastInserter",
+    "LastOnTimeout",
     "Arbiter",
 ]
 
@@ -252,6 +253,54 @@ def elaborate(self, platform):
         return m
 
 
+class LastOnTimeout(Elaboratable):
+    """ This module injects a `last` signal into the source stream
+    if no more data is received from the sink stream since more than
+    `timeout` clock cycles.
+
+    Typically this module can be connected to a UART receiver to delimit
+    packets based on the idle time between packets, or to to realign
+    a converter when no more data is received.
+    """
+    def __init__(self, layout, timeout=1):
+        self.timeout = timeout
+        self.sink = stream.Endpoint(layout)
+        self.source = stream.Endpoint(layout)
+
+    def elaborate(self, platform):
+        sink = self.sink
+        source = self.source
+
+        m = Module()
+
+        m.submodules.timer = timer = WaitTimer(self.timeout)
+        m.d.comb += timer.wait.eq(~sink.valid)
+
+        last_r = Signal()
+        loaded = Signal()
+
+        with m.If(sink.valid & sink.ready):
+            m.d.sync += [
+                source.data.eq(sink.data),
+                last_r     .eq(sink.last),
+                loaded     .eq(1),
+            ]
+        with m.Elif(source.valid & source.ready):
+            m.d.sync += loaded.eq(0)
+
+        with m.If(~source.valid | source.ready):
+            m.d.comb += sink.ready.eq(1)
+
+        # We send the data to the source stream immediately
+        # when it was the last data, or when more data is incoming,
+        # or if we waited until the timeout expired.
+        with m.If(loaded & (last_r | sink.valid | timer.done)):
+            m.d.comb += source.valid.eq(1)
+        m.d.comb += source.last.eq(last_r | timer.done)
+
+        return m
+
+
 class RoundRobin(Elaboratable):
     """ Fair round robin.
     """

lambdalib/tests/test_interface_stream.py

@@ -117,7 +117,30 @@ def test_last_inserter():
         sim.run()
 
 
+def test_last_timeout():
+    lot = LastOnTimeout([("data", 8)], timeout=10)
+    sim = Simulator(lot)
+
+    data = {
+        "data": range(20),
+        "last": [0]*3 + [1] + [0]*10 + [1] + [0]*5,
+    }
+
+    length = len(data["data"])
+    sender = StreamSimSender(lot.sink, data, speed=0.2)
+    receiver = StreamSimReceiver(lot.source,
+                                 length=length,
+                                 speed=0.8, verbose=True)
+
+    sim.add_clock(1e-6)
+    sim.add_sync_process(sender.sync_process)
+    sim.add_sync_process(receiver.sync_process)
+    with sim.write_vcd("tests/test_stream_last_on_timeout.vcd"):
+        sim.run()
+
+
 if __name__ == "__main__":
     test_splitter(); print()
     test_merger(); print()
     test_last_inserter(); print()
+    test_last_timeout(); print()