test(icp): add comprehensive tests for full-scene ICP pipeline + update docs
This commit is contained in:
@@ -437,215 +437,213 @@ def test_refine_with_icp_synthetic_offset():
|
||||
|
||||
finally:
|
||||
aruco.ground_plane.unproject_depth_to_points = orig_unproject
|
||||
if "orig_global_reg" in locals():
|
||||
aruco.icp_registration.global_registration = locals()["orig_global_reg"]
|
||||
|
||||
|
||||
def test_refine_with_icp_no_overlap():
|
||||
def test_refine_with_icp_floor_mode_regression(monkeypatch):
|
||||
import aruco.icp_registration
|
||||
import aruco.ground_plane
|
||||
|
||||
mock_extract = []
|
||||
|
||||
def mock_extract_scene_points(
|
||||
points, floor_y, floor_normal, mode="floor", band_height=0.3
|
||||
):
|
||||
mock_extract.append(mode)
|
||||
return points
|
||||
|
||||
mock_overlap_xz = []
|
||||
|
||||
def mock_compute_overlap_xz(pa, pb, margin=0.0):
|
||||
mock_overlap_xz.append((len(pa), len(pb)))
|
||||
return 10.0
|
||||
|
||||
mock_overlap_3d = []
|
||||
|
||||
def mock_compute_overlap_3d(pa, pb, margin=0.0):
|
||||
mock_overlap_3d.append((len(pa), len(pb)))
|
||||
return 10.0
|
||||
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "extract_scene_points", mock_extract_scene_points
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "compute_overlap_xz", mock_compute_overlap_xz
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "compute_overlap_3d", mock_compute_overlap_3d
|
||||
)
|
||||
|
||||
# Mock unproject to return enough points
|
||||
def mock_unproject(depth, K, stride=1, **kwargs):
|
||||
if depth[0, 0] == 1.0:
|
||||
return np.random.rand(200, 3) + [0, -1, 0]
|
||||
else:
|
||||
return np.random.rand(200, 3) + [10, -1, 0]
|
||||
return np.random.rand(200, 3)
|
||||
|
||||
orig_unproject = aruco.ground_plane.unproject_depth_to_points
|
||||
aruco.ground_plane.unproject_depth_to_points = mock_unproject
|
||||
monkeypatch.setattr(aruco.ground_plane, "unproject_depth_to_points", mock_unproject)
|
||||
|
||||
try:
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.zeros((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=1.0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=1.0),
|
||||
}
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
}
|
||||
|
||||
config = ICPConfig(min_overlap_area=1.0)
|
||||
new_extrinsics, metrics = refine_with_icp(
|
||||
camera_data, extrinsics, floor_planes, config
|
||||
config = ICPConfig(region="floor", overlap_mode="xz")
|
||||
|
||||
refine_with_icp(camera_data, extrinsics, floor_planes, config)
|
||||
|
||||
assert "floor" in mock_extract
|
||||
assert len(mock_overlap_xz) > 0
|
||||
assert len(mock_overlap_3d) == 0
|
||||
|
||||
|
||||
def test_refine_with_icp_hybrid_mode_integration(monkeypatch):
|
||||
import aruco.icp_registration
|
||||
import aruco.ground_plane
|
||||
|
||||
mock_extract = []
|
||||
|
||||
def mock_extract_scene_points(
|
||||
points, floor_y, floor_normal, mode="floor", band_height=0.3
|
||||
):
|
||||
mock_extract.append(mode)
|
||||
return points
|
||||
|
||||
mock_overlap_3d = []
|
||||
|
||||
def mock_compute_overlap_3d(pa, pb, margin=0.0):
|
||||
mock_overlap_3d.append((len(pa), len(pb)))
|
||||
return 10.0
|
||||
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "extract_scene_points", mock_extract_scene_points
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "compute_overlap_3d", mock_compute_overlap_3d
|
||||
)
|
||||
|
||||
# Mock unproject to return enough points
|
||||
def mock_unproject(depth, K, stride=1, **kwargs):
|
||||
return np.random.rand(200, 3)
|
||||
|
||||
monkeypatch.setattr(aruco.ground_plane, "unproject_depth_to_points", mock_unproject)
|
||||
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
}
|
||||
|
||||
config = ICPConfig(region="hybrid", overlap_mode="3d")
|
||||
|
||||
refine_with_icp(camera_data, extrinsics, floor_planes, config)
|
||||
|
||||
assert "hybrid" in mock_extract
|
||||
assert len(mock_overlap_3d) > 0
|
||||
|
||||
|
||||
def test_refine_with_icp_sor_preprocessing(monkeypatch):
|
||||
import aruco.icp_registration
|
||||
import aruco.ground_plane
|
||||
|
||||
mock_preprocess = []
|
||||
|
||||
def mock_preprocess_point_cloud(pcd, voxel_size):
|
||||
mock_preprocess.append(voxel_size)
|
||||
return pcd
|
||||
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "preprocess_point_cloud", mock_preprocess_point_cloud
|
||||
)
|
||||
|
||||
# Mock unproject to return enough points
|
||||
def mock_unproject(depth, K, stride=1, **kwargs):
|
||||
return np.random.rand(200, 3)
|
||||
|
||||
monkeypatch.setattr(aruco.ground_plane, "unproject_depth_to_points", mock_unproject)
|
||||
|
||||
# Mock extract_scene_points to return enough points
|
||||
def mock_extract_scene_points(points, *args, **kwargs):
|
||||
return points
|
||||
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "extract_scene_points", mock_extract_scene_points
|
||||
)
|
||||
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
}
|
||||
|
||||
config = ICPConfig(region="floor")
|
||||
|
||||
refine_with_icp(camera_data, extrinsics, floor_planes, config)
|
||||
|
||||
assert len(mock_preprocess) > 0
|
||||
assert mock_preprocess[0] == config.voxel_size
|
||||
|
||||
|
||||
def test_per_pair_logging_all_pairs(monkeypatch):
|
||||
"""Verify all pairs are logged regardless of convergence."""
|
||||
import aruco.icp_registration
|
||||
import aruco.ground_plane
|
||||
|
||||
# Mock unproject & extract
|
||||
monkeypatch.setattr(
|
||||
aruco.ground_plane,
|
||||
"unproject_depth_to_points",
|
||||
lambda *args, **kwargs: np.random.rand(200, 3),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration,
|
||||
"extract_scene_points",
|
||||
lambda points, *args, **kwargs: points,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
aruco.icp_registration, "compute_overlap_xz", lambda *args, **kwargs: 10.0
|
||||
)
|
||||
|
||||
# Mock pairwise_icp to FAIL
|
||||
def mock_pairwise_icp_fail(*args, **kwargs):
|
||||
return ICPResult(
|
||||
transformation=np.eye(4),
|
||||
fitness=0.0,
|
||||
inlier_rmse=1.0,
|
||||
information_matrix=np.eye(6),
|
||||
converged=False,
|
||||
)
|
||||
|
||||
assert metrics.num_cameras_optimized == 1
|
||||
assert metrics.success
|
||||
monkeypatch.setattr(aruco.icp_registration, "pairwise_icp", mock_pairwise_icp_fail)
|
||||
|
||||
finally:
|
||||
aruco.ground_plane.unproject_depth_to_points = orig_unproject
|
||||
|
||||
|
||||
def test_refine_with_icp_single_camera():
|
||||
camera_data = {"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)}}
|
||||
extrinsics = {"cam1": np.eye(4)}
|
||||
floor_planes = {"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=1.0)}
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.zeros((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.zeros((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=0),
|
||||
}
|
||||
|
||||
config = ICPConfig()
|
||||
new_extrinsics, metrics = refine_with_icp(
|
||||
camera_data, extrinsics, floor_planes, config
|
||||
)
|
||||
new_ext, metrics = refine_with_icp(camera_data, extrinsics, floor_planes, config)
|
||||
|
||||
assert metrics.num_cameras_optimized == 1
|
||||
assert metrics.success
|
||||
|
||||
|
||||
def test_compute_fpfh_features():
|
||||
pcd = create_box_pcd()
|
||||
voxel_size = 0.05
|
||||
pcd_down = pcd.voxel_down_sample(voxel_size)
|
||||
|
||||
fpfh = compute_fpfh_features(pcd_down, voxel_size)
|
||||
|
||||
assert fpfh.dimension() == 33
|
||||
assert fpfh.num() == len(pcd_down.points)
|
||||
|
||||
|
||||
def test_global_registration_known_transform():
|
||||
source = create_box_pcd(size=1.0, num_points=1000)
|
||||
|
||||
# Create target with significant transform (rotation + translation)
|
||||
T_true = np.eye(4)
|
||||
# 30 degree rotation around Y
|
||||
T_true[:3, :3] = Rotation.from_euler("y", 30, degrees=True).as_matrix()
|
||||
T_true[:3, 3] = [0.5, 0.0, 0.2]
|
||||
|
||||
target = o3d.geometry.PointCloud()
|
||||
target.points = o3d.utility.Vector3dVector(
|
||||
(np.asarray(source.points) @ T_true[:3, :3].T) + T_true[:3, 3]
|
||||
)
|
||||
|
||||
voxel_size = 0.05
|
||||
source_down = source.voxel_down_sample(voxel_size)
|
||||
target_down = target.voxel_down_sample(voxel_size)
|
||||
|
||||
source_fpfh = compute_fpfh_features(source_down, voxel_size)
|
||||
target_fpfh = compute_fpfh_features(target_down, voxel_size)
|
||||
|
||||
result = global_registration(
|
||||
source_down, target_down, source_fpfh, target_fpfh, voxel_size
|
||||
)
|
||||
|
||||
assert result.fitness > 0.1
|
||||
# RANSAC is stochastic, but with clean data it should be reasonably close
|
||||
# We check if it found a transform close to truth
|
||||
T_est = result.transformation
|
||||
|
||||
# Check rotation difference
|
||||
R_diff = T_est[:3, :3] @ T_true[:3, :3].T
|
||||
rot_diff = Rotation.from_matrix(R_diff).as_euler("xyz", degrees=True)
|
||||
assert np.linalg.norm(rot_diff) < 5.0 # Within 5 degrees
|
||||
|
||||
# Check translation difference
|
||||
trans_diff = np.linalg.norm(T_est[:3, 3] - T_true[:3, 3])
|
||||
assert trans_diff < 0.1 # Within 10cm
|
||||
|
||||
|
||||
def test_refine_with_icp_global_init_success():
|
||||
import aruco.icp_registration
|
||||
import aruco.ground_plane
|
||||
|
||||
# Create two overlapping point clouds with large offset
|
||||
box_points = create_box_pcd(size=1.0, num_points=2000).points
|
||||
box_points = np.asarray(box_points)
|
||||
|
||||
# Mock unproject to return these points
|
||||
def mock_unproject(depth, K, stride=1, **kwargs):
|
||||
if depth[0, 0] == 1.0: # cam1
|
||||
return box_points
|
||||
else: # cam2
|
||||
# Apply large transform to simulate misaligned camera
|
||||
# Rotate 90 deg and translate
|
||||
R = Rotation.from_euler("y", 90, degrees=True).as_matrix()
|
||||
return (box_points @ R.T) + [2.0, 0, 0]
|
||||
|
||||
orig_unproject = aruco.ground_plane.unproject_depth_to_points
|
||||
aruco.ground_plane.unproject_depth_to_points = mock_unproject
|
||||
|
||||
try:
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.zeros((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
|
||||
# Initial extrinsics are identity (very wrong for cam2)
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
}
|
||||
|
||||
# Enable global init
|
||||
config = ICPConfig(
|
||||
global_init=True,
|
||||
min_overlap_area=0.0, # Disable overlap check for this test since initial overlap is zero
|
||||
min_fitness=0.1,
|
||||
voxel_size=0.05,
|
||||
max_rotation_deg=180.0, # Allow large rotation for this test
|
||||
max_translation_m=5.0, # Allow large translation
|
||||
)
|
||||
|
||||
new_extrinsics, metrics = refine_with_icp(
|
||||
camera_data, extrinsics, floor_planes, config
|
||||
)
|
||||
|
||||
assert metrics.success
|
||||
assert metrics.num_cameras_optimized == 2
|
||||
|
||||
# Check if cam2 moved significantly from identity
|
||||
T_cam2 = new_extrinsics["cam2"]
|
||||
assert np.linalg.norm(T_cam2[:3, 3]) > 1.0
|
||||
|
||||
finally:
|
||||
aruco.ground_plane.unproject_depth_to_points = orig_unproject
|
||||
|
||||
|
||||
def test_refine_with_icp_global_init_disabled():
|
||||
"""Verify that global registration is skipped when disabled."""
|
||||
import aruco.icp_registration
|
||||
import aruco.ground_plane
|
||||
|
||||
# Mock global_registration to raise error if called
|
||||
orig_global_reg = aruco.icp_registration.global_registration
|
||||
|
||||
def mock_global_reg(*args, **kwargs):
|
||||
raise RuntimeError("Global registration should not be called")
|
||||
|
||||
aruco.icp_registration.global_registration = mock_global_reg
|
||||
|
||||
# Mock unproject
|
||||
box_points = create_box_pcd(size=0.5).points
|
||||
box_points = np.asarray(box_points)
|
||||
|
||||
def mock_unproject(depth, K, stride=1, **kwargs):
|
||||
if depth[0, 0] == 1.0:
|
||||
return box_points
|
||||
else:
|
||||
return box_points # Perfect overlap
|
||||
|
||||
orig_unproject = aruco.ground_plane.unproject_depth_to_points
|
||||
aruco.ground_plane.unproject_depth_to_points = mock_unproject
|
||||
|
||||
try:
|
||||
camera_data = {
|
||||
"cam1": {"depth": np.ones((10, 10)), "K": np.eye(3)},
|
||||
"cam2": {"depth": np.zeros((10, 10)), "K": np.eye(3)},
|
||||
}
|
||||
extrinsics = {"cam1": np.eye(4), "cam2": np.eye(4)}
|
||||
floor_planes = {
|
||||
"cam1": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
"cam2": FloorPlane(normal=np.array([0, 1, 0]), d=0.0),
|
||||
}
|
||||
|
||||
# Default config (global_init=False)
|
||||
config = ICPConfig(min_overlap_area=0.01)
|
||||
|
||||
# Should not raise RuntimeError
|
||||
refine_with_icp(camera_data, extrinsics, floor_planes, config)
|
||||
|
||||
finally:
|
||||
aruco.ground_plane.unproject_depth_to_points = orig_unproject
|
||||
aruco.icp_registration.global_registration = orig_global_reg
|
||||
# Even though it failed, it should be in per_pair_results
|
||||
assert ("cam1", "cam2") in metrics.per_pair_results
|
||||
assert metrics.per_pair_results[("cam1", "cam2")].converged is False
|
||||
assert metrics.num_pairs_converged == 0
|
||||
|
||||
|
||||
def test_refine_with_icp_global_init_fallback_low_fitness():
|
||||
|
||||
Reference in New Issue
Block a user