Topic List

dimos/core/native_module.py

@@ -54,13 +54,13 @@ class MyCppModule(NativeModule):
 import time
 from typing import IO, Any
 
-from pydantic import Field, model_validator
+from pydantic import BaseModel, Field, model_validator
 
 from dimos.constants import DEFAULT_THREAD_JOIN_TIMEOUT
 from dimos.core.core import rpc
 from dimos.core.global_config import global_config
 from dimos.core.module import Module, ModuleConfig
-from dimos.core.transport_factory import session_config
+from dimos.core.transport_factory import channel_for, session_config
 from dimos.protocol.service.spec import SessionConfig
 from dimos.utils.logging_config import setup_logger
 
@@ -114,6 +114,18 @@ class LogFormat(enum.Enum):
 }
 
 
+class TopicGroup(BaseModel):
+    """Several same-typed channels a native port fans into one handler.
+
+    The module declares no In stream for these, so nothing here participates in
+    blueprint autoconnection: the names are resolved to wire channels and handed
+    to the native process, which subscribes them itself.
+    """
+
+    names: list[str]
+    msg_type: type[Any] | None = None
+
+
 class NativeModuleConfig(ModuleConfig):
     """Configuration for a native subprocess module."""
 
@@ -131,6 +143,9 @@ class NativeModuleConfig(ModuleConfig):
 
     stdin_config: bool = False
 
+    # Port name -> the group of channels that port's single handler receives.
+    topic_groups: dict[str, TopicGroup] = Field(default_factory=dict)
+
     cli_exclude: frozenset[str] = frozenset()
     cli_name_override: dict[str, str] = Field(default_factory=dict)
 
@@ -201,6 +216,9 @@ class NativeModule(Module):
     ``DIMOS_TRANSPORT`` env var. With ``stdin_config``, the topics, config,
     publisher QoS and session settings also arrive as one JSON line on stdin.
 
+    A port named in ``topic_groups`` gets a list of channels instead of one, so
+    several same-typed topics reach a single native handler.
+
     The native process should parse whichever it uses and pub/sub on the given
     topics directly. On ``stop()``, the process receives SIGTERM.
     """
@@ -258,16 +276,17 @@ def _session(self) -> SessionConfig:
         # A blueprint builds its config before global config is settled.
         return pinned.rebased()
 
-    def _argv(self, topics: dict[str, str]) -> list[str]:
+    def _argv(self, topics: dict[str, str | list[str]]) -> list[str]:
         """The command line the native process is spawned with."""
         cmd = [self.config.executable]
         for name, topic_str in topics.items():
-            cmd.extend([f"--{name}", topic_str])
+            joined = ",".join(topic_str) if isinstance(topic_str, list) else topic_str
+            cmd.extend([f"--{name}", joined])
         cmd.extend(self.config.to_cli_args())
         cmd.extend(self.config.extra_args)
         return cmd
 
-    def _stdin_blob(self, topics: dict[str, str]) -> bytes:
+    def _stdin_blob(self, topics: dict[str, str | list[str]]) -> bytes:
         """The JSON line the native process reads its launch from."""
         config_dict = self.config.to_config_dict()
         blob: dict[str, Any] = {
@@ -513,8 +532,8 @@ def _maybe_build(self) -> None:
             duration_sec=round(build_elapsed, 3),
         )
 
-    def _collect_topics(self) -> dict[str, str]:
-        topics: dict[str, str] = {}
+    def _collect_topics(self) -> dict[str, str | list[str]]:
+        topics: dict[str, str | list[str]] = {}
         for name in list(self.inputs) + list(self.outputs) + list(self.ios):
             stream = getattr(self, name, None)
             if stream is None:
@@ -525,6 +544,13 @@ def _collect_topics(self) -> dict[str, str]:
             channel = getattr(transport, "channel", None)
             if channel is not None:
                 topics[name] = channel
+        for port, group in self.config.topic_groups.items():
+            if port in topics:
+                raise ValueError(
+                    f"[{self._module_label}] topic group {port!r} collides with the "
+                    "port of the same name declared as a stream"
+                )
+            topics[port] = [channel_for(n, group.msg_type) for n in group.names]
         return topics
 
     def _collect_output_qos(self) -> dict[str, dict[str, str]]:

dimos/core/test_native_module.py

@@ -36,7 +36,7 @@
 from dimos.core.core import rpc
 from dimos.core.global_config import GlobalConfig, TransportBackend
 from dimos.core.module import Module
-from dimos.core.native_module import LogFormat, NativeModule, NativeModuleConfig
+from dimos.core.native_module import LogFormat, NativeModule, NativeModuleConfig, TopicGroup
 from dimos.core.stream import IO, In, Out
 from dimos.core.transport import LCMTransport, ZenohTransport
 from dimos.core.transport_factory import make_transport, transport_topic
@@ -211,6 +211,53 @@ def test_tf_topic_comes_from_the_declared_port_only() -> None:
             transport.stop()
 
 
+def test_a_topic_group_reaches_the_native_process_as_a_list(monkeypatch) -> None:
+    """A group's channels are resolved without the module declaring a stream each."""
+    monkeypatch.setattr(native_module_mod.global_config, "transport", "lcm")
+    module = StubNativeModule(
+        executable=_ECHO,
+        topic_groups={"cams": TopicGroup(names=["/cam0/imu", "/cam1/imu"], msg_type=Imu)},
+    )
+    try:
+        topics = module._collect_topics()
+        assert topics["cams"] == ["/cam0/imu#sensor_msgs.Imu", "/cam1/imu#sensor_msgs.Imu"]
+        assert module._argv(topics)[1:3] == [
+            "--cams",
+            "/cam0/imu#sensor_msgs.Imu,/cam1/imu#sensor_msgs.Imu",
+        ]
+    finally:
+        module.stop()
+
+
+def test_a_topic_group_cannot_shadow_a_declared_port() -> None:
+    module = StubNativeModule(
+        executable=_ECHO,
+        topic_groups={"cmd_vel": TopicGroup(names=["/other"], msg_type=Twist)},
+    )
+    transport = LCMTransport("/cmd_vel", Twist)
+    try:
+        module.set_transport("cmd_vel", transport)
+        with pytest.raises(ValueError, match="collides"):
+            module._collect_topics()
+    finally:
+        module.stop()
+        with contextlib.suppress(Exception):
+            transport.stop()
+
+
+def test_a_topic_group_is_not_a_native_config_field() -> None:
+    """The group is wiring, so it belongs in `topics`, not the config struct."""
+    module = StubNativeModule(
+        executable=_ECHO,
+        topic_groups={"cams": TopicGroup(names=["/cam0/imu"], msg_type=Imu)},
+    )
+    try:
+        assert "topic_groups" not in module.config.to_config_dict()
+        assert "--topic_groups" not in module._argv({})
+    finally:
+        module.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))

dimos/core/test_transport_factory.py

@@ -25,6 +25,7 @@
 )
 from dimos.core.transport_factory import (
     apply_transport_arg,
+    channel_for,
     default_zenoh_qos,
     make_transport,
     rpc_backend,
@@ -80,6 +81,14 @@ def test_make_transport_zenoh_pickled() -> None:
     assert t.topic == "dimos/human_input"
 
 
+@pytest.mark.parametrize("g", [LCM, ZENOH])
+@pytest.mark.parametrize(("name", "msg_type"), [("/camera/color", Image), ("/human_input", None)])
+def test_channel_for_matches_the_transport_it_skips_building(
+    g: GlobalConfig, name: str, msg_type: type | None
+) -> None:
+    assert channel_for(name, msg_type, g=g) == make_transport(name, msg_type, g=g).channel
+
+
 def test_default_zenoh_qos_high_rate_sensor_types_drop() -> None:
     assert default_zenoh_qos("/camera/color", Image) == QOS_LATEST_WINS
 

dimos/core/transport_factory.py

@@ -27,6 +27,7 @@
     pLCMTransport,
     pZenohTransport,
 )
+from dimos.protocol.pubsub.impl.lcmpubsub import Topic as LCMTopic
 from dimos.protocol.pubsub.impl.zenohpubsub import (
     QOS_LATEST_WINS,
     QOS_NEVER_DROP,
@@ -112,6 +113,22 @@ def make_transport(
     return LCMTransport(topic, msg_type)
 
 
+def channel_for(name: str, msg_type: type | None = None, *, g: GlobalConfig = global_config) -> str:
+    """The wire channel `make_transport` would land on, without building one.
+
+    Lets a caller name a channel it never subscribes to itself, such as the extra
+    topics a native module fans into one handler.
+    """
+    use_pickled = msg_type is None or getattr(msg_type, "lcm_encode", None) is None
+    topic = transport_topic(name, g)
+    if g.transport == "zenoh":
+        return str(ZenohTopic(topic, None if use_pickled else msg_type).key_expr)
+    if use_pickled:
+        return topic
+    assert msg_type is not None
+    return str(LCMTopic(topic, msg_type))
+
+
 def _transport_arg_error(argv: list[str], message: str) -> NoReturn:
     """Print an argparse-style CLI error for `--transport` and exit(2)."""
     prog = os.path.basename(argv[0]) if argv else "dimos"

native/rust/README.md

@@ -62,6 +62,7 @@ Every transport is compiled into the binary. `run_with_transport` opens the one
 - `#[derive(Module)]`: on the struct. Required.
 - `#[module(setup = fn, teardown = fn)]`: on the struct. Both optional. Names methods on `Self`. `setup` runs once before the input dispatch loop starts (use it to spawn background tasks or initialize resources); `teardown` runs once after the loop exits (use it for cleanup).
 - `#[input(decode = fn, handler = fn)]`: on a field of type `Input<T>`. `decode` is required; `handler` defaults to `handle_<field_name>`.
+- `#[input_group(decode = fn, handler = fn)]`: on a field of type `InputGroup<T>`, one port fed by several topics of the same message type (see [Topic groups](#topic-groups)). `decode` is required; `handler` defaults to `handle_<field_name>` and takes `(index, msg)`.
 - `#[output(encode = fn)]`: on a field of type `Output<T>`. `encode` is required.
 - `#[io(decode = fn, encode = fn, handler = fn)]`: on a field of type `Io<T>`, a port that publishes to and subscribes on one topic. `decode` and `encode` are required; `handler` defaults to `handle_<field_name>`. The transports deliver a message back to its own sender, so the handler also sees what the module publishes. Use `#[output]` instead when the module only publishes.
 - `#[config]`: on one field. The type must be defined with `#[native_config]` (see [Config](#config)). At most one per struct. If absent, `Config` defaults to `dimos_module::NoConfig`.
@@ -108,6 +109,34 @@ At runtime `run()` enforces the mapping on the Python payload: deserialization r
 
 Field name = port name. Ports map to topics via the stdin JSON; unmapped ports fall back to `/{port}`.
 
+## Topic groups
+
+A rig with N identical sensors would otherwise need N ports and N near-identical handlers. An `InputGroup<T>` is one port wired to a list of topics that all carry `T`, delivered to one handler in arrival order. Each message is tagged with the index of the topic it arrived on, so the handler can tell the sources apart.
+
+```rust
+#[derive(Module)]
+struct MultiCam {
+    #[input_group(decode = Image::decode)]
+    cameras: InputGroup<Image>,
+}
+
+impl MultiCam {
+    async fn handle_cameras(&mut self, index: usize, image: Image) {
+        let topic = self.cameras.topic(index);
+    }
+}
+```
+
+The Python wrapper supplies the topics with `topic_groups`, keyed by port name:
+
+```python
+MultiCam.blueprint(
+    topic_groups={"cameras": TopicGroup(names=["/cam0/color", "/cam1/color"], msg_type=Image)},
+)
+```
+
+Those channels are not declared as `In` streams, so they take no part in blueprint autoconnection: the names are resolved to wire channels and the native process subscribes them itself. On the launch line the port's value is an array rather than a string. A group configured with no topics still claims its port but never yields.
+
 ## Transforms
 
 A `#[tf]` field gives a module a view of the transform graph, the Rust counterpart to Python's `tf.get()` and `tf.publish()`. It subscribes to the `tf` topic (mapped like any other port, default `/tf`), buffers each `parent -> child` edge it sees, and answers queries by composing transforms along the shortest path through the graph.

native/rust/dimos-module-macros/src/lib.rs

@@ -17,7 +17,7 @@ use proc_macro2::TokenStream as TokenStream2;
 use quote::{format_ident, quote};
 use syn::{parse_macro_input, Data, DeriveInput, Field, Fields, Ident, LitStr, Path, Type};
 
-#[proc_macro_derive(Module, attributes(input, output, io, config, tf, module))]
+#[proc_macro_derive(Module, attributes(input, input_group, output, io, config, tf, module))]
 pub fn derive_module(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
     match expand(input) {
@@ -175,7 +175,7 @@ fn is_option(ty: &Type) -> bool {
 }
 
 const ONE_ATTR_ONLY: &str = "field has multiple module attributes; only one of #[input], \
-                             #[output], #[io], #[config], #[tf] is allowed";
+                             #[input_group], #[output], #[io], #[config], #[tf] is allowed";
 
 /// What a `#[tf]` port carries, for the Cargo.toml registry cross-check.
 const TF_PAYLOAD_TYPE: &str = "TFMessage";
@@ -185,6 +185,10 @@ enum FieldKind {
         decode: Path,
         handler: Ident,
     },
+    InputGroup {
+        decode: Path,
+        handler: Ident,
+    },
     Output {
         encode: Path,
     },
@@ -323,6 +327,9 @@ fn expand(input: DeriveInput) -> syn::Result<TokenStream2> {
             FieldKind::Input { decode, .. } => {
                 quote!(#name: builder.input(#name_str, #decode))
             }
+            FieldKind::InputGroup { decode, .. } => {
+                quote!(#name: builder.input_group(#name_str, #decode))
+            }
             FieldKind::Output { encode } => {
                 quote!(#name: builder.output(#name_str, #encode))
             }
@@ -336,26 +343,29 @@ fn expand(input: DeriveInput) -> syn::Result<TokenStream2> {
     });
 
     // Every port that receives messages gets an arm in the select! loop.
-    let handled_fields: Vec<(&Ident, &Ident)> = classified
+    let handle_arms: Vec<TokenStream2> = classified
         .iter()
-        .filter_map(|f| match &f.kind {
-            FieldKind::Input { handler, .. } | FieldKind::Io { handler, .. } => {
-                Some((f.name, handler))
+        .filter_map(|f| {
+            let name = f.name;
+            match &f.kind {
+                FieldKind::Input { handler, .. } | FieldKind::Io { handler, .. } => Some(quote!(
+                    ::core::option::Option::Some(msg) = self.#name.recv() => {
+                        self.#handler(msg).await
+                    }
+                )),
+                FieldKind::InputGroup { handler, .. } => Some(quote!(
+                    ::core::option::Option::Some((index, msg)) = self.#name.recv() => {
+                        self.#handler(index, msg).await
+                    }
+                )),
+                _ => None,
             }
-            _ => None,
         })
         .collect();
 
-    let handle_body = if handled_fields.is_empty() {
+    let handle_body = if handle_arms.is_empty() {
         quote!(::std::future::pending::<()>().await)
     } else {
-        let handle_arms = handled_fields.iter().map(|(name, handler)| {
-            quote!(
-                ::core::option::Option::Some(msg) = self.#name.recv() => {
-                    self.#handler(msg).await
-                }
-            )
-        });
         quote! {
             loop {
                 ::tokio::select! {
@@ -434,7 +444,7 @@ fn port_decls(classified: &[ClassifiedField], want_input: bool) -> Vec<PortDecl>
     classified
         .iter()
         .filter(|f| match f.kind {
-            FieldKind::Input { .. } => want_input,
+            FieldKind::Input { .. } | FieldKind::InputGroup { .. } => want_input,
             FieldKind::Output { .. } => !want_input,
             // A `#[tf]` field is a port like any other as far as the registry
             // goes: bake has to put the topic in the host's map or the module
@@ -577,6 +587,32 @@ fn classify_field(field: &Field, name: &Ident) -> syn::Result<FieldAttr> {
                 .ok_or_else(|| syn::Error::new_spanned(attr, "#[input] requires `decode = ...`"))?;
             let handler = handler.unwrap_or_else(|| format_ident!("handle_{}", name));
             found = Some(FieldKind::Input { decode, handler });
+        } else if path.is_ident("input_group") {
+            if found.is_some() {
+                return Err(syn::Error::new_spanned(attr, ONE_ATTR_ONLY));
+            }
+            let mut decode: Option<Path> = None;
+            let mut handler: Option<Ident> = None;
+            attr.parse_nested_meta(|meta| {
+                if meta.path.is_ident("decode") {
+                    decode = Some(meta.value()?.parse()?);
+                } else if meta.path.is_ident("handler") {
+                    handler = Some(meta.value()?.parse()?);
+                } else if meta.path.is_ident("msg") {
+                    msg = Some(meta.value()?.parse::<LitStr>()?.value());
+                } else {
+                    return Err(meta.error(
+                        "unrecognized #[input_group] argument; expected `decode = ...`, \
+                         `handler = ...` or `msg = ...`",
+                    ));
+                }
+                Ok(())
+            })?;
+            let decode = decode.ok_or_else(|| {
+                syn::Error::new_spanned(attr, "#[input_group] requires `decode = ...`")
+            })?;
+            let handler = handler.unwrap_or_else(|| format_ident!("handle_{}", name));
+            found = Some(FieldKind::InputGroup { decode, handler });
         } else if path.is_ident("output") {
             if found.is_some() {
                 return Err(syn::Error::new_spanned(attr, ONE_ATTR_ONLY));