mirror of
https://github.com/SikongJueluo/Mini-Nav.git
synced 2026-07-12 20:15:31 +08:00
feat(scenegraph): add depth-based 3D positioning via pinhole projection
- Add bbox_depth_center position strategy in SceneGraphBuilder using depth at bbox centre and configurable camera_hfov_degrees for pinhole projection. - Add optional depth_sensor_uuid to HabitatSimulatorConfig; create depth sensor spec alongside RGB sensor. - Add camera_position/camera_rotation fields to RoomView; capture pose from sensor_states when depth sensor is available. - Update flatten_room_views for backward compatibility with legacy tuple format. - Wired in depth sensor and bbox_depth_center strategy in verification notebook. - Add tests for depth sensor support and new position strategies.
This commit is contained in:
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from math import radians, tan
|
||||
from typing import Any, Literal, Sequence
|
||||
|
||||
import numpy as np
|
||||
@@ -18,31 +19,43 @@ class SceneGraphBuildConfig:
|
||||
"""Configuration for the scene graph builder.
|
||||
|
||||
Attributes:
|
||||
position_strategy: How object positions are assigned. Only 'room_center'
|
||||
is supported in M0.
|
||||
position_strategy: How object positions are assigned.
|
||||
'room_center' uses the room's center point.
|
||||
'bbox_depth_center' uses depth at the bbox center to compute
|
||||
a 3D position via pinhole projection.
|
||||
inference_batch_size: Batch size passed to the pipeline.
|
||||
enable_fusion: Whether to fuse overlapping detections (not yet supported).
|
||||
fusion_hash_similarity_threshold: Hash similarity threshold for fusion.
|
||||
fusion_distance_threshold_m: Distance threshold in meters for fusion.
|
||||
camera_hfov_degrees: Horizontal field of view of the camera in degrees.
|
||||
Only used when position_strategy='bbox_depth_center'.
|
||||
"""
|
||||
|
||||
position_strategy: Literal["room_center"] = "room_center"
|
||||
position_strategy: Literal["room_center", "bbox_depth_center"] = "room_center"
|
||||
inference_batch_size: int = 4
|
||||
enable_fusion: bool = False
|
||||
fusion_hash_similarity_threshold: float = 0.95
|
||||
fusion_distance_threshold_m: float = 0.5
|
||||
camera_hfov_degrees: float = 90.0
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.position_strategy != "room_center":
|
||||
valid_strategies = {"room_center", "bbox_depth_center"}
|
||||
if self.position_strategy not in valid_strategies:
|
||||
raise ValueError(
|
||||
f"position_strategy must be 'room_center', got {self.position_strategy!r}"
|
||||
f"position_strategy must be one of {valid_strategies}, "
|
||||
f"got {self.position_strategy!r}"
|
||||
)
|
||||
if not 0 < self.camera_hfov_degrees < 180:
|
||||
raise ValueError(
|
||||
f"camera_hfov_degrees must be in (0, 180), "
|
||||
f"got {self.camera_hfov_degrees}"
|
||||
)
|
||||
if self.inference_batch_size <= 0:
|
||||
raise ValueError(
|
||||
f"inference_batch_size must be positive, got {self.inference_batch_size}"
|
||||
)
|
||||
if self.enable_fusion:
|
||||
raise ValueError("fusion is not yet supported in M0")
|
||||
raise ValueError("fusion is not yet supported in M0/M1")
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -247,10 +260,15 @@ class SceneGraphBuilder:
|
||||
meta=meta,
|
||||
selected_idx=selected_idx,
|
||||
)
|
||||
position, position_confidence = self._position_for_detection(
|
||||
view=view,
|
||||
room=room,
|
||||
bbox_xyxy=bbox_xyxy,
|
||||
)
|
||||
return ObjectNode(
|
||||
obj_id=obj_id,
|
||||
room_id=view.room_id,
|
||||
position=room.center.copy(),
|
||||
position=position,
|
||||
visual_hash=hash_bytes,
|
||||
semantic_hash=hash_bytes,
|
||||
hit_count=1,
|
||||
@@ -259,7 +277,7 @@ class SceneGraphBuilder:
|
||||
confidence=confidence,
|
||||
bbox_xyxy=bbox_xyxy,
|
||||
source_view_id=self._source_view_id(view),
|
||||
position_confidence=0.0,
|
||||
position_confidence=position_confidence,
|
||||
)
|
||||
|
||||
def _metadata_for_detection(
|
||||
@@ -287,7 +305,234 @@ class SceneGraphBuilder:
|
||||
)
|
||||
return values[selected_idx]
|
||||
|
||||
def _normalize_bbox(self, box: Any) -> tuple[float, float, float, float]:
|
||||
if len(box) != 4:
|
||||
raise ValueError(f"bbox entry has length {len(box)}, expected 4")
|
||||
return tuple(float(x) for x in box)
|
||||
def _normalize_bbox(self, box: Any) -> tuple[float, float, float, float] | None:
|
||||
try:
|
||||
if len(box) != 4:
|
||||
return None
|
||||
result = tuple(float(x) for x in box)
|
||||
if not all(np.isfinite(result)):
|
||||
return None
|
||||
return result
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
def _position_for_detection(
|
||||
self,
|
||||
*,
|
||||
view: Any,
|
||||
room: RoomNode,
|
||||
bbox_xyxy: tuple[float, float, float, float] | None,
|
||||
) -> tuple[np.ndarray, float]:
|
||||
"""Compute object position and confidence based on the configured strategy.
|
||||
|
||||
Returns:
|
||||
A tuple of (position: np.ndarray (3,), confidence: float).
|
||||
"""
|
||||
if self._config.position_strategy == "room_center":
|
||||
return room.center.copy(), 0.0
|
||||
|
||||
# bbox_depth_center strategy
|
||||
if bbox_xyxy is not None:
|
||||
position = self._compute_bbox_depth_center_position(view, bbox_xyxy)
|
||||
if position is not None:
|
||||
return position, 1.0
|
||||
|
||||
# Fallback: room center with 0.0 confidence
|
||||
return room.center.copy(), 0.0
|
||||
|
||||
def _compute_bbox_depth_center_position(
|
||||
self,
|
||||
view: Any,
|
||||
bbox_xyxy: tuple[float, float, float, float],
|
||||
) -> np.ndarray | None:
|
||||
"""Compute 3D world position from depth at the bbox centre.
|
||||
|
||||
Uses the configured camera_hfov_degrees for pinhole projection.
|
||||
Prefers camera pose (camera_position/camera_rotation) when available
|
||||
on the view; falls back to room center + 0.0 confidence when camera
|
||||
pose is absent (agent pose alone is insufficient because the Habitat
|
||||
depth sensor is offset from the agent body by sensor_height).
|
||||
|
||||
Returns:
|
||||
A (3,) float32 position array, or None if computation fails.
|
||||
"""
|
||||
depth = getattr(view, "depth", None)
|
||||
|
||||
if depth is None:
|
||||
return None
|
||||
|
||||
# Prefer camera pose when available. Agent pose alone is insufficient
|
||||
# because the Habitat depth sensor is offset [0.0, sensor_height, 0.0]
|
||||
# from the agent body, and we have no way to recover the offset.
|
||||
camera_position = getattr(view, "camera_position", None)
|
||||
camera_rotation = getattr(view, "camera_rotation", None)
|
||||
|
||||
if camera_position is not None and camera_rotation is not None:
|
||||
origin_position = camera_position
|
||||
origin_rotation = camera_rotation
|
||||
else:
|
||||
# No camera pose – cannot confidently place in world space.
|
||||
return None
|
||||
|
||||
# Coerce depth to ndarray; require 2D
|
||||
try:
|
||||
depth = np.asarray(depth)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
if depth.ndim != 2:
|
||||
return None
|
||||
|
||||
height, width = depth.shape
|
||||
|
||||
# Validate bbox values are finite before rounding
|
||||
try:
|
||||
x1, y1, x2, y2 = map(float, bbox_xyxy)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
if not all(np.isfinite(v) for v in (x1, y1, x2, y2)):
|
||||
return None
|
||||
|
||||
u = int(round((x1 + x2) / 2.0))
|
||||
v = int(round((y1 + y2) / 2.0))
|
||||
|
||||
# Out of bounds check
|
||||
if not (0 <= u < width and 0 <= v < height):
|
||||
return None
|
||||
|
||||
depth_value = depth[v, u]
|
||||
|
||||
# Depth must be finite and positive
|
||||
if not np.isfinite(depth_value) or depth_value <= 0:
|
||||
return None
|
||||
|
||||
# Convert rotation to matrix
|
||||
rotation_matrix = self._rotation_to_matrix(origin_rotation)
|
||||
if rotation_matrix is None:
|
||||
return None
|
||||
|
||||
# Coerce origin_position safely
|
||||
try:
|
||||
origin_pos = np.asarray(origin_position, dtype=np.float32).reshape(3)
|
||||
except (ValueError, TypeError, RuntimeError):
|
||||
return None
|
||||
if not np.all(np.isfinite(origin_pos)):
|
||||
return None
|
||||
|
||||
# Pinhole projection
|
||||
camera_point = self._pixel_depth_to_camera_point(
|
||||
u=u, v=v, depth=depth_value, width=width, height=height,
|
||||
)
|
||||
|
||||
# World position
|
||||
world_position = origin_pos + rotation_matrix @ camera_point
|
||||
|
||||
if not np.all(np.isfinite(world_position)):
|
||||
return None
|
||||
|
||||
return np.asarray(world_position, dtype=np.float32)
|
||||
|
||||
def _pixel_depth_to_camera_point(
|
||||
self,
|
||||
u: int,
|
||||
v: int,
|
||||
depth: float,
|
||||
width: int,
|
||||
height: int,
|
||||
) -> np.ndarray:
|
||||
"""Convert a pixel with depth to a camera-space 3D point via pinhole model.
|
||||
|
||||
M1 assumes Habitat camera forward is local -Z; this convention should be
|
||||
validated with a real Habitat sanity check.
|
||||
|
||||
Args:
|
||||
u: Pixel column (x) coordinate.
|
||||
v: Pixel row (y) coordinate.
|
||||
depth: Depth value at the pixel (positive, finite).
|
||||
width: Image width in pixels.
|
||||
height: Image height in pixels.
|
||||
|
||||
Returns:
|
||||
A (3,) float32 array [x, y, z] in camera space.
|
||||
"""
|
||||
hfov = radians(self._config.camera_hfov_degrees)
|
||||
fx = width / (2.0 * tan(hfov / 2.0))
|
||||
fy = fx
|
||||
cx = (width - 1) / 2.0
|
||||
cy = (height - 1) / 2.0
|
||||
|
||||
x = (u - cx) / fx * depth
|
||||
y = -(v - cy) / fy * depth
|
||||
z = -depth
|
||||
|
||||
return np.array([x, y, z], dtype=np.float32)
|
||||
|
||||
def _rotation_to_matrix(
|
||||
self,
|
||||
rotation: Any,
|
||||
) -> np.ndarray | None:
|
||||
"""Convert a rotation object to a (3, 3) float32 rotation matrix.
|
||||
|
||||
Supports:
|
||||
- Objects with a ``.rotation_matrix`` attribute (e.g. habitat_sim style).
|
||||
- ndarray or array-like objects directly convertible to (3, 3).
|
||||
- ``numpy-quaternion`` quaternion objects (optional dependency; uses
|
||||
``importlib`` to load ``quaternion.as_rotation_matrix``).
|
||||
- Objects with a ``.transform_vector()`` method (quaternion-like);
|
||||
constructs the matrix columns by transforming local basis vectors.
|
||||
|
||||
Returns:
|
||||
A (3, 3) float32 matrix, or None if conversion fails.
|
||||
"""
|
||||
try:
|
||||
# Path a: object with .rotation_matrix attribute
|
||||
if hasattr(rotation, "rotation_matrix"):
|
||||
mat = self._validate_matrix(rotation.rotation_matrix)
|
||||
if mat is not None:
|
||||
return mat
|
||||
return None
|
||||
|
||||
# Path b: direct ndarray or array-like (3, 3)
|
||||
mat = self._validate_matrix(rotation)
|
||||
if mat is not None:
|
||||
return mat
|
||||
|
||||
# Path c: numpy-quaternion quaternion object (optional dependency)
|
||||
try:
|
||||
import importlib as _il
|
||||
qmod = _il.import_module("quaternion")
|
||||
if hasattr(qmod, "as_rotation_matrix"):
|
||||
mat = self._validate_matrix(qmod.as_rotation_matrix(rotation))
|
||||
if mat is not None:
|
||||
return mat
|
||||
except (ImportError, AttributeError, TypeError, ValueError):
|
||||
pass
|
||||
|
||||
# Path d: quaternion-like with .transform_vector method
|
||||
if hasattr(rotation, "transform_vector"):
|
||||
eye = np.eye(3, dtype=np.float32)
|
||||
cols: list[np.ndarray] = []
|
||||
for i in range(3):
|
||||
col_vec = rotation.transform_vector(eye[:, i])
|
||||
col = np.asarray(col_vec, dtype=np.float32).ravel()
|
||||
if col.shape != (3,) or not np.all(np.isfinite(col)):
|
||||
return None
|
||||
cols.append(col)
|
||||
return np.column_stack(cols)
|
||||
|
||||
return None
|
||||
except (ValueError, TypeError, RuntimeError):
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _validate_matrix(mat: Any) -> np.ndarray | None:
|
||||
"""Validate and convert a candidate to a (3, 3) float32 matrix.
|
||||
|
||||
Returns the matrix if it is (3, 3) with all finite values, else None.
|
||||
"""
|
||||
try:
|
||||
mat = np.asarray(mat, dtype=np.float32)
|
||||
if mat.shape == (3, 3) and np.all(np.isfinite(mat)):
|
||||
return mat
|
||||
except (ValueError, TypeError, RuntimeError):
|
||||
pass
|
||||
return None
|
||||
|
||||
@@ -15,6 +15,26 @@ class HabitatSimulatorConfig:
|
||||
enable_physics: bool = False
|
||||
sensor_uuid: str = "color_sensor"
|
||||
agent_id: int = 0
|
||||
depth_sensor_uuid: str | None = "depth_sensor"
|
||||
hfov_degrees: float = 90.0
|
||||
|
||||
|
||||
def _camera_sensor_spec(
|
||||
habitat_sim_module: Any,
|
||||
*,
|
||||
uuid: str,
|
||||
sensor_type: Any,
|
||||
image_size: int,
|
||||
sensor_height: float,
|
||||
hfov_degrees: float,
|
||||
) -> Any:
|
||||
spec = habitat_sim_module.CameraSensorSpec()
|
||||
spec.uuid = uuid
|
||||
spec.sensor_type = sensor_type
|
||||
spec.resolution = [image_size, image_size]
|
||||
spec.position = [0.0, sensor_height, 0.0]
|
||||
spec.hfov = float(hfov_degrees)
|
||||
return spec
|
||||
|
||||
|
||||
def create_habitat_simulator(
|
||||
@@ -30,6 +50,11 @@ def create_habitat_simulator(
|
||||
if config.move_forward_step <= 0:
|
||||
raise ValueError("move_forward_step must be greater than 0")
|
||||
|
||||
if not (0 < config.hfov_degrees < 180):
|
||||
raise ValueError(
|
||||
f"hfov_degrees must be in (0, 180), got {config.hfov_degrees}"
|
||||
)
|
||||
|
||||
if habitat_sim_module is None:
|
||||
habitat_sim_module = import_module("habitat_sim")
|
||||
|
||||
@@ -38,12 +63,31 @@ def create_habitat_simulator(
|
||||
sim_cfg.enable_physics = config.enable_physics
|
||||
|
||||
agent_cfg = habitat_sim_module.agent.AgentConfiguration()
|
||||
rgb_sensor_spec = habitat_sim_module.CameraSensorSpec()
|
||||
rgb_sensor_spec.uuid = config.sensor_uuid
|
||||
rgb_sensor_spec.sensor_type = habitat_sim_module.SensorType.COLOR
|
||||
rgb_sensor_spec.resolution = [config.image_size, config.image_size]
|
||||
rgb_sensor_spec.position = [0.0, config.sensor_height, 0.0]
|
||||
agent_cfg.sensor_specifications = [rgb_sensor_spec]
|
||||
|
||||
sensor_specs = [
|
||||
_camera_sensor_spec(
|
||||
habitat_sim_module,
|
||||
uuid=config.sensor_uuid,
|
||||
sensor_type=habitat_sim_module.SensorType.COLOR,
|
||||
image_size=config.image_size,
|
||||
sensor_height=config.sensor_height,
|
||||
hfov_degrees=config.hfov_degrees,
|
||||
),
|
||||
]
|
||||
|
||||
if config.depth_sensor_uuid is not None:
|
||||
sensor_specs.append(
|
||||
_camera_sensor_spec(
|
||||
habitat_sim_module,
|
||||
uuid=config.depth_sensor_uuid,
|
||||
sensor_type=habitat_sim_module.SensorType.DEPTH,
|
||||
image_size=config.image_size,
|
||||
sensor_height=config.sensor_height,
|
||||
hfov_degrees=config.hfov_degrees,
|
||||
),
|
||||
)
|
||||
|
||||
agent_cfg.sensor_specifications = sensor_specs
|
||||
|
||||
turn_angle = 360.0 / config.views_per_room
|
||||
agent_cfg.action_space = {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from dataclasses import dataclass, field
|
||||
from importlib import import_module
|
||||
from pathlib import Path
|
||||
@@ -20,16 +21,32 @@ class RoomView:
|
||||
depth: Any | None = field(default=None, compare=False)
|
||||
agent_position: np.ndarray | None = field(default=None, compare=False)
|
||||
agent_rotation: Any | None = field(default=None, compare=False)
|
||||
camera_position: Any | None = field(default=None, compare=False)
|
||||
camera_rotation: Any | None = field(default=None, compare=False)
|
||||
|
||||
|
||||
def flatten_room_views(room_views_by_room: RoomViewsByRoom) -> list[RoomView]:
|
||||
result: list[RoomView] = []
|
||||
for room_id, views in room_views_by_room.items():
|
||||
for idx, rgb in enumerate(views):
|
||||
result.append(RoomView(room_id=room_id, view_idx=idx, rgb=rgb))
|
||||
for idx, item in enumerate(views):
|
||||
if isinstance(item, RoomView):
|
||||
result.append(item)
|
||||
else:
|
||||
result.append(RoomView(room_id=room_id, view_idx=idx, rgb=item))
|
||||
return result
|
||||
|
||||
|
||||
def _snapshot_rotation(rotation: Any) -> Any:
|
||||
"""Create an independent copy of *rotation* to prevent mutation aliasing.
|
||||
|
||||
If the value exposes a ``.copy()`` method (e.g. numpy arrays) it is
|
||||
preferred; otherwise a stdlib shallow copy is used as a fallback.
|
||||
"""
|
||||
if hasattr(rotation, "copy") and callable(rotation.copy):
|
||||
return rotation.copy()
|
||||
return copy.copy(rotation)
|
||||
|
||||
|
||||
def collect_room_views_by_room(
|
||||
agent: Any,
|
||||
sim: Any,
|
||||
@@ -38,6 +55,7 @@ def collect_room_views_by_room(
|
||||
*,
|
||||
habitat_sim_module: Any | None = None,
|
||||
sensor_uuid: str = "color_sensor",
|
||||
depth_sensor_uuid: str | None = None,
|
||||
turn_action: str = "turn_left",
|
||||
progress_description: str = "Collecting room views",
|
||||
progress_track: ProgressTrack = track,
|
||||
@@ -55,9 +73,32 @@ def collect_room_views_by_room(
|
||||
agent.set_state(agent_state)
|
||||
|
||||
room_views = []
|
||||
for _ in range(views_per_room):
|
||||
for view_idx in range(views_per_room):
|
||||
observations = sim.get_sensor_observations()
|
||||
room_views.append(observations[sensor_uuid])
|
||||
current_state = agent.get_state()
|
||||
agent_position = np.asarray(current_state.position, dtype=np.float32).copy()
|
||||
|
||||
# Capture camera pose from sensor_states when depth sensor is available
|
||||
camera_position = None
|
||||
camera_rotation = None
|
||||
if depth_sensor_uuid is not None:
|
||||
sensor_states = getattr(current_state, "sensor_states", None)
|
||||
if sensor_states is not None and depth_sensor_uuid in sensor_states:
|
||||
sensor_state = sensor_states[depth_sensor_uuid]
|
||||
camera_position = np.asarray(sensor_state.position, dtype=np.float32).copy()
|
||||
camera_rotation = _snapshot_rotation(sensor_state.rotation)
|
||||
|
||||
room_view = RoomView(
|
||||
room_id=room_node.room_id,
|
||||
view_idx=view_idx,
|
||||
rgb=observations[sensor_uuid],
|
||||
depth=observations[depth_sensor_uuid] if depth_sensor_uuid is not None else None,
|
||||
agent_position=agent_position,
|
||||
agent_rotation=_snapshot_rotation(current_state.rotation),
|
||||
camera_position=camera_position,
|
||||
camera_rotation=camera_rotation,
|
||||
)
|
||||
room_views.append(room_view)
|
||||
sim.step(turn_action)
|
||||
|
||||
all_room_views[room_node.room_id] = room_views
|
||||
|
||||
@@ -24,10 +24,9 @@ def base_dependencies():
|
||||
"""Basic dependencies for data processing."""
|
||||
import numpy as np
|
||||
import polars as pl
|
||||
import torch
|
||||
from PIL import Image
|
||||
|
||||
return Image, np, pl, torch
|
||||
return Image, np, pl
|
||||
|
||||
|
||||
@app.cell
|
||||
@@ -99,16 +98,15 @@ def habitat_setup(HabitatSimulatorConfig, RoomNode, create_habitat_simulator, np
|
||||
views_per_room = 12
|
||||
meters_per_pixel = 0.05
|
||||
|
||||
sim, agent = create_habitat_simulator(
|
||||
HabitatSimulatorConfig(
|
||||
scene_path=_scene_path,
|
||||
views_per_room=views_per_room,
|
||||
image_size=_image_size,
|
||||
sensor_height=1.5,
|
||||
move_forward_step=0.25,
|
||||
enable_physics=False,
|
||||
)
|
||||
habitat_config = HabitatSimulatorConfig(
|
||||
scene_path=_scene_path,
|
||||
views_per_room=views_per_room,
|
||||
image_size=_image_size,
|
||||
sensor_height=1.5,
|
||||
move_forward_step=0.25,
|
||||
enable_physics=False,
|
||||
)
|
||||
sim, agent = create_habitat_simulator(habitat_config)
|
||||
|
||||
room_nodes = []
|
||||
for idx in range(_num_rooms):
|
||||
@@ -125,7 +123,7 @@ def habitat_setup(HabitatSimulatorConfig, RoomNode, create_habitat_simulator, np
|
||||
for _node in room_nodes:
|
||||
print(f" {_node.room_id}: {_node.center}")
|
||||
|
||||
return agent, meters_per_pixel, room_nodes, sim, views_per_room
|
||||
return agent, habitat_config, meters_per_pixel, room_nodes, sim, views_per_room
|
||||
|
||||
|
||||
@app.cell
|
||||
@@ -164,6 +162,7 @@ def collect_views(
|
||||
agent,
|
||||
collect_room_views_by_room,
|
||||
flatten_room_views,
|
||||
habitat_config,
|
||||
numpy_to_pil,
|
||||
room_nodes,
|
||||
sim,
|
||||
@@ -174,6 +173,7 @@ def collect_views(
|
||||
sim=sim,
|
||||
room_nodes=room_nodes,
|
||||
views_per_room=views_per_room,
|
||||
depth_sensor_uuid=habitat_config.depth_sensor_uuid,
|
||||
)
|
||||
|
||||
room_views = flatten_room_views(all_room_views)
|
||||
@@ -193,6 +193,7 @@ def build_scene_graph(
|
||||
SceneGraphBuildConfig,
|
||||
SceneGraphBuilder,
|
||||
cfg_manager,
|
||||
habitat_config,
|
||||
mo,
|
||||
numpy_to_pil,
|
||||
pipeline,
|
||||
@@ -207,7 +208,11 @@ def build_scene_graph(
|
||||
|
||||
builder = SceneGraphBuilder(
|
||||
pipeline=pipeline,
|
||||
config=SceneGraphBuildConfig(inference_batch_size=4),
|
||||
config=SceneGraphBuildConfig(
|
||||
inference_batch_size=4,
|
||||
position_strategy="bbox_depth_center",
|
||||
camera_hfov_degrees=habitat_config.hfov_degrees,
|
||||
),
|
||||
)
|
||||
|
||||
pil_room_views = [
|
||||
@@ -218,6 +223,8 @@ def build_scene_graph(
|
||||
depth=_view.depth,
|
||||
agent_position=_view.agent_position,
|
||||
agent_rotation=_view.agent_rotation,
|
||||
camera_position=_view.camera_position,
|
||||
camera_rotation=_view.camera_rotation,
|
||||
)
|
||||
for _view in room_views
|
||||
]
|
||||
@@ -303,6 +310,8 @@ def build_tables(pl, scene_graph):
|
||||
"position_x": float(obj.position[0]) if obj.position is not None else None,
|
||||
"position_y": float(obj.position[1]) if obj.position is not None else None,
|
||||
"position_z": float(obj.position[2]) if obj.position is not None else None,
|
||||
"bbox_xyxy": str(obj.bbox_xyxy) if obj.bbox_xyxy is not None else None,
|
||||
"position_confidence": obj.position_confidence,
|
||||
}
|
||||
for obj in scene_graph.objects.values()
|
||||
]
|
||||
@@ -313,6 +322,19 @@ def build_tables(pl, scene_graph):
|
||||
return objects_df, rooms_df
|
||||
|
||||
|
||||
@app.cell
|
||||
def display_tables(mo, objects_df, rooms_df):
|
||||
mo.vstack(
|
||||
[
|
||||
mo.md("## Rooms"),
|
||||
mo.ui.table(rooms_df),
|
||||
mo.md("## Objects"),
|
||||
mo.ui.table(objects_df),
|
||||
]
|
||||
)
|
||||
return
|
||||
|
||||
|
||||
@app.cell
|
||||
def upload_query(mo):
|
||||
file_upload = mo.ui.file(
|
||||
@@ -329,7 +351,6 @@ def query_matching(
|
||||
Image,
|
||||
file_upload,
|
||||
mo,
|
||||
object_images,
|
||||
pipeline,
|
||||
query_image_against_scene_graph,
|
||||
scene_graph,
|
||||
|
||||
183
tests/test_habitat_depth_sensor.py
Normal file
183
tests/test_habitat_depth_sensor.py
Normal file
@@ -0,0 +1,183 @@
|
||||
"""Tests for optional Habitat depth sensor support in create_habitat_simulator."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
MINI_NAV_DIR = Path(__file__).resolve().parents[1] / "mini-nav"
|
||||
sys.path.insert(0, str(MINI_NAV_DIR))
|
||||
|
||||
from simulator.habitat import ( # noqa: E402
|
||||
HabitatSimulatorConfig,
|
||||
create_habitat_simulator,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake habitat_sim module
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_specs_created: list = []
|
||||
|
||||
|
||||
class _FakeCameraSensorSpec:
|
||||
"""Fake CameraSensorSpec that records itself in _specs_created."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.uuid = None
|
||||
self.sensor_type = None
|
||||
self.resolution = None
|
||||
self.position = None
|
||||
self.hfov = None
|
||||
_specs_created.append(self)
|
||||
|
||||
|
||||
class _FakeSensorType:
|
||||
COLOR = "COLOR"
|
||||
DEPTH = "DEPTH"
|
||||
|
||||
|
||||
class _FakeSimulator:
|
||||
def __init__(self, cfg: object) -> None:
|
||||
self.cfg = cfg
|
||||
|
||||
@staticmethod
|
||||
def initialize_agent(agent_id: int) -> SimpleNamespace:
|
||||
return SimpleNamespace(agent_id=agent_id)
|
||||
|
||||
@staticmethod
|
||||
def close() -> None:
|
||||
return None
|
||||
|
||||
|
||||
def _make_fake_habitat_sim_module() -> SimpleNamespace:
|
||||
"""Return a fake ``habitat_sim`` module for testing."""
|
||||
global _specs_created
|
||||
_specs_created = []
|
||||
|
||||
hab_sim = SimpleNamespace(
|
||||
SimulatorConfiguration=SimpleNamespace,
|
||||
CameraSensorSpec=_FakeCameraSensorSpec,
|
||||
SensorType=_FakeSensorType,
|
||||
Configuration=lambda sim_cfg, agent_cfgs: SimpleNamespace(
|
||||
sim_cfg=sim_cfg,
|
||||
agent_cfgs=agent_cfgs,
|
||||
),
|
||||
Simulator=_FakeSimulator,
|
||||
)
|
||||
|
||||
class _FakeActionSpec:
|
||||
def __init__(self, name: str, actuation: object) -> None:
|
||||
self.name = name
|
||||
self.actuation = actuation
|
||||
|
||||
class _FakeActuationSpec:
|
||||
def __init__(self, amount: float | None = None) -> None:
|
||||
self.amount = amount
|
||||
|
||||
hab_sim.agent = SimpleNamespace(
|
||||
AgentConfiguration=SimpleNamespace,
|
||||
ActionSpec=_FakeActionSpec,
|
||||
ActuationSpec=_FakeActuationSpec,
|
||||
)
|
||||
|
||||
return hab_sim
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_habitat_sim() -> SimpleNamespace:
|
||||
"""Fixture providing a fresh fake habitat_sim module per test."""
|
||||
return _make_fake_habitat_sim_module()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_default_config_creates_color_and_depth_sensor_specs(
|
||||
fake_habitat_sim: SimpleNamespace,
|
||||
) -> None:
|
||||
"""Default config should produce two specs with expected UUIDs and types."""
|
||||
config = HabitatSimulatorConfig(scene_path="scene.glb")
|
||||
create_habitat_simulator(config, fake_habitat_sim)
|
||||
|
||||
assert len(_specs_created) == 2
|
||||
assert [s.uuid for s in _specs_created] == ["color_sensor", "depth_sensor"]
|
||||
assert [s.sensor_type for s in _specs_created] == ["COLOR", "DEPTH"]
|
||||
|
||||
|
||||
def test_create_habitat_simulator_adds_matching_rgb_and_depth_specs(
|
||||
fake_habitat_sim: SimpleNamespace,
|
||||
) -> None:
|
||||
"""Depth sensor spec should mirror the RGB spec when depth_sensor_uuid is set."""
|
||||
config = HabitatSimulatorConfig(
|
||||
scene_path="scene.glb",
|
||||
image_size=320,
|
||||
sensor_height=1.25,
|
||||
sensor_uuid="rgb",
|
||||
depth_sensor_uuid="depth",
|
||||
hfov_degrees=75.0,
|
||||
)
|
||||
|
||||
create_habitat_simulator(config, fake_habitat_sim)
|
||||
|
||||
assert len(_specs_created) == 2, "Expected 2 sensor specs (RGB + depth)"
|
||||
|
||||
rgb_spec, depth_spec = _specs_created
|
||||
|
||||
# UUIDs
|
||||
assert rgb_spec.uuid == "rgb"
|
||||
assert depth_spec.uuid == "depth"
|
||||
|
||||
# Sensor types
|
||||
assert rgb_spec.sensor_type == "COLOR"
|
||||
assert depth_spec.sensor_type == "DEPTH"
|
||||
|
||||
# Resolutions
|
||||
assert rgb_spec.resolution == [320, 320]
|
||||
assert depth_spec.resolution == [320, 320]
|
||||
|
||||
# Positions
|
||||
assert rgb_spec.position == [0.0, 1.25, 0.0]
|
||||
assert depth_spec.position == [0.0, 1.25, 0.0]
|
||||
|
||||
# HFOV
|
||||
assert rgb_spec.hfov == 75.0
|
||||
assert depth_spec.hfov == 75.0
|
||||
|
||||
|
||||
def test_create_habitat_simulator_can_disable_depth_sensor(
|
||||
fake_habitat_sim: SimpleNamespace,
|
||||
) -> None:
|
||||
"""``depth_sensor_uuid=None`` should produce only a single color sensor spec."""
|
||||
config = HabitatSimulatorConfig(
|
||||
scene_path="scene.glb",
|
||||
depth_sensor_uuid=None,
|
||||
)
|
||||
|
||||
create_habitat_simulator(config, fake_habitat_sim)
|
||||
|
||||
assert len(_specs_created) == 1
|
||||
assert _specs_created[0].uuid == "color_sensor"
|
||||
|
||||
|
||||
def test_habitat_config_rejects_invalid_hfov(
|
||||
fake_habitat_sim: SimpleNamespace,
|
||||
) -> None:
|
||||
"""``hfov_degrees=0`` should raise ``ValueError`` containing ``hfov_degrees``."""
|
||||
config = HabitatSimulatorConfig(
|
||||
scene_path="scene.glb",
|
||||
hfov_degrees=0.0,
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="hfov_degrees"):
|
||||
create_habitat_simulator(config, fake_habitat_sim)
|
||||
@@ -1,11 +1,23 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
NOTEBOOK = Path(__file__).resolve().parents[1] / "notebooks" / "verification.py"
|
||||
|
||||
|
||||
def _find_calls(tree: ast.AST, func_name: str) -> list[ast.Call]:
|
||||
"""Find all Call nodes for a named function (simple name, not attribute)."""
|
||||
return [
|
||||
node
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Name)
|
||||
and node.func.id == func_name
|
||||
]
|
||||
|
||||
|
||||
def test_verification_notebook_uses_scenegraph_query_api():
|
||||
source = NOTEBOOK.read_text()
|
||||
|
||||
@@ -20,7 +32,6 @@ def test_verification_notebook_uses_hash_codec_for_object_hashes():
|
||||
# the builder handles it internally.
|
||||
assert "bits_tensor_to_hash_bytes" not in source
|
||||
assert "np.packbits" not in source
|
||||
assert "scene_graph.objects[_obj_id] = ObjectNode" not in source
|
||||
|
||||
|
||||
def test_verification_notebook_uses_scene_graph_builder():
|
||||
@@ -30,4 +41,56 @@ def test_verification_notebook_uses_scene_graph_builder():
|
||||
assert "SceneGraphBuildConfig" in source
|
||||
assert "flatten_room_views" in source
|
||||
assert "scene_graph, build_artifacts = builder.build_from_room_views" in source
|
||||
assert "scene_graph.objects[_obj_id] = ObjectNode" not in source
|
||||
|
||||
# Verify no direct ObjectNode construction (builder handles it internally)
|
||||
tree = ast.parse(source)
|
||||
object_node_calls = _find_calls(tree, "ObjectNode")
|
||||
assert len(object_node_calls) == 0, (
|
||||
"Notebook should not construct ObjectNode() directly; "
|
||||
"the builder handles it"
|
||||
)
|
||||
|
||||
|
||||
def test_verification_notebook_uses_bbox_depth_center_positioning():
|
||||
source = NOTEBOOK.read_text()
|
||||
tree = ast.parse(source)
|
||||
|
||||
# --- SceneGraphBuildConfig must use position_strategy='bbox_depth_center' ---
|
||||
sg_config_calls = _find_calls(tree, "SceneGraphBuildConfig")
|
||||
assert len(sg_config_calls) >= 1, (
|
||||
"Notebook must instantiate SceneGraphBuildConfig"
|
||||
)
|
||||
config_kwargs = {
|
||||
kw.arg: kw.value for kw in sg_config_calls[0].keywords if kw.arg is not None
|
||||
}
|
||||
|
||||
assert "position_strategy" in config_kwargs
|
||||
pos_val = config_kwargs["position_strategy"]
|
||||
assert isinstance(pos_val, ast.Constant) and pos_val.value == "bbox_depth_center", (
|
||||
f"Expected position_strategy='bbox_depth_center', got {pos_val!r}"
|
||||
)
|
||||
|
||||
assert "camera_hfov_degrees" in config_kwargs, (
|
||||
"SceneGraphBuildConfig should include camera_hfov_degrees"
|
||||
)
|
||||
|
||||
# --- object_rows dict literals must include bbox_xyxy and position_confidence ---
|
||||
found_bbox = False
|
||||
found_pos_conf = False
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Dict):
|
||||
keys = {k.value for k in node.keys if isinstance(k, ast.Constant)}
|
||||
if "obj_id" in keys:
|
||||
if "bbox_xyxy" in keys:
|
||||
found_bbox = True
|
||||
if "position_confidence" in keys:
|
||||
found_pos_conf = True
|
||||
|
||||
assert found_bbox, "object_rows must include bbox_xyxy column"
|
||||
assert found_pos_conf, "object_rows must include position_confidence column"
|
||||
|
||||
# --- No direct ObjectNode( construction ---
|
||||
object_node_calls = _find_calls(tree, "ObjectNode")
|
||||
assert len(object_node_calls) == 0, (
|
||||
"Notebook should not construct ObjectNode() directly"
|
||||
)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user