Topic List

dimos/core/baked_host.py

@@ -24,7 +24,7 @@
 
 from pydantic import Field, create_model
 
-from dimos.core.native_module import NativeModule, NativeModuleConfig
+from dimos.core.native_module import NativeModule, NativeModuleConfig, TopicsMap, TopicValue
 from dimos.core.stream import IO, In, Out
 
 
@@ -83,23 +83,23 @@ class BakedHost(NativeModule):
     _members: Mapping[str, type[NativeModule]] = {}
     _remaps: Mapping[tuple[str, str], str] = {}
 
-    def _member_topics(self, instance: str, topics: Mapping[str, str]) -> dict[str, str]:
+    def _member_topics(self, instance: str, topics: TopicsMap) -> dict[str, TopicValue]:
         member = self._members[instance]
-        resolved = {}
+        resolved: dict[str, TopicValue] = {}
         for port in _member_ports(member):
             name = self._remaps.get((instance, port), port)
             if name in topics:
                 resolved[port] = topics[name]
         return resolved
 
-    def _argv(self, topics: dict[str, str]) -> list[str]:
+    def _argv(self, topics: TopicsMap) -> list[str]:
         """A baked host takes its whole wiring on the launch line, so no flags.
 
         The binary rejects an unknown argument rather than ignoring it.
         """
         return [self.config.executable, *self.config.extra_args]
 
-    def _stdin_blob(self, topics: dict[str, str]) -> bytes:
+    def _stdin_blob(self, topics: TopicsMap) -> bytes:
         sections: dict[str, object] = {}
         for instance in self._members:
             member_config = getattr(self.config, f"{instance}_config")

dimos/core/coordination/blueprints.py

@@ -27,7 +27,7 @@
 if TYPE_CHECKING:
     from dimos.protocol.service.system_configurator.base import SystemConfigurator
 
-from dimos.core.module import ModuleBase, is_module_type
+from dimos.core.module import ModuleBase, TopicFunnel, is_module_type
 from dimos.core.stream import IO, In, Out, Transport
 from dimos.spec.utils import Spec, is_spec
 from dimos.utils.logging_config import setup_logger
@@ -231,13 +231,39 @@ def global_config(self, **kwargs: Any) -> "Blueprint":
     def remappings(
         self,
         remappings: Sequence[
-            tuple[type[ModuleBase] | str, str, str | type[ModuleBase] | type[Spec]]
+            tuple[
+                type[ModuleBase] | str,
+                str,
+                str
+                | Sequence[str]
+                | Mapping[str, Mapping[str, Any]]
+                | type[ModuleBase]
+                | type[Spec],
+            ]
         ],
     ) -> "Blueprint":
+        """Rewrite what a module's streams and module refs connect to.
+
+        A string (or Spec/Module class) renames one stream (or redirects one ref).
+        A list of names, or a dict of name -> arbitrary info, instead fans several
+        same-typed streams into that one port, so its handler serves all of them:
+
+            .remappings([
+                (Fuser, "scan", ["front_lidar", "rear_lidar"]),
+                (Fuser, "image", {"depth": {"meters_per_unit": 0.001}, "color": {}}),
+            ])
+        """
         remappings_dict = dict(self.remapping_map)
+        atoms = list(self.blueprints)
         for module, old, new in remappings:
-            remappings_dict[(self._instance_key(module), old)] = new
-        return replace(self, remapping_map=MappingProxyType(remappings_dict))
+            instance_key = self._instance_key(module)
+            if isinstance(new, (str, type)):
+                remappings_dict[instance_key, old] = new
+            else:
+                atoms = _fan_in(atoms, instance_key, old, TopicFunnel.of(new))
+        return replace(
+            self, blueprints=tuple(atoms), remapping_map=MappingProxyType(remappings_dict)
+        )
 
     def _instance_key(self, module: type[ModuleBase] | str) -> str:
         if isinstance(module, str):
@@ -381,6 +407,54 @@ def autoconnect(*blueprints: Blueprint) -> Blueprint:
     )
 
 
+def _fan_in(
+    atoms: list[BlueprintAtom], instance_key: str, port: str, funnel: TopicFunnel
+) -> list[BlueprintAtom]:
+    """Point one declared input at several streams instead of one.
+
+    The port stops being a stream of its own; each entry takes its place as an
+    `In` of the same type, so autoconnect, `.namespace()` and transport pins
+    treat the entries as ordinary streams. The module keeps the port as the
+    single place messages from all of them arrive.
+    """
+    updated = []
+    matched = False
+    for atom in atoms:
+        declared = next(
+            (
+                stream
+                for stream in atom.streams
+                if atom.name == instance_key
+                and stream.name == port
+                and stream.direction in ("in", "inout")
+            ),
+            None,
+        )
+        if declared is None:
+            updated.append(atom)
+            continue
+        matched = True
+        entries = tuple(
+            StreamRef(name=name, type=declared.type, direction="in") for name in funnel.names
+        )
+        updated.append(
+            replace(
+                atom,
+                kwargs={
+                    **atom.kwargs,
+                    "topic_funnels": {**atom.kwargs.get("topic_funnels", {}), port: funnel},
+                },
+                streams=tuple(s for s in atom.streams if s is not declared) + entries,
+            )
+        )
+    if not matched:
+        raise ValueError(
+            f"cannot fan {port!r} in: {instance_key} declares no such In/IO stream "
+            "in this blueprint (already fanned in, or a typo?)"
+        )
+    return updated
+
+
 def _eliminate_duplicates(blueprints: list[BlueprintAtom]) -> list[BlueprintAtom]:
     # The duplicates are eliminated in reverse so that newer blueprints override older ones.
     seen = set()