From 6f18aab2f03541f675e05d9ec8733863bf5b25e7 Mon Sep 17 00:00:00 2001 From: jwgcurrie Date: Sun, 26 Jul 2026 10:14:19 +0200 Subject: [PATCH 1/3] add unit and integration tests for the perception service --- tests/guard.py | 76 ++++++++++++++++++++++++++++ tests/run-in-docker.sh | 19 +++++++ tests/test_service_contract.py | 92 ++++++++++++++++++++++++++++++++++ tests/test_yolo_parsing.py | 70 ++++++++++++++++++++++++++ 4 files changed, 257 insertions(+) create mode 100644 tests/guard.py create mode 100755 tests/run-in-docker.sh create mode 100644 tests/test_service_contract.py create mode 100644 tests/test_yolo_parsing.py diff --git a/tests/guard.py b/tests/guard.py new file mode 100644 index 0000000..e983b5a --- /dev/null +++ b/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) diff --git a/tests/run-in-docker.sh b/tests/run-in-docker.sh new file mode 100755 index 0000000..161c798 --- /dev/null +++ b/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 "$@"' -- "$@" diff --git a/tests/test_service_contract.py b/tests/test_service_contract.py new file mode 100644 index 0000000..ab4cdc3 --- /dev/null +++ b/tests/test_service_contract.py @@ -0,0 +1,92 @@ +"""Integration guard: the ZMQ REP contract the robot clients depend on. + +Needs a running service; `data` varies by backend, so the guard reads the backend from the reply. +""" +from guard import (Guard, blank_image, encode, main, noise_image, request, + service_reachable) + +try: + import pytest + pytestmark = pytest.mark.skipif(not service_reachable(), reason="requires a running service") +except ImportError: + pass + +LANDMARK_KEYS = {"face_landmarks", "left_hand_landmarks", + "pose_landmarks", "right_hand_landmarks"} +DETECTION_KEYS = {"class", "confidence", "bbox"} + + +def check_blank_shape(g, backend, data): + """A blank frame must yield an empty detection set in whichever shape the backend uses.""" + if backend == "yolo": + g.check("yolo data is a list", isinstance(data, list), f"{type(data).__name__}") + g.check("blank frame yields no detections", data == [], f"{data}") + return + g.check("data is a dict", isinstance(data, dict), f"{type(data).__name__}") + if not isinstance(data, dict): + return + expected = LANDMARK_KEYS | ({"detections"} if backend == "combined" else set()) + # Clients index these keys directly (a rename is a breaking change). + g.check(f"{backend} data carries the documented keys", set(data) == expected, + f"unexpected={sorted(set(data) ^ expected)}") + if backend == "combined": + g.check("detections is a list", isinstance(data.get("detections"), list)) + g.check("blank frame yields no detections", data.get("detections") == [], + f"{data.get('detections')}") + for key in sorted(LANDMARK_KEYS): + g.check(f"{key} is None on a blank frame", data.get(key) is None) + + +def test_service_contract(): + g = Guard("perception contract") + + reply = request([b'{"meta":"x"}', encode(blank_image())]) + g.check("service answers a valid request", reply is not None) + if reply is None: + g.finish() + return + + backend = reply.get("backend") + g.check("status is success", reply.get("status") == "success", f"{reply.get('status')}") + g.check("backend is one of the supported three", backend in ("yolo", "mediapipe", "combined"), + f"{backend}") + g.check("inference_ms is a positive number", + isinstance(reply.get("inference_ms"), (int, float)) and reply["inference_ms"] > 0, + f"{reply.get('inference_ms')}") + check_blank_shape(g, backend, reply.get("data")) + + # The last frame is the image (a request without metadata work). + single = request([encode(noise_image())]) + g.check("single-frame request is accepted", + single is not None and single.get("status") == "success", + f"{single.get('status') if single else 'timeout'}") + + # Any detection returned must carry the fields clients read. A synthetic frame rarely trips the detector, so this is opportunistic; tests/test_yolo_parsing.py pins the shape. + if single and backend in ("yolo", "combined"): + found = single["data"] if backend == "yolo" else single["data"]["detections"] + if found: + g.check("detections have the documented fields", + all(DETECTION_KEYS <= set(d) for d in found), f"n={len(found)}") + g.check("bbox is four numbers", + all(len(d["bbox"]) == 4 for d in found), f"n={len(found)}") + else: + print("[SKIP] detection field shape: probe frame produced none " + "(pinned by tests/test_yolo_parsing.py)") + + # An undecodable payload must return a structured error. + bad = request([b"not-an-image"]) + g.check("undecodable payload returns an error", + bad is not None and bad.get("status") == "error", f"{bad}") + if bad: + g.check("error carries a message", bool(bad.get("message")), f"{bad.get('message')}") + + # The service must still serve after an error, or one bad frame kills the session. + after = request([encode(blank_image())]) + g.check("service still serves after a bad request", + after is not None and after.get("status") == "success") + + g.finish() + + +if __name__ == "__main__": + main(test_service_contract) diff --git a/tests/test_yolo_parsing.py b/tests/test_yolo_parsing.py new file mode 100644 index 0000000..ade142c --- /dev/null +++ b/tests/test_yolo_parsing.py @@ -0,0 +1,70 @@ +"""Unit test: YOLODetector.detect turns ultralytics results into the documented dict shape. + +Deterministic, unlike the integration guard, which only sees detections when a probe frame trips one. +""" +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from perception_service.detectors.yolo_detector import YOLODetector # noqa: E402 + + +class _Tensor(list): + def tolist(self): + return list(self) + + +class _Box: + def __init__(self, xyxy, conf, cls): + self.xyxy = [_Tensor(xyxy)] + self.conf = [conf] + self.cls = [cls] + + +class _Result: + def __init__(self, boxes): + self.boxes = boxes + + +class _FakeModel: + """Stands in for ultralytics YOLO: callable, and exposes a class-id to name mapping.""" + + names = {0: "person", 32: "sports ball"} + + def __init__(self, results): + self._results = results + + def __call__(self, image, **kwargs): + return self._results + + +def _detector(results): + # Bypass __init__ (no weights are loaded). + detector = object.__new__(YOLODetector) + detector.model = _FakeModel(results) + return detector + + +def test_detect_returns_documented_fields(): + out = _detector([_Result([_Box([1.0, 2.0, 3.0, 4.0], 0.9, 0)])]).detect(None) + assert out == [{"class": "person", "confidence": 0.9, "bbox": [1.0, 2.0, 3.0, 4.0]}] + + +def test_detect_returns_an_empty_list_when_nothing_is_found(): + assert _detector([_Result([])]).detect(None) == [] + + +def test_detect_flattens_multiple_results_and_boxes(): + out = _detector([ + _Result([_Box([0.0, 0.0, 1.0, 1.0], 0.5, 0)]), + _Result([_Box([2.0, 2.0, 3.0, 3.0], 0.25, 32)]), + ]).detect(None) + assert [d["class"] for d in out] == ["person", "sports ball"] + assert all(len(d["bbox"]) == 4 for d in out) + + +def test_detect_coerces_confidence_and_class_id_types(): + out = _detector([_Result([_Box([0, 0, 1, 1], 0.75, 0)])]).detect(None) + assert isinstance(out[0]["confidence"], float) + assert out[0]["class"] == "person" From 9b507d9e2042bb978b0021391905c7a302367252 Mon Sep 17 00:00:00 2001 From: jwgcurrie Date: Sun, 26 Jul 2026 10:14:19 +0200 Subject: [PATCH 2/3] remove dead code and trim verbose comments --- debug_client.py | 19 ++++---------- docker-compose.yml | 1 - perception_service/detectors/base.py | 10 +------- .../detectors/mediapipe_detector.py | 16 +----------- perception_service/detectors/yolo_detector.py | 21 +++------------- perception_service/main.py | 25 ++++++------------- test_client.py | 1 - webcam_client.py | 2 -- 8 files changed, 18 insertions(+), 77 deletions(-) diff --git a/debug_client.py b/debug_client.py index 7f6f345..be60d20 100644 --- a/debug_client.py +++ b/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 diff --git a/docker-compose.yml b/docker-compose.yml index 1f63595..95cf17f 100644 --- a/docker-compose.yml +++ b/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: diff --git a/perception_service/detectors/base.py b/perception_service/detectors/base.py index 11fa06f..2b7daeb 100644 --- a/perception_service/detectors/base.py +++ b/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 diff --git a/perception_service/detectors/mediapipe_detector.py b/perception_service/detectors/mediapipe_detector.py index db1d99d..5bca78b 100644 --- a/perception_service/detectors/mediapipe_detector.py +++ b/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) diff --git a/perception_service/detectors/yolo_detector.py b/perception_service/detectors/yolo_detector.py index fbf202b..02a36d8 100644 --- a/perception_service/detectors/yolo_detector.py +++ b/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 = [] diff --git a/perception_service/main.py b/perception_service/main.py index baea6ec..4fe0e0d 100644 --- a/perception_service/main.py +++ b/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,7 +40,6 @@ 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") @@ -49,18 +47,15 @@ def main(): 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: diff --git a/test_client.py b/test_client.py index 3c2e8e9..0c1cf21 100644 --- a/test_client.py +++ b/test_client.py @@ -2,7 +2,6 @@ import cv2 import json import sys -import numpy as np def main(): if len(sys.argv) < 2: diff --git a/webcam_client.py b/webcam_client.py index 311c8e2..20d5563 100644 --- a/webcam_client.py +++ b/webcam_client.py @@ -1,8 +1,6 @@ import cv2 import zmq -import numpy as np import time -import json # MediaPipe Pose Connections (Simplified) POSE_CONNECTIONS = [ From 3362262fcbd5f0848b4fa34db2a87d411214caee Mon Sep 17 00:00:00 2001 From: jwgcurrie Date: Sun, 26 Jul 2026 10:14:19 +0200 Subject: [PATCH 3/3] set zmq linger to zero so client shutdown cannot block --- perception_service/client.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/perception_service/client.py b/perception_service/client.py index e5677f5..f299ae3 100644 --- a/perception_service/client.py +++ b/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: