LCM/Zenoh transport stub

Author: leshy

dimos/multiprocess/actors3/__init__.py

@@ -1,7 +1,7 @@
 import pytest
 from dask.distributed import Client, LocalCluster
 
-from dimos.multiprocess.actors3.base import In, LCMTransport, Out, RemoteOut, rpc
+from dimos.multiprocess.actors3.base import In, LCMTransport, Out, RemoteOut, ZenohTransport, rpc
 from dimos.multiprocess.actors3.module_dask import Module
 
 

dimos/multiprocess/actors3/base.py

@@ -60,7 +60,7 @@ def connect(self, selfstream: RemoteIn[T], otherstream: RemoteOut[T]) -> None:
     def broadcast(self, selfstream: Out[T], value: T): ...
 
 
-class LCMTransport(Transport[T]):
+class PubSubTransport(Transport[T]):
     topic: str
     type: type
 
@@ -69,9 +69,17 @@ def __init__(self, topic: str, type: type):
         self.type = type
 
     def __str__(self) -> str:
-        return colors.green("LCM(") + colors.blue(self.topic) + colors.green(")")
+        return (
+            colors.green(f"{self.__class__.__name__}(")
+            + colors.blue(self.topic)
+            + colors.green(")")
+        )
 
-    def broadcast(self, selfstream: Out[T], value: T): ...
+
+class LCMTransport(PubSubTransport[T]): ...
+
+
+class ZenohTransport(PubSubTransport[T]): ...
 
 
 class State(enum.Enum):
@@ -176,8 +184,6 @@ class In(Stream[T]):
     def __str__(self):
         return super().__str__() + ("" if not self.connection else f" <- {self.connection}")
 
-    def subscribe(self, cb): ...
-
     def __reduce__(self):  # noqa: D401
         if self.owner is None or not hasattr(self.owner, "ref"):
             raise ValueError("Cannot serialise Out without an owner ref")
@@ -187,10 +193,18 @@ def __reduce__(self):  # noqa: D401
     def state(self) -> State:  # noqa: D401
         return State.UNBOUND if self.owner is None else State.READY
 
+    # actual message passing implementation
+    def connect_remote(self):
+        self._transport.connect(self.connection)
+
+    def disconnect_remote(self):
+        self._transport.disconnect()
+
+    def subscribe(self, cb): ...
+
 
 class RemoteIn(RemoteStream[T]):
     def connect(self, other: RemoteOut[T]) -> None:
-        print("CONENCT REQU", other)
         return self.owner.connect_stream(self.name, other).result()
 
 

dimos/multiprocess/actors3/test_base.py

@@ -15,7 +15,16 @@
 import time
 from threading import Event, Thread
 
-from dimos.multiprocess.actors3 import In, LCMTransport, Module, Out, RemoteOut, dimos, rpc
+from dimos.multiprocess.actors3 import (
+    In,
+    LCMTransport,
+    Module,
+    Out,
+    RemoteOut,
+    ZenohTransport,
+    dimos,
+    rpc,
+)
 from dimos.robot.unitree_webrtc.type.lidar import LidarMessage
 from dimos.robot.unitree_webrtc.type.odometry import Odometry
 from dimos.types.vector import Vector
@@ -108,6 +117,7 @@ def test_deployment(dimos):
     target_stream = RemoteOut[Vector](Vector, "target")
 
     robot.lidar.transport = LCMTransport("/lidar", LidarMessage)
+    robot.odometry.transport = ZenohTransport("/odom", LidarMessage)
 
     print("\n")
     print("lidar stream", robot.lidar)