add loop arg to player script

Author: zariiii9003

can/player.py

@@ -7,9 +7,10 @@
 
 import argparse
 import errno
+import math
 import sys
 from datetime import datetime
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Union, cast
 
 from can import LogReader, MessageSync
 from can.cli import (
@@ -26,19 +27,22 @@
     from can import Message
 
 
+def parse_loop(value: str) -> Union[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 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",
@@ -75,7 +79,14 @@ def main() -> None:
         default=60 * 60 * 24,
         help="<s> skip gaps greater than 's' seconds",
     )
-
+    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 +114,34 @@ 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: Union[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,
+                    )
+
+                    progress = (
+                        " [infinite loop]"
+                        if math.isinf(loop_count)
+                        else f" [loop {iteration + 1}/{int(loop_count)}]"
+                    )
+                    print(f"Can LogReader (Started on {datetime.now()}){progress}")
+
+                    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__":

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,51 @@ 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.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):
     """