fix mypy (#1167)

* old numpy forced old version of rerun

* fix other mypy issues

Author: paul-nechifor

dimos/models/depth/metric3d.py

@@ -78,7 +78,10 @@ def infer_depth(self, img, debug: bool = False):  # type: ignore[no-untyped-def]
         try:
             if isinstance(img, str):
                 print(f"Image type string: {type(img)}")
-                self.rgb_origin = cv2.imread(img)[:, :, ::-1]
+                img_data = cv2.imread(img)
+                if img_data is None:
+                    raise ValueError(f"Failed to load image from {img}")
+                self.rgb_origin = img_data[:, :, ::-1]
             else:
                 # print(f"Image type not string: {type(img)}, cv2 conversion assumed to be handled. If not, this will throw an error")
                 self.rgb_origin = img
@@ -172,9 +175,11 @@ def unpad_transform_depth(self, pred_depth):  # type: ignore[no-untyped-def]
 
     def eval_predicted_depth(self, depth_file, pred_depth) -> None:  # type: ignore[no-untyped-def]
         if depth_file is not None:
-            gt_depth = cv2.imread(depth_file, -1)
-            gt_depth = gt_depth / self.gt_depth_scale  # type: ignore[assignment]
-            gt_depth = torch.from_numpy(gt_depth).float().to(self.device)  # type: ignore[assignment]
+            gt_depth_np = cv2.imread(depth_file, -1)
+            if gt_depth_np is None:
+                raise ValueError(f"Failed to load depth file from {depth_file}")
+            gt_depth_scaled = gt_depth_np / self.gt_depth_scale
+            gt_depth = torch.from_numpy(gt_depth_scaled).float().to(self.device)
             assert gt_depth.shape == pred_depth.shape
 
             mask = gt_depth > 1e-8

dimos/models/segmentation/edge_tam.py

@@ -36,7 +36,7 @@
 from dimos.utils.logging_config import setup_logger
 
 if TYPE_CHECKING:
-    from sam2.sam2_video_predictor import SAM2VideoPredictor  # type: ignore[import-untyped]
+    from sam2.sam2_video_predictor import SAM2VideoPredictor
 
 os.environ['TQDM_DISABLE'] = '1'
 

dimos/perception/detection/type/detection2d/seg.py

@@ -183,8 +183,10 @@ def to_points_annotation(self) -> list[PointsAnnotation]:
             approx = cv2.approxPolyDP(contour, epsilon, True)
 
             points = []
-            for pt in approx:
-                points.append(Point2(x=float(pt[0][0]), y=float(pt[0][1])))
+            for i in range(len(approx)):
+                x_coord = float(approx[i, 0, 0])
+                y_coord = float(approx[i, 0, 1])
+                points.append(Point2(x=x_coord, y=y_coord))
 
             if len(points) < 3:
                 continue

dimos/protocol/pubsub/benchmark/testdata.py

@@ -14,7 +14,7 @@
 
 from collections.abc import Generator
 from contextlib import contextmanager
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import numpy as np
 
@@ -211,13 +211,21 @@ def redis_msggen(size: int) -> tuple[str, Any]:
     ROSTopic,
 )
 
+if TYPE_CHECKING:
+    from numpy.typing import NDArray
+
 if ROS_AVAILABLE:
-    from rclpy.qos import QoSDurabilityPolicy, QoSHistoryPolicy, QoSProfile, QoSReliabilityPolicy
-    from sensor_msgs.msg import Image as ROSImage
+    from rclpy.qos import (  # type: ignore[no-untyped-call]
+        QoSDurabilityPolicy,
+        QoSHistoryPolicy,
+        QoSProfile,
+        QoSReliabilityPolicy,
+    )
+    from sensor_msgs.msg import Image as ROSImage  # type: ignore[attr-defined,no-untyped-call]
 
     @contextmanager
     def ros_best_effort_pubsub_channel() -> Generator[RawROS, None, None]:
-        qos = QoSProfile(
+        qos = QoSProfile(  # type: ignore[no-untyped-call]
             reliability=QoSReliabilityPolicy.BEST_EFFORT,
             history=QoSHistoryPolicy.KEEP_LAST,
             durability=QoSDurabilityPolicy.VOLATILE,
@@ -230,7 +238,7 @@ def ros_best_effort_pubsub_channel() -> Generator[RawROS, None, None]:
 
     @contextmanager
     def ros_reliable_pubsub_channel() -> Generator[RawROS, None, None]:
-        qos = QoSProfile(
+        qos = QoSProfile(  # type: ignore[no-untyped-call]
             reliability=QoSReliabilityPolicy.RELIABLE,
             history=QoSHistoryPolicy.KEEP_LAST,
             durability=QoSDurabilityPolicy.VOLATILE,
@@ -245,21 +253,21 @@ def ros_msggen(size: int) -> tuple[RawROSTopic, ROSImage]:
         import numpy as np
 
         # Create image data
-        data = np.frombuffer(make_data_bytes(size), dtype=np.uint8).reshape(-1)
-        padded_size = ((len(data) + 2) // 3) * 3
-        data = np.pad(data, (0, padded_size - len(data)))
-        pixels = len(data) // 3
+        raw_data: NDArray[np.uint8] = np.frombuffer(make_data_bytes(size), dtype=np.uint8)
+        padded_size = ((len(raw_data) + 2) // 3) * 3
+        padded_data: NDArray[np.uint8] = np.pad(raw_data, (0, padded_size - len(raw_data)))
+        pixels = len(padded_data) // 3
         height = max(1, int(pixels**0.5))
         width = pixels // height
-        data = data[: height * width * 3]
+        final_data: NDArray[np.uint8] = padded_data[: height * width * 3]
 
         # Create ROS Image message
-        msg = ROSImage()
+        msg = ROSImage()  # type: ignore[no-untyped-call]
         msg.height = height
         msg.width = width
         msg.encoding = "rgb8"
         msg.step = width * 3
-        msg.data = data.tobytes()
+        msg.data = bytes(final_data)
 
         topic = RawROSTopic(topic="/benchmark/ros", ros_type=ROSImage)
         return (topic, msg)
@@ -280,7 +288,7 @@ def ros_msggen(size: int) -> tuple[RawROSTopic, ROSImage]:
 
     @contextmanager
     def dimos_ros_best_effort_pubsub_channel() -> Generator[DimosROS, None, None]:
-        qos = QoSProfile(
+        qos = QoSProfile(  # type: ignore[no-untyped-call]
             reliability=QoSReliabilityPolicy.BEST_EFFORT,
             history=QoSHistoryPolicy.KEEP_LAST,
             durability=QoSDurabilityPolicy.VOLATILE,
@@ -293,7 +301,7 @@ def dimos_ros_best_effort_pubsub_channel() -> Generator[DimosROS, None, None]:
 
     @contextmanager
     def dimos_ros_reliable_pubsub_channel() -> Generator[DimosROS, None, None]:
-        qos = QoSProfile(
+        qos = QoSProfile(  # type: ignore[no-untyped-call]
             reliability=QoSReliabilityPolicy.RELIABLE,
             history=QoSHistoryPolicy.KEEP_LAST,
             durability=QoSDurabilityPolicy.VOLATILE,

dimos/protocol/pubsub/impl/rospubsub.py

@@ -104,7 +104,7 @@ def __init__(self, node_name: str | None = None, qos: "QoSProfile | None" = None
         if qos is not None:
             self._qos = qos
         else:
-            self._qos = QoSProfile(
+            self._qos = QoSProfile(  # type: ignore[no-untyped-call]
                 # Haven't noticed any difference between BEST_EFFORT and RELIABLE for local comms in our tests
                 # ./bin/dev python -m pytest -svm tool -k ros dimos/protocol/pubsub/benchmark/test_benchmark.py
                 #
@@ -120,7 +120,7 @@ def start(self) -> None:
         if self._spin_thread is not None:
             return
 
-        if not rclpy.ok():
+        if not rclpy.ok():  # type: ignore[attr-defined]
             rclpy.init()
 
         self._stop_event.clear()
@@ -160,7 +160,7 @@ def stop(self) -> None:
                 self._node.destroy_publisher(publisher)
 
         if self._node:
-            self._node.destroy_node()
+            self._node.destroy_node()  # type: ignore[no-untyped-call]
             self._node = None
 
         self._executor = None

dimos/spec/utils.py

@@ -15,7 +15,7 @@
 import inspect
 from typing import Any, Protocol, runtime_checkable
 
-from annotation_protocol import AnnotationProtocol  # type: ignore[import-untyped]
+from annotation_protocol import AnnotationProtocol
 from typing_extensions import is_protocol
 
 

dimos/utils/docs/doclinks.py

@@ -529,7 +529,7 @@ def process_file(md_path: Path, quiet: bool = False) -> tuple[bool, list[str]]:
 
         watch_paths = args.paths if args.paths else [str(root / "docs")]
 
-        class MarkdownHandler(FileSystemEventHandler):
+        class MarkdownHandler(FileSystemEventHandler):  # type: ignore[misc]
             def on_modified(self, event: Any) -> None:
                 if not event.is_directory and event.src_path.endswith(".md"):
                     process_file(Path(event.src_path))

docker/ros/Dockerfile

@@ -43,9 +43,6 @@ RUN apt-get install -y \
     qtbase5-dev-tools \
     supervisor
 
-# Install specific numpy version first
-RUN pip install 'numpy<2.0.0'
-
 # Add ROS2 apt repository
 RUN curl -sSL https://raw.githubusercontent.com/ros/rosdistro/master/ros.key -o /usr/share/keyrings/ros-archive-keyring.gpg && \
     echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/ros-archive-keyring.gpg] http://packages.ros.org/ros2/ubuntu $(lsb_release -cs) main" | tee /etc/apt/sources.list.d/ros2.list > /dev/null
@@ -87,5 +84,3 @@ RUN rosdep update
 
 # Source ROS2 and workspace in bashrc
 RUN echo "source /opt/ros/${ROS_DISTRO}/setup.bash" >> /root/.bashrc
-
-# Trigger docker workflow rerun 1

pyproject.toml

@@ -323,27 +323,31 @@ exclude = "^dimos/models/Detic(/|$)|^dimos/rxpy_backpressure(/|$)|.*/test_.|.*/c
 
 [[tool.mypy.overrides]]
 module = [
-    "rclpy.*",
-    "std_msgs.*",
+    "annotation_protocol",
+    "dimos_lcm.*",
+    "etils",
     "geometry_msgs.*",
-    "sensor_msgs.*",
-    "nav_msgs.*",
-    "tf2_msgs.*",
     "mujoco",
     "mujoco_playground.*",
-    "etils",
-    "xarm.*",
-    "dimos_lcm.*",
+    "nav_msgs.*",
+    "open_clip",
     "piper_sdk.*",
+    "plotext",
     "plum.*",
-    "pycuda.*",
     "pycuda",
-    "plotext",
-    "torchreid",
-    "open_clip",
-    "pyzed.*",
+    "pycuda.*",
     "pyzed",
+    "pyzed.*",
+    "rclpy.*",
+    "sam2.*",
+    "sensor_msgs.*",
+    "std_msgs.*",
+    "tf2_msgs.*",
+    "torchreid",
+    "ultralytics.*",
     "unitree_webrtc_connect.*",
+    "watchdog.*",
+    "xarm.*",
 ]
 ignore_missing_imports = true