navigation tuning

dimos/core/global_config.py

@@ -68,5 +68,5 @@ def mujoco_start_pos_float(self) -> tuple[float, float]:
     @cached_property
     def mujoco_camera_position_float(self) -> tuple[float, ...]:
         if self.mujoco_camera_position is None:
-            return [-0.906, 0.008, 1.101, 4.931, 89.749, -46.378]
+            return (-0.906, 0.008, 1.101, 4.931, 89.749, -46.378)
         return tuple(_get_all_numbers(self.mujoco_camera_position))

dimos/mapping/costmapper.py

@@ -0,0 +1,67 @@
+# Copyright 2025 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 dataclasses import asdict, dataclass, field
+
+from reactivex import operators as ops
+
+from dimos.core import In, Module, Out, rpc
+from dimos.core.module import ModuleConfig
+from dimos.mapping.pointclouds.occupancy import (
+    OCCUPANCY_ALGOS,
+    HeightCostConfig,
+    OccupancyConfig,
+)
+from dimos.msgs.nav_msgs import OccupancyGrid
+from dimos.robot.unitree_webrtc.type.lidar import LidarMessage
+from dimos.utils.reactive import backpressure
+
+
+@dataclass
+class Config(ModuleConfig):
+    algo: str = "height_cost"
+    config: OccupancyConfig = field(default_factory=HeightCostConfig)
+
+
+class CostMapper(Module):
+    default_config = Config
+    config: Config
+
+    global_map: In[LidarMessage]
+    global_costmap: Out[OccupancyGrid]
+
+    @rpc
+    def start(self) -> None:
+        super().start()
+
+        self._disposables.add(
+            backpressure(
+                self.global_map.observable()  # type: ignore[no-untyped-call]
+            )
+            .pipe(ops.map(self._calculate_costmap))
+            .subscribe(
+                self.global_costmap.publish,
+            )
+        )
+
+    @rpc
+    def stop(self) -> None:
+        super().stop()
+
+    def _calculate_costmap(self, msg: LidarMessage) -> OccupancyGrid:
+        fn = OCCUPANCY_ALGOS[self.config.algo]
+        return fn(msg, **asdict(self.config.config))
+
+
+cost_mapper = CostMapper.blueprint