add loop arg to player script (#1986)

Author: zariiii9003

can/player.py

@@ -7,6 +7,7 @@
 
 import argparse
 import errno
+import math
 import sys
 from datetime import datetime
 from typing import TYPE_CHECKING, cast
@@ -26,19 +27,41 @@
     from can import Message
 
 
+def _parse_loop(value: str) -> int | float:
+    """Parse the loop argument, allowing integer or 'i' for infinite."""
+    if value == "i":
+        return float("inf")
+    try:
+        return int(value)
+    except ValueError as exc:
+        err_msg = "Loop count must be an integer or 'i' for infinite."
+        raise argparse.ArgumentTypeError(err_msg) from exc
+
+
+def _format_player_start_message(iteration: int, loop_count: int | float) -> str:
+    """
+    Generate a status message indicating the start of a CAN log replay iteration.
+
+    :param iteration:
+        The current loop iteration (zero-based).
+    :param loop_count:
+        Total number of replay loops, or infinity for endless replay.
+    :return:
+        A formatted string describing the replay start and loop information.
+    """
+    if loop_count < 2:
+        loop_info = ""
+    else:
+        loop_val = "∞" if math.isinf(loop_count) else str(loop_count)
+        loop_info = f" [loop {iteration + 1}/{loop_val}]"
+    return f"Can LogReader (Started on {datetime.now()}){loop_info}"
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description="Replay CAN traffic.")
 
     player_group = parser.add_argument_group("Player arguments")
 
-    player_group.add_argument(
-        "-f",
-        "--file_name",
-        dest="log_file",
-        help="Path and base log filename, for supported types see can.LogReader.",
-        default=None,
-    )
-
     player_group.add_argument(
         "-v",
         action="count",
@@ -73,9 +96,20 @@ def main() -> None:
         "--skip",
         type=float,
         default=60 * 60 * 24,
-        help="<s> skip gaps greater than 's' seconds",
+        help="Skip gaps greater than 's' seconds between messages. "
+        "Default is 86400 (24 hours), meaning only very large gaps are skipped. "
+        "Set to 0 to never skip any gaps (all delays are preserved). "
+        "Set to a very small value (e.g., 1e-4) "
+        "to skip all gaps and send messages as fast as possible.",
+    )
+    player_group.add_argument(
+        "-l",
+        "--loop",
+        type=_parse_loop,
+        metavar="NUM",
+        default=1,
+        help="Replay file NUM times. Use 'i' for infinite loop (default: 1)",
     )
-
     player_group.add_argument(
         "infile",
         metavar="input-file",
@@ -103,25 +137,28 @@ def main() -> None:
     error_frames = results.error_frames
 
     with create_bus_from_namespace(results) as bus:
-        with LogReader(results.infile, **additional_config) as reader:
-            in_sync = MessageSync(
-                cast("Iterable[Message]", reader),
-                timestamps=results.timestamps,
-                gap=results.gap,
-                skip=results.skip,
-            )
-
-            print(f"Can LogReader (Started on {datetime.now()})")
-
-            try:
-                for message in in_sync:
-                    if message.is_error_frame and not error_frames:
-                        continue
-                    if verbosity >= 3:
-                        print(message)
-                    bus.send(message)
-            except KeyboardInterrupt:
-                pass
+        loop_count: int | float = results.loop
+        iteration = 0
+        try:
+            while iteration < loop_count:
+                with LogReader(results.infile, **additional_config) as reader:
+                    in_sync = MessageSync(
+                        cast("Iterable[Message]", reader),
+                        timestamps=results.timestamps,
+                        gap=results.gap,
+                        skip=results.skip,
+                    )
+                    print(_format_player_start_message(iteration, loop_count))
+
+                    for message in in_sync:
+                        if message.is_error_frame and not error_frames:
+                            continue
+                        if verbosity >= 3:
+                            print(message)
+                        bus.send(message)
+                iteration += 1
+        except KeyboardInterrupt:
+            pass
 
 
 if __name__ == "__main__":

doc/changelog.d/1815.added.md

@@ -0,0 +1 @@
+Added support for replaying CAN log files multiple times or infinitely in the player script via the new --loop/-l argument.

doc/changelog.d/1815.removed.md

@@ -0,0 +1 @@
+Removed the unused --file_name/-f argument from the player CLI.

test/test_player.py

@@ -11,6 +11,8 @@
 from unittest import mock
 from unittest.mock import Mock
 
+from parameterized import parameterized
+
 import can
 import can.player
 
@@ -38,7 +40,7 @@ def assertSuccessfulCleanup(self):
         self.mock_virtual_bus.__exit__.assert_called_once()
 
     def test_play_virtual(self):
-        sys.argv = self.baseargs + [self.logfile]
+        sys.argv = [*self.baseargs, self.logfile]
         can.player.main()
         msg1 = can.Message(
             timestamp=2.501,
@@ -65,8 +67,8 @@ def test_play_virtual(self):
         self.assertSuccessfulCleanup()
 
     def test_play_virtual_verbose(self):
-        sys.argv = self.baseargs + ["-v", self.logfile]
-        with unittest.mock.patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+        sys.argv = [*self.baseargs, "-v", self.logfile]
+        with mock.patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
             can.player.main()
         self.assertIn("09 08 07 06 05 04 03 02", mock_stdout.getvalue())
         self.assertIn("05 0c 00 00 00 00 00 00", mock_stdout.getvalue())
@@ -76,7 +78,7 @@ def test_play_virtual_verbose(self):
     def test_play_virtual_exit(self):
         self.MockSleep.side_effect = [None, KeyboardInterrupt]
 
-        sys.argv = self.baseargs + [self.logfile]
+        sys.argv = [*self.baseargs, self.logfile]
         can.player.main()
         assert self.mock_virtual_bus.send.call_count <= 2
         self.assertSuccessfulCleanup()
@@ -85,7 +87,7 @@ def test_play_skip_error_frame(self):
         logfile = os.path.join(
             os.path.dirname(__file__), "data", "logfile_errorframes.asc"
         )
-        sys.argv = self.baseargs + ["-v", logfile]
+        sys.argv = [*self.baseargs, "-v", logfile]
         can.player.main()
         self.assertEqual(self.mock_virtual_bus.send.call_count, 9)
         self.assertSuccessfulCleanup()
@@ -94,11 +96,52 @@ def test_play_error_frame(self):
         logfile = os.path.join(
             os.path.dirname(__file__), "data", "logfile_errorframes.asc"
         )
-        sys.argv = self.baseargs + ["-v", "--error-frames", logfile]
+        sys.argv = [*self.baseargs, "-v", "--error-frames", logfile]
         can.player.main()
         self.assertEqual(self.mock_virtual_bus.send.call_count, 12)
         self.assertSuccessfulCleanup()
 
+    @parameterized.expand([0, 1, 2, 3])
+    def test_play_loop(self, loop_val):
+        sys.argv = [*self.baseargs, "--loop", str(loop_val), self.logfile]
+        can.player.main()
+        msg1 = can.Message(
+            timestamp=2.501,
+            arbitration_id=0xC8,
+            is_extended_id=False,
+            is_fd=False,
+            is_rx=False,
+            channel=1,
+            dlc=8,
+            data=[0x9, 0x8, 0x7, 0x6, 0x5, 0x4, 0x3, 0x2],
+        )
+        msg2 = can.Message(
+            timestamp=17.876708,
+            arbitration_id=0x6F9,
+            is_extended_id=False,
+            is_fd=False,
+            is_rx=True,
+            channel=0,
+            dlc=8,
+            data=[0x5, 0xC, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0],
+        )
+        for i in range(loop_val):
+            self.assertTrue(
+                msg1.equals(self.mock_virtual_bus.send.mock_calls[2 * i + 0].args[0])
+            )
+            self.assertTrue(
+                msg2.equals(self.mock_virtual_bus.send.mock_calls[2 * i + 1].args[0])
+            )
+        self.assertEqual(self.mock_virtual_bus.send.call_count, 2 * loop_val)
+        self.assertSuccessfulCleanup()
+
+    def test_play_loop_infinite(self):
+        self.mock_virtual_bus.send.side_effect = [None] * 99 + [KeyboardInterrupt]
+        sys.argv = [*self.baseargs, "-l", "i", self.logfile]
+        can.player.main()
+        self.assertEqual(self.mock_virtual_bus.send.call_count, 100)
+        self.assertSuccessfulCleanup()
+
 
 class TestPlayerCompressedFile(TestPlayerScriptModule):
     """