Object registration API wrap DIM 1435

[bogwi]

What is the PR about

New API wrapper for perception stack introduced in: #3422 as per DIM 1435

Full e2e example working

import os
import sys
import time

from dimos.memory.store.memory import MemoryStore
from dimos.memory.store.sqlite import SqliteStore
from dimos.msgs.sensor_msgs.Image import Image
from dimos.msgs.tf2_msgs.TFMessage import TFMessage
from dimos.perception.memory.dandetect import DanDetector
from dimos.utils.data import get_data

recording = SqliteStore(
    path=get_data(
        "xarm6_worldbelief_realsense_d435i_stationery_calibrated/"
        "xarm6_worldbelief_20260729_203624_161992.db"
    )
)
lo, _ = recording.streams.color_image.get_time_range()

# we simulate live stream for this example
live = MemoryStore()
tf_live = live.stream("tf", TFMessage)
for obs in recording.streams.tf.before(lo + 130):
    tf_live.append(obs.data, ts=obs.ts, pose=None)
color_live = live.stream("color_image", Image)

with DanDetector() as detector:
  
    # SIM block ===========================================
    embedded = detector.embed(live, live=True)

    # fed frames into the live store for this example
    fed = 0
    for obs in recording.streams.color_image.after(lo + 53).before(lo + 129):
        color_live.append(obs.data, ts=obs.ts, pose=obs.pose)
        fed += 1
        time.sleep(0.04)
    print(f"fed {fed} frames into the live store", flush=True)

    # The embed pipeline runs on a background thread and a live stream never 
    # completes, so there is no signal to wait on. 
    # In this example we treat the index as done 
    # once its count has stopped changing for a few seconds.
    stable = embedded.count()
    quiet = time.time()
    while time.time() - quiet < 5.0:
        time.sleep(0.5)
        n = embedded.count()
        if n != stable:
            stable, quiet = n, time.time()
    print(f"color_image_embedded holds {stable} frames", flush=True)

    # main API mechanics ====================================
    index = live.streams.color_image_embedded

    # single query
    hit = detector.localize(recording, "book", index=index)
    if hit is not None:
        print(f"book: {hit.position_world_xyz} score={hit.semantic_score:.2f}")

    # multi query
    queries = ["pen", "red marker"]
    hits = detector.localize(recording, queries, index=index)
    for query, hit in zip(queries, hits):
        if hit is None:
            print(f"{query}: not found")
        else:
            print(f"{query}: {hit.position_world_xyz} score={hit.semantic_score:.2f}")

sys.stdout.flush()
os._exit(0)

A more elaborate example of how API works in the module context

SIM block are needed here to drive the example without a robot as in the previous example
Ignore them.
As you can see the real API surface is small

"""Live DanDetector example, runnable on a laptop against a recording.

Blocks marked API are the code you would actually write on a robot.
Blocks marked SIM ONLY fake what a robot provides for free (a recorder
that has been filling the store, a camera producing frames).

Run from the dimos repo root: uv run python <path to this file>
"""

import os
import sys
import time

from dimos.agents.annotation import skill
from dimos.core.core import rpc
from dimos.memory.module import MemoryModule
from dimos.memory.store.sqlite import SqliteStore
from dimos.msgs.sensor_msgs.Image import Image
from dimos.msgs.tf2_msgs.TFMessage import TFMessage
from dimos.perception.memory.dandetect import DanDetector
from dimos.perception.memory.types import Localization
from dimos.utils.data import get_data

# =============================================================================
# SIM ONLY: fake the recorder's past.
# On a robot the Recorder has been writing tf, camera_info, depth_image and
# color_image into the store since startup. Here we copy them out of a
# recording into a fresh db so the module starts with the same state.
# =============================================================================

DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "objectmemory.db")

recording = SqliteStore(
    path=get_data(
        "xarm6_worldbelief_realsense_d435i_stationery_calibrated/"
        "xarm6_worldbelief_20260729_203624_161992.db"
    )
)
lo, _ = recording.streams.color_image.get_time_range()

if os.path.exists(DB):
    os.unlink(DB)
seed = SqliteStore(path=DB)
seed.start()

tf_seed = seed.stream("tf", TFMessage)
for obs in recording.streams.tf.before(lo + 130):
    tf_seed.append(obs.data, ts=obs.ts, pose=None)
ci = recording.streams.camera_info.first()
seed.stream("camera_info", type(ci.data)).append(ci.data, ts=ci.ts, pose=None)

# depth must stay lossless, the default Image codec is jpeg
depth_seed = seed.stream("depth_image", Image, codec="lz4+lcm")
for obs in recording.streams.depth_image.after(lo + 52).before(lo + 130):
    depth_seed.append(obs.data, ts=obs.ts, pose=None)

# the live tail subscribes to this stream, so it has to exist before start()
seed.stream("color_image", Image)
seed.stop()
print(f"seeded {DB} ({os.path.getsize(DB) / 1e6:.0f} MB)", flush=True)


# =============================================================================
# API: the deployment shape.
# One module owns the store and the models. embed(live=True) starts a
# background thread that keeps filling color_image_embedded until the module
# stops. localize_object answers from whatever is embedded so far. On a robot
# this is one line in a blueprint: blueprint.add(ObjectMemory, db_path=...).
# =============================================================================


class ObjectMemory(MemoryModule):
    @rpc
    def start(self) -> None:
        super().start()
        self.detector = self.register_disposable(DanDetector())
        self.detector.start()
        self.detector.embed(self.store, live=True)

    @skill
    def localize_object(self, query: str) -> Localization | None:
        index = self.store.streams.color_image_embedded
        return self.detector.localize(self.store, query, index=index)


module = ObjectMemory(db_path=DB)
module.start()
print("ObjectMemory started, live embed tailing color_image", flush=True)

# =============================================================================
# SIM ONLY: fake the camera.
# Replays the recording's color frames into the module's store in time
# order. On a robot the camera driver and recorder do this, and this is
# the only stream the module consumes live.
# =============================================================================

color = module.store.stream("color_image", Image)
fed = 0
for obs in recording.streams.color_image.after(lo + 53).before(lo + 129):
    color.append(obs.data, ts=obs.ts, pose=obs.pose)
    fed += 1
    time.sleep(0.04)
print(f"fed {fed} frames through the module store", flush=True)

# =============================================================================
# SIM ONLY: wait for the tail to run dry.
# The feed above is finite, so once the embedded count stops moving the
# pipeline has processed everything it will ever get. A robot never has
# this wait: frames never stop, a query just reads what is there.
# =============================================================================

embedded = module.store.streams.color_image_embedded
stable = embedded.count()
quiet = time.time()
while time.time() - quiet < 3.0:
    time.sleep(0.5)
    n = embedded.count()
    if n != stable:
        stable, quiet = n, time.time()
print(f"color_image_embedded holds {stable} frames", flush=True)

# =============================================================================
# API: query the memory, then shut the module down.
# =============================================================================

hit = module.localize_object("book")
if hit is None:
    print("book: not found")
else:
    print(f"book: {hit.position_world_xyz} score={hit.semantic_score:.2f}")

module.stop()
print("module stopped", flush=True)

# =============================================================================
# SIM ONLY: hard exit. The embed thread is still blocked waiting for a next
# frame that never comes, so a normal exit would hang. On a robot the process
# runs forever anyway, here we flush and kill it.
# =============================================================================

sys.stdout.flush()
os._exit(0)

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ All tests successful. No failed tests found.

@@            Coverage Diff             @@
##             main    #3496      +/-   ##
==========================================
+ Coverage   74.05%   76.04%   +1.98%     
==========================================
  Files        1283     1224      -59     
  Lines      124704   118742    -5962     
  Branches    11141    10654     -487     
==========================================
- Hits        92349    90292    -2057     
+ Misses      29493    25373    -4120     
- Partials     2862     3077     +215     

Flag	Coverage Δ
OS-ubuntu-24.04-arm	70.43% <ø> (ø)
OS-ubuntu-latest	72.22% <ø> (-0.01%)	⬇️
Py-3.10	72.21% <ø> (-0.01%)	⬇️
Py-3.11	72.21% <ø> (+<0.01%)	⬆️
Py-3.12	72.21% <ø> (-0.01%)	⬇️
Py-3.13	72.21% <ø> (-0.01%)	⬇️
Py-3.14	72.22% <ø> (+<0.01%)	⬆️
Py-3.14t	72.21% <ø> (-0.01%)	⬇️
SelfHosted-Large	29.75% <ø> (+0.04%)	⬆️
SelfHosted-Linux	35.84% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.
see 83 files with indirect coverage changes

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.