Detect types in _parse_additonal_config

can/logger.py

@@ -13,12 +13,12 @@
 
 Dynamic Controls 2010
 """
-
+import re
 import sys
 import argparse
 from datetime import datetime
 import errno
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Union, Sequence, Tuple
 
 import can
 from . import Bus, BusState, Logger, SizedRotatingLogger
@@ -132,11 +132,32 @@ def _parse_filters(parsed_args: Any) -> CanFilters:
     return can_filters
 
 
-def _parse_additonal_config(unknown_args):
-    return dict(
-        (arg.split("=", 1)[0].lstrip("--").replace("-", "_"), arg.split("=", 1)[1])
-        for arg in unknown_args
-    )
+def _parse_additional_config(
+    unknown_args: Sequence[str],
+) -> Dict[str, Union[str, int, float, bool]]:
+    for arg in unknown_args:
+        if not re.match(r"^--[a-zA-Z\-]*?=\S*?$", arg):
+            raise ValueError(f"Parsing argument {arg} failed")
+
+    def _split_arg(_arg: str) -> Tuple[str, str]:
+        left, right = _arg.split("=", 1)
+        return left.lstrip("--").replace("-", "_"), right
+
+    args: Dict[str, Union[str, int, float, bool]] = {}
+    for key, string_val in map(_split_arg, unknown_args):
+        if re.match(r"^[-+]?\d+$", string_val):
+            # value is integer
+            args[key] = int(string_val)
+        elif re.match(r"^[-+]?\d*\.\d+$", string_val):
+            # value is float
+            args[key] = float(string_val)
+        elif re.match(r"^(?:True|False)$", string_val):
+            # value is bool
+            args[key] = string_val == "True"
+        else:
+            # value is string
+            args[key] = string_val
+    return args
 
 
 def main() -> None:
@@ -190,7 +211,7 @@ def main() -> None:
         raise SystemExit(errno.EINVAL)
 
     results, unknown_args = parser.parse_known_args()
-    additional_config = _parse_additonal_config(unknown_args)
+    additional_config = _parse_additional_config(unknown_args)
     bus = _create_bus(results, can_filters=_parse_filters(results), **additional_config)
 
     if results.active:

can/player.py

@@ -13,7 +13,7 @@
 
 from can import LogReader, Message, MessageSync
 
-from .logger import _create_base_argument_parser, _create_bus
+from .logger import _create_base_argument_parser, _create_bus, _parse_additional_config
 
 
 def main() -> None:
@@ -78,13 +78,14 @@ def main() -> None:
         parser.print_help(sys.stderr)
         raise SystemExit(errno.EINVAL)
 
-    results = parser.parse_args()
+    results, unknown_args = parser.parse_known_args()
+    additional_config = _parse_additional_config(unknown_args)
 
     verbosity = results.verbosity
 
     error_frames = results.error_frames
 
-    with _create_bus(results) as bus:
+    with _create_bus(results, **additional_config) as bus:
         with LogReader(results.infile) as reader:
 
             in_sync = MessageSync(

can/viewer.py

@@ -35,7 +35,7 @@
     _parse_filters,
     _append_filter_argument,
     _create_base_argument_parser,
-    _parse_additonal_config,
+    _parse_additional_config,
 )
 
 
@@ -540,7 +540,7 @@ def parse_args(args):
             else:
                 data_structs[key] = struct.Struct(fmt)
 
-    additional_config = _parse_additonal_config(unknown_args)
+    additional_config = _parse_additional_config(unknown_args)
     return parsed_args, can_filters, data_structs, additional_config
 
 

test/test_logger.py

@@ -10,6 +10,9 @@
 import gzip
 import os
 import sys
+
+import pytest
+
 import can
 import can.logger
 
@@ -105,6 +108,49 @@ def test_log_virtual_sizedlogger(self):
         self.assertSuccessfullCleanup()
         self.mock_logger_sized.assert_called_once()
 
+    def test_parse_additional_config(self):
+        unknown_args = [
+            "--app-name=CANalyzer",
+            "--serial=5555",
+            "--receive-own-messages=True",
+            "--false-boolean=False",
+            "--offset=1.5",
+        ]
+        parsed_args = can.logger._parse_additional_config(unknown_args)
+
+        assert "app_name" in parsed_args
+        assert parsed_args["app_name"] == "CANalyzer"
+
+        assert "serial" in parsed_args
+        assert parsed_args["serial"] == 5555
+
+        assert "receive_own_messages" in parsed_args
+        assert (
+            isinstance(parsed_args["receive_own_messages"], bool)
+            and parsed_args["receive_own_messages"] is True
+        )
+
+        assert "false_boolean" in parsed_args
+        assert (
+            isinstance(parsed_args["false_boolean"], bool)
+            and parsed_args["false_boolean"] is False
+        )
+
+        assert "offset" in parsed_args
+        assert parsed_args["offset"] == 1.5
+
+        with pytest.raises(ValueError):
+            can.logger._parse_additional_config(["--wrong-format"])
+
+        with pytest.raises(ValueError):
+            can.logger._parse_additional_config(["-wrongformat=value"])
+
+        with pytest.raises(ValueError):
+            can.logger._parse_additional_config(["--wrongformat=value1 value2"])
+
+        with pytest.raises(ValueError):
+            can.logger._parse_additional_config(["wrongformat="])
+
 
 class TestLoggerCompressedFile(unittest.TestCase):
     def setUp(self) -> None: