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)

Important

This PR and the previous that has landed in main #3422 => introduce NOT a fully generalized perception stack that can be thrown onto xArm AND Go2 | Go1 | Any robot platform, no. These are xArm tailored and tested well. Generalized perception stack will land in a while.

[codecov]

Codecov Report

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

@@           Coverage Diff           @@
##             main    #3496   +/-   ##
=======================================
  Coverage   76.13%   76.13%           
=======================================
  Files        1228     1228           
  Lines      119174   119174           
  Branches    10684    10684           
=======================================
+ Hits        90731    90733    +2     
+ Misses      25342    25341    -1     
+ Partials     3101     3100    -1     

Flag	Coverage Δ
OS-ubuntu-24.04-arm	70.55% <ø> (+<0.01%)	⬆️
OS-ubuntu-latest	72.33% <ø> (+<0.01%)	⬆️
Py-3.10	72.32% <ø> (+<0.01%)	⬆️
Py-3.11	72.32% <ø> (+<0.01%)	⬆️
Py-3.12	72.31% <ø> (-0.01%)	⬇️
Py-3.13	72.32% <ø> (+<0.01%)	⬆️
Py-3.14	72.33% <ø> (+<0.01%)	⬆️
Py-3.14t	72.32% <ø> (ø)
SelfHosted-Large	29.74% <ø> (+<0.01%)	⬆️
SelfHosted-Linux	35.77% <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.
see 1 file 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.

[leshy]

strict improvement, stable API, merging now, will validate IRL etc later. Probably parts of this PR description belong in a nice readme on this.

for synthetic live data, we can use actual IRL replay of mem2 store (like go2 hk office)

[greptile-apps]

Greptile Summary

This change consolidates embedding, localization, and inventory model ownership in DanDetector and updates memory localization to use OWLv2 detections. Two reproduced correctness failures need attention before merge: localization can return a weaker recent support instead of the best query match without marking the result ambiguous, and replay embedding without explicit bounds silently indexes no frames.

Confidence Score: 3/5

Not safe to merge until localization ranking and unbounded replay behavior are corrected.

Focused executable checks reproduced two independent functional failures: one selects a weaker localization support based on recency, and the other drops all replay frames when optional bounds are omitted.

Files Needing Attention: dimos/perception/memory/localize.py needs a query-evidence-based winner and ambiguity comparison; dimos/perception/memory/dandetect.py needs concrete replay-bound handling.

T-Rex Logs

What T-Rex did

• T-Rex produced a proof for a posted P1 finding and included evidence from the focused localization harness and its successful localization failure output.
• T-Rex produced a second P1 finding proof that includes the no-bounds replay harness and the numeric-bounds replay control harness, along with their execution outputs.
• T-Rex produced a third P1 finding proof backed by the third proof, with no artifacts attached.
• T-Rex performed a general contract validation showing before/no-bounds and after/control-bounds runs, with harness sources uploaded and corresponding execution logs linked.

_{Ran code and verified through T-Rex}

Comments Outside Diff (1)

1. General comment

   Unbounded replay embedding silently filters every frame

   • Bug
     • Calling DanDetector.embed(store, live=False) without optional bounds passes None to the replay index. The downstream index calls .after(t0).before(t1) with those values; under SQLite-style numeric timestamp predicates this produces no matching frames. The executable no-bounds run demonstrated embed_index_args after=None before=None and filtered_timestamps=[], while the numeric control retained timestamp 15.0.
   • Cause
     • Lines 118-119 use cast("float", after/before), which changes only static typing and leaves default None unchanged at runtime. embed_index requires floats and applies both parameters directly as time filters at dimos/perception/memory/localize.py:254-255.
   • Fix
     • Require both replay bounds before calling embed_index, or derive concrete store time-range defaults for omitted bounds; do not use typing.cast as runtime normalization. Add a regression test covering embed(store, live=False) with omitted bounds.

   _{Ran code and verified through T-Rex}

_{Reviews (1): Last reviewed commit: "Merge branch 'main' into danvi/dim1435/o..." | Re-trigger Greptile}