feat: rust transform support

dimos/core/native_module.py

@@ -12,12 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""NativeModule: blueprint-integrated wrapper for native (C/C++) executables.
+"""NativeModule: blueprint-integrated wrapper for native executables.
 
-A NativeModule is a thin Python Module subclass that declares In/Out ports
+A NativeModule is a thin Python Module subclass that declares In/Out/IO ports
 for blueprint wiring but delegates all real work to a managed subprocess.
-The native process receives its LCM topic names via CLI args and does
-pub/sub directly on the LCM multicast bus.
+The native process receives its topic names via CLI args, or as a JSON line on
+stdin when ``stdin_config`` is set, and does pub/sub on them directly.
 
 Example usage::
 
@@ -178,15 +178,19 @@ class NativeModule(Module):
     """
     Module that wraps a native executable as a managed subprocess.
 
-    Subclass this, declare In/Out ports, and annotate ``config`` with a
+    Subclass this, declare In/Out/IO ports, and annotate ``config`` with a
     :class:`NativeModuleConfig` subclass pointing at the executable.
 
     On ``start()``, the binary is launched with CLI args::
 
-        <executable> --<port_name> <lcm_topic_string> ... <extra_args>
+        <executable> --<port_name> <topic> ... --<config_field> <value> ... <extra_args>
 
-    The native process should parse these args and pub/sub on the given
-    LCM topics directly.  On ``stop()``, the process receives SIGTERM.
+    Each topic is the wire channel for that port on the transport named by the
+    ``DIMOS_TRANSPORT`` env var. With ``stdin_config``, those same topics plus
+    the config and any publisher QoS also arrive as one JSON line on stdin.
+
+    The native process should parse whichever it uses and pub/sub on the given
+    topics directly.  On ``stop()``, the process receives SIGTERM.
     """
 
     config: NativeModuleConfig
@@ -471,7 +475,7 @@ def _maybe_build(self) -> None:
 
     def _collect_topics(self) -> dict[str, str]:
         topics: dict[str, str] = {}
-        for name in list(self.inputs) + list(self.outputs):
+        for name in list(self.inputs) + list(self.outputs) + list(self.ios):
             stream = getattr(self, name, None)
             if stream is None:
                 continue
@@ -484,9 +488,9 @@ def _collect_topics(self) -> dict[str, str]:
         return topics
 
     def _collect_output_qos(self) -> dict[str, dict[str, str]]:
-        """Publisher QoS per output channel, keyed by channel."""
+        """Publisher QoS per published channel, keyed by channel."""
         qos_map: dict[str, dict[str, str]] = {}
-        for name in self.outputs:
+        for name in list(self.outputs) + list(self.ios):
             stream = getattr(self, name, None)
             if stream is None:
                 continue

dimos/core/test_native_module.py

@@ -18,6 +18,7 @@
 The echo script writes received CLI args to a temp file for assertions.
 """
 
+import contextlib
 from io import BytesIO
 import json
 from pathlib import Path
@@ -33,11 +34,13 @@
 from dimos.core.core import rpc
 from dimos.core.module import Module
 from dimos.core.native_module import LogFormat, NativeModule, NativeModuleConfig
-from dimos.core.stream import In, Out
-from dimos.core.transport import LCMTransport
+from dimos.core.stream import IO, In, Out
+from dimos.core.transport import LCMTransport, ZenohTransport
 from dimos.msgs.geometry_msgs.Twist import Twist
 from dimos.msgs.sensor_msgs.Imu import Imu
 from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
+from dimos.msgs.tf2_msgs.TFMessage import TFMessage
+from dimos.protocol.pubsub.impl.zenohpubsub import QOS_NEVER_DROP, Topic as ZenohTopic
 
 _ECHO = str(Path(__file__).parent / "demos" / "native_echo.py")
 
@@ -87,6 +90,12 @@ class StubNativeModule(NativeModule):
     cmd_vel: In[Twist]
 
 
+class StubIoModule(NativeModule):
+    config: StubNativeConfig
+    cmd_vel: In[Twist]
+    tf: IO[TFMessage]
+
+
 class StubConsumer(Module):
     pointcloud: In[PointCloud2]
     imu: In[Imu]
@@ -160,6 +169,54 @@ def test_manual(dimos_cluster: ModuleCoordinator, args_file: str) -> None:
     }
 
 
+def test_io_port_topic_reaches_the_native_process() -> None:
+    """An IO port is both a subscriber and a publisher, so it needs its topic."""
+    module = StubIoModule(executable=_ECHO)
+    transports = [LCMTransport("/cmd_vel", Twist), LCMTransport("/tf", TFMessage)]
+    try:
+        module.set_transport("cmd_vel", transports[0])
+        module.set_transport("tf", transports[1])
+
+        assert module._collect_topics() == {
+            "cmd_vel": "/cmd_vel#geometry_msgs.Twist",
+            "tf": "/tf#tf2_msgs.TFMessage",
+        }
+    finally:
+        module.stop()
+        for transport in transports:
+            with contextlib.suppress(Exception):
+                transport.stop()
+
+
+def test_tf_topic_comes_from_the_declared_port_only() -> None:
+    """No tf port declared means no tf topic, rather than a silently injected one."""
+    module = StubNativeModule(executable=_ECHO)
+    transport = LCMTransport("/cmd_vel", Twist)
+    try:
+        module.set_transport("cmd_vel", transport)
+
+        assert module._collect_topics() == {"cmd_vel": "/cmd_vel#geometry_msgs.Twist"}
+    finally:
+        module.stop()
+        with contextlib.suppress(Exception):
+            transport.stop()
+
+
+def test_io_port_publisher_qos_reaches_the_native_process() -> None:
+    module = StubIoModule(executable=_ECHO)
+    transport = ZenohTransport(ZenohTopic("/tf", TFMessage, qos=QOS_NEVER_DROP))
+    try:
+        module.set_transport("tf", transport)
+
+        assert module._collect_output_qos() == {
+            transport.channel: {"reliability": "reliable", "congestion_control": "block"},
+        }
+    finally:
+        module.stop()
+        with contextlib.suppress(Exception):
+            transport.stop()
+
+
 def test_autoconnect(args_file: str) -> None:
     """autoconnect passes correct topic args to the native subprocess."""
     blueprint = autoconnect(