feat: #403 annotationless scenario less mode (#405)

Author: hayato-m126

docs/overview/command.en.md

@@ -67,6 +67,15 @@ dlr simulation run -p default -l play_rate:=0.5 -l input_pointcloud:=/sensing/li
 
 # Set perception_mode to camera_lidar_fusion
 dlr simulation run -p default -l perception_mode:=camera_lidar_fusion
+
+# Dry-run mode to obtain metrics without creating a scenario file
+# set bag, map, sensor_model, vehicle_model, [and vehicle_id] from arguments
+dlr simulation dry-run -p ${profile} -u ${use_case} -l sensor_model:=${sensor_model} -l vehicle_model:=${vehicle_model} -l map_path:=${map_path} -l input_bag:=${bag_path} [-l vehicle_id:=${vehicle_id}]
+
+# Command example
+# The -p option is used to determine the output destination for bag and result.jsonl. If omitted, the output is sent to the output_directory of the default profile.
+# Currently, use_case is only annotationless_perception, so if -u is omitted, it automatically becomes annotationless_perception.
+dlr simulation dry-run -l input_bag:=$HOME/dlr_data/auto/annotationless/sample/input_bag -l sensor_model:=sample_sensor_kit -l vehicle_model:=sample_vehicle -l map_path:=$HOME/map/sample_map
 ```
 
 The arguments that can be specified can be displayed by using the -s option of ros2 launch.

docs/overview/command.ja.md

@@ -48,6 +48,15 @@ dlr simulation show-result ${output_directory}
 
 # 結果ファイルをjsonに変換する
 dlr simulation convert-result ${output_directory}
+
+# シナリオファイルを作成せずに、メトリクスを得るために実行する予行演習モード
+# bag, map, sensor_model, vehicle_model [, vehicle_id] を引数で指定する
+dlr simulation dry-run -p ${profile} -u ${use_case} -l sensor_model:=${sensor_model} -l vehicle_model:=${vehicle_model} -l map_path:=${map_path} -l input_bag:=${bag_path} [-l vehicle_id:=${vehicle_id}]
+
+# コマンド例
+# -pのオプションは、bagとresult.jsonlの出力先を決めるために使われる。省略するとdefaultプロファイルのoutput_directoryに出力される。
+# 現時点ではuse_caseはannotationless_perceptionのみなので-uを省略すると自動でanontationless_perceptionになる
+dlr simulation dry-run -l input_bag:=$HOME/dlr_data/auto/annotationless/sample/input_bag -l sensor_model:=sample_sensor_kit -l vehicle_model:=sample_vehicle -l map_path:=$HOME/map/sample_map
 ```
 
 #### dlr simulation run launch argument option

driving_log_replayer_cli/core/scenario.py

@@ -19,14 +19,14 @@ class Scenario(BaseModel):
     SensorModel: str
     VehicleModel: str
     VehicleId: str | None = None
-    LocalMapPath: Path | None = None
+    LocalMapPath: Path | str = ""
     Evaluation: dict
 
     @field_validator("LocalMapPath", mode="before")
     @classmethod
-    def validate_local_path(cls, v: str | None) -> Path | None:
-        if v is None:
-            return None
+    def validate_local_path(cls, v: str | None) -> Path | str:
+        if v == "":
+            return ""
         normal_path = Path(expandvars(v))
         if normal_path.exists():
             return normal_path
@@ -49,6 +49,14 @@ def load_scenario(scenario_path: Path) -> Scenario:
         return Scenario(**yaml.safe_load(scenario_file))
 
 
+def get_dry_run_scenario_path(use_case: str) -> Path:
+    return (
+        Path(__file__)
+        .parent.parent.resolve()
+        .joinpath("resources", "sample", use_case, "dry_run.yaml")
+    )
+
+
 def backup_scenario_file(scenario_path: Path) -> None:
     bak_name = scenario_path.name + f".{datetime.now().strftime('%Y%m%d%H%M%S')}.bak"  # noqa
     backup_file_path = scenario_path.parent.joinpath(bak_name)

driving_log_replayer_cli/resources/__init__.py

@@ -0,0 +1,2 @@
+# Save resource files to be used from cli in this directory.
+# refer https://github.com/psf/black/tree/main/src/black/resources

driving_log_replayer_cli/resources/sample

@@ -0,0 +1 @@
+../../sample

driving_log_replayer_cli/simulation/__init__.py

@@ -6,6 +6,7 @@
 from driving_log_replayer_cli.core.config import load_config
 from driving_log_replayer_cli.core.result import convert_all
 from driving_log_replayer_cli.core.result import display_all
+from driving_log_replayer_cli.simulation.run import dry_run as sim_dry_run
 from driving_log_replayer_cli.simulation.run import run as sim_run
 from driving_log_replayer_cli.simulation.update import update_annotationless_scenario_condition
 
@@ -40,6 +41,28 @@ def run(
     )
 
 
+@simulation.command(context_settings=CONTEXT_SETTINGS)
+@click.option("--profile", "-p", type=str, default="default")
+@click.option(
+    "--use-case",
+    "-u",
+    type=click.Choice(["annotationless_perception"]),
+    default="annotationless_perception",
+)
+@click.option("--launch_args", "-l", multiple=True, default=[])
+def dry_run(
+    profile: str,
+    use_case: str,
+    launch_args: list[str],
+) -> None:
+    config: Config = load_config(profile)
+    sim_dry_run(
+        config,
+        use_case,
+        launch_args,
+    )
+
+
 @simulation.command(context_settings=CONTEXT_SETTINGS)
 @click.argument(
     "output_directory",

driving_log_replayer_cli/simulation/run.py

@@ -10,6 +10,7 @@
 from driving_log_replayer_cli.core.result import convert_all
 from driving_log_replayer_cli.core.result import display_all
 from driving_log_replayer_cli.core.scenario import Datasets
+from driving_log_replayer_cli.core.scenario import get_dry_run_scenario_path
 from driving_log_replayer_cli.core.scenario import load_scenario
 from driving_log_replayer_cli.core.scenario import Scenario
 from driving_log_replayer_cli.simulation.update import update_annotationless_scenario_condition
@@ -48,19 +49,20 @@ def run(
             continue
         scenario = load_scenario(scenario_file)
         launch_cmd = None
+        launch_arg_dict = args_to_dict(launch_args)
         if scenario.Evaluation["UseCaseName"] in USE_T4_DATASET:
             launch_cmd = cmd_use_t4_dataset(
                 scenario_file,
                 output_case,
                 config.autoware_path,
-                launch_args,
+                launch_arg_dict,
             )
         else:
             launch_cmd = cmd_use_bag_only(
                 scenario_file,
                 output_case,
                 config.autoware_path,
-                launch_args,
+                launch_arg_dict,
             )
 
         if launch_cmd is None:
@@ -93,6 +95,54 @@ def run(
     display_all(output_dir_by_time)
 
 
+def dry_run(
+    config: Config,
+    use_case: str,
+    launch_args: list[str],
+) -> None:
+    output_dir_by_time = create_output_dir_by_time(config.output_directory)
+    output_case = output_dir_by_time.joinpath("dry_run")
+    scenario_file = get_dry_run_scenario_path(use_case)
+    scenario = load_scenario(scenario_file)
+    launch_cmd = None
+    launch_arg_dict = args_to_dict(launch_args)
+    if scenario.Evaluation["UseCaseName"] in USE_T4_DATASET:
+        launch_cmd = cmd_use_t4_dataset(
+            scenario_file,
+            output_case,
+            config.autoware_path,
+            launch_arg_dict,
+        )
+    else:
+        launch_cmd = cmd_use_bag_only(
+            scenario_file,
+            output_case,
+            config.autoware_path,
+            launch_arg_dict,
+        )
+
+    if launch_cmd is None:
+        return
+
+    # create dir for output bag and result.jsonl
+    output_case.mkdir(exist_ok=True)
+
+    # save command as bash script
+    run_script = output_case.joinpath("run.bash")
+    with run_script.open("w") as f:
+        f.write(launch_cmd)
+
+    # run simulation
+    cmd = ["/bin/bash", run_script.as_posix()]
+    try:
+        run_with_log(cmd, output_case.joinpath("console.log"))
+    except KeyboardInterrupt:
+        termcolor.cprint("Simulation execution canceled by Ctrl+C", "red")
+    # convert result file and display result
+    convert_all(output_dir_by_time)
+    display_all(output_dir_by_time)
+
+
 def create_output_dir_by_time(base_path: Path) -> Path:
     output_dir_by_time = base_path.joinpath(
         datetime.datetime.now().strftime("%Y-%m%d-%H%M%S")  # noqa
@@ -117,22 +167,29 @@ def get_scenario_file(dataset_path: Path) -> Path | None:
     return scenario_file_path
 
 
-def extract_arg(launch_args: list[str]) -> str:
-    extract_launch_arg = ""
+def args_to_dict(launch_args: list[str]) -> dict[str, str]:
+    launch_arg_dict = {}
     for l_arg in launch_args:
-        if len(l_arg.split(":=")) != 2:  # noqa
+        try:
+            key, value = l_arg.split(":=")
+            launch_arg_dict[key] = value
+        except ValueError:
             # invalid argument
             termcolor.cprint(
                 f"{l_arg} is ignored because it is invalid",
                 "red",
             )
-        else:  # noqa
-            # Enclose in single quotes to avoid being treated as special characters by shell
-            if "{" in l_arg or "[" in l_arg:
-                extract_launch_arg += f" '{l_arg}'"
-            else:
-                extract_launch_arg += f" {l_arg}"
-    return extract_launch_arg
+    return launch_arg_dict
+
+
+def launch_dict_to_str(launch_arg_dict: dict) -> str:
+    rtn_str = ""
+    for k, v in launch_arg_dict.items():
+        if isinstance(v, str) and ("{" in v or "[" in v):
+            rtn_str += f" '{k}:={v}'"
+        else:
+            rtn_str += f" {k}:={v}"
+    return rtn_str
 
 
 def clean_up_cmd() -> str:
@@ -155,30 +212,40 @@ def cmd_use_bag_only(
     scenario_path: Path,
     output_path: Path,
     autoware_path: Path,
-    launch_args: str,
+    launch_args_dict: dict[str, str],
 ) -> str | None:
     scenario: Scenario = load_scenario(scenario_path)
-    launch_command = f"source {autoware_path.joinpath('install', 'setup.bash').as_posix()}\n"
-    launch_command += f"ros2 launch driving_log_replayer {scenario.Evaluation['UseCaseName']}.launch.py map_path:={scenario.LocalMapPath} vehicle_model:={scenario.VehicleModel} sensor_model:={scenario.SensorModel} vehicle_id:={scenario.VehicleId}"
-    launch_command += f" scenario_path:={scenario_path.as_posix()} result_json_path:={output_path.joinpath('result.json').as_posix()} input_bag:={scenario_path.parent.joinpath('input_bag').as_posix()} result_bag_path:={output_path.joinpath('result_bag').as_posix()}"
+    launch_command = f"source {autoware_path.joinpath('install', 'setup.bash')}\n"
+    launch_command += (
+        f"ros2 launch driving_log_replayer {scenario.Evaluation['UseCaseName']}.launch.py"
+    )
+    launch_arg_dict_scenario = {
+        "map_path": scenario.LocalMapPath,
+        "vehicle_model": scenario.VehicleModel,
+        "sensor_model": scenario.SensorModel,
+        "vehicle_id": scenario.VehicleId,
+        "scenario_path": scenario_path,
+        "result_json_path": output_path.joinpath("result.json"),
+        "input_bag": scenario_path.parent.joinpath("input_bag"),
+        "result_bag_path": output_path.joinpath("result_bag"),
+    }
     launch_localization = scenario.Evaluation.get("LaunchLocalization")
     if launch_localization is not None:
-        launch_command += f" localization:={launch_localization}"
-    launch_command += extract_arg(launch_args)
+        launch_arg_dict_scenario["localization"] = launch_localization
+    launch_arg_dict_scenario.update(launch_args_dict)
+    launch_command += launch_dict_to_str(launch_arg_dict_scenario) + "\n"
     return launch_command + clean_up_cmd()
 
 
 def cmd_use_t4_dataset(
     scenario_path: Path,
     output_path: Path,
     autoware_path: Path,
-    launch_args: str,
+    launch_args_dict: dict[str, str],
 ) -> str | None:
     dataset_path = scenario_path.parent
     scenario: Scenario = load_scenario(scenario_path)
-    launch_command_for_all_dataset = (
-        f"source {autoware_path.joinpath('install', 'setup.bash').as_posix()}\n"
-    )
+    launch_command_for_all_dataset = f"source {autoware_path.joinpath('install', 'setup.bash')}\n"
     t4_dataset_base_path = dataset_path.joinpath("t4_dataset")
     try:
         t4_datasets: Datasets = Datasets(Datasets=scenario.Evaluation["Datasets"])
@@ -202,12 +269,25 @@ def cmd_use_t4_dataset(
                 )
                 continue
 
-            launch_command = f"ros2 launch driving_log_replayer {scenario.Evaluation['UseCaseName']}.launch.py map_path:={map_path} vehicle_model:={scenario.VehicleModel} sensor_model:={scenario.SensorModel} vehicle_id:={vehicle_id}"
-            launch_command += f" scenario_path:={scenario_path.as_posix()} result_json_path:={output_dir_per_dataset.joinpath('result.json').as_posix()} input_bag:={t4_dataset_path.joinpath('input_bag').as_posix()} result_bag_path:={output_dir_per_dataset.joinpath('result_bag').as_posix()}"
-            launch_command += f" t4_dataset_path:={t4_dataset_path} result_archive_path:={output_dir_per_dataset.joinpath('result_archive').as_posix()}"
+            launch_command = (
+                f"ros2 launch driving_log_replayer {scenario.Evaluation['UseCaseName']}.launch.py"
+            )
+            launch_arg_dict_dataset = {
+                "map_path": map_path,
+                "vehicle_model": scenario.VehicleModel,
+                "sensor_model": scenario.SensorModel,
+                "vehicle_id": vehicle_id,
+                "scenario_path": scenario_path,
+                "result_json_path": output_dir_per_dataset.joinpath("result.json"),
+                "input_bag": t4_dataset_path.joinpath("input_bag"),
+                "result_bag_path": output_dir_per_dataset.joinpath("result_bag"),
+                "t4_dataset_path": t4_dataset_path,
+                "result_archive_path": output_dir_per_dataset.joinpath("result_archive"),
+            }
             if t4_dataset[key].LaunchSensing is not None:
-                launch_command += f" sensing:={t4_dataset[key].LaunchSensing}"
-            launch_command += extract_arg(launch_args)
+                launch_arg_dict_dataset["sensing"] = t4_dataset[key].LaunchSensing
+            launch_arg_dict_dataset.update(launch_args_dict)
+            launch_command += launch_dict_to_str(launch_arg_dict_dataset) + "\n"
             launch_command += clean_up_cmd()
             launch_command_for_all_dataset += launch_command
         if is_database_evaluation:
@@ -218,6 +298,8 @@ def cmd_use_t4_dataset(
                 "driving_log_replayer",
                 "perception_database_result.py",
             )
-            database_result_command = f"python3 {database_result_script_path.as_posix()} -s {scenario_path} -r {output_path}\n"
+            database_result_command = (
+                f"python3 {database_result_script_path} -s {scenario_path} -r {output_path}\n"
+            )
             launch_command_for_all_dataset += database_result_command
         return launch_command_for_all_dataset

driving_log_replayer_cli/test/simulation/test_run.py

@@ -1,31 +1,32 @@
-from driving_log_replayer_cli.simulation.run import extract_arg
+from driving_log_replayer_cli.simulation.run import args_to_dict
+from driving_log_replayer_cli.simulation.run import launch_dict_to_str
 
 
 def test_extract_single_arg() -> None:
-    args = extract_arg(["rate:=0.5"])
-    assert args == " rate:=0.5"
+    args = args_to_dict(["rate:=0.5"])
+    assert launch_dict_to_str(args) == " rate:=0.5"
 
 
 def test_extract_multiple_args() -> None:
-    args = extract_arg(["rate:=0.5", "delay:=2.0"])
-    assert args == " rate:=0.5 delay:=2.0"
+    args = args_to_dict(["rate:=0.5", "delay:=2.0"])
+    assert launch_dict_to_str(args) == " rate:=0.5 delay:=2.0"
 
 
 def test_extract_no_arg() -> None:
-    args = extract_arg("not_arg")
-    assert args == ""
+    args = args_to_dict("not_arg")
+    assert launch_dict_to_str(args) == ""
 
 
 def test_extract_dict_string() -> None:
     # annotationless perception argument
-    args = extract_arg(
+    args = args_to_dict(
         [
             'annotationless_pass_range:={"CAR":{"min":"0.0-1.1","max":"0.0-1.2","mean":"0.5-1.3"},"BUS":{"min":"0.0-1.1","max":"0.0-1.2","mean":"0.5-1.3"}}',
             "annotationless_threshold_file:=/home/autoware/result.jsonl",
         ],
     )
     assert (
-        args
+        launch_dict_to_str(args)
         == ' \'annotationless_pass_range:={"CAR":{"min":"0.0-1.1","max":"0.0-1.2","mean":"0.5-1.3"},"BUS":{"min":"0.0-1.1","max":"0.0-1.2","mean":"0.5-1.3"}}\' annotationless_threshold_file:=/home/autoware/result.jsonl'
     )
     # confirm json load
@@ -37,5 +38,15 @@ def test_extract_dict_string() -> None:
 
 def test_extract_array_string() -> None:
     # pose_initializer argument
-    args = extract_arg(["initial_pose:=[89571.150,42301.188,-3.159,-0.009,-0.002,0.288,0.958]"])
-    assert args == " 'initial_pose:=[89571.150,42301.188,-3.159,-0.009,-0.002,0.288,0.958]'"
+    args = args_to_dict(["initial_pose:=[89571.150,42301.188,-3.159,-0.009,-0.002,0.288,0.958]"])
+    assert (
+        launch_dict_to_str(args)
+        == " 'initial_pose:=[89571.150,42301.188,-3.159,-0.009,-0.002,0.288,0.958]'"
+    )
+
+
+def test_override_arg() -> None:
+    scenario_arg = {"vehicle_model": "sample_vehicle"}
+    args = args_to_dict(["vehicle_model:=lexus"])
+    scenario_arg.update(args)
+    assert launch_dict_to_str(scenario_arg) == " vehicle_model:=lexus"

sample/annotationless_perception/dry_run.yaml

@@ -0,0 +1,11 @@
+ScenarioFormatVersion: 2.2.0
+ScenarioName: dry_run_annotationless_perception
+ScenarioDescription: dry_run_annotationless_perception
+SensorModel: dry_run_sensor
+VehicleModel: dry_run_vehicle
+VehicleId: default
+Evaluation:
+  UseCaseName: annotationless_perception
+  UseCaseFormatVersion: 0.2.0
+  Conditions:
+    ClassConditions: {}