Add MCAP support and use osi-python

pyproject.toml

@@ -13,7 +13,7 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.10"
 asam-qc-baselib = {git = "https://github.com/asam-ev/qc-baselib-py.git", rev = "main"}
-open-simulation-interface = {git = "https://github.com/OpenSimulationInterface/open-simulation-interface.git", rev = "v3.7.0"}
+osi-python = {git = "https://github.com/OpenSimulationInterface/osi-python.git", rev = "main"}
 pyyaml = "^6.0.0"
 iso3166 = "^2.1.1"
 

qc_ositrace/checks/deserialization/deserialization_checker.py

@@ -68,7 +68,9 @@ def run_checks(config: Configuration, result: Result) -> None:
     try:
         try:
             trace = OSITrace(
-                config.get_config_param("InputFile"), config.get_config_param("osiType")
+                path=config.get_config_param("InputFile"),
+                type_name=config.get_config_param("osiType"),
+                topic=config.get_config_param("osiTopic"),
             )
         except Exception as e:
             logging.error(f"Error reading input file: {e}")

qc_ositrace/checks/osirules/osirules_checker.py

@@ -406,7 +406,11 @@ def check_message_against_rules(
                         )
                     else:
                         # Check if referred message matches one of the expected types
-                        expected_types = ["osi3." + t for t in rule['refers_to']] if isinstance(rule['refers_to'], list) else ["osi3." + rule['refers_to']]
+                        expected_types = (
+                            ["osi3." + t for t in rule["refers_to"]]
+                            if isinstance(rule["refers_to"], list)
+                            else ["osi3." + rule["refers_to"]]
+                        )
                         if referred_message.DESCRIPTOR.full_name not in expected_types:
                             register_issue(
                                 result,
@@ -469,7 +473,11 @@ def run_checks(config: Configuration, result: Result) -> None:
 
     try:
         try:
-            trace = OSITrace(config.get_config_param("InputFile"), expected_type_name)
+            trace = OSITrace(
+                path=config.get_config_param("InputFile"),
+                type_name=expected_type_name,
+                topic=config.get_config_param("osiTopic"),
+            )
         except Exception as e:
             logging.error(f"Error reading input file: {e}")
             raise RuntimeError(f"Error reading input file: {e}") from e

qc_ositrace/main.py

@@ -28,6 +28,11 @@ def args_entrypoint() -> argparse.Namespace:
         type=pathlib.Path,
         help="Path to the input OSI Trace file.",
     )
+    parser.add_argument(
+        "--osiTopic",
+        type=str,
+        help="Channel topic of a multi-trace OSI Trace file to select.",
+    )
     parser.add_argument(
         "--osiType",
         type=str,
@@ -101,6 +106,9 @@ def main():
         logging.info("Setting input file: %s", args.input_file)
         config.set_config_param("InputFile", str(args.input_file))
 
+    if args.osiTopic:
+        logging.info("Setting OSI Topic: %s", args.osiTopic)
+        config.set_config_param("osiTopic", args.osiTopic)
     if args.osiType:
         logging.info("Setting OSI Type: %s", args.osiType)
         config.set_config_param("osiType", args.osiType)

tests/test_deserialization_checks.py

@@ -4,6 +4,9 @@
 
 from typing import List
 
+from qc_ositrace.checks.deserialization.deserialization_constants import (
+    CHECKER_ID as deserialization_checker_id,
+)
 from qc_baselib import IssueSeverity
 
 from test_setup import *
@@ -58,3 +61,70 @@ def test_deserialization_expected_version(
     launch_main(monkeypatch)
     check_issues(rule_uid, issue_count, issue_severity)
     cleanup_files()
+
+
+@pytest.mark.parametrize(
+    "target_file,target_topic,target_type,target_version,issue_count",
+    [
+        ("360", "MySensorView", "SensorView", "3.5.0", 547),
+        ("360", "MySensorView", "SensorView", "3.6.0", 0),
+        ("360", "Foo", "SensorView", "3.6.0", -1),
+    ],
+)
+def test_deserialization_mcap_topic(
+    target_file: str,
+    target_topic: str,
+    target_type: str,
+    target_version: str,
+    issue_count: int,
+    monkeypatch,
+) -> None:
+    base_path = "tests/data/deserialization_expected_version/"
+    target_file_name = f"deserialization_expected_version_{target_file}.mcap"
+    rule_uid = "asam.net:osi:3.0.0:deserialization.expected_version"
+    issue_severity = IssueSeverity.ERROR
+
+    target_file_path = os.path.join(base_path, target_file_name)
+    create_test_config(
+        target_file_path, target_type, target_version, None, target_topic
+    )
+    launch_main(monkeypatch)
+    if issue_count < 0:
+        check_failure(deserialization_checker_id)
+    else:
+        check_issues(rule_uid, issue_count, issue_severity)
+    cleanup_files()
+
+
+@pytest.mark.parametrize(
+    "target_file,target_topic,target_type,target_version,issue_count",
+    [
+        ("360", "MySensorView", "SensorView", "3.5.0", 547),
+        ("360", "MySensorView", "SensorView", "3.6.0", 0),
+        ("360", "Foo", "SensorView", "3.5.0", 547),
+        ("360", "Foo", "SensorView", "3.6.0", 0),
+    ],
+)
+def test_deserialization_single_topic(
+    target_file: str,
+    target_topic: str,
+    target_type: str,
+    target_version: str,
+    issue_count: int,
+    monkeypatch,
+) -> None:
+    base_path = "tests/data/deserialization_expected_version/"
+    target_file_name = f"deserialization_expected_version_{target_file}.osi"
+    rule_uid = "asam.net:osi:3.0.0:deserialization.expected_version"
+    issue_severity = IssueSeverity.ERROR
+
+    target_file_path = os.path.join(base_path, target_file_name)
+    create_test_config(
+        target_file_path, target_type, target_version, None, target_topic
+    )
+    launch_main(monkeypatch)
+    if issue_count < 0:
+        check_failure(deserialization_checker_id)
+    else:
+        check_issues(rule_uid, issue_count, issue_severity)
+    cleanup_files()

tests/test_osirules_checks.py

@@ -91,6 +91,20 @@ def test_osirules_expected_version(
             "sensorview.mounting_position.is_set",
             547,
         ),
+        (
+            "deserialization_expected_version/deserialization_expected_version_360.mcap",
+            "SensorView",
+            "3.6.0",
+            "sensorview.mounting_position.is_set",
+            0,
+        ),
+        (
+            "deserialization_expected_version/deserialization_expected_version_360.mcap",
+            "SensorView",
+            "3.7.0",
+            "sensorview.mounting_position.is_set",
+            547,
+        ),
     ],
 )
 def test_osirules_automatic_rules(

tests/test_setup.py

@@ -3,10 +3,12 @@
 
 from typing import Optional
 
+from contextlib import suppress
+
 import qc_ositrace.main as main
 
 from qc_ositrace import constants
-from qc_baselib import Configuration, Result, IssueSeverity
+from qc_baselib import Configuration, Result, IssueSeverity, StatusType
 
 CONFIG_FILE_PATH = "bundle_config.xml"
 REPORT_FILE_PATH = "osi_bundle_report.xqar"
@@ -17,6 +19,7 @@ def create_test_config(
     target_file_type: str,
     target_file_version: Optional[str] = None,
     target_file_rules: Optional[str] = None,
+    target_file_topic: Optional[str] = None,
 ):
     test_config = Configuration()
     test_config.set_config_param(name="InputFile", value=target_file_path)
@@ -25,6 +28,8 @@ def create_test_config(
         test_config.set_config_param(name="osiVersion", value=target_file_version)
     if target_file_rules is not None:
         test_config.set_config_param(name="osiRulesFile", value=target_file_rules)
+    if target_file_topic is not None:
+        test_config.set_config_param(name="osiTopic", value=target_file_topic)
     test_config.register_checker_bundle(checker_bundle_name=constants.BUNDLE_NAME)
     test_config.set_checker_bundle_param(
         checker_bundle_name=constants.BUNDLE_NAME,
@@ -47,6 +52,14 @@ def check_issues(rule_uid: str, issue_count: int, severity: IssueSeverity):
         assert issue.level == severity
 
 
+def check_failure(checker_id: str):
+    result = Result()
+    result.load_from_file(REPORT_FILE_PATH)
+
+    status = result.get_checker_status(checker_id)
+    assert status == StatusType.ERROR
+
+
 def launch_main(monkeypatch):
     monkeypatch.setattr(
         sys,
@@ -61,5 +74,6 @@ def launch_main(monkeypatch):
 
 
 def cleanup_files():
-    os.remove(REPORT_FILE_PATH)
-    os.remove(CONFIG_FILE_PATH)
+    with suppress(Exception):
+        os.remove(REPORT_FILE_PATH)
+        os.remove(CONFIG_FILE_PATH)