CommonRoad Scenario Factory
Our Scenario Factory provides the entire pipeline from automatic map generation to generating critical scenarios. It bundles and integrates scenario syntheses from formal specifications and falsification techniques of the CommonRoad ecosystem. It introduces simulation modes that make microscopic traffic simulation applicable in different stages of the scenario generation process. The Scenario Factory further supports autonomous driving researchers with utilities such as automatic map creation from OpenStreetMap data, scenario labeling and criticality assessment, testing of motion planners, and visualization.
Use case
from pathlib import Path
from scenario_factory.globetrotter import Coordinates, OsmApiMapProvider, RegionMetadata
from scenario_factory.pipeline_steps import (
SimulateScenarioArguments,
WriteScenarioToFileArguments,
pipeline_add_metadata_to_scenario,
pipeline_assign_tags_to_scenario,
pipeline_simulate_scenario_with_sumo,
pipeline_write_scenario_to_file,
)
from scenario_factory.pipelines import (
create_globetrotter_pipeline,
create_scenario_generation_pipeline,
)
from scenario_factory.scenario_config import ScenarioFactoryConfig
from scenario_factory.simulation import (
SimulationConfig,
SimulationMode,
)
output_path = Path("/tmp/use_case")
output_path.mkdir(parents=True, exist_ok=True)
# Get map (Autobahn, near Garching)
coords = "48.264165183714745, 11.64459089457302" # Garching
simulation_config = SimulationConfig(
mode=SimulationMode.RANDOM_TRAFFIC_GENERATION, simulation_steps=600
)
globetrotter_pipeline = (
create_globetrotter_pipeline(0.2, OsmApiMapProvider())
.map(pipeline_add_metadata_to_scenario)
.map(pipeline_simulate_scenario_with_sumo(SimulateScenarioArguments(config=simulation_config)))
)
globetrotter_inputs = [RegionMetadata.from_coordinates(Coordinates.from_str(coords))]
# Create scenarios
scenario_config = ScenarioFactoryConfig(seed=1234, cr_scenario_time_steps=150)
scenario_generation_pipeline = create_scenario_generation_pipeline(
scenario_config.criterions, scenario_config.filters
).map(pipeline_assign_tags_to_scenario)
pipeline = globetrotter_pipeline.chain(scenario_generation_pipeline).map(
pipeline_write_scenario_to_file(WriteScenarioToFileArguments(output_path))
)
result = pipeline.execute(globetrotter_inputs)
result.print_cum_time_per_step()