feat: result test (#246)

Author: hayato-m126

.driving_log_replayer.cspell.json

@@ -1,5 +1,6 @@
 {
   "words": [
+    "autouse",
     "bbox",
     "kaleido",
     "xaxes",

.gitignore

@@ -1,4 +1,5 @@
 log/
 .cspell.json
+*.pyc
 # MkDocs's dir
 site/

driving_log_replayer/CMakeLists.txt

@@ -23,6 +23,10 @@ rclcpp_components_register_node(obstacle_segmentation_evaluator_component
 if(BUILD_TESTING)
   find_package(ament_lint_auto REQUIRED)
   ament_lint_auto_find_test_dependencies()
+  find_package(ament_cmake_pytest)
+  ament_add_pytest_test(driving_log_replayer
+    test
+  )
 endif()
 
 ament_python_install_package(${PROJECT_NAME})

driving_log_replayer/driving_log_replayer/evaluator.py

@@ -15,7 +15,6 @@
 from abc import ABC
 from abc import abstractmethod
 from collections.abc import Callable
-import os
 from os.path import expandvars
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -69,7 +68,7 @@ def __init__(self, name: str) -> None:
 
         self._scenario_yaml_obj = None
         try:
-            with open(self._scenario_path) as scenario_file:
+            with Path(self._scenario_path).open() as scenario_file:
                 self._scenario_yaml_obj = yaml.safe_load(scenario_file)
         except (FileNotFoundError, PermissionError, yaml.YAMLError) as e:
             self.get_logger().error(f"An error occurred while loading the scenario. {e}")
@@ -229,8 +228,8 @@ def check_scenario(self) -> None:
 
     @classmethod
     def get_goal_pose_from_t4_dataset(cls, dataset_path: str) -> PoseStamped:
-        ego_pose_json_path = os.path.join(dataset_path, "annotation", "ego_pose.json")
-        with open(ego_pose_json_path) as ego_pose_file:
+        ego_pose_json_path = Path(dataset_path).joinpath("annotation", "ego_pose.json")
+        with ego_pose_json_path.open() as ego_pose_file:
             ego_pose_json = json.load(ego_pose_file)
             last_ego_pose = ego_pose_json[-1]
             goal_pose = PoseStamped()

driving_log_replayer/driving_log_replayer/result.py

@@ -14,8 +14,10 @@
 
 from abc import ABC
 from abc import abstractmethod
-import os
+from os.path import expandvars
+from pathlib import Path
 import pickle
+from typing import Any
 
 from rclpy.clock import Clock
 from rclpy.clock import ClockType
@@ -28,12 +30,15 @@ def __init__(self) -> None:
         self._summary = "NoData"
         self._frame = {}
 
+    @property
     def success(self) -> bool:
         return self._success
 
+    @property
     def summary(self) -> str:
         return self._summary
 
+    @property
     def frame(self) -> dict:
         return self._frame
 
@@ -48,33 +53,45 @@ def set_frame(self) -> None:
 
 class ResultWriter:
     def __init__(self, result_json_path: str, ros_clock: Clock, condition: dict) -> None:
-        # 拡張子を書き換える
-        result_file = os.path.splitext(os.path.expandvars(result_json_path))[0] + ".jsonl"
-        self._result_file = open(result_file, "w")  # noqa
+        self._result_path = self.create_jsonl_path(result_json_path)
+        self._result_file = self._result_path.open("w")
         self._ros_clock = ros_clock
         self._system_clock = Clock(clock_type=ClockType.SYSTEM_TIME)
-        self.write_condition(condition)
-        self.write_header()
+        self.write_line({"Condition": condition})
+        self.write_line(self.get_header())
+
+    @property
+    def result_path(self) -> Path:
+        return self._result_path
+
+    def create_jsonl_path(self, result_json_path: str) -> Path:
+        # For compatibility with previous versions.
+        # If a json file name is passed, replace it with the filename + jsonl
+        original_path = Path(expandvars(result_json_path))
+        return original_path.parent.joinpath(original_path.stem + ".jsonl")
 
     def close(self) -> None:
         self._result_file.close()
 
-    def write_condition(self, condition: dict) -> None:
-        dict_condition = {"Condition": condition}
-        str_condition = json.dumps(dict_condition, ignore_nan=True) + "\n"
-        self._result_file.write(str_condition)
+    def delete_result_file(self) -> None:
+        self._result_path.unlink()
+
+    def write_line(self, write_obj: Any) -> None:
+        str_record = json.dumps(write_obj, ignore_nan=True) + "\n"
+        self._result_file.write(str_record)
+
+    def write_result(self, result: ResultBase) -> None:
+        self.write_line(self.get_result(result))
 
-    def write_header(self) -> None:
+    def get_header(self) -> dict:
         system_time = self._system_clock.now()
-        dict_header = {
+        return {
             "Result": {"Success": False, "Summary": "NoData"},
             "Stamp": {"System": system_time.nanoseconds / pow(10, 9)},
             "Frame": {},
         }
-        str_header = json.dumps(dict_header, ignore_nan=True) + "\n"
-        self._result_file.write(str_header)
 
-    def write(self, result: ResultBase) -> None:
+    def get_result(self, result: ResultBase) -> dict:
         system_time = self._system_clock.now()
         time_dict = {"System": system_time.nanoseconds / pow(10, 9)}
         if self._ros_clock.ros_time_is_active:
@@ -83,16 +100,14 @@ def write(self, result: ResultBase) -> None:
         else:
             time_dict["ROS"] = 0.0
 
-        dict_record = {
-            "Result": {"Success": result.success(), "Summary": result.summary()},
+        return {
+            "Result": {"Success": result.success, "Summary": result.summary},
             "Stamp": time_dict,
-            "Frame": result.frame(),
+            "Frame": result.frame,
         }
-        str_record = json.dumps(dict_record, ignore_nan=True) + "\n"
-        self._result_file.write(str_record)
 
 
 class PickleWriter:
-    def __init__(self, out_pkl_path: str, write_object) -> None:  # noqa
-        with open(os.path.expandvars(out_pkl_path), "wb") as pkl_file:
+    def __init__(self, out_pkl_path: str, write_object: Any) -> None:
+        with Path(expandvars(out_pkl_path)).open("wb") as pkl_file:
             pickle.dump(write_object, pkl_file)

driving_log_replayer/package.xml

@@ -46,8 +46,10 @@
   <exec_depend>rosidl_runtime_py</exec_depend>
 
   <test_depend>ament_cmake_gtest</test_depend>
+  <test_depend>ament_cmake_pytest</test_depend>
   <test_depend>ament_lint_auto</test_depend>
   <test_depend>autoware_lint_common</test_depend>
+  <test_depend>launch_pytest</test_depend>
 
   <export>
     <build_type>ament_cmake</build_type>

driving_log_replayer/scripts/eagleye_evaluator_node.py

@@ -80,7 +80,7 @@ def check_scenario(self) -> None:
 
     def diagnostics_cb(self, msg: DiagnosticArray) -> None:
         self.__result.set_frame(msg)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
 
 @evaluator_main

driving_log_replayer/scripts/localization_evaluator_node.py

@@ -310,7 +310,7 @@ def tp_cb(self, msg: Float32Stamped) -> None:
             DLREvaluator.transform_stamped_with_euler_angle(map_to_baselink),
             self.__latest_nvtl,
         )
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
     def nvtl_cb(self, msg: Float32Stamped) -> None:
         self.__latest_nvtl = msg
@@ -323,7 +323,7 @@ def nvtl_cb(self, msg: Float32Stamped) -> None:
             DLREvaluator.transform_stamped_with_euler_angle(map_to_baselink),
             self.__latest_tp,
         )
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
     def pose_cb(self, msg: PoseStamped) -> None:
         map_to_baselink = self.lookup_transform(msg.header.stamp)
@@ -335,11 +335,11 @@ def pose_cb(self, msg: PoseStamped) -> None:
             self.__latest_iteration_num,
         )
         self.__pub_lateral_distance.publish(msg_lateral_distance)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
     def diagnostics_cb(self, msg: DiagnosticArray) -> None:
         self.__result.set_frame(msg)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
 
 @evaluator_main

driving_log_replayer/scripts/obstacle_segmentation_evaluator_node.py

@@ -603,7 +603,7 @@ def obstacle_segmentation_input_cb(self, msg: ObstacleSegmentationInput) -> None
             pcd_header,
             topic_rate=msg.topic_rate,
         )
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
         topic_rate_data = ObstacleSegmentationMarker()
         topic_rate_data.header = msg.pointcloud.header

driving_log_replayer/scripts/perception_2d_evaluator_node.py

@@ -218,7 +218,7 @@ def write_metrics(self) -> None:
             conf_mat_dict = conf_mat_df.to_dict()
         final_metrics = {"Score": score_dict, "ConfusionMatrix": conf_mat_dict}
         self.__result.set_final_metrics(final_metrics)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
     def list_dynamic_object_2d_from_ros_msg(
         self,
@@ -281,7 +281,7 @@ def detected_objs_cb(self, msg: DetectedObjectsWithFeature, camera_type: str) ->
                 DLREvaluator.transform_stamped_with_euler_angle(map_to_baselink),
                 camera_type,
             )
-            self._result_writer.write(self.__result)
+            self._result_writer.write_result(self.__result)
 
     def get_final_result(self) -> MetricsScore:
         final_metric_score = self.__evaluator.get_scene_result()

driving_log_replayer/scripts/perception_evaluator_node.py

@@ -227,7 +227,7 @@ def write_metrics(self) -> None:
         if self.__evaluation_task == "fp_validation":
             final_metrics = self.get_fp_result()
             self.__result.set_final_metrics(final_metrics)
-            self._result_writer.write(self.__result)
+            self._result_writer.write_result(self.__result)
         else:
             self.get_final_result()
             score_dict = {}
@@ -243,7 +243,7 @@ def write_metrics(self) -> None:
                 )
             final_metrics = {"Score": score_dict, "Error": error_dict}
             self.__result.set_final_metrics(final_metrics)
-            self._result_writer.write(self.__result)
+            self._result_writer.write_result(self.__result)
 
     def list_dynamic_object_from_ros_msg(
         self,
@@ -330,7 +330,7 @@ def perception_cb(self, msg: DetectedObjects | TrackedObjects) -> None:
             msg.header,
             DLREvaluator.transform_stamped_with_euler_angle(map_to_baselink),
         )
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
         self.__pub_marker_ground_truth.publish(marker_ground_truth)
         self.__pub_marker_results.publish(marker_results)
 

driving_log_replayer/scripts/performance_diag_evaluator_node.py

@@ -361,7 +361,7 @@ def diag_cb(self, msg: DiagnosticArray) -> None:
         for k, v in msg_blockage_levels.items():
             if v is not None:
                 self.__pub_blockage_levels[k].publish(v)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
 
 @evaluator_main

driving_log_replayer/scripts/traffic_light_evaluator_node.py

@@ -184,7 +184,7 @@ def write_metrics(self) -> None:
             conf_mat_dict = conf_mat_df.to_dict()
         final_metrics = {"Score": score_dict, "ConfusionMatrix": conf_mat_dict}
         self.__result.set_final_metrics(final_metrics)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
     def list_dynamic_object_2d_from_ros_msg(
         self,
@@ -235,7 +235,7 @@ def traffic_signals_cb(self, msg: TrafficSignalArray) -> None:
                 msg.header,
                 DLREvaluator.transform_stamped_with_euler_angle(map_to_baselink),
             )
-            self._result_writer.write(self.__result)
+            self._result_writer.write_result(self.__result)
 
     def get_final_result(self) -> MetricsScore:
         final_metric_score = self.__evaluator.get_scene_result()

driving_log_replayer/scripts/yabloc_evaluator_node.py

@@ -81,7 +81,7 @@ def check_scenario(self) -> None:
 
     def diagnostics_cb(self, msg: DiagnosticArray) -> None:
         self.__result.set_frame(msg)
-        self._result_writer.write(self.__result)
+        self._result_writer.write_result(self.__result)
 
 
 @evaluator_main

driving_log_replayer/test/unittest/test_result.py

@@ -0,0 +1,70 @@
+# Copyright (c) 2023 TIER IV.inc
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from collections.abc import Callable
+from os.path import expandvars
+
+import pytest
+from rclpy.clock import ClockType
+from rclpy.clock import ROSClock
+from rclpy.time import Time
+
+from driving_log_replayer.result import ResultBase
+from driving_log_replayer.result import ResultWriter
+
+
+class SampleResult(ResultBase):
+    def __init__(self) -> None:
+        super().__init__()
+
+    def update(self) -> None:
+        self._success = True
+        self._summary = "Sample OK"
+
+    def set_frame(self) -> None:
+        self._frame = {"Test": "ok"}
+
+
+@pytest.fixture()
+def create_writer() -> ResultWriter:
+    json_path = "$HOME/dlr_result.json"
+    ros_clock = ROSClock()
+    ros_time = Time(seconds=123, nanoseconds=456, clock_type=ClockType.ROS_TIME)
+    ros_clock.set_ros_time_override(ros_time)
+    ros_clock._set_ros_time_is_active(True)  # noqa
+    condition = {"sample_condition": "sample"}
+    writer = ResultWriter(json_path, ros_clock, condition)
+    yield writer
+    # delete created jsonl file
+    writer.delete_result_file()
+
+
+def test_create_jsonl_path(create_writer: Callable) -> None:
+    writer: ResultWriter = create_writer
+    assert writer.result_path.as_posix() == expandvars("$HOME/dlr_result.jsonl")
+
+
+def test_get_result(create_writer: Callable) -> None:
+    writer: ResultWriter = create_writer
+    sample_result = SampleResult()
+    sample_result.set_frame()
+    sample_result.update()
+    record = writer.get_result(sample_result)
+    writer.write_line(record)
+    assert record["Result"]["Success"] == True  # noqa
+    assert record["Result"]["Summary"] == "Sample OK"
+    assert record["Frame"] == {"Test": "ok"}
+    # The system time stamp is not checked since it changes with each execution.
+    # Since the logic is the same as that of ros time, it is sufficient to check ros time.
+    assert record["Stamp"]["ROS"] == 123.000000456  # noqa

pyproject.toml

@@ -51,7 +51,7 @@ show-source = true
 # https://beta.ruff.rs/docs/rules/
 select = ["A", "B", "C", "D", "E", "F", "G", "I", "N", "Q", "S", "T", "W", "ANN", "ARG", "BLE", "COM", "DJ", "DTZ", "EM", "ERA", "EXE", "FBT", "ICN", "ISC", "NPY", "PGH", "PIE", "PL", "PT", "PYI", "RET", "RSE", "RUF", "SIM", "SLF", "TCH", "TID", "TRY", "UP", "YTT"]
 # "INP", "PD", "PTH"
-ignore = ["Q000", "ANN101", "ANN102", "PGH004", "E501", "PLR0913", "S301", "S602", "D100", "D101", "D102", "D103", "D104", "D105", "D106", "D107", "D203", "D212", "D404", "D417"]
+ignore = ["Q000", "ANN101", "ANN102", "ANN401", "PGH004", "E501", "PLR0913", "S101", "S301", "S602", "SIM115", "D100", "D101", "D102", "D103", "D104", "D105", "D106", "D107", "D203", "D212", "D404", "D417"]
 fixable=["D", "I", "ANN", "COM", "EXE", "PIE"]
 
 [tool.ruff.isort]