diff --git a/dimos/models/embedding/siglip.py b/dimos/models/embedding/siglip.py index 1b36cf090e..225c98a04f 100644 --- a/dimos/models/embedding/siglip.py +++ b/dimos/models/embedding/siglip.py @@ -53,11 +53,6 @@ def _model(self) -> HFSiglipModel: def _processor(self) -> SiglipProcessor: return SiglipProcessor.from_pretrained(self.config.model_name, use_fast=True) - @property - def logit_scale(self) -> float: - """Trained sigmoid temperature: exp of the model's raw logit scale.""" - return float(self._model.logit_scale.exp()) - @overload def embed(self, image: Image, /) -> Embedding: ... @overload diff --git a/dimos/perception/memory/dandetect.py b/dimos/perception/memory/dandetect.py new file mode 100644 index 0000000000..081cb52e61 --- /dev/null +++ b/dimos/perception/memory/dandetect.py @@ -0,0 +1,180 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""One disposable resource wrapping the memory perception API. + +``DanDetector`` owns the models behind :func:`embed_index`, :func:`localize`, +and :func:`inventory`: enter once, query many times on warm weights, and +``stop()`` (or leave the ``with`` block) releases whatever loaded. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Literal, cast, overload + +from dimos.core.resource import Resource +from dimos.memory.embed import EmbedImages +from dimos.memory.tf import StreamTF +from dimos.memory.transform import throttle +from dimos.perception.memory import gates +from dimos.perception.memory.gates import OPTICAL_FRAME, TF_TOLERANCE, WORLD_FRAME +from dimos.perception.memory.inventory import DEFAULT_VOCABULARY, NamingVocabulary, inventory +from dimos.perception.memory.localize import EMBED_HZ, embed_index, localize + +if TYPE_CHECKING: + from reactivex.abc import DisposableBase + + from dimos.memory.stream import Stream + from dimos.models.embedding.siglip import SigLIPModel + from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter + from dimos.perception.detection.detectors.owlv2 import Owlv2Detector + from dimos.perception.memory.types import Instance, Localization + + +class DanDetector(Resource): + """The perception models as one resource. + + ``start()`` constructs SigLIP, OWLv2, and EdgeTAM. The two + HuggingFace models load lazily on first use, so an inventory-only + caller never pays for SigLIP; ``stop()`` releases whatever loaded. + """ + + siglip: SigLIPModel + detector: Owlv2Detector + segmenter: EdgeTAMImageSegmenter + + def start(self) -> None: + from dimos.models.embedding.siglip import SigLIPModel + from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter + from dimos.perception.detection.detectors.owlv2 import Owlv2Detector + + self.siglip = SigLIPModel() + self.detector = Owlv2Detector() + self.segmenter = EdgeTAMImageSegmenter() + self._live: list[DisposableBase] = [] + + def stop(self) -> None: + for disposable in self._live: + disposable.dispose() + self.siglip.stop() + self.detector.stop() + del self.segmenter + + @overload + def embed( + self, + store: Any, + after: float, + before: float, + *, + live: Literal[False] = False, + optical_frame: str = ..., + world_frame: str = ..., + tf_tolerance: float = ..., + ) -> Stream[Any, Any]: ... + @overload + def embed( + self, + store: Any, + *, + live: Literal[True], + optical_frame: str = ..., + world_frame: str = ..., + tf_tolerance: float = ..., + ) -> Stream[Any, Any]: ... + def embed( + self, + store: Any, + after: float | None = None, + before: float | None = None, + *, + live: bool = False, + optical_frame: str = OPTICAL_FRAME, + world_frame: str = WORLD_FRAME, + tf_tolerance: float = TF_TOLERANCE, + ) -> Stream[Any, Any]: + """SigLIP-embedded, world-posed frame index for :meth:`localize`. + + Replay mode indexes ``[after, before]`` in memory and returns when + done. ``live=True`` instead tails ``color_image`` and keeps saving + into the store's named ``color_image_embedded`` stream on a + background thread; the returned stream is that named stream. + """ + if not live: + return embed_index( + store, + self.siglip, + cast("float", after), + cast("float", before), + optical_frame=optical_frame, + world_frame=world_frame, + tf_tolerance=tf_tolerance, + ) + + from dimos.msgs.sensor_msgs.Image import Image + + tf = StreamTF.from_store(store) + if tf is None: + raise ValueError("store has no tf stream") + embedded: Stream[Any, Any] = store.stream("color_image_embedded", Image) + pipeline = ( + store.streams.color_image.live() + .transform(throttle(1.0 / EMBED_HZ)) + .map( + lambda obs: obs.derive( + data=obs.data, + pose=gates.camera_pose(tf, obs.ts, optical_frame, world_frame, tf_tolerance), + ) + ) + .filter(lambda obs: obs.pose is not None) + .transform(EmbedImages(self.siglip, batch_size=1)) + .save(embedded) + ) + self._live.append(pipeline.drain_thread()) + return embedded + + def localize( + self, + store: Any, + query: str | list[str], + *, + index: Stream[Any, Any], + **kwargs: Any, + ) -> Localization | list[Localization | None] | None: + """:func:`localize` on this resource's models.""" + return localize( + store, + query, + index=index, + siglip=self.siglip, + detector=self.detector, + segmenter=self.segmenter, + **kwargs, + ) + + def inventory( + self, + store: Any, + *, + naming_vocabulary: NamingVocabulary = DEFAULT_VOCABULARY, + **kwargs: Any, + ) -> list[Instance]: + """:func:`inventory` on this resource's models.""" + return inventory( + store, + segmenter=self.segmenter, + detector=self.detector, + naming_vocabulary=naming_vocabulary, + **kwargs, + ) diff --git a/dimos/perception/memory/inventory.py b/dimos/perception/memory/inventory.py index 13a0e81734..f256701ee1 100644 --- a/dimos/perception/memory/inventory.py +++ b/dimos/perception/memory/inventory.py @@ -19,7 +19,7 @@ instance table. Existence is decoupled from naming and the ordering is a constraint, not a preference: propose (EdgeTAM automatic masks), lift (masked depth to world supports), associate (hard constraints before any -score), and only then name (OmDet-Turbo, labels as metadata). Labels and +score), and only then name (OWLv2, labels as metadata). Labels and appearance never enter association; position and same-frame co-occurrence decide everything, which is what keeps two identical objects two instances. @@ -61,7 +61,7 @@ from dimos_lcm.sensor_msgs import CameraInfo from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter - from dimos.perception.detection.detectors.omdet import OmDetDetector + from dimos.perception.detection.detectors.owlv2 import Owlv2Detector from dimos.perception.detection.type.detection2d.seg import Detection2DSeg from dimos.protocol.tf.tf import TFLookup @@ -676,11 +676,11 @@ def _name_and_suppress( tracks: list[_Track], tracks_2d: list[_Track2D], store: Any, - detector: OmDetDetector, + detector: Owlv2Detector, vocabulary: NamingVocabulary, policy: InventoryPolicy, ) -> None: - """OmDet naming per instance on keyframes, person/hand suppressing observations. + """OWLv2 naming per instance on keyframes, person/hand suppressing observations. Runs after association by construction: association consumed unnamed supports, so per-view label instability cannot starve existence or split @@ -705,7 +705,7 @@ def _name_and_suppress( queries, starts, canonical = _flatten(vocabulary) all_ts = sorted(set(frame_members) | set(frame_members_2d)) logger.info( - f"naming: OmDet over {len(all_ts)} keyframes, " + f"naming: OWLv2 over {len(all_ts)} keyframes, " f"{len(canonical)} groups of {len(queries)} queries" ) for ts in all_ts: @@ -766,7 +766,7 @@ def inventory( store: Any, *, segmenter: EdgeTAMImageSegmenter, - detector: OmDetDetector, + detector: Owlv2Detector, naming_vocabulary: NamingVocabulary, after: float | None = None, before: float | None = None, diff --git a/dimos/perception/memory/localize.py b/dimos/perception/memory/localize.py index 7442e16d0a..e6182fd8d6 100644 --- a/dimos/perception/memory/localize.py +++ b/dimos/perception/memory/localize.py @@ -15,9 +15,8 @@ """Query-time object localization: text prompt to latest 3D pose and cloud. Search memory with embeddings (SigLIP, -frame-level), open-vocabulary detection (OmDet-Turbo, per-box scores), -segmentation (EdgeTAM), projection to 3D through aligned depth, referent -identity on the observation crops (SigLIP, pairwise margins). Two +frame-level), open-vocabulary detection (OWLv2, calibrated per-box scores), +segmentation (EdgeTAM), projection to 3D through aligned depth. Two algorithm rules distinguish it from a best-crop search: * **Latest-pose semantics.** Among verified observations of the chosen @@ -41,7 +40,7 @@ from dimos.memory.transform import throttle from dimos.perception.detection.project import sees from dimos.perception.detection.type.detection2d.imageDetections2D import ImageDetections2D -from dimos.perception.detection.type.detection3d.pointcloud import Detection3DPC +from dimos.perception.detection.type.detection3d.imageDetections3DPC import ImageDetections3DPC from dimos.perception.memory import gates from dimos.perception.memory.gates import OPTICAL_FRAME, TF_TOLERANCE, WORLD_FRAME from dimos.perception.memory.types import Localization, LocalizePolicy, Support @@ -51,10 +50,10 @@ from dimos_lcm.sensor_msgs import CameraInfo from dimos.memory.stream import Stream - from dimos.models.embedding.base import Embedding from dimos.models.embedding.siglip import SigLIPModel from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter - from dimos.perception.detection.detectors.omdet import OmDetDetector + from dimos.perception.detection.detectors.owlv2 import Owlv2Detector + from dimos.perception.detection.type.detection3d.pointcloud import Detection3DPC from dimos.protocol.tf.tf import TFLookup logger = setup_logger() @@ -75,7 +74,6 @@ class _ClusterObservation: cloud: Any camera_position: np.ndarray detection: Detection3DPC - crop: Any # Image cut to the 2D box, for crop-level identity @dataclass @@ -160,7 +158,7 @@ def _axes_observed( class _DetectionCache: - """One OmDet + EdgeTAM pass per unique frame, shared across queries and clusters.""" + """One OWLv2 + EdgeTAM pass per unique frame, shared across queries and clusters.""" def __init__(self, detector: Any, segmenter: Any, queries: list[str], floor: float) -> None: self.detector = detector @@ -201,8 +199,8 @@ def _lift( tf_tolerance: float, policy: LocalizePolicy, plane: Any | None = None, -) -> list[tuple[Detection3DPC, np.ndarray, Any]]: - """Depth-lift 2D detections; returns valid (detection3d, camera_position, detection2d).""" +) -> list[tuple[Detection3DPC, np.ndarray]]: + """Depth-lift 2D detections; returns valid (detection3d, camera_position) pairs.""" depth = gates.depth_at(store, detections.ts) transform = tf.get(optical_frame, world_frame, detections.ts, tf_tolerance) if depth is None or transform is None: @@ -212,11 +210,9 @@ def _lift( return [] camera = np.array([pose.position.x, pose.position.y, pose.position.z]) - valid: list[tuple[Detection3DPC, np.ndarray, Any]] = [] - for det2d in detections: - det3d = Detection3DPC.from_depth(det2d, depth, camera_info, transform) - if det3d is None: - continue + lifted = ImageDetections3DPC.from_depth(detections, depth, camera_info, transform) + valid: list[tuple[Detection3DPC, np.ndarray]] = [] + for det3d in lifted: points = np.asarray(det3d.pointcloud.pointcloud.points) if len(points) < policy.min_depth_points: continue @@ -232,18 +228,10 @@ def _lift( high = float(np.quantile(heights, 0.95)) if low > policy.surface_patch_min_drop_m and high < policy.surface_patch_max_rise_m: continue - valid.append((det3d, camera, det2d)) + valid.append((det3d, camera)) return valid -def _cluster_identity(cluster: _Cluster, siglip: SigLIPModel, query_embedding: Embedding) -> float: - """Best SigLIP crop-to-query cosine over the cluster's observations.""" - embeddings = siglip.embed(*[o.crop for o in cluster.observations]) - if not isinstance(embeddings, list): - embeddings = [embeddings] - return max(e @ query_embedding for e in embeddings) - - def embed_index( store: Any, siglip: SigLIPModel, @@ -330,7 +318,7 @@ def localize( *, index: Stream[Any, Any], siglip: SigLIPModel, - detector: OmDetDetector, + detector: Owlv2Detector, segmenter: EdgeTAMImageSegmenter, require_pose: bool = True, policy: LocalizePolicy | None = None, @@ -349,7 +337,7 @@ def localize( ``refusal_margin`` - a flagged hit, never a silent guess. A list *query* runs every label through one shared detection cache - - OmDet takes the whole list per frame - and returns one result per label, + OWLv2 takes the whole list per frame - and returns one result per label, in input order; ``trace`` then takes a list of the same length. The index and the three models belong to the caller: nothing here is @@ -420,7 +408,7 @@ def _localize_one( store, tf, camera_info, frames, optical_frame, world_frame, tf_tolerance ) - # Pass 2 - OmDet + EdgeTAM: detect, segment, lift, verify. + # Pass 2 - OWLv2 + EdgeTAM: detect, segment, lift, verify. clusters: list[_Cluster] = [] ungrounded_best: tuple[float, float] | None = None # (score, ts) processed: set[float] = set() @@ -447,12 +435,11 @@ def _absorb(frame_obs: Any, is_verify: bool) -> None: plane, ) for det2d in detections: - if not any(d.track_id == det2d.track_id for d, _, _ in lifted): + if not any(d.track_id == det2d.track_id for d, _ in lifted): best = (det2d.confidence, det2d.ts) if ungrounded_best is None or best[0] > ungrounded_best[0]: ungrounded_best = best - for det3d, camera, det2d in lifted: - x1, y1, x2, y2 = det2d.bbox + for det3d, camera in lifted: observation = _ClusterObservation( ts=det3d.ts, score=det3d.confidence, @@ -460,7 +447,6 @@ def _absorb(frame_obs: Any, is_verify: bool) -> None: cloud=det3d.pointcloud, camera_position=camera, detection=det3d, - crop=det2d.image.crop(int(x1), int(y1), int(x2 - x1) + 1, int(y2 - y1) + 1), ) if trace is not None: (trace.verified if is_verify else trace.matched).append((det3d.ts, det3d)) @@ -547,37 +533,15 @@ def _absorb(frame_obs: Any, is_verify: bool) -> None: ) return None - # Detector scores verify supports but do not bind attributes ("red - # marker" ranks a black pen high): the referent is chosen by SigLIP crop - # identity, compared pairwise through the trained sigmoid temperature so - # margins are probability differences and the bias term cancels. - identity = {id(c): _cluster_identity(c, siglip, query_embedding) for c in verified} - scale = siglip.logit_scale - - def dominance(a: float, b: float) -> float: - return math.tanh(scale * (a - b) / 2) - - best_identity = max(identity.values()) - candidates = [ - c for c in verified if dominance(best_identity, identity[id(c)]) < policy.refusal_margin - ] - winner = max(candidates, key=lambda c: c.latest.ts) + winner = max(verified, key=lambda c: c.latest.ts) w_lo, w_hi = winner.interval - rival_identities = [ - identity[id(c)] + rival_scores = [ + c.max_score for c in verified if c is not winner and not (c.interval[1] < w_lo or c.interval[0] > w_hi) # coexisting in time ] - margin = min( - (dominance(identity[id(winner)], r) for r in rival_identities), - default=1.0, - ) - logger.info( - "identity: " - + ", ".join(f"cos={identity[id(c)]:.3f} score={c.max_score:.2f}" for c in verified) - + f" margin={margin:.2f}" - ) + margin = winner.max_score - max(rival_scores) if rival_scores else 1.0 reason = "ambiguous_between_coexisting_candidates" if margin < policy.refusal_margin else None latest = winner.latest diff --git a/dimos/perception/memory/tool_inventory.py b/dimos/perception/memory/tool_inventory.py index 715789b004..dc8c20afaf 100644 --- a/dimos/perception/memory/tool_inventory.py +++ b/dimos/perception/memory/tool_inventory.py @@ -58,7 +58,7 @@ from dimos.memory.tf import StreamTF from dimos.memory.transform import throttle from dimos.perception.memory import gates -from dimos.perception.memory.inventory import DEFAULT_VOCABULARY, NamingVocabulary, inventory +from dimos.perception.memory.inventory import DEFAULT_VOCABULARY, NamingVocabulary from dimos.perception.memory.types import Instance, SupportObservation from dimos.utils.data import get_data @@ -261,25 +261,17 @@ def main() -> int: else: naming_vocabulary = DEFAULT_VOCABULARY - from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter - from dimos.perception.detection.detectors.omdet import OmDetDetector - - segmenter = EdgeTAMImageSegmenter() - detector = OmDetDetector() - - instances = inventory( - store, - segmenter=segmenter, - detector=detector, - naming_vocabulary=naming_vocabulary, - after=after, - before=before, - include_ungrounded=args.include_ungrounded, - log_progress=args.log_progress, - ) + from dimos.perception.memory.dandetect import DanDetector - detector.stop() - del segmenter + with DanDetector() as dan: + instances = dan.inventory( + store, + naming_vocabulary=naming_vocabulary, + after=after, + before=before, + include_ungrounded=args.include_ungrounded, + log_progress=args.log_progress, + ) print(f"instances: {len(instances)}") for i, instance in enumerate(instances): diff --git a/dimos/perception/memory/tool_localize.py b/dimos/perception/memory/tool_localize.py index f706ea9f1a..b1ec9e9190 100644 --- a/dimos/perception/memory/tool_localize.py +++ b/dimos/perception/memory/tool_localize.py @@ -35,7 +35,7 @@ from dimos.memory.tf import StreamTF from dimos.memory.transform import throttle from dimos.perception.memory import gates -from dimos.perception.memory.localize import LocalizeTrace, embed_index, localize +from dimos.perception.memory.localize import LocalizeTrace from dimos.perception.memory.types import Localization from dimos.utils.data import get_data @@ -217,53 +217,38 @@ def main() -> int: after = lo + args.start before = lo + args.start + args.duration if args.duration is not None else hi - from dimos.models.embedding.siglip import SigLIPModel - from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter - from dimos.perception.detection.detectors.omdet import OmDetDetector - - siglip = SigLIPModel() - detector = OmDetDetector() - segmenter = EdgeTAMImageSegmenter() - index = embed_index(store, siglip, after, before) + from dimos.perception.memory.dandetect import DanDetector traces: list[tuple[str, LocalizeTrace]] = [] hits = 0 - if args.multi: - qtraces = [LocalizeTrace() for _ in queries] - results = localize( - store, - queries, - index=index, - siglip=siglip, - detector=detector, - segmenter=segmenter, - require_pose=not args.allow_no_pose, - trace=qtraces, - ) - for query, qtrace, hit in zip(queries, qtraces, results, strict=True): - if report(query, hit, lo): - hits += 1 - traces.append((query, qtrace)) - else: - for query in queries: - trace = LocalizeTrace() - hit = localize( + with DanDetector() as dan: + index = dan.embed(store, after, before) + if args.multi: + qtraces = [LocalizeTrace() for _ in queries] + results = dan.localize( store, - query, + queries, index=index, - siglip=siglip, - detector=detector, - segmenter=segmenter, require_pose=not args.allow_no_pose, - trace=trace, + trace=qtraces, ) - if report(query, hit, lo): - hits += 1 - traces.append((query, trace)) - - siglip.stop() - detector.stop() - del segmenter + for query, qtrace, hit in zip(queries, qtraces, results, strict=True): + if report(query, hit, lo): + hits += 1 + traces.append((query, qtrace)) + else: + for query in queries: + trace = LocalizeTrace() + hit = dan.localize( + store, + query, + index=index, + require_pose=not args.allow_no_pose, + trace=trace, + ) + if report(query, hit, lo): + hits += 1 + traces.append((query, trace)) if not hits: return 1