Deduplicate Unitree connections/entrypoints.

dimos/agents/modules/base_agent.py

@@ -125,7 +125,7 @@ def start(self) -> None:
         # Connect response output
         if self.response_out:
             disposable = self.response_subject.subscribe(
-                lambda response: self.response_out.publish(response)  # type: ignore[no-untyped-call]
+                lambda response: self.response_out.publish(response)
             )
             self._module_disposables.append(disposable)
 

dimos/agents2/skills/demo_robot.py

@@ -31,7 +31,7 @@ def stop(self) -> None:
         super().stop()
 
     def _publish_gps_location(self) -> None:
-        self.gps_location.publish(LatLon(lat=37.78092426217621, lon=-122.40682866540769))  # type: ignore[no-untyped-call]
+        self.gps_location.publish(LatLon(lat=37.78092426217621, lon=-122.40682866540769))
 
 
 demo_robot = DemoRobot.blueprint

dimos/agents2/skills/gps_nav_skill.py

@@ -84,7 +84,7 @@ def set_gps_travel_points(self, *points: dict[str, float]) -> str:
         logger.info(f"Set travel points: {new_points}")
 
         if self.gps_goal._transport is not None:
-            self.gps_goal.publish(new_points)  # type: ignore[no-untyped-call]
+            self.gps_goal.publish(new_points)
 
         if self._set_gps_travel_goal_points:
             self._set_gps_travel_goal_points(new_points)

dimos/core/stream.py

@@ -28,6 +28,7 @@
 from reactivex.disposable import Disposable
 
 import dimos.core.colors as colors
+from dimos.utils.logging_config import setup_logger
 import dimos.utils.reactive as reactive
 from dimos.utils.reactive import backpressure
 
@@ -37,6 +38,9 @@
 T = TypeVar("T")
 
 
+logger = setup_logger(__file__)
+
+
 class ObservableMixin(Generic[T]):
     # subscribes and returns the first value it receives
     # might be nicer to write without rxpy but had this snippet ready
@@ -162,9 +166,10 @@ def __reduce__(self):  # type: ignore[no-untyped-def]
             ),
         )
 
-    def publish(self, msg):  # type: ignore[no-untyped-def]
+    def publish(self, msg) -> None:  # type: ignore[no-untyped-def]
         if not hasattr(self, "_transport") or self._transport is None:
-            raise Exception(f"{self} transport for stream is not specified,")
+            logger.warning(f"Trying to publish on Out {self} without a transport")
+            return
         self._transport.broadcast(self, msg)
 
 

dimos/core/testing.py

@@ -75,9 +75,9 @@ def odomloop(self) -> None:
                     return
                 print(odom)
                 odom.pubtime = time.perf_counter()
-                self.odometry.publish(odom)  # type: ignore[no-untyped-call]
+                self.odometry.publish(odom)
 
                 lidarmsg = next(lidariter)
                 lidarmsg.pubtime = time.perf_counter()  # type: ignore[union-attr]
-                self.lidar.publish(lidarmsg)  # type: ignore[no-untyped-call]
+                self.lidar.publish(lidarmsg)
                 time.sleep(0.1)

dimos/core/transport.py

@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import traceback
 from typing import Any, TypeVar
 
 import dimos.core.colors as colors
@@ -26,7 +25,7 @@
     TypeVar,
 )
 
-from dimos.core.stream import In, RemoteIn, Transport
+from dimos.core.stream import In, Transport
 from dimos.protocol.pubsub.jpeg_shm import JpegSharedMemory
 from dimos.protocol.pubsub.lcmpubsub import LCM, JpegLCM, PickleLCM, Topic as LCMTopic
 from dimos.protocol.pubsub.shmpubsub import PickleSharedMemory, SharedMemory

dimos/hardware/camera/module.py

@@ -50,7 +50,7 @@ class CameraModuleConfig(ModuleConfig):
 
 
 class CameraModule(Module, spec.Camera):
-    image: Out[Image] = None  # type: ignore[assignment]
+    color_image: Out[Image] = None  # type: ignore[assignment]
     camera_info: Out[CameraInfo] = None  # type: ignore[assignment]
 
     hardware: Callable[[], CameraHardware] | CameraHardware = None  # type: ignore[assignment, type-arg]
@@ -75,7 +75,7 @@ def start(self) -> str:  # type: ignore[return]
         self._disposables.add(self.camera_info_stream().subscribe(self.publish_info))
 
         stream = self.hardware.image_stream().pipe(sharpness_barrier(self.config.frequency))  # type: ignore[attr-defined, union-attr]
-        self._disposables.add(stream.subscribe(self.image.publish))
+        self._disposables.add(stream.subscribe(self.color_image.publish))
 
     @rpc
     def stop(self) -> None:
@@ -92,7 +92,7 @@ def video_stream(self) -> Image:  # type: ignore[misc]
         yield from iter(_queue.get, None)
 
     def publish_info(self, camera_info: CameraInfo) -> None:
-        self.camera_info.publish(camera_info)  # type: ignore[no-untyped-call]
+        self.camera_info.publish(camera_info)
 
         if self.config.transform is None:  # type: ignore[attr-defined]
             return

dimos/hardware/camera/zed/camera.py

@@ -738,7 +738,7 @@ def _publish_color_image(self, image: np.ndarray, header: Header) -> None:  # ty
                 ts=header.ts,
             )
 
-            self.color_image.publish(msg)  # type: ignore[no-untyped-call]
+            self.color_image.publish(msg)
 
         except Exception as e:
             logger.error(f"Error publishing color image: {e}")
@@ -753,7 +753,7 @@ def _publish_depth_image(self, depth: np.ndarray, header: Header) -> None:  # ty
                 frame_id=header.frame_id,
                 ts=header.ts,
             )
-            self.depth_image.publish(msg)  # type: ignore[no-untyped-call]
+            self.depth_image.publish(msg)
 
         except Exception as e:
             logger.error(f"Error publishing depth image: {e}")
@@ -831,7 +831,7 @@ def _publish_camera_info(self) -> None:
                 binning_y=0,
             )
 
-            self.camera_info.publish(msg)  # type: ignore[no-untyped-call]
+            self.camera_info.publish(msg)
 
         except Exception as e:
             logger.error(f"Error publishing camera info: {e}")
@@ -844,7 +844,7 @@ def _publish_pose(self, pose_data: dict[str, Any], header: Header) -> None:
 
             # Create PoseStamped message
             msg = PoseStamped(ts=header.ts, position=position, orientation=rotation)
-            self.pose.publish(msg)  # type: ignore[no-untyped-call]
+            self.pose.publish(msg)
 
             # Publish TF transform
             camera_tf = Transform(

dimos/hardware/fake_zed_module.py

@@ -214,7 +214,7 @@ def start(self) -> None:
         try:
             # Color image stream
             unsub = self._get_color_stream().subscribe(
-                lambda msg: self.color_image.publish(msg) if self._running else None  # type: ignore[no-untyped-call]
+                lambda msg: self.color_image.publish(msg) if self._running else None
             )
             self._disposables.add(unsub)
             logger.info("Started color image replay stream")
@@ -224,7 +224,7 @@ def start(self) -> None:
         try:
             # Depth image stream
             unsub = self._get_depth_stream().subscribe(
-                lambda msg: self.depth_image.publish(msg) if self._running else None  # type: ignore[no-untyped-call]
+                lambda msg: self.depth_image.publish(msg) if self._running else None
             )
             self._disposables.add(unsub)
             logger.info("Started depth image replay stream")
@@ -244,7 +244,7 @@ def start(self) -> None:
         try:
             # Camera info stream
             unsub = self._get_camera_info_stream().subscribe(
-                lambda msg: self.camera_info.publish(msg) if self._running else None  # type: ignore[no-untyped-call]
+                lambda msg: self.camera_info.publish(msg) if self._running else None
             )
             self._disposables.add(unsub)
             logger.info("Started camera info replay stream")
@@ -265,7 +265,7 @@ def stop(self) -> None:
     def _publish_pose(self, msg) -> None:  # type: ignore[no-untyped-def]
         """Publish pose and TF transform."""
         if msg:
-            self.pose.publish(msg)  # type: ignore[no-untyped-call]
+            self.pose.publish(msg)
 
             # Publish TF transform from world to camera
             import time

dimos/hardware/gstreamer_camera.py

@@ -286,7 +286,7 @@ def _on_new_sample(self, appsink):  # type: ignore[no-untyped-def]
 
             # Publish the image
             if self.video and self.running:
-                self.video.publish(image_msg)  # type: ignore[no-untyped-call]
+                self.video.publish(image_msg)
 
             # Log statistics periodically
             self.frame_count += 1

dimos/manipulation/visual_servoing/manipulation_module.py

@@ -904,7 +904,7 @@ def _publish_visualization(self, viz_image: np.ndarray) -> None:  # type: ignore
         try:
             viz_rgb = cv2.cvtColor(viz_image, cv2.COLOR_BGR2RGB)
             msg = Image.from_numpy(viz_rgb)
-            self.viz_image.publish(msg)  # type: ignore[no-untyped-call]
+            self.viz_image.publish(msg)
         except Exception as e:
             logger.error(f"Error publishing visualization: {e}")
 

dimos/models/manipulation/contact_graspnet_pytorch/inference.py

@@ -4,8 +4,12 @@
 
 from contact_graspnet_pytorch import config_utils  # type: ignore[import-not-found]
 from contact_graspnet_pytorch.checkpoints import CheckpointIO  # type: ignore[import-not-found]
-from contact_graspnet_pytorch.contact_grasp_estimator import GraspEstimator  # type: ignore[import-not-found]
-from contact_graspnet_pytorch.data import load_available_input_data  # type: ignore[import-not-found]
+from contact_graspnet_pytorch.contact_grasp_estimator import (  # type: ignore[import-not-found]
+    GraspEstimator,
+)
+from contact_graspnet_pytorch.data import (  # type: ignore[import-not-found]
+    load_available_input_data,
+)
 import numpy as np
 
 from dimos.utils.data import get_data