scene_service.ingest

Ingest layer — async tasks that pull observations into the registry.

v2 (current): scene runs in its own docker container with rclpy on the host DDS bus. ROS2 subscribers are the primary ingest path; VLM perception consumes the latest RGB frame each tick and produces Detections that go through state/data_assoc.

v1 (removed): MCP-based polling of primitive caps via atlas. MCP is for pilot only; scene’s own data fetching uses the fast direct path.

class scene_service.ingest.ConceptGraphsDetector(*, rgb_fetcher_msg: Callable[[], Any | None], depth_fetcher_msg: Callable[[], Any | None], camera_info_fetcher: Callable[[], _CamIntrinsics | None], chassis_pose_fn: Callable[[], tuple[float, float, float, float] | None], on_detections: Callable[[list[Detection]], Awaitable[None]], registry: ObjectRegistry, world_frame_fn: Callable[[], str] | None = None, period_s: float = 0.6, confidence_threshold: float = 0.3, camera_height_m: float = 1.1, max_detections: int = 30, yolo_weights_path: str | None = None, sam_weights_path: str | None = None, clip_model_name: str | None = None, clip_pretrained: str | None = None, cfg_overrides: dict | None = None, hub: Any = None, camera_frame: str = 'camera_optical_frame', base_frame: str | None = None)

Bases: object

Per-frame detector that runs the canonical concept-graphs merge pipeline. Owns a persistent MapObjectList and projects it into scene’s ObjectRegistry once per tick so the web UI/MCP API stay consistent.

on_detections is kept for backwards compat but no longer used — we write to the registry directly via _project_to_registry. The caller still passes a registry via this side channel.

embed_text(texts: list[str]) → list[list[float]] | None

Encode texts with the already-loaded open_clip text encoder, L2-normalized to match the per-object image features. Returns one 512-d vector per input, or None when CLIP isn’t loaded (models unavailable / detector not started) so the caller can fall back.

Reuses the model loaded in start() — no second model instance. Runs synchronously under torch.no_grad on self._device.

export_3d_snapshot(max_points_per_obj: int = 256) → dict

latest_frame_bundle()

Return (rgb_bgr, K, T_cam_map) for projecting map-frame points into the current camera image, or None when any piece is unavailable.

Consumed by the scene-graph builder’s image-grounded relation pass. Reads the latest RGB frame, intrinsics, and camera→map transform WITHOUT holding _inference_lock — same rationale as export_3d_snapshot: the lock is held by the worker tick for ~100 ms of YOLO/SAM and the asyncio-loop caller must not block on it. A one-tick mismatch between frame and transform is immaterial (poses move slowly relative to the 30 s rebuild cadence).

async start() → None

async stop() → None

class scene_service.ingest.SubscribersHub(specs: list[TopicSpec], *, node_name: str = 'scene_subscribers')

Bases: object

Owns the rclpy.Node + spin thread + per-kind latest slots. Lifecycle:

hub = SubscribersHub(specs=[…]) await hub.start() # spawns spin thread, creates subs … await hub.stop() # destroys node, joins thread

add_spec(spec: TopicSpec) → bool

Add a new (kind, topic) subscription to a hub already up. Used by the background reconciler to absorb topics that come online after start(). Returns True if added, False if the kind was already known — or if the subscription could not be created (missing msg class): committing the slot anyway would make has_kinds() claim a subscription that doesn’t exist and pair a success log with the skip warning.

has(kind: str) → bool

has_kinds() → set[str]

latest(kind: str) → tuple[Any, float, int]

Returns (msg, stamp_unix, count). msg is None until the first callback arrives; readers should branch on stamp_unix > 0.

lookup_transform_4x4(target_frame: str, source_frame: str = 'map')

Return a 4x4 numpy homogeneous transform that maps points from target_frame into source_frame. None when tf isn’t available. Used by perception to project camera-optical depth points directly into map frame without composing odom- frame chassis pose by hand.

lookup_xy_yaw(target_frame: str = 'base_link', source_frame: str = 'map') → tuple[float, float, float, float] | None

Return (x, y, z, yaw) of target_frame expressed in source_frame via tf2, or None when the transform isn’t available yet. This is the SLAM-corrected pose: it goes through /tf (rtabmap publishes map→odom; chassis publishes odom→base_link), so it tracks rviz exactly. Reading /odom instead leaves the consumer in odom-frame and drifts away from the map once SLAM corrects.

async start() → None

async stop() → None

class scene_service.ingest.TopicSpec(kind: str, topic: str, msg_type: str, qos_profile: str = 'default')

Bases: object

One observation kind’s wiring. kind is the abstract name (rgb / depth / lidar2d / pose / odom); topic is the concrete ROS topic; msg_type selects which sensor_msgs / nav_msgs class to import. Optional qos_profile carries the Atlas declaration; when it is absent, sensor streams use best-effort/volatile and map lifecycle data uses reliable/transient-local semantics.

kind: str

msg_type: str

qos_profile: str = 'default'

topic: str

class scene_service.ingest.VLMObjectDetector(*, rgb_fetcher: Callable[[], bytes | None], chassis_pose_fn: Callable[[], tuple[float, float, float, float] | None], on_detections: Callable[[list[Detection]], Awaitable[None]], period_s: float = 3.0, camera_frame_id: str = 'camera_optical_frame', intrinsics: _CamIntrinsics | None = None)

Bases: object

Runs the RGB-poll → VLM-call → Detection-list pipeline as one asyncio task. Calls back into on_detections with a batch of Detection objects at each successful tick.

The detector reads camera/snapshot via the existing PrimitivePoller machinery (passed in as rgb_fetcher) so we don’t duplicate the atlas connect logic.

async start() → None

async stop() → None

Modules

capabilities	Hardware-capability probe → perception tier selection.
perception_concept_graphs	ConceptGraphs perception and persistent object-map integration.
perception_vlm	VLM-based object detection: take an RGB frame, ask an OpenAI-compatible vision model to enumerate visible objects with approximate image coordinates, parse the JSON response, and (when depth is available) reproject to world coordinates.
ros_subscribers	ROS2 subscribers — scene's primary ingest path.