refactor: add generated occupancy value helpers with legacy primitive reuse

.github/workflows/message-codegen.yml

@@ -215,3 +215,9 @@ jobs:
           MYPYPATH="$PWD/build/message-source/python" .native-review/bin/python -m mypy --follow-imports=silent dimos/perception/depth2depth_cloud/module.py
           cargo test --locked --manifest-path dimos/perception/depth2depth_cloud/rust/Cargo.toml -j 2 --lib
           cargo test --locked --manifest-path native/rust/dimos-module/Cargo.toml -j 2 pointcloud::tests
+      - name: Verify generated occupancy values and retained numeric helpers
+        if: runner.os == 'Linux'
+        run: |
+          .native-review/bin/python -I -c 'import dimos.msgs.occupancy as values; import pytest; import sys; assert "site-packages" in str(values.__file__); raise SystemExit(pytest.main(sys.argv[1:]))' --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/msgs/test_occupancy.py"
+          .native-review/bin/python -I -c 'import dimos.msgs.nav_msgs.OccupancyGrid; import pytest; import sys; raise SystemExit(pytest.main(sys.argv[1:]))' --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/msgs/nav_msgs/test_OccupancyGrid.py" -k block_max_reduce
+          MYPYPATH="$PWD/build/message-source/python" .native-review/bin/python -m mypy --follow-imports=silent dimos/msgs/occupancy.py

dimos/models/segmentation/edge_tam.py

@@ -20,14 +20,14 @@
 import tempfile
 from typing import TYPE_CHECKING, Any, Protocol, TypedDict
 
-from dimos.messages.sensor_msgs.msg import Image
 from hydra.utils import instantiate
 import numpy as np
 from numpy.typing import NDArray
 from omegaconf import OmegaConf
 from PIL import Image as PILImage
 import torch
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import default_torch_device
 from dimos.msgs.image import image_to_bgr, image_to_jpeg
 from dimos.perception.detection.detectors.base import Detector

dimos/models/segmentation/test_yoloe.py

@@ -16,11 +16,11 @@
 from typing import Any
 from unittest.mock import MagicMock
 
-from dimos.messages.sensor_msgs.msg import Image
-from dimos.messages.std_msgs.msg import Header
 import numpy as np
 from ultralytics.models.yolo.yoloe import YOLOEVPSegPredictor
 
+from dimos.messages.sensor_msgs.msg import Image
+from dimos.messages.std_msgs.msg import Header
 from dimos.models.segmentation.yoloe import YoloeBoxSegmenter
 from dimos.msgs.image import image_from_array, image_to_bgr
 from dimos.msgs.time import time_from_seconds, to_seconds

dimos/models/segmentation/yoloe.py

@@ -19,11 +19,11 @@
 from dataclasses import replace
 from typing import Protocol
 
-from dimos.messages.sensor_msgs.msg import Image
 import numpy as np
 from numpy.typing import NDArray
 from ultralytics.models.yolo.yoloe import YOLOEVPSegPredictor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.msgs.image import image_to_bgr
 from dimos.perception.detection.detectors.yoloe import Yoloe2DDetector, YoloePromptMode
 from dimos.perception.detection.type.detection2d.bbox import Bbox, Detection2DBBox

dimos/msgs/nav_msgs/OccupancyGrid.py

@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-from enum import IntEnum
 from functools import lru_cache
 import time
 from typing import TYPE_CHECKING, BinaryIO
@@ -28,6 +27,7 @@
 
 from dimos.msgs.geometry_msgs.Pose import Pose
 from dimos.msgs.geometry_msgs.Vector3 import Vector3, VectorLike
+from dimos.msgs.occupancy import CostValues as CostValues, block_max_reduce as block_max_reduce
 from dimos.types.timestamped import Timestamped
 
 
@@ -97,44 +97,6 @@ def _build_occupancy_lut(
     from rerun._baseclasses import Archetype
 
 
-def block_max_reduce(cells: NDArray[np.int8], factor: int) -> NDArray[np.int8]:
-    """Coarsen an occupancy grid by taking the max of factor x factor blocks.
-
-    Block maximum (not mean) so coarsening never erases an obstacle; a block
-    is unknown (-1) only when every cell in it is unknown. Trailing remainder
-    rows/cols are trimmed, keeping row 0/col 0 - the origin corner - so the
-    grid origin stays valid. Grids thinner than `factor` pass through.
-    """
-    h, w = cells.shape[:2]
-    new_h, new_w = h // factor, w // factor
-    if new_h == 0 or new_w == 0:
-        return cells
-    trimmed = cells[: new_h * factor, : new_w * factor]
-    blocks = trimmed.reshape(new_h, factor, new_w, factor)
-    # Sink unknown below every known value for the max, then map it back.
-    as_int = blocks.astype(np.int16)
-    known = np.where(as_int < 0, -1000, as_int)
-    reduced = known.max(axis=(1, 3))
-    reduced[reduced == -1000] = -1
-    result: NDArray[np.int8] = reduced.astype(np.int8)
-    return result
-
-
-class CostValues(IntEnum):
-    """Standard cost values for occupancy grid cells.
-
-    These values follow the ROS nav_msgs/OccupancyGrid convention:
-    - 0: Free space
-    - 1-99: Occupied space with varying cost levels
-    - 100: Lethal obstacle (definitely occupied)
-    - -1: Unknown space
-    """
-
-    UNKNOWN = -1  # Unknown space
-    FREE = 0  # Free space
-    OCCUPIED = 100  # Occupied/lethal space
-
-
 class OccupancyGrid(Timestamped):
     """
     Convenience wrapper for nav_msgs/OccupancyGrid with numpy array support.

dimos/msgs/occupancy.py

@@ -0,0 +1,158 @@
+# 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.
+
+"""Array operations on generated occupancy grids."""
+
+from copy import deepcopy
+from enum import IntEnum
+import math
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+from numpy.typing import NDArray
+from PIL import Image
+
+from dimos.messages.geometry_msgs.msg import Point, Pose, Quaternion
+from dimos.messages.nav_msgs.msg import MapMetaData, OccupancyGrid
+from dimos.messages.std_msgs.msg import Header
+from dimos.msgs.geometry import pose_matrix
+
+
+class CostValues(IntEnum):
+    """Standard cost values for occupancy grid cells.
+
+    These values follow the ROS nav_msgs/OccupancyGrid convention:
+    - 0: Free space
+    - 1-99: Occupied space with varying cost levels
+    - 100: Lethal obstacle (definitely occupied)
+    - -1: Unknown space
+    """
+
+    UNKNOWN = -1  # Unknown space
+    FREE = 0  # Free space
+    OCCUPIED = 100  # Occupied/lethal space
+
+
+def occupancy_view(msg: OccupancyGrid) -> NDArray[np.int8]:
+    """Borrow a read-only row-major occupancy grid, retaining its owner."""
+    if len(msg.data) != msg.info.height * msg.info.width:
+        raise ValueError("occupancy data length does not match dimensions")
+    view = msg.data.view().reshape(msg.info.height, msg.info.width)
+    view.setflags(write=False)
+    return view
+
+
+def block_max_reduce(cells: NDArray[np.int8], factor: int) -> NDArray[np.int8]:
+    """Coarsen an occupancy grid by taking the max of factor x factor blocks.
+
+    Block maximum (not mean) so coarsening never erases an obstacle; a block
+    is unknown (-1) only when every cell in it is unknown. Trailing remainder
+    rows/cols are trimmed, keeping row 0/col 0 - the origin corner - so the
+    grid origin stays valid. Grids thinner than `factor` pass through.
+    """
+    h, w = cells.shape[:2]
+    new_h, new_w = h // factor, w // factor
+    if new_h == 0 or new_w == 0:
+        return cells
+    trimmed = cells[: new_h * factor, : new_w * factor]
+    blocks = trimmed.reshape(new_h, factor, new_w, factor)
+    # Sink unknown below every known value for the max, then map it back.
+    as_int = blocks.astype(np.int16)
+    known = np.where(as_int < 0, -1000, as_int)
+    reduced = known.max(axis=(1, 3))
+    reduced[reduced == -1000] = -1
+    result: NDArray[np.int8] = reduced.astype(np.int8)
+    return result
+
+
+def occupancy_extent(message: OccupancyGrid) -> tuple[float, float]:
+    """Return physical width and height, rejecting an invalid grid resolution."""
+    info = message.info
+    if not math.isfinite(info.resolution) or info.resolution <= 0:
+        raise ValueError("OccupancyGrid resolution must be finite and positive")
+    return info.width * info.resolution, info.height * info.resolution
+
+
+def world_to_grid(message: OccupancyGrid, point: Point) -> tuple[float, float]:
+    """Convert a world point to continuous grid coordinates, including origin rotation."""
+    occupancy_extent(message)
+    matrix = pose_matrix(message.info.origin)
+    coordinates = np.array([point.x, point.y, point.z])
+    if not np.isfinite(coordinates).all():
+        raise ValueError("grid coordinates must be finite")
+    local = matrix[:3, :3].T @ (coordinates - matrix[:3, 3])
+    if not np.isfinite(local).all():
+        raise ValueError("grid coordinates must be finite")
+    return float(local[0] / message.info.resolution), float(local[1] / message.info.resolution)
+
+
+def grid_to_world(message: OccupancyGrid, coordinates: tuple[float, float]) -> Point:
+    """Convert continuous grid coordinates to the grid plane in world coordinates."""
+    occupancy_extent(message)
+    x, y = coordinates
+    if not math.isfinite(x) or not math.isfinite(y):
+        raise ValueError("grid coordinates must be finite")
+    result = pose_matrix(message.info.origin) @ np.array(
+        [x * message.info.resolution, y * message.info.resolution, 0.0, 1.0]
+    )
+    return Point(x=float(result[0]), y=float(result[1]), z=float(result[2]))
+
+
+def occupancy_from_file(path: Path, *, header: Header, resolution: float = 0.05) -> OccupancyGrid:
+    """Load a static .npy or grayscale PNG grid with explicit frame/time metadata.
+
+    Cells must be ROS occupancy values (-1 unknown, 0..100 known); PNG 255 is
+    unknown. NumPy object/pickle loading is disabled. Row order is unchanged.
+    """
+    if not math.isfinite(resolution) or resolution <= 0:
+        raise ValueError("OccupancyGrid resolution must be finite and positive")
+    cells: NDArray[Any]
+    if path.suffix == ".npy":
+        cells = np.load(path, allow_pickle=False)
+    elif path.suffix == ".png":
+        with Image.open(path) as image:
+            cells = np.asarray(image.convert("L")).astype(np.int16)
+        cells[cells == 255] = -1
+    else:
+        raise ValueError(f"Unsupported occupancy file format: {path.suffix}")
+    return occupancy_from_array(cells, header=header, resolution=resolution)
+
+
+def occupancy_from_array(
+    cells: NDArray[Any], *, header: Header, resolution: float, origin: Pose | None = None
+) -> OccupancyGrid:
+    """Copy a 2D ROS occupancy array with explicit map metadata."""
+    if not math.isfinite(resolution) or resolution <= 0:
+        raise ValueError("OccupancyGrid resolution must be finite and positive")
+    if cells.ndim != 2 or cells.dtype.kind not in "iu":
+        raise ValueError("Occupancy cells must be a 2D integer array")
+    if np.any((cells < -1) | (cells > 100)):
+        raise ValueError("Occupancy cells must be -1 or 0..100")
+    return OccupancyGrid(
+        header=deepcopy(header),
+        info=MapMetaData(
+            map_load_time=deepcopy(header.stamp),
+            resolution=resolution,
+            height=cells.shape[0],
+            width=cells.shape[1],
+            origin=deepcopy(origin)
+            if origin is not None
+            else Pose(
+                orientation=Quaternion(w=1, x=0.0, y=0.0, z=0.0),
+                position=Point(x=0.0, y=0.0, z=0.0),
+            ),
+        ),
+        data=np.asarray(cells.astype(np.int8).reshape(-1), dtype=np.int8),
+    )