Chore/refactor

debug_client.py

@@ -1,9 +1,6 @@
 import zmq
 import cv2
 import numpy as np
-import time
-import json
-import threading
 
 def main():
     # Context
@@ -42,14 +39,10 @@ def nothing(x): pass
             if video_sub.poll(100):
                 try:
                     parts = video_sub.recv_multipart(flags=zmq.NOBLOCK)
-                    timestamp = 0.0
-                    if len(parts) == 3:
-                        topic, header, msg = parts 
-                        # timestamp = struct.unpack('d', header)[0]
-                    elif len(parts) == 2:
-                        topic, msg = parts
-                    else:
+                    # [topic, header, image] or [topic, image] (always image last).
+                    if len(parts) not in (2, 3):
                         continue
+                    msg = parts[-1]
                 except zmq.Again:
                     continue
 
@@ -86,8 +79,7 @@ def nothing(x): pass
                 # Format is already BGR for display/encode
 
 
-                # B. Send to Perception (Needs JPG bytes)
-                # CV2.imencode expects BGR input.
+                # Perception wants JPG bytes, and imencode expects BGR input.
                 _, jpg_encoded = cv2.imencode('.jpg', display_frame)
 
                 # REQ/REP is blocking.
@@ -141,8 +133,7 @@ def nothing(x): pass
                 if not isinstance(response_data, list):
                     pose = response_data.get("pose_landmarks")
                 if pose:
-                    # Define simple skeleton topology (indices based on Mediapipe Pose)
-                    # 0: Nose, 11: Yes, 12: Right Shoulder, 13: Left Elbow, 14: Right Elbow, ...
+                    # MediaPipe Pose indices: 0 nose, 11/12 shoulders, 13/14 elbows, 15/16 wrists, 23/24 hips.
                     connections = [
                         (11, 12), (11, 13), (13, 15), # Left Arm
                         (12, 14), (14, 16),           # Right Arm

docker-compose.yml

@@ -8,7 +8,6 @@ services:
     command: [ "python", "-m", "perception_service.main", "--backend", "combined" ]
     ports:
       - "5557:5557" # REP socket for on-demand requests
-      - "5558:5558" # PUB socket for continuous streams (optional)
     volumes:
       - .:/app
     environment:

perception_service/client.py

@@ -1,20 +1,25 @@
 import zmq
 import json
 import cv2
-import time
 
 class PerceptionClient:
     def __init__(self, service_uri="tcp://localhost:5557"):
         self.service_uri = service_uri
         self.context = zmq.Context()
-        self.socket = self.context.socket(zmq.REQ)
-        self.socket.connect(service_uri)
+        self.socket = self._new_socket()
         print(f"PerceptionClient: Connected to {service_uri}")
 
+    def _new_socket(self):
+        """REQ socket with LINGER 0; the default (-1) makes close() block on an abandoned request."""
+        socket = self.context.socket(zmq.REQ)
+        socket.setsockopt(zmq.LINGER, 0)
+        socket.connect(self.service_uri)
+        return socket
+
     def detect(self, img_bgr, target_label=None):
-        """
-        Sends image to perception service.
-        Returns list of detections or None on timeout/error.
+        """Send a BGR frame and return the service's `data` payload, or None on timeout/error.
+
+        `data` is a list for the yolo backend and a dict for mediapipe and combined.
         """
         try:
             # Encode to JPG
@@ -34,10 +39,9 @@ def detect(self, img_bgr, target_label=None):
                 return result.get("data", {})
             else:
                 print("PerceptionClient: Timeout")
-                # Reset socket on timeout
+                # REQ is strict-alternating, so a timed-out socket cannot be reused.
                 self.socket.close()
-                self.socket = self.context.socket(zmq.REQ)
-                self.socket.connect(self.service_uri)
+                self.socket = self._new_socket()
                 return None
 
         except Exception as e:

perception_service/detectors/base.py

@@ -3,13 +3,5 @@
 class BaseDetector(ABC):
     @abstractmethod
     def detect(self, image):
-        """
-        Process the image and return results.
-
-        Args:
-            image (numpy.ndarray): Input image in BGR format.
-
-        Returns:
-            dict or list: Detection results (format depends on implementation).
-        """
+        """Detect on a BGR ndarray; returns a list (yolo) or dict (mediapipe), per implementation."""
         pass

perception_service/detectors/mediapipe_detector.py

@@ -1,6 +1,5 @@
 import mediapipe as mp
 import cv2
-import numpy as np
 from .base import BaseDetector
 
 class MediaPipeDetector(BaseDetector):
@@ -24,20 +23,7 @@ def _process_landmarks(self, landmarks):
         return [{"x": lm.x, "y": lm.y, "z": lm.z, "visibility": lm.visibility} for lm in landmarks.landmark]
 
     def detect(self, image):
-        """
-        Run Holistic inference.
-
-        Args:
-            image: BGR image.
-
-        Returns:
-            dict: {
-                "face_landmarks": [...],
-                "pose_landmarks": [...],
-                "left_hand_landmarks": [...],
-                "right_hand_landmarks": [...]
-            }
-        """
+        """Run Holistic on a BGR image; returns pose/face/left_hand/right_hand landmark lists, None when absent."""
         # Convert BGR to RGB
         image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
 

perception_service/detectors/yolo_detector.py

@@ -3,29 +3,14 @@
 
 class YOLODetector(BaseDetector):
     def __init__(self, model_name='yolov8n.pt'):
-        """
-        Initialize the YOLO object detector.
-
-        Args:
-            model_name (str): The name or path of the YOLO model to load.
-        """
+        """Load the YOLO model named or pathed by `model_name`."""
         print(f"Loading YOLO model: {model_name}...")
         self.model = YOLO(model_name)
         print("Model loaded successfully.")
 
     def detect(self, image):
-        """
-        Run inference on an image.
-
-        Args:
-            image (numpy.ndarray): The input image (BGR format from OpenCV).
-
-        Returns:
-            list: A list of dictionaries containing detection results.
-                  Each dict has: 'class', 'confidence', 'bbox' [x1, y1, x2, y2].
-        """
-        # Run inference
-        # Tuned: Lowered confidence to 0.1 to catch objects in low-res streams
+        """Run inference on a BGR image; returns dicts of 'class', 'confidence', 'bbox' [x1,y1,x2,y2]."""
+        # conf=0.1 is deliberately low to catch objects in the robot's low-res stream.
         results = self.model(image, verbose=False, conf=0.1)
 
         detections = []

perception_service/main.py

@@ -1,14 +1,13 @@
-import zmq
+import argparse
+import sys
+import time
+
 import cv2
 import numpy as np
-import json
-import os
-import sys
-import argparse
+import zmq
 
-# Import new detectors
-from .detectors.yolo_detector import YOLODetector
 from .detectors.base import BaseDetector
+from .detectors.yolo_detector import YOLODetector
 
 class CombinedDetector(BaseDetector):
     def __init__(self, model_name='yolov8n.pt'):
@@ -41,26 +40,22 @@ def get_detector(backend, model_name):
         raise ValueError(f"Unknown backend: {backend}")
 
 def main():
-    import time # Re-import locally to force availability
     parser = argparse.ArgumentParser(description="Pepper Perception Service")
     parser.add_argument("--port", type=int, default=5557, help="ZMQ REP port")
     parser.add_argument("--backend", type=str, default="yolo", choices=['yolo', 'mediapipe', 'combined'], help="Detection backend")
     parser.add_argument("--model", type=str, default="yolov8n.pt", help="Model name (for YOLO)")
     args = parser.parse_args()
 
     print(f"Starting Perception Service on port {args.port} using {args.backend}...")
-    print(f"Time module loaded: {time}")
 
-    # Initialize ZMQ
     context = zmq.Context()
     socket = context.socket(zmq.REP)
     socket.bind(f"tcp://*:{args.port}")
 
-    # Initialize Detector
     try:
-        import traceback
         detector = get_detector(args.backend, args.model)
     except Exception as e:
+        import traceback
         print(f"CRITICAL Error loading backend {args.backend}: {e}")
         traceback.print_exc()
         sys.exit(1)
@@ -71,12 +66,8 @@ def main():
         while True:
             # Wait for next request
             msg_parts = socket.recv_multipart()
-
-            if not msg_parts:
-                socket.send_json({"error": "Empty message"})
-                continue
 
-            # First part is optional metadata, Last part is image
+            # A ZMQ message always carries at least one frame; the last one is the image.
             img_data = msg_parts[-1]
 
             try:

test_client.py

@@ -2,7 +2,6 @@
 import cv2
 import json
 import sys
-import numpy as np
 
 def main():
     if len(sys.argv) < 2:

tests/guard.py

@@ -0,0 +1,76 @@
+"""Shared helpers for the perception service guards.
+
+Each guard runs as a script (exit 1 on failure) and under pytest (skipped when no service answers).
+"""
+import sys
+
+import cv2
+import numpy as np
+import zmq
+
+ADDRESS = "tcp://localhost:5557"
+TIMEOUT_MS = 30000
+
+
+def request(parts, address=ADDRESS, timeout_ms=TIMEOUT_MS):
+    """Send one multipart REQ and return the decoded JSON reply, or None on timeout."""
+    ctx = zmq.Context.instance()
+    sock = ctx.socket(zmq.REQ)
+    sock.setsockopt(zmq.RCVTIMEO, timeout_ms)
+    sock.setsockopt(zmq.LINGER, 0)
+    sock.connect(address)
+    try:
+        sock.send_multipart(parts)
+        return sock.recv_json()
+    except zmq.Again:
+        return None
+    finally:
+        sock.close()
+
+
+def service_reachable():
+    """True if the perception service answers a trivial request."""
+    return request([encode(blank_image())], timeout_ms=5000) is not None
+
+
+def blank_image(width=320, height=240):
+    return np.zeros((height, width, 3), np.uint8)
+
+
+def noise_image(width=320, height=240, seed=0):
+    return np.random.RandomState(seed).randint(0, 255, (height, width, 3), dtype=np.uint8)
+
+
+def encode(image):
+    ok, buffer = cv2.imencode(".jpg", image)
+    if not ok:
+        raise RuntimeError("cv2.imencode failed")
+    return buffer.tobytes()
+
+
+class Guard:
+    """Accumulates named checks and fails once at the end with the full list."""
+
+    def __init__(self, name):
+        self.name = name
+        self.failures = []
+
+    def check(self, label, ok, detail=""):
+        print(f"[{'PASS' if ok else 'FAIL'}] {label}" + (f" ({detail})" if detail else ""))
+        if not ok:
+            self.failures.append(label)
+
+    def finish(self):
+        if self.failures:
+            raise AssertionError(
+                f"{self.name}: {len(self.failures)} check(s) failed: {self.failures}")
+        print(f"\n--- {self.name}: all checks passed ---")
+
+
+def main(fn):
+    """Run a guard function as a script, exiting 1 on failure."""
+    try:
+        fn()
+    except AssertionError as exc:
+        print(f"\n--- FAILURE ---\n{exc}")
+        sys.exit(1)

tests/run-in-docker.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+# Run pytest inside the pepper-perception image, mounting the working tree over the image's copy.
+#
+# Usage: ./tests/run-in-docker.sh [pytest-args...]
+#   ./tests/run-in-docker.sh tests/test_yolo_parsing.py -v
+#
+# The integration guard needs a running service; run it directly:
+#   python3 tests/test_service_contract.py
+
+set -e
+
+SCRIPT_DIR="$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )"
+REPO_ROOT="$(dirname "$SCRIPT_DIR")"
+
+exec docker run --rm \
+    -v "$REPO_ROOT:/app" \
+    -w /app \
+    ghcr.io/action-prediction-lab/pepper-perception:latest \
+    bash -c 'python -m pip install --quiet --user pytest && python -m pytest "$@"' -- "$@"