Replay CAN frames in real-time

pycyphal/transport/can/media/candump/_candump.py

@@ -23,7 +23,7 @@
 
 class CandumpMedia(Media):
     """
-    This is a pseudo-media layer that reads standard SocketCAN candump log files.
+    This is a pseudo-media layer that replays standard SocketCAN candump log files.
     It can be used to perform postmortem analysis of a Cyphal/CAN network based on the standard log files
     collected by ``candump``.
 
@@ -58,10 +58,22 @@ class CandumpMedia(Media):
 
     Each line contains a CAN frame which is reported as received with the specified wall (system) timestamp.
     This media layer, naturally, cannot accept outgoing frames, so they are dropped (and logged).
+
+    Usage example with `Yakut <https://github.com/OpenCyphal/yakut>`_::
+
+        export UAVCAN__CAN__IFACE='candump:verification/integration/candump.log'
+        y sub uavcan.node.heartbeat 10:reg.udral.service.common.readiness 130:reg.udral.service.actuator.common.status
+        y mon
+
+    ..  warning::
+
+        The API of this class is experimental and subject to breaking changes.
     """
 
     GLOB_PATTERN = "candump*.log"
 
+    _BATCH_SIZE_LIMIT = 100
+
     def __init__(self, file: str | Path | TextIO) -> None:
         self._f: TextIO = (
             open(file, "r", encoding="utf8")  # pylint: disable=consider-using-with
@@ -117,16 +129,20 @@ def _is_closed(self) -> bool:
         return self._thread is None
 
     def _thread_function(self, handler: Media.ReceivedFramesHandler, loop: asyncio.AbstractEventLoop) -> None:
-        def forward(rec: DataFrameRecord) -> None:
-            frm = (
-                rec.ts,
-                Envelope(
-                    frame=DataFrame(format=rec.fmt, identifier=rec.can_id, data=bytearray(rec.can_payload)),
-                    loopback=False,
-                ),
-            )
+        def forward(batch: list[DataFrameRecord]) -> None:
             if not self._is_closed:  # Don't call after closure to prevent race conditions and use-after-close.
-                pycyphal.util.broadcast([handler])([frm])
+                pycyphal.util.broadcast([handler])(
+                    [
+                        (
+                            rec.ts,
+                            Envelope(
+                                frame=DataFrame(format=rec.fmt, identifier=rec.can_id, data=bytearray(rec.can_payload)),
+                                loopback=False,
+                            ),
+                        )
+                        for rec in batch
+                    ]
+                )
 
         try:
             _logger.debug("%r: Waiting for the acceptance filters to be configured before proceeding...", self)
@@ -138,6 +154,8 @@ def forward(rec: DataFrameRecord) -> None:
                 else:
                     break
             _logger.debug("%r: Acceptance filters configured, starting to read frames", self)
+            batch: list[DataFrameRecord] = []
+            time_offset: float | None = None
             for idx, line in enumerate(self._f):
                 rec = Record.parse(line)
                 if not rec:
@@ -158,7 +176,19 @@ def forward(rec: DataFrameRecord) -> None:
                         self._iface_name,
                     )
                     continue
-                loop.call_soon_threadsafe(forward, rec)
+                now_mono = time.monotonic()
+                ts = float(rec.ts.system)
+                if time_offset is None:
+                    time_offset = ts - now_mono
+                target_mono = ts - time_offset
+                sleep_duration = target_mono - now_mono
+                if sleep_duration > 0 or len(batch) > self._BATCH_SIZE_LIMIT:
+                    loop.call_soon_threadsafe(forward, batch)
+                    batch = []
+                    if sleep_duration > 0:
+                        time.sleep(sleep_duration)
+                batch.append(rec)
+            loop.call_soon_threadsafe(forward, batch)
         except BaseException as ex:  # pylint: disable=broad-except
             if not self._is_closed:
                 _logger.exception("%r: Log file reader failed: %s", self, ex)

tests/application/transport_factory_candump.py

@@ -2,6 +2,7 @@
 # This software is distributed under the terms of the MIT License.
 # Author: Pavel Kirienko <pavel@opencyphal.org>
 
+from __future__ import annotations
 import typing
 import asyncio
 from decimal import Decimal
@@ -41,38 +42,41 @@ def handle_capture(cap: Capture) -> None:
         captures.append(cap)
 
     tr.begin_capture(handle_capture)
-    await asyncio.sleep(5.0)
-    tr.close()
-
-    assert len(captures) == 4
+    await asyncio.sleep(4.0)
+    assert len(captures) == 2
 
-    assert captures[0].timestamp.system == Decimal("1657800496.359233")
+    assert captures[0].timestamp.system == Decimal("1657800490.360135")
     assert captures[0].frame.identifier == 0x0C60647D
     assert captures[0].frame.format == pycyphal.transport.can.media.FrameFormat.EXTENDED
     assert captures[0].frame.data == bytes.fromhex("020000FB")
 
-    assert captures[1].timestamp.system == Decimal("1657800496.360136")
+    assert captures[1].timestamp.system == Decimal("1657800490.360136")
     assert captures[1].frame.identifier == 0x10606E7D
     assert captures[1].frame.format == pycyphal.transport.can.media.FrameFormat.EXTENDED
     assert captures[1].frame.data == bytes.fromhex("00000000000000BB")
 
-    assert captures[2].timestamp.system == Decimal("1657800496.360152")
-    assert captures[2].frame.identifier == 0x10606E7D
-    assert captures[2].frame.format == pycyphal.transport.can.media.FrameFormat.EXTENDED
-    assert captures[2].frame.data == bytes.fromhex("000000000000003B")
+    captures.clear()
+    await asyncio.sleep(10.0)
+    tr.close()
+    assert len(captures) == 2
+
+    assert captures[0].timestamp.system == Decimal("1657800499.360152")
+    assert captures[0].frame.identifier == 0x10606E7D
+    assert captures[0].frame.format == pycyphal.transport.can.media.FrameFormat.EXTENDED
+    assert captures[0].frame.data == bytes.fromhex("000000000000003B")
 
-    assert captures[3].timestamp.system == Decimal("1657800496.360317")
-    assert captures[3].frame.identifier == 0x1060787D
-    assert captures[3].frame.format == pycyphal.transport.can.media.FrameFormat.EXTENDED
-    assert captures[3].frame.data == bytes.fromhex("0000C07F147CB71B")
+    assert captures[1].timestamp.system == Decimal("1657800499.360317")
+    assert captures[1].frame.identifier == 0x1060787D
+    assert captures[1].frame.format == pycyphal.transport.can.media.FrameFormat.EXTENDED
+    assert captures[1].frame.data == bytes.fromhex("0000C07F147CB71B")
 
 
 _CANDUMP_TEST_DATA = """
-(1657800496.359233) slcan0 0C60647D#020000FB
-(1657800496.360136) slcan0 10606E7D#00000000000000BB
-(1657800496.360149) slcan1 10606E7D#000000000000001B
-(1657800496.360152) slcan0 10606E7D#000000000000003B
-(1657800496.360305) slcan2 1060787D#00000000000000BB
-(1657800496.360317) slcan0 1060787D#0000C07F147CB71B
-(1657800496.361011) slcan1 1060787D#412BCC7B
+(1657800490.360135) slcan0 0C60647D#020000FB
+(1657800490.360136) slcan0 10606E7D#00000000000000BB
+(1657800490.360149) slcan1 10606E7D#000000000000001B
+(1657800499.360152) slcan0 10606E7D#000000000000003B
+(1657800499.360305) slcan2 1060787D#00000000000000BB
+(1657800499.360317) slcan0 1060787D#0000C07F147CB71B
+(1657800499.361011) slcan1 1060787D#412BCC7B
 """