feat(sim): dimsim as connection (#1735)

Co-authored-by: Viswajit Nair <viswajitnair@gmail.com>

Author: paul-nechifor

.github/workflows/ci.yml

@@ -76,9 +76,6 @@ jobs:
           - os: ubuntu
             pyver: "3.14t"
             experimental: false
-          - os: ubuntu
-            pyver: "3.15"
-            experimental: true
       fail-fast: true
     runs-on: ${{ matrix.os }}-latest
     continue-on-error: ${{ matrix.experimental }}
@@ -112,13 +109,13 @@ jobs:
         if: ${{ endsWith(matrix.pyver, 't') }}
         run: echo "PYTHON_GIL=0" >> $GITHUB_ENV
       - name: Run tests
-        run: uv run pytest --numprocesses=3 --cov=dimos/ --junitxml=junit.xml -m 'not (tool or self_hosted or mujoco)'
+        run: uv run pytest --numprocesses=3 --cov=dimos/ --junitxml=junit.xml -m 'not (tool or self_hosted or mujoco or dimsim)'
       - name: Re-run the failing tests with maximum verbosity
         if: failure()
         env:
           COLOR: yes
         run: >-  # `exit 1` makes sure that the job remains red with flaky runs
-          uv run pytest --no-cov -vvvvv --lf  -m 'not (tool or self_hosted or mujoco)' && exit 1
+          uv run pytest --no-cov -vvvvv --lf  -m 'not (tool or self_hosted or mujoco or dimsim)' && exit 1
         shell: bash
       - name: Turn coverage into xml
         run: uv run python -m coverage xml

dimos/agents/skills/person_follow.py

@@ -78,14 +78,18 @@ def __init__(self, **kwargs: Any) -> None:
         self._should_stop: Event = Event()
         self._lock = RLock()
 
-        # Use MuJoCo camera intrinsics in simulation mode
+        # Use simulator camera intrinsics in simulation mode
         camera_info = self.config.camera_info
-        if self.config.g.simulation:
+        if self.config.g.simulation == "mujoco":
             from dimos.robot.unitree.mujoco_connection import MujocoConnection
 
             camera_info = MujocoConnection.camera_info_static
+        elif self.config.g.simulation == "dimsim":
+            from dimos.robot.unitree.dimsim_connection import DimSimConnection
 
-        self._visual_servo = VisualServoing2D(camera_info, self.config.g.simulation)
+            camera_info = DimSimConnection.camera_info_static
+
+        self._visual_servo = VisualServoing2D(camera_info, bool(self.config.g.simulation))
         self._detection_navigation = DetectionNavigation(self.tf, camera_info)
 
     @rpc

dimos/conftest.py

@@ -20,19 +20,27 @@
 import tempfile
 import threading
 
-# Pick a per-xdist-worker bucket and pin env vars *before* any dimos
-# module is imported. ``LCMConfig`` captures ``LCM_DEFAULT_URL`` at import
-# time; ``GlobalConfig`` captures ``MCP_PORT``; ``run_registry`` captures
-# ``XDG_STATE_HOME``. ``LCM_DEFAULT_URL`` in particular has to be an env
-# var (not just a fixture) because subprocess workers spawned by
+# With pytest-xdist, pick a per-worker bucket and pin env vars *before*
+# any dimos module is imported, so parallel workers don't share LCM bus,
+# MCP port, or state directory. ``LCMConfig`` captures ``LCM_DEFAULT_URL``
+# at import time; ``GlobalConfig`` captures ``MCP_PORT``; ``run_registry``
+# captures ``XDG_STATE_HOME``. ``LCM_DEFAULT_URL`` in particular has to be
+# an env var (not just a fixture) because subprocess workers spawned by
 # ``ModuleCoordinator`` create their own ``LCMConfig`` / ``LCMRPC``
 # instances internally and can't receive a fixture value — they inherit
 # our env at fork time.
-_worker = os.environ.get("PYTEST_XDIST_WORKER", "main")
-_BUCKET = int.from_bytes(hashlib.blake2b(_worker.encode(), digest_size=2).digest(), "big") % 5000
-os.environ["LCM_DEFAULT_URL"] = f"udpm://239.255.76.67:{7700 + _BUCKET}?ttl=0"
-os.environ["MCP_PORT"] = str(20000 + _BUCKET)
-os.environ["XDG_STATE_HOME"] = tempfile.mkdtemp(prefix=f"dimos-test-state-{_worker}-")
+#
+# Single-worker runs (no xdist) keep the defaults, so external processes
+# with hard-coded ports (e.g. the dimsim Deno bridge, which binds to LCM
+# 7667) can still talk to the test bus.
+_worker = os.environ.get("PYTEST_XDIST_WORKER")
+if _worker:
+    _BUCKET = (
+        int.from_bytes(hashlib.blake2b(_worker.encode(), digest_size=2).digest(), "big") % 5000
+    )
+    os.environ["LCM_DEFAULT_URL"] = f"udpm://239.255.76.67:{7700 + _BUCKET}?ttl=0"
+    os.environ["MCP_PORT"] = str(20000 + _BUCKET)
+    os.environ["XDG_STATE_HOME"] = tempfile.mkdtemp(prefix=f"dimos-test-state-{_worker}-")
 
 # Raise the open-file limit. Each LCM transport opens at least one
 # multicast socket; with pytest-xdist workers running many in parallel,
@@ -77,6 +85,7 @@ def pytest_configure(config):
         "self_hosted: tests that need the self-hosted runner (LFS, ROS, CUDA, etc.)",
     )
     config.addinivalue_line("markers", "mujoco: tests which open mujoco")
+    config.addinivalue_line("markers", "dimsim: tests which require dimsim")
     config.addinivalue_line("markers", "skipif_in_ci: skip when CI env var is set")
     config.addinivalue_line("markers", "skipif_no_openai: skip when OPENAI_API_KEY is not set")
     config.addinivalue_line("markers", "skipif_no_alibaba: skip when ALIBABA_API_KEY is not set")
@@ -90,20 +99,20 @@ def pytest_configure(config):
 
 @pytest.fixture(scope="session")
 def mcp_port() -> int:
-    """The MCP server port pinned for this xdist worker."""
-    return 20000 + _BUCKET
+    """The MCP server port pinned for this xdist worker (or the default)."""
+    return int(os.environ.get("MCP_PORT", "9990"))
 
 
 @pytest.fixture(scope="session")
 def mcp_url(mcp_port: int) -> str:
-    """The MCP server URL pinned for this xdist worker."""
+    """The MCP server URL pinned for this xdist worker (or the default)."""
     return f"http://localhost:{mcp_port}/mcp"
 
 
 @pytest.fixture(scope="session")
 def lcm_url() -> str:
-    """The LCM bus URL pinned for this xdist worker."""
-    return f"udpm://239.255.76.67:{7700 + _BUCKET}?ttl=0"
+    """The LCM bus URL pinned for this xdist worker (or the default)."""
+    return os.environ.get("LCM_DEFAULT_URL", "udpm://239.255.76.67:7667?ttl=0")
 
 
 @pytest.hookimpl(tryfirst=True)

dimos/core/global_config.py

@@ -36,7 +36,7 @@ class GlobalConfig(BaseSettings):
     xarm7_ip: str | None = None
     xarm6_ip: str | None = None
     can_port: str | None = None
-    simulation: bool = False
+    simulation: str = ""
     replay: bool = False
     replay_db: str = "go2_short"
     new_memory: bool = False
@@ -65,6 +65,8 @@ class GlobalConfig(BaseSettings):
     obstacle_avoidance: bool = True
     detection_model: VlModelName = "moondream"
     listen_host: str = "127.0.0.1"
+    dimsim_scene: str = "apt"
+    dimsim_port: int = 8090
 
     model_config = SettingsConfigDict(
         env_file=".env",
@@ -84,7 +86,7 @@ def unitree_connection_type(self) -> str:
         if self.replay:
             return "replay"
         if self.simulation:
-            return "mujoco"
+            return self.simulation
         return "webrtc"
 
     @property

dimos/core/transport.py

@@ -133,10 +133,12 @@ def broadcast(self, _, msg) -> None:  # type: ignore[no-untyped-def]
 
         self.lcm.publish(self.topic, msg)
 
-    def subscribe(self, callback: Callable[[T], None], selfstream: In[T] = None) -> None:  # type: ignore[assignment, override]
+    def subscribe(
+        self, callback: Callable[[T], Any], selfstream: Stream[T] | None = None
+    ) -> Callable[[], None]:
         if not self._started:
             self.start()
-        return self.lcm.subscribe(self.topic, lambda msg, topic: callback(msg))  # type: ignore[arg-type, return-value]
+        return self.lcm.subscribe(self.topic, lambda msg, topic: callback(msg))  # type: ignore[arg-type]
 
 
 class JpegLcmTransport(LCMTransport):  # type: ignore[type-arg]

dimos/e2e_tests/conftest.py

@@ -20,6 +20,7 @@
 
 from dimos.core.transport import pLCMTransport
 from dimos.e2e_tests.conf_types import StartPersonTrack
+from dimos.e2e_tests.dim_sim_client import DimSimClient
 from dimos.e2e_tests.dimos_cli_call import DimosCliCall
 from dimos.e2e_tests.lcm_spy import LcmSpy
 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
@@ -66,12 +67,17 @@ def fun(*, points: list[tuple[float, float, float]], fail_message: str) -> None:
 
 
 @pytest.fixture
-def start_blueprint() -> Iterator[Callable[..., DimosCliCall]]:
+def start_blueprint(mcp_port: int) -> Iterator[Callable[..., DimosCliCall]]:
     dimos_robot_call = DimosCliCall()
+    dimos_robot_call.mcp_port = mcp_port
 
-    def set_name_and_start(*demo_args: str, mcp_port: int | None = None) -> DimosCliCall:
+    def set_name_and_start(
+        *demo_args: str,
+        simulator: str | None = None,
+    ) -> DimosCliCall:
         dimos_robot_call.demo_args = list(demo_args)
-        dimos_robot_call.mcp_port = mcp_port
+        if simulator is not None:
+            dimos_robot_call.simulator = simulator
         dimos_robot_call.start()
         return dimos_robot_call
 
@@ -153,3 +159,93 @@ def explore() -> None:
         direct_cmd_vel_explorer.follow_points(points)
 
     return explore
+
+
+@pytest.fixture
+def dim_sim():
+    client = DimSimClient()
+    client.start()
+    yield client
+    client.stop()
+
+
+@pytest.fixture
+def spawn_wall_on_pose(lcm_spy: LcmSpy, dim_sim: DimSimClient):
+    """Spawn a dim_sim wall when the robot's /odom comes within `threshold` metres of `point`."""
+    odom_topic = "/odom#geometry_msgs.PoseStamped"
+    stop_event = threading.Event()
+    workers: list[threading.Thread] = []
+    errors: list[BaseException] = []
+
+    def spawn(*, point, threshold, wall) -> None:
+        px, py = point
+        threshold_sq = threshold * threshold
+        triggered = threading.Event()
+
+        def on_odom(data):
+            if triggered.is_set():
+                return
+            pose = PoseStamped.lcm_decode(data)
+            dx = pose.x - px
+            dy = pose.y - py
+            if dx * dx + dy * dy < threshold_sq:
+                triggered.set()
+
+        def worker():
+            try:
+                with lcm_spy.topic_listener(odom_topic, on_odom):
+                    while not stop_event.is_set():
+                        if triggered.wait(timeout=0.1):
+                            dim_sim.add_wall(*wall)
+                            return
+            except BaseException as e:
+                errors.append(e)
+
+        thread = threading.Thread(target=worker, daemon=True)
+        thread.start()
+        workers.append(thread)
+
+    yield spawn
+
+    stop_event.set()
+    for thread in workers:
+        thread.join(timeout=5.0)
+    if errors:
+        raise errors[0]
+
+
+@pytest.fixture
+def explore_house(
+    direct_cmd_vel_explorer: DirectCmdVelExplorer,
+) -> Callable[[], None]:
+    points = [
+        (3.881, 4.803),
+        (4.160, 1.615),
+        (1.596, 1.505),
+        (1.649, 0.137),
+        (-3.644, -0.064),
+        (-3.759, -2.661),
+        (-4.186, -4.830),
+        (-3.759, -2.661),
+        (-1.070, -3.285),
+        (-2.504, -2.452),
+        (-2.647, 5.243),
+        (-3.663, 3.591),
+        (-1.178, 1.974),
+        (-2.416, 2.629),
+        (-2.581, 0.164),
+        (1.834, 0.072),
+        (3.010, -3.883),
+        (1.756, -3.742),
+        (6.336, -4.077),
+        (8.264, -5.119),
+        (6.258, -0.964),
+        (6.453, 5.327),
+    ]
+
+    direct_cmd_vel_explorer.linear_speed = 0.5
+
+    def explore() -> None:
+        direct_cmd_vel_explorer.follow_points(points)
+
+    return explore

dimos/e2e_tests/dim_sim_client.py

@@ -0,0 +1,56 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from dimos.core.transport import LCMTransport
+from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
+from dimos.simulation.dimsim.scene_client import SceneClient
+
+
+class DimSimClient:
+    _client: SceneClient | None = None
+
+    def __init__(self) -> None:
+        self._client = None
+        self._goal_request: LCMTransport[PoseStamped] = LCMTransport("/goal_request", PoseStamped)
+
+    def start(self) -> None:
+        # self.client should be started lazily to avoid starting the dimsim
+        # process before pytest fixtures are ready
+        self._goal_request.start()
+
+    def stop(self) -> None:
+        self.client.stop()
+        self._goal_request.stop()
+
+    @property
+    def client(self) -> SceneClient:
+        if self._client is None:
+            self._client = SceneClient()
+            self._client.start()
+        return self._client
+
+    def set_agent_position(self, x: float, y: float, z: float = 0.52) -> None:
+        self.client.set_agent_position(y, z, x)
+
+    def add_wall(self, x1: float, y1: float, x2: float, y2: float) -> None:
+        self.client.add_wall(y1, x1, y2, x2)
+
+    def publish_goal(self, x: float, y: float) -> None:
+        self._goal_request.publish(
+            PoseStamped(
+                position=(x, y, 0),
+                orientation=(0, 0, 0, 1),
+                frame_id="world",
+            )
+        )

dimos/e2e_tests/dimos_cli_call.py

@@ -22,6 +22,7 @@ class DimosCliCall:
     process: subprocess.Popen[bytes] | None
     demo_args: list[str] | None = None
     mcp_port: int | None = None
+    simulator: str = "mujoco"
 
     def __init__(self) -> None:
         self.process = None
@@ -48,7 +49,14 @@ def start(self) -> None:
             ]
 
         self.process = subprocess.Popen(
-            ["dimos", *global_overrides, "--simulation", *args, *blueprint_overrides],
+            [
+                "dimos",
+                *global_overrides,
+                "--simulation",
+                self.simulator,
+                *args,
+                *blueprint_overrides,
+            ],
             start_new_session=True,
         )
 

dimos/e2e_tests/lcm_spy.py

@@ -73,25 +73,21 @@ def save_topic(self, topic: str) -> None:
         with self._saved_topics_lock:
             self._saved_topics.add(topic)
 
-    def register_topic_listener(self, topic: str, listener: Callable[[bytes], None]) -> int:
+    def register_topic_listener(self, topic: str, listener: Callable[[bytes], None]) -> None:
         with self._topic_listeners_lock:
-            listeners = self._topic_listeners.setdefault(topic, [])
-            listener_index = len(listeners)
-            listeners.append(listener)
-            return listener_index
+            self._topic_listeners.setdefault(topic, []).append(listener)
 
-    def unregister_topic_listener(self, topic: str, listener_index: int) -> None:
+    def unregister_topic_listener(self, topic: str, listener: Callable[[bytes], None]) -> None:
         with self._topic_listeners_lock:
-            listeners = self._topic_listeners[topic]
-            listeners.pop(listener_index)
+            self._topic_listeners[topic].remove(listener)
 
     @contextmanager
     def topic_listener(self, topic: str, listener: Callable[[bytes], None]) -> Iterator[None]:
-        listener_index = self.register_topic_listener(topic, listener)
+        self.register_topic_listener(topic, listener)
         try:
             yield
         finally:
-            self.unregister_topic_listener(topic, listener_index)
+            self.unregister_topic_listener(topic, listener)
 
     def wait_until(
         self,