Skip to content
Open
8 changes: 8 additions & 0 deletions opencda/core/attack/advcp/adv_coperception_model_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
AgentId,
AttackerId,
)
from opencda.core.attack.advcp.utils.runtime_assets import AdvCPRuntimeAssetHelper
from opencda.core.common.coperception_model_manager import (
CoperceptionInferenceResult,
CoperceptionModelManager,
Expand Down Expand Up @@ -407,6 +408,10 @@ def load_config(config_path: str | None, assets_dir: str | None = None) -> AdvCP
step_value = cast(int | str, config["step"])
config.setdefault("lr", 1.0 if int(step_value) <= 2 else 0.05)
config.setdefault("feature_size", 10)
# Runtime asset generation writes to user-set spoof paths only; bundle defaults are never overwritten.
spoof_paths_explicit = ("car_mesh_path" in config or "model_path" in config) and (
"car_mesh_divide_path" in config or "mesh_divide_path" in config
)
config.setdefault("car_mesh_path", config.get("model_path", str(local_model_root / "car_mesh_0200.ply")))
config.setdefault(
"car_mesh_divide_path",
Expand Down Expand Up @@ -448,6 +453,9 @@ def load_config(config_path: str | None, assets_dir: str | None = None) -> AdvCP
path = Path(str(path_value)).expanduser()
if not path.is_absolute():
config[optional_path_key] = str((config_dir / path).resolve())

# Opt-in on-demand generation (``asset_runtime_generation: true``); points the asset paths at the result.
AdvCPRuntimeAssetHelper.prepare_assets(config, config_dir, spoof_paths_explicit)
return cast(AdvCPConfig, config)

def validate_advcp_agents(self, valid_agent_ids: list[AgentId]) -> bool:
Expand Down
25 changes: 21 additions & 4 deletions opencda/core/attack/advcp/attack_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -1224,10 +1224,22 @@ def build_real_car_mesh_pieces(cls, spoof_box: BoxLwhBottomCenter, advcp_config:
car_mesh_pieces = [car_mesh.select_by_index(vertex_indices) for vertex_indices in cls._CAR_MESH_DIVIDE_CACHE[car_mesh_divide_path]]
else:
car_mesh_pieces = [car_mesh]
return cls.post_process_car_meshes(car_mesh_pieces, spoof_box, car_mesh_name)
car_mesh_bbox = cls.CAR_MESH_3D_EXAMPLES.get(car_mesh_name)
if car_mesh_bbox is None:
# Unknown (e.g. generated) mesh: measure its own extents instead
# of borrowing the reference box of a different car model.
extent = np.asarray(car_mesh.get_max_bound(), dtype=np.float64) - np.asarray(car_mesh.get_min_bound(), dtype=np.float64)
car_mesh_bbox = np.array([0.0, 0.0, 0.0, *extent, 0.0], dtype=np.float32)
return cls.post_process_car_meshes(car_mesh_pieces, spoof_box, car_mesh_name, car_mesh_bbox)

@classmethod
def post_process_car_meshes(cls, car_mesh_pieces: list[Any], spoof_box: BoxLwhBottomCenter, car_mesh_name: str) -> list[Any]:
def post_process_car_meshes(
cls,
car_mesh_pieces: list[Any],
spoof_box: BoxLwhBottomCenter,
car_mesh_name: str,
car_mesh_bbox: npt.NDArray | None = None,
) -> list[Any]:
"""
Fit each mesh piece into the configured target box.

Expand All @@ -1242,15 +1254,20 @@ def post_process_car_meshes(cls, car_mesh_pieces: list[Any], spoof_box: BoxLwhBo
Target box.
car_mesh_name : str
Stem of the mesh file; used to look up the reference
bounding box in ``CAR_MESH_3D_EXAMPLES``.
bounding box in ``CAR_MESH_3D_EXAMPLES`` when
``car_mesh_bbox`` is not given.
car_mesh_bbox : Optional[npt.NDArray]
Reference bounding box ``[x, y, z, l, w, h, yaw]`` of the
source mesh. Takes precedence over the name lookup.

Returns
-------
list of Open3D meshes
Transformed mesh pieces ready for ray tracing.
"""
processed_meshes = []
car_mesh_bbox = cls.CAR_MESH_3D_EXAMPLES.get(car_mesh_name, cls.CAR_MESH_3D_EXAMPLES["car_mesh_0200"])
if car_mesh_bbox is None:
car_mesh_bbox = cls.CAR_MESH_3D_EXAMPLES.get(car_mesh_name, cls.CAR_MESH_3D_EXAMPLES["car_mesh_0200"])
# Use the smallest per-axis scale ratio to preserve aspect.
scale = float(np.min(spoof_box[3:6] / car_mesh_bbox[3:6]))
rotation = np.array(
Expand Down
35 changes: 13 additions & 22 deletions opencda/core/attack/advcp/early_fusion_attack.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,11 @@
AttackerId,
BoxLwhBottomCenter,
)
from opencda.core.attack.advcp.utils.asset_utils import (
ADVSHAPE_TEMPLATE_DIMENSIONS_M,
advshape_default_divide,
advshape_template_mesh,
)

logger = logging.getLogger("cavise.opencda.opencda.core.attack.advcp.early_fusion_attack")

Expand Down Expand Up @@ -1013,14 +1018,13 @@ def _resolve_optional_advshape_path(path_value: Any) -> Path | None:
def _create_adv_shape_template_mesh(cls, advcp_config: AdvCPConfig) -> Any:
import open3d as o3d

# Matches AdvCP canonical adversarial shape template dimensions.
template_box = np.array([0.0, 0.0, 0.0, 4.9, 2.5, 2.0, 0.0], dtype=np.float32)
mesh = AdvCPCarMeshHelper.build_box_piece_mesh(
template_box,
(float(template_box[3]), float(template_box[4]), float(template_box[5])),
(0.0, 0.0, float(template_box[5]) / 2.0),
# Shared with the asset generators so that positional perturbation
# and divide assets address the same vertices.
template = advshape_template_mesh()
mesh = o3d.geometry.TriangleMesh(
o3d.utility.Vector3dVector(template.vertices),
o3d.utility.Vector3iVector(template.faces),
)
mesh = mesh.subdivide_midpoint(2)

perturb_path = cls._resolve_optional_advshape_path(advcp_config.get("remove_adv_shape_perturb_path"))
if perturb_path is not None and perturb_path.exists():
Expand Down Expand Up @@ -1049,25 +1053,12 @@ def _resolve_remove_adv_shape_divide(

@staticmethod
def _default_adv_shape_divide(template_mesh: Any) -> list[npt.NDArray]:
vertices = np.asarray(template_mesh.vertices, dtype=np.float64)
bbox = np.array([4.9, 2.5, 2.0], dtype=np.float64)
return [
np.argwhere(vertices[:, 0] > bbox[0] / 2.0 - 0.01).reshape(-1),
np.argwhere(vertices[:, 0] < -bbox[0] / 2.0 + 0.01).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] >= 0.0, vertices[:, 1] > bbox[1] / 2.0 - 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] <= 0.0, vertices[:, 1] > bbox[1] / 2.0 - 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] >= 0.0, vertices[:, 1] < -bbox[1] / 2.0 + 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] <= 0.0, vertices[:, 1] < -bbox[1] / 2.0 + 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] >= 0.0, vertices[:, 2] > bbox[2] - 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] <= 0.0, vertices[:, 2] > bbox[2] - 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] >= 0.0, vertices[:, 2] < 0.01)).reshape(-1),
np.argwhere(np.logical_and(vertices[:, 0] <= 0.0, vertices[:, 2] < 0.01)).reshape(-1),
]
return advshape_default_divide(np.asarray(template_mesh.vertices, dtype=np.float64))

@staticmethod
def _apply_adv_shape_transform(mesh_pieces: list[Any], removal_box: npt.NDArray) -> list[Any]:
transformed_meshes = []
reference_size = np.array([4.9, 2.5, 2.0], dtype=np.float64)
reference_size = np.asarray(ADVSHAPE_TEMPLATE_DIMENSIONS_M, dtype=np.float64)
scale = float(np.max((np.asarray(removal_box[3:6], dtype=np.float64) + 0.6) / reference_size))
rotation = np.array(
[
Expand Down
18 changes: 18 additions & 0 deletions opencda/core/attack/advcp/test/test_advcp.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import pytest
from unittest.mock import MagicMock, patch
from importlib import import_module
from pathlib import Path
import numpy as np

# The production code imports are now safe because pytest_configure in conftest.py
Expand Down Expand Up @@ -975,6 +976,23 @@ def test_load_config_uses_resolved_advcp_asset_bundle(self, tmp_path):
assert loaded_config["car_mesh_path"] == str(tmp_path / "car_mesh_0200.ply")
assert loaded_config["car_mesh_divide_path"] == str(tmp_path / "spoof" / "car_mesh_divide.pkl")

def test_load_config_generates_runtime_assets_when_enabled(self, tmp_path):
config_path = tmp_path / "advcp.yaml"
config_path.write_text(
"asset_runtime_generation: true\nvehicle_blueprint: vehicle.audi.a2\nadvshape: true\nasset_cache_dir: cache\n",
encoding="utf-8",
)
bundle_dir = tmp_path / "bundle"

loaded_config = AdvCoperceptionModelManager.load_config(str(config_path), assets_dir=str(bundle_dir))

cache_dir = (tmp_path / "cache").resolve()
for key in ("car_mesh_path", "car_mesh_divide_path", "remove_adv_shape_divide_path"):
path = Path(loaded_config[key])
assert path.exists()
assert cache_dir in path.parents
assert not bundle_dir.exists()


class TestAdvCoperceptionVisualizer:
def test_advcp_visualizer_includes_removed_boxes_from_context(self):
Expand Down
Loading