qodo-code-review fixes

config.yaml

@@ -117,7 +117,7 @@ vision:
       height: 640
     "imx500":
       width: 640
-      height: 640
+      height: 480
     "default":
       width: 1080
       height: 720

src/vision/video_capture.py

@@ -12,6 +12,7 @@
 from typing import TYPE_CHECKING, Any
 
 import cv2
+import numpy as np
 
 # Ensure 'src' is in sys.path
 sys.path.insert(0, str(pathlib.Path(__file__).parent.parent.parent.resolve()))
@@ -24,8 +25,6 @@
 if TYPE_CHECKING:
     from collections.abc import Generator
 
-    import numpy as np
-
     from src.utils.config import Config
 
 config.setup_python_path()
@@ -36,6 +35,16 @@
 logger = logging.getLogger(lib_name)
 
 
+def _is_valid_frame(frame: Any) -> bool:
+    """Check if frame is a non-empty array or a valid test object (e.g. MagicMock)."""
+    if frame is None:
+        return False
+    size = getattr(frame, "size", None)
+    if isinstance(size, (int, float, np.integer)):
+        return size > 0
+    return True
+
+
 class VideoCapture:
     """Simplified Video Capture class orchestrating modular frame processing."""
 
@@ -151,17 +160,19 @@ def capture_frame(self) -> np.ndarray | None:
         if self.camera is None:
             return None
         frame, metadata = self.camera.read()
-        if frame is not None:
+        if _is_valid_frame(frame):
             with self.lock:
                 self.latest_frame = frame
                 self.latest_metadata = metadata
             logger.debug("Frame captured")
-        else:
-            logger.error("Failed to capture frame")
-        return frame
+            return frame
+        logger.error("Failed to capture frame")
+        return None
 
     def process_frame(self, frame: np.ndarray) -> np.ndarray:
         """Process frame using active frame processors."""
+        if not _is_valid_frame(frame):
+            return frame
         annotated_frame = frame.copy()
 
         # 1. Run YOLO Object Detection if enabled
@@ -209,7 +220,7 @@ def process_frame(self, frame: np.ndarray) -> np.ndarray:
 
     def _add_performance_overlay(self, frame: np.ndarray) -> None:
         """Add premium status and speed information using Ultralytics metrics."""
-        if self.latest_results is None:
+        if self.latest_results is None or not _is_valid_frame(frame):
             return
 
         # Speed metrics (ms)
@@ -231,7 +242,14 @@ def _add_performance_overlay(self, frame: np.ndarray) -> None:
         color: tuple[int, int, int] = (255, 255, 255)
         cv2.putText(frame, f"FPS: {self.fps:.1f}", (10, 25), font, 0.6, color, 1)
         cv2.putText(frame, f"Inf: {inference:.1f}ms", (10, 50), font, 0.5, color, 1)
-        cv2.putText(frame, f"Objs: {len(self.latest_results.boxes)}", (10, 75), font, 0.5, color, 1)
+        boxes = getattr(self.latest_results, "boxes", None)
+        if hasattr(self.latest_results, "detections") and isinstance(self.latest_results.detections, (list, tuple)):
+            obj_count = len(self.latest_results.detections)
+        elif boxes is not None and hasattr(boxes, "__len__"):
+            obj_count = len(boxes)
+        else:
+            obj_count = 0
+        cv2.putText(frame, f"Objs: {obj_count}", (10, 75), font, 0.5, color, 1)
 
     def benchmark(self, iterations: int = 100) -> None:
         """Run a performance benchmark of the current model."""
@@ -264,7 +282,14 @@ def capture_photo(self) -> tuple[bool, str]:
         results.save(filename=str(annotated_path))  # save to disk
         self.capture_count += 1
         logger.info("Saved annotated frame to %s", annotated_path)
-        return True, f"Captured result with {len(results.boxes)} object(s)"
+        boxes = getattr(results, "boxes", None)
+        if hasattr(results, "detections") and isinstance(results.detections, (list, tuple)):
+            obj_count = len(results.detections)
+        elif boxes is not None and hasattr(boxes, "__len__"):
+            obj_count = len(boxes)
+        else:
+            obj_count = 0
+        return True, f"Captured result with {obj_count} object(s)"
 
     def generate_frames(self) -> Generator[bytes, None, None]:
         """Generate frames for Flask video streaming.

src/vision/yolo_cpu.py

@@ -12,7 +12,6 @@
 
 import cv2
 import numpy as np
-import onnxruntime as ort
 from ultralytics import YOLO
 from ultralytics.engine.results import Results
 
@@ -39,6 +38,14 @@ def __init__(self, model_path: str, names: dict[int, str] | None = None) -> None
 
         """
         self.model_path = model_path
+        try:
+            import onnxruntime as ort
+        except ImportError as err:
+            msg = (
+                "onnxruntime is required for LibreYoloOnnxPredictor when libreyolo package is not installed. "
+                "Please install onnxruntime (e.g. pip install onnxruntime)."
+            )
+            raise ImportError(msg) from err
         self.session = ort.InferenceSession(model_path, providers=["CPUExecutionProvider"])
         self.input_name = self.session.get_inputs()[0].name
         self.names = names or {i: f"class_{i}" for i in range(80)}

src/vision/yolo_imx500.py

@@ -39,9 +39,13 @@ def __init__(self) -> None:
         self.conf = []
         self.cls = []
 
+    def __len__(self) -> int:
+        """Return number of bounding boxes."""
+        return len(self.xyxy)
+
     def __iter__(self) -> Any:
-        """Return empty iterator over bounding boxes."""
-        return iter([])
+        """Return iterator over bounding boxes."""
+        return iter(self.xyxy)
 
 
 class DummyResults:

tests/utils/test_config.py

@@ -305,9 +305,10 @@ def test_get_model_resolution() -> None:
     assert cfg.vision.get_model_resolution("LibreYOLOXn.onnx") == (416, 416)
     assert cfg.vision.get_model_resolution("LibreYOLOXn") == (416, 416)
 
-    # Test lookup for yolo26n / buffalo_l
+    # Test lookup for yolo26n / buffalo_l / imx500
     assert cfg.vision.get_model_resolution("yolo26n.onnx") == (640, 640)
     assert cfg.vision.get_model_resolution("buffalo_l") == (640, 640)
+    assert cfg.vision.get_model_resolution("imx500") == (640, 480)
 
     # Test default fallback for unknown model
     assert cfg.vision.get_model_resolution("unknown_model_xyz") == (1080, 720)

tests/vision/test_libreyolo.py

@@ -43,3 +43,14 @@ def test_yolo_cpu_detector_libreyolo_load() -> None:
         assert dets[0]["score"] == 0.95
         assert dets[0]["class_id"] == 0
         assert dets[0]["label"] == "person"
+
+
+def test_libre_yolo_onnx_predictor_missing_onnxruntime() -> None:
+    """Test that LibreYoloOnnxPredictor raises ImportError when onnxruntime is missing."""
+    from src.vision.yolo_cpu import LibreYoloOnnxPredictor
+
+    with (
+        patch.dict("sys.modules", {"onnxruntime": None}),
+        pytest.raises(ImportError, match="onnxruntime is required for LibreYoloOnnxPredictor"),
+    ):
+        LibreYoloOnnxPredictor("dummy.onnx")

tests/vision/test_vision_imx500.py

@@ -97,3 +97,21 @@ def test_imx500_detector_get_labels_dash_filter() -> None:
     detector = Imx500Detector(Imx500Config(), imx500=mock_imx500)
     labels = detector.get_labels()
     assert labels == ["person", "car"]
+
+
+def test_dummy_boxes_len_and_iter() -> None:
+    """Test DummyBoxes __len__ and __iter__ functionality."""
+    from src.vision.yolo_imx500 import DummyBoxes, DummyResults
+
+    boxes = DummyBoxes()
+    assert len(boxes) == 0
+    assert list(boxes) == []
+
+    class MockDet:
+        x1, y1, x2, y2 = 10.0, 20.0, 30.0, 40.0
+        score = 0.9
+        cls = 0
+
+    res = DummyResults([MockDet()], (480, 640))
+    assert len(res.boxes) == 1
+    assert list(res.boxes) == [[10.0, 20.0, 30.0, 40.0]]