scene_service.ingest.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.
This lives inside scene/ deliberately — system/ services should not reverse-depend on a service-layer perception package. The detector calls the same OpenAI-compatible endpoint that pilot already uses (VLM_BASE_URL / VLM_API_KEY / VLM_MODEL), so no new credentials.
The polling loop keeps a perceptual fingerprint of the last successful frame, so a cached camera image or low-level JPEG noise does not spend another model call. Failures retry with bounded exponential backoff.
Classes
|
Runs the RGB-poll → VLM-call → Detection-list pipeline as one asyncio task. |
- class scene_service.ingest.perception_vlm.VLMObjectDetector(*, rgb_fetcher: ~typing.Callable[[], bytes | tuple[bytes, int | float] | None], camera_to_world_fn: ~typing.Callable[[], tuple[object, str] | None], on_detections: ~typing.Callable[[list[~scene_service.state.data_assoc.Detection]], ~typing.Awaitable[None]], period_s: float = 3.0, intrinsics: ~scene_service.ingest.perception_vlm._CamIntrinsics | None = None, intrinsics_fn: ~typing.Callable[[], ~scene_service.ingest.perception_vlm._CamIntrinsics | None] | None = None, frame_change_threshold: float | None = None, cache_max_age_s: float | None = None, failure_backoff_base_s: float | None = None, failure_backoff_max_s: float | None = None, clock: ~typing.Callable[[], float] = <built-in function monotonic>)[source]¶
Bases:
objectRuns 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.
- property inference_counts: dict[str, int]¶