diff --git a/.github/workflows/message-codegen.yml b/.github/workflows/message-codegen.yml index da305cbc9d3..53952452bdc 100644 --- a/.github/workflows/message-codegen.yml +++ b/.github/workflows/message-codegen.yml @@ -142,14 +142,14 @@ jobs: cmake -S build/sdk-deps/zenoh-cpp -B build/sdk-deps/zenoh-cpp-build -DCMAKE_PREFIX_PATH="$PWD/build/sdk-deps/zenoh-c" -DCMAKE_INSTALL_PREFIX="$PWD/build/sdk-deps/zenoh-cpp-install" -DZENOHCXX_ENABLE_TESTS=OFF -DZENOHCXX_ENABLE_EXAMPLES=OFF cmake --install build/sdk-deps/zenoh-cpp-build .native-review/bin/python -m pip install pytest pytest-mock pytest-env - - name: Verify explicit CDR transports and actual SDK coordinator example + - name: Verify default CDR transports and actual SDK coordinator example if: runner.os == 'Linux' env: DIMOS_EXTERNAL_NATIVE_PYTHON: ${{ github.workspace }}/.native-review/bin/python DIMOS_MESSAGE_NATIVE_CACHE: ${{ github.workspace }}/build/native-project-cache DIMOS_CDR_NATIVE_SDK_PREFIX: ${{ github.workspace }}/build/sdk-deps/zenoh-c;${{ github.workspace }}/build/sdk-deps/zenoh-cpp-install run: | - .native-review/bin/python -I -m pytest --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/core/test_cdr_transport.py" + LCM_DEFAULT_URL=memq:// .native-review/bin/python -I -m pytest --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/core/test_cdr_transport.py" "$PWD/dimos/core/test_transport_factory.py" .venv/bin/python -m pytest --noconftest --import-mode=importlib -o addopts='' -o 'markers=native_e2e: compiled native consumer acceptance' -q dimos/message_codegen/test_project_native.py -k actual_coordinator cargo test --manifest-path native/rust/dimos-module/Cargo.toml --locked --features cdr --lib --jobs 2 - name: Verify generated browser schemas, relay encoding and web consumers diff --git a/dimos/core/cdr_transport.py b/dimos/core/cdr_transport.py index af8eaf8ffbc..8604812b9bc 100644 --- a/dimos/core/cdr_transport.py +++ b/dimos/core/cdr_transport.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Explicit CDR encoding over existing raw transports; no default cutover.""" +"""CDR encoding for generated messages over existing raw transports.""" from __future__ import annotations @@ -36,7 +36,9 @@ class CdrTransport(PubSubTransport[T]): - """Opt-in plain XCDR1 values over raw LCM or Zenoh. + """Plain XCDR1 values over raw LCM or Zenoh. + + The default transport factory selects this class for generated message types. The explicit /cdr//msg/ suffix separates these channels from existing LCM-encoded streams. Both peers must select the same schema; diff --git a/dimos/core/test_transport_factory.py b/dimos/core/test_transport_factory.py index 42504371c96..8763244b8bd 100644 --- a/dimos/core/test_transport_factory.py +++ b/dimos/core/test_transport_factory.py @@ -14,8 +14,14 @@ from __future__ import annotations +from contextlib import ExitStack +import pickle + +from dimos_cdrgen.registry import encode import pytest +import dimos.core.cdr_transport as cdr_module +from dimos.core.cdr_transport import CdrTransport from dimos.core.global_config import GlobalConfig from dimos.core.transport import ( LCMTransport, @@ -31,6 +37,8 @@ session_config, transport_topic, ) +from dimos.messages.sensor_msgs.msg import Image as CdrImage +from dimos.messages.std_msgs.msg import Int32 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped from dimos.msgs.geometry_msgs.Twist import Twist from dimos.msgs.sensor_msgs.Image import Image @@ -155,3 +163,35 @@ def test_apply_transport_arg_bad_value_exits(argv: list[str]) -> None: apply_transport_arg(argv, g=g) assert exc.value.code == 2 assert g.transport == "lcm" # config left untouched + + +@pytest.mark.parametrize("config", [LCM, ZENOH]) +def test_generated_factory_publishes_cdr_instead_of_pickle(config, mocker): + boundary = "LCMPubSubBase" if config.transport == "lcm" else "ZenohPubSubBase" + factory = mocker.patch.object(cdr_module, boundary, autospec=True) + transport = make_transport("reading", Int32, g=config) + with ExitStack() as resources: + resources.callback(transport.stop) + transport.publish(Int32(42)) + factory.return_value.publish.assert_called_once_with(transport.topic, encode(Int32(42))) + prefix = "dimos/" if config.transport == "zenoh" else "" + assert transport.channel == prefix + "reading/cdr/std_msgs/msg/Int32" + restored = pickle.loads(pickle.dumps(transport)) + assert type(restored) is CdrTransport + assert restored.channel == transport.channel + + +def test_generated_sensor_qos_survives_factory_and_pickle(): + sensor = make_transport("camera", CdrImage, g=ZENOH) + seed = make_transport("loaded_map", CdrImage, g=ZENOH) + assert sensor.publish_qos == QOS_LATEST_WINS.to_wire() + assert pickle.loads(pickle.dumps(seed)).publish_qos == QOS_NEVER_DROP.to_wire() + assert make_transport("camera", CdrImage, g=LCM).publish_qos is None + + +def test_unregistered_generated_claim_is_rejected_instead_of_pickled(): + class Unregistered: + __msgtype__ = "std_msgs/msg/Int32" + + with pytest.raises(TypeError, match="registry-owned"): + make_transport("reading", Unregistered, g=ZENOH) diff --git a/dimos/core/transport_factory.py b/dimos/core/transport_factory.py index b86b271682a..22e892e6fcc 100644 --- a/dimos/core/transport_factory.py +++ b/dimos/core/transport_factory.py @@ -20,6 +20,7 @@ import sys from typing import TYPE_CHECKING, Any, NoReturn, get_args +from dimos.core.cdr_transport import CdrTransport from dimos.core.global_config import GlobalConfig, TransportBackend, global_config from dimos.core.transport import ( LCMTransport, @@ -58,7 +59,12 @@ def transport_topic(name: str, g: GlobalConfig = global_config) -> str: # High-rate sensor streams: drop stale frames under congestion, never stall the # publisher. Matched by message type since that is what makes them high-rate. -_LATEST_WINS_TYPES = ("sensor_msgs.Image", "sensor_msgs.PointCloud2") +_LATEST_WINS_TYPES = ( + "sensor_msgs.Image", + "sensor_msgs.PointCloud2", + "sensor_msgs/msg/Image", + "sensor_msgs/msg/PointCloud2", +) # Low-rate channels where a drop loses something that never comes back: a whole # turn of agent/human conversation, a one-shot robot action verb, a push-to-talk # chunk where one gap discards the whole utterance, or a region of a seeded map @@ -91,7 +97,9 @@ def default_zenoh_qos_for(name: str, msg_name: str) -> ZenohQoS | None: def default_zenoh_qos(name: str, msg_type: type | None = None) -> ZenohQoS | None: """Default publisher QoS for a logical channel; None = zenoh defaults.""" - return default_zenoh_qos_for(name, getattr(msg_type, "msg_name", "")) + return default_zenoh_qos_for( + name, getattr(msg_type, "__msgtype__", getattr(msg_type, "msg_name", "")) + ) def make_transport( @@ -99,8 +107,10 @@ def make_transport( ) -> PubSubTransport[Any]: """Construct the active-backend pub/sub transport for a logical channel. - A pickled (self-describing) transport is used when no `msg_type` is given or - the type has no `lcm_encode`. Otherwise a typed transport is used. + Registry-owned generated messages use CDR on the selected raw backend. + Legacy types with `lcm_encode` retain their explicit legacy encoding; other + Python values use pickle. CDR channels have a separate `/cdr/` suffix; + there is no encoding negotiation or fallback decoding. A channel name alone doesn't fully define a topic: backends have per-topic settings (Zenoh publisher QoS, for one) that live on their Topic objects. @@ -110,8 +120,15 @@ def make_transport( transport map. LCM (UDP multicast) has no per-topic settings. """ - use_pickled = msg_type is None or getattr(msg_type, "lcm_encode", None) is None topic = transport_topic(name, g) + if msg_type is not None and hasattr(msg_type, "__msgtype__"): + return CdrTransport( + topic, + msg_type, + backend=g.transport, + qos=default_zenoh_qos(name, msg_type) if g.transport == "zenoh" else None, + ) + use_pickled = msg_type is None or getattr(msg_type, "lcm_encode", None) is None if g.transport == "zenoh": ztopic = ZenohTopic( topic, None if use_pickled else msg_type, qos=default_zenoh_qos(name, msg_type) diff --git a/dimos/models/embedding/clip.py b/dimos/models/embedding/clip.py index ad5e3e0cc6c..f429dc62a93 100644 --- a/dimos/models/embedding/clip.py +++ b/dimos/models/embedding/clip.py @@ -17,12 +17,12 @@ from functools import cached_property from typing import overload -from dimos.messages.sensor_msgs.msg import Image from PIL import Image as PILImage import torch import torch.nn.functional as functional from transformers import CLIPModel as HFCLIPModel, CLIPProcessor +from dimos.messages.sensor_msgs.msg import Image from dimos.models.base import HuggingFaceModel from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig from dimos.msgs.image import image_to_rgb diff --git a/dimos/models/embedding/dino.py b/dimos/models/embedding/dino.py index 5b0d24fd137..022a442ebf7 100644 --- a/dimos/models/embedding/dino.py +++ b/dimos/models/embedding/dino.py @@ -17,12 +17,12 @@ from functools import cached_property from typing import Any, overload -from dimos.messages.sensor_msgs.msg import Image from PIL import Image as PILImage import torch import torch.nn.functional as functional from transformers import AutoImageProcessor, AutoModel +from dimos.messages.sensor_msgs.msg import Image from dimos.models.base import HuggingFaceModel from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig from dimos.msgs.image import image_to_rgb diff --git a/dimos/models/embedding/siglip.py b/dimos/models/embedding/siglip.py index bb0ed762943..8cabb1afdf4 100644 --- a/dimos/models/embedding/siglip.py +++ b/dimos/models/embedding/siglip.py @@ -17,12 +17,12 @@ from functools import cached_property from typing import overload -from dimos.messages.sensor_msgs.msg import Image from PIL import Image as PILImage import torch import torch.nn.functional as functional from transformers import SiglipModel as HFSiglipModel, SiglipProcessor +from dimos.messages.sensor_msgs.msg import Image from dimos.models.base import HuggingFaceModel from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig from dimos.msgs.image import image_to_rgb diff --git a/dimos/models/embedding/siglip2.py b/dimos/models/embedding/siglip2.py index f8f1fe8861b..c85932f9176 100644 --- a/dimos/models/embedding/siglip2.py +++ b/dimos/models/embedding/siglip2.py @@ -17,12 +17,12 @@ from functools import cached_property from typing import overload -from dimos.messages.sensor_msgs.msg import Image from PIL import Image as PILImage import torch import torch.nn.functional as functional from transformers import SiglipModel as HFSiglipModel, SiglipProcessor +from dimos.messages.sensor_msgs.msg import Image from dimos.models.base import HuggingFaceModel from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig from dimos.msgs.image import image_to_rgb diff --git a/dimos/models/embedding/test_generated_images.py b/dimos/models/embedding/test_generated_images.py index e3c1245ab07..a51f9778ba4 100644 --- a/dimos/models/embedding/test_generated_images.py +++ b/dimos/models/embedding/test_generated_images.py @@ -29,20 +29,25 @@ from dimos.msgs.image import image_from_array, image_view -@pytest.mark.parametrize("model_type", [CLIPModel,DINOModel,SigLIPModel,SigLIP2Model]) -def test_generated_embeddings_keep_zero_source_time_and_convert_rgb(mocker,model_type): - processor,network=mocker.Mock(),mocker.Mock() - features=torch.tensor([[3.0,4.0]]) - processor.return_value=BatchFeature(data={"pixel_values":torch.zeros(1,3,4,8)}) - network.get_image_features.return_value=features - network.return_value=BaseModelOutputWithPooling(last_hidden_state=features[:,None,:],pooler_output=features) - mocker.patch.object(model_type,"_processor",new_callable=PropertyMock,return_value=processor) - mocker.patch.object(model_type,"_model",new_callable=PropertyMock,return_value=network) - image=image_from_array(np.full((4,8,3),[0,0,255],dtype=np.uint8),encoding="bgr8", - header=Header(frame_id="camera",stamp=Time(sec=0,nanosec=0))) - model=model_type(device="cpu") - result=model.embed(image) +@pytest.mark.parametrize("model_type", [CLIPModel, DINOModel, SigLIPModel, SigLIP2Model]) +def test_generated_embeddings_keep_zero_source_time_and_convert_rgb(mocker, model_type): + processor, network = mocker.Mock(), mocker.Mock() + features = torch.tensor([[3.0, 4.0]]) + processor.return_value = BatchFeature(data={"pixel_values": torch.zeros(1, 3, 4, 8)}) + network.get_image_features.return_value = features + network.return_value = BaseModelOutputWithPooling( + last_hidden_state=features[:, None, :], pooler_output=features + ) + mocker.patch.object(model_type, "_processor", new_callable=PropertyMock, return_value=processor) + mocker.patch.object(model_type, "_model", new_callable=PropertyMock, return_value=network) + image = image_from_array( + np.full((4, 8, 3), [0, 0, 255], dtype=np.uint8), + encoding="bgr8", + header=Header(frame_id="camera", stamp=Time(sec=0, nanosec=0)), + ) + model = model_type(device="cpu") + result = model.embed(image) assert result.timestamp == 0.0 - torch.testing.assert_close(result.vector,torch.tensor([0.6,0.8])) - assert processor.call_args.kwargs["images"][0].getpixel((0,0)) == (255,0,0) - np.testing.assert_array_equal(image_view(image)[0,0],[0,0,255]) + torch.testing.assert_close(result.vector, torch.tensor([0.6, 0.8])) + assert processor.call_args.kwargs["images"][0].getpixel((0, 0)) == (255, 0, 0) + np.testing.assert_array_equal(image_view(image)[0, 0], [0, 0, 255]) diff --git a/dimos/models/embedding/treid.py b/dimos/models/embedding/treid.py index ce36b5d004b..3ce7331eec3 100644 --- a/dimos/models/embedding/treid.py +++ b/dimos/models/embedding/treid.py @@ -19,11 +19,11 @@ from functools import cached_property -from dimos.messages.sensor_msgs.msg import Image import torch import torch.nn.functional as functional from torchreid import utils as torchreid_utils +from dimos.messages.sensor_msgs.msg import Image from dimos.models.base import LocalModel from dimos.models.embedding.base import Embedding, EmbeddingModel, EmbeddingModelConfig from dimos.msgs.image import image_to_bgr diff --git a/dimos/models/segmentation/edge_tam.py b/dimos/models/segmentation/edge_tam.py index e240a6529d1..b65977283c0 100644 --- a/dimos/models/segmentation/edge_tam.py +++ b/dimos/models/segmentation/edge_tam.py @@ -20,7 +20,6 @@ import tempfile from typing import TYPE_CHECKING, Any, Protocol, TypedDict -from dimos.messages.sensor_msgs.msg import Image from hydra.utils import instantiate import numpy as np from numpy.typing import NDArray @@ -28,6 +27,7 @@ from PIL import Image as PILImage import torch +from dimos.messages.sensor_msgs.msg import Image from dimos.models.base import default_torch_device from dimos.msgs.image import image_to_bgr, image_to_jpeg from dimos.perception.detection.detectors.base import Detector diff --git a/dimos/models/segmentation/test_yoloe.py b/dimos/models/segmentation/test_yoloe.py index 345fc79ac1c..5d2efaffa50 100644 --- a/dimos/models/segmentation/test_yoloe.py +++ b/dimos/models/segmentation/test_yoloe.py @@ -16,11 +16,11 @@ from typing import Any from unittest.mock import MagicMock -from dimos.messages.sensor_msgs.msg import Image -from dimos.messages.std_msgs.msg import Header import numpy as np from ultralytics.models.yolo.yoloe import YOLOEVPSegPredictor +from dimos.messages.sensor_msgs.msg import Image +from dimos.messages.std_msgs.msg import Header from dimos.models.segmentation.yoloe import YoloeBoxSegmenter from dimos.msgs.image import image_from_array, image_to_bgr from dimos.msgs.time import time_from_seconds, to_seconds diff --git a/dimos/models/segmentation/yoloe.py b/dimos/models/segmentation/yoloe.py index d19ecc15701..fd075227d27 100644 --- a/dimos/models/segmentation/yoloe.py +++ b/dimos/models/segmentation/yoloe.py @@ -19,11 +19,11 @@ from dataclasses import replace from typing import Protocol -from dimos.messages.sensor_msgs.msg import Image import numpy as np from numpy.typing import NDArray from ultralytics.models.yolo.yoloe import YOLOEVPSegPredictor +from dimos.messages.sensor_msgs.msg import Image from dimos.msgs.image import image_to_bgr from dimos.perception.detection.detectors.yoloe import Yoloe2DDetector, YoloePromptMode from dimos.perception.detection.type.detection2d.bbox import Bbox, Detection2DBBox diff --git a/examples/external-messages/demo_modules.py b/examples/external-messages/demo_modules.py index 43ef27967b9..dab0a5a848b 100644 --- a/examples/external-messages/demo_modules.py +++ b/examples/external-messages/demo_modules.py @@ -30,7 +30,6 @@ from reactivex.disposable import Disposable import zenoh -from dimos.core.cdr_transport import CdrTransport from dimos.core.coordination.blueprints import autoconnect from dimos.core.coordination.module_coordinator import ModuleCoordinator from dimos.core.core import rpc @@ -112,40 +111,22 @@ def run(cpp: Path, rust: Path) -> None: gossip=False, connect_timeout=10, ) - blueprint = ( - autoconnect( - Probe.blueprint(), - CppRelay.blueprint( - executable=str(cpp.resolve()), stdin_config=True, session=session - ), - RustRelay.blueprint( - executable=str(rust.resolve()), stdin_config=True, session=session - ), - ) - .transports( - { - (name, DeviceReading): CdrTransport.spec( - "example/" + name, - DeviceReading, - backend="zenoh", - **session.model_dump(), - ) - for name in ("raw", "cpp", "done") - } - ) - .global_config( - viewer="none", - n_workers=3, - transport="zenoh", - robot_ip=None, - robot_ips="", - zenoh_connect=endpoint, - zenoh_mode="client", - zenoh_multicast=False, - zenoh_gossip=False, - zenoh_scouting=False, - zenoh_connect_timeout=10, - ) + blueprint = autoconnect( + Probe.blueprint(), + CppRelay.blueprint(executable=str(cpp.resolve()), stdin_config=True, session=session), + RustRelay.blueprint(executable=str(rust.resolve()), stdin_config=True, session=session), + ).global_config( + viewer="none", + n_workers=3, + transport="zenoh", + robot_ip=None, + robot_ips="", + zenoh_connect=endpoint, + zenoh_mode="client", + zenoh_multicast=False, + zenoh_gossip=False, + zenoh_scouting=False, + zenoh_connect_timeout=10, ) coordinator = ModuleCoordinator.build(blueprint) resources.callback(coordinator.stop)