PR #14348 from AviaAv: Create iq_helper.py, detect page on depth test

Author: Nir-Az

unit-tests/live/image-quality/iq_helper.py

@@ -0,0 +1,103 @@
+# License: Apache 2.0. See LICENSE file in root directory.
+# Copyright(c) 2025 RealSense, Inc. All Rights Reserved.
+
+from rspy import log, test
+import numpy as np
+import cv2
+import time
+import pyrealsense2 as rs
+
+
+# standard size to display / process the target
+WIDTH = 1280
+HEIGHT = 720
+
+# transformation matrix from frame to aligned region of interest
+M = None
+
+def compute_homography(pts):
+    """
+    Given 4 points (the detected ArUco marker centers), find the 3×3 matrix that stretches/rotates
+    the four ArUco points so they become the corners of an A4 page (used to "flatten" the page in an image)
+    """
+    pts_sorted = sorted(pts, key=lambda p: (p[1], p[0]))
+    top_left, top_right = sorted(pts_sorted[:2], key=lambda p: p[0])
+    bottom_left, bottom_right = sorted(pts_sorted[2:], key=lambda p: p[0])
+
+    src = np.array([top_left, top_right, bottom_right, bottom_left], dtype=np.float32)
+    dst = np.array([[0,0],[WIDTH-1,0],[WIDTH-1,HEIGHT-1],[0,HEIGHT-1]], dtype=np.float32)
+    M = cv2.getPerspectiveTransform(src, dst)
+    return M  # we later use M to get our roi
+
+
+def detect_a4_page(img, required_ids):
+    """
+    Detect ArUco markers and return center of each one
+    Returns None if not all required markers are found
+    """
+    # init aruco detector
+    aruco = cv2.aruco
+    dict_type = cv2.aruco.DICT_4X4_1000
+    dictionary = aruco.getPredefinedDictionary(dict_type)
+    try:
+        # new API (OpenCV >= 4.7)
+        parameters = aruco.DetectorParameters()
+        detector = aruco.ArucoDetector(dictionary, parameters)
+        corners, ids, _ = detector.detectMarkers(img)
+    except AttributeError:
+        # legacy API (OpenCV <= 4.6) - used on some of our machines
+        parameters = aruco.DetectorParameters_create()
+        corners, ids, _ = aruco.detectMarkers(img, dictionary, parameters=parameters)
+
+    if ids is None or not all(rid in ids for rid in required_ids):
+        return None
+
+    id_to_corner = dict(zip(ids.flatten(), corners))  # map id to corners
+    values = [id_to_corner[rid][0].mean(axis=0) for rid in required_ids] # for each required id, get center of marker coords
+
+    return np.array(values, dtype=np.float32)
+
+
+def find_roi_location(pipeline, required_ids, DEBUG_MODE=False):
+    """
+    Returns a matrix that transforms from frame to region of interest
+    This matrix will later be used with cv2.warpPerspective()
+    """
+    global M
+    # stream until page found
+    page_pts = None
+    start_time = time.time()
+    while page_pts is None and time.time() - start_time < 5:
+        frames = pipeline.wait_for_frames()
+        aruco_detectable_streams = (rs.stream.color, rs.stream.infrared) # we need one of those streams to detect ArUco markers
+        frame = next(f for f in frames if f.get_profile().stream_type() in aruco_detectable_streams)
+        img_bgr = np.asanyarray(frame.get_data())
+
+        if DEBUG_MODE:
+            cv2.imshow("PageDetect - waiting for page", img_bgr)
+            cv2.waitKey(1)
+
+        page_pts = detect_a4_page(img_bgr, required_ids)
+
+    if page_pts is None:
+        log.e("Failed to detect page within timeout")
+        test.fail()
+        raise Exception("Page not found")
+
+    # page found - use it to calculate transformation matrix from frame to region of interest
+    M = compute_homography(page_pts)
+    cv2.destroyAllWindows()
+    return M, page_pts
+
+def get_roi_from_frame(frame):
+    """
+    Apply the previously computed transformation matrix to the given frame
+    to get the region of interest (A4 page)
+    """
+    global M
+    if M is None:
+        raise Exception("Transformation matrix not computed yet")
+
+    np_frame = np.asanyarray(frame.get_data())
+    warped = cv2.warpPerspective(np_frame, M, (WIDTH, HEIGHT)) # using A4 size for its ratio
+    return warped

unit-tests/live/image-quality/test-basic-color.py

@@ -7,17 +7,13 @@
 from rspy import log, test
 import numpy as np
 import cv2
-import time
+from iq_helper import find_roi_location, get_roi_from_frame, WIDTH, HEIGHT
 
 NUM_FRAMES = 100 # Number of frames to check
 COLOR_TOLERANCE = 60 # Acceptable per-channel deviation in RGB values
 FRAMES_PASS_THRESHOLD =0.8 # Percentage of frames that needs to pass
 DEBUG_MODE = False
 
-# A4 size in pixels at 96 DPI
-A4_WIDTH = 794
-A4_HEIGHT = 1123
-
 # expected colors (insertion order -> mapped row-major to 3x3 grid)
 expected_colors = {
     "red":   (132, 60, 60),
@@ -35,38 +31,23 @@
 
 # we are given a 3x3 grid, we split it using 2 vertical and 2 horizontal separators
 # we also calculate the center of each grid cell for sampling from it for the test
-xs = [1.5 * A4_WIDTH / 6.0, A4_WIDTH / 2.0, 4.5 * A4_WIDTH / 6.0]
-ys = [1.5 * A4_HEIGHT / 6.0, A4_HEIGHT / 2.0, 4.5 * A4_HEIGHT / 6.0]
+xs = [1.5 * WIDTH / 6.0, WIDTH / 2.0, 4.5 * WIDTH / 6.0]
+ys = [1.5 * HEIGHT / 6.0, HEIGHT / 2.0, 4.5 * HEIGHT / 6.0]
 centers = [(x, y) for y in ys for x in xs]
 
 dev, ctx = test.find_first_device_or_exit()
 
 def is_color_close(actual, expected, tolerance):
     return all(abs(int(a) - int(e)) <= tolerance for a, e in zip(actual, expected))
 
-def compute_homography(pts):
-    """
-    Given 4 points (the detected ArUco marker centers), find the 3×3 matrix that stretches/rotates
-    the four ArUco points so they become the corners of an A4 page (used to "flatten" the page in an image)
-    """
-    pts_sorted = sorted(pts, key=lambda p: (p[1], p[0]))
-    top_left, top_right = sorted(pts_sorted[:2], key=lambda p: p[0])
-    bottom_left, bottom_right = sorted(pts_sorted[2:], key=lambda p: p[0])
-
-    src = np.array([top_left, top_right, bottom_right, bottom_left], dtype=np.float32)
-    dst = np.array([[0,0],[A4_WIDTH-1,0],[A4_WIDTH-1,A4_HEIGHT-1],[0,A4_HEIGHT-1]], dtype=np.float32)
-    M = cv2.getPerspectiveTransform(src, dst)
-    return M  # we later use M to get our roi
-
-
 def draw_debug(frame_bgr, a4_page_bgr):
     """
     Simple debug view:
       - left: camera frame
       - right: focused view on the A4 page with grid and color names
     """
-    vertical_lines = [A4_WIDTH / 3.0, 2.0 * A4_WIDTH / 3.0]
-    horizontal_lines = [A4_HEIGHT / 3.0, 2.0 * A4_HEIGHT / 3.0]
+    vertical_lines = [WIDTH / 3.0, 2.0 * WIDTH / 3.0]
+    horizontal_lines = [HEIGHT / 3.0, 2.0 * HEIGHT / 3.0]
     H, W = a4_page_bgr.shape[:2]
 
     # draw grid on a4 page image
@@ -93,63 +74,6 @@ def draw_debug(frame_bgr, a4_page_bgr):
     return np.hstack([left, right])
 
 
-def detect_a4_page(img, dict_type=cv2.aruco.DICT_4X4_1000, required_ids=(0,1,2,3)):
-    """
-    Detect ArUco markers and return center of each one
-    Returns None if not all required markers are found
-    """
-    # init aruco detector
-    aruco = cv2.aruco
-    dictionary = aruco.getPredefinedDictionary(dict_type)
-    try:
-        # new API (OpenCV >= 4.7)
-        parameters = aruco.DetectorParameters()
-        detector = aruco.ArucoDetector(dictionary, parameters)
-        corners, ids, _ = detector.detectMarkers(img)
-    except AttributeError:
-        # legacy API (OpenCV <= 4.6) - used on some of our machines
-        parameters = aruco.DetectorParameters_create()
-        corners, ids, _ = aruco.detectMarkers(img, dictionary, parameters=parameters)
-
-    if ids is None or not all(rid in ids for rid in required_ids):
-        return None
-
-    id_to_corner = dict(zip(ids.flatten(), corners))  # map id to corners
-    values = [id_to_corner[rid][0].mean(axis=0) for rid in required_ids] # for each required id, get center of marker coords
-
-    return np.array(values, dtype=np.float32)
-
-
-def find_roi_location(pipeline):
-    """
-    Returns a matrix that transforms from frame to region of interest
-    This matrix will later be used with cv2.warpPerspective()
-    """
-    # stream until page found
-    page_pts = None
-    start_time = time.time()
-    while page_pts is None and time.time() - start_time < 5:
-        frames = pipeline.wait_for_frames()
-        color_frame = frames.get_color_frame()
-        img_bgr = np.asanyarray(color_frame.get_data())
-
-        if DEBUG_MODE:
-            cv2.imshow("PageDetect - waiting for page", img_bgr)
-            cv2.waitKey(1)
-
-        page_pts = detect_a4_page(img_bgr)
-
-    if page_pts is None:
-        log.e("Failed to detect page within timeout")
-        test.fail()
-        raise Exception("Page not found")
-
-    # page found - use it to calculate transformation matrix from frame to region of interest
-    M = compute_homography(page_pts)
-    cv2.destroyAllWindows()
-    return M, page_pts
-
-
 def is_cfg_supported(resolution, fps):
     color_sensor = dev.first_color_sensor()
     for p in color_sensor.get_stream_profiles():
@@ -170,35 +94,32 @@ def run_test(resolution, fps):
     for i in range(30):  # skip initial frames
         pipeline.wait_for_frames()
     try:
-
         # find region of interest (page) and get the transformation matrix
-        # page_pts is only used for debug display
-        M, page_pts = find_roi_location(pipeline)
+        find_roi_location(pipeline, (0, 1, 2, 3), DEBUG_MODE) # markers in the lab are 0,1,2,3
 
         # sampling loop
         for i in range(NUM_FRAMES):
             frames = pipeline.wait_for_frames()
             color_frame = frames.get_color_frame()
             img_bgr = np.asanyarray(color_frame.get_data())
 
-            # use M to get the region of interest - our colored grid printed in the lab
-            a4_bgr = cv2.warpPerspective(img_bgr, M, (A4_WIDTH, A4_HEIGHT))
+            color_frame_roi = get_roi_from_frame(color_frame)
 
             # sample each grid center and compare to expected color by row-major insertion order
             for idx, (x, y) in enumerate(centers):
                 color = color_names[idx] if idx < len(color_names) else str(idx)
                 expected_rgb = expected_colors[color]
                 x = int(round(x))
                 y = int(round(y))
-                b, g, r = (int(v) for v in a4_bgr[y, x])  # stream is BGR, convert to RGB
+                b, g, r = (int(v) for v in color_frame_roi[y, x])  # stream is BGR, convert to RGB
                 pixel = (r, g, b)
                 if is_color_close(pixel, expected_rgb, COLOR_TOLERANCE):
                     color_match_count[color] += 1
                 else:
                     log.d(f"Frame {i} - {color} at ({x},{y}) sampled: {pixel} too far from expected {expected_rgb}")
 
             if DEBUG_MODE:
-                dbg = draw_debug(img_bgr, a4_bgr)
+                dbg = draw_debug(img_bgr, color_frame_roi)
                 cv2.imshow("PageDetect - camera | A4", dbg)
                 cv2.waitKey(1)
 

unit-tests/live/image-quality/test-basic-depth.py

@@ -3,21 +3,18 @@
 
 # test:device D400*
 # test:donotrun
+
 import pyrealsense2 as rs
 from rspy import log, test
 import numpy as np
+import cv2
 import time
+from iq_helper import find_roi_location, get_roi_from_frame
 
-NUM_FRAMES = 10 # Number of frames to check
+NUM_FRAMES = 100 # Number of frames to check
 DEPTH_TOLERANCE = 0.05  # Acceptable deviation from expected depth in meters
 FRAMES_PASS_THRESHOLD =0.8 # Percentage of frames that needs to pass
-
-# Known pixel positions and expected depth values (in meters)
-depth_points = {
-    "front": ((538, 315), 0.524 ),
-    "middle": ((437, 252), 0.655),
-    "back": ((750, 63), 1.058 )
-}
+DEBUG_MODE = False
 
 test.start("Basic Depth Image Quality Test")
 
@@ -26,24 +23,35 @@
 
     pipeline = rs.pipeline(ctx)
     cfg = rs.config()
-    cfg.enable_stream(rs.stream.depth, 848, 480, rs.format.z16, 30)
+    cfg.enable_stream(rs.stream.depth, 640, 480, rs.format.z16, 30)
+    cfg.enable_stream(rs.stream.infrared, 1, 640, 480, rs.format.y8, 30)  # needed for finding the ArUco markers
     profile = pipeline.start(cfg)
-    frames = pipeline.wait_for_frames()
     time.sleep(2)
 
     depth_sensor = profile.get_device().first_depth_sensor()
     depth_scale = depth_sensor.get_depth_scale()
 
-    depth_passes = {name: 0 for name in depth_points}
-
+    # find region of interest (page) and get the transformation matrix
+    find_roi_location(pipeline, (4,5,6,7), DEBUG_MODE)  # markers in the lab are 4,5,6,7
+    depth_passes = {}
     for i in range(NUM_FRAMES):
         frames = pipeline.wait_for_frames()
         depth_frame = frames.get_depth_frame()
+        infrared_frame = frames.get_infrared_frame()
         if not depth_frame:
             continue
 
-        depth_image = np.asanyarray(depth_frame.get_data())
+        depth_image = get_roi_from_frame(depth_frame)
 
+        # Known pixel positions and expected depth values (in meters)
+        # Using temporary values until setup in lab is completed
+        h, w = depth_image.shape
+        depth_points = {
+            "cube":  ((h // 2, w // 2), 0.45), # center of page, cube at 0.45m
+            "background": ((h // 2, int(w * 0.1)), 0.6), # left edge, background at 0.6m
+        }
+        if not depth_passes:
+            depth_passes = {name: 0 for name in depth_points}
         for point_name, ((x, y), expected_depth) in depth_points.items():
             raw_depth = depth_image[y, x]
             depth_value = raw_depth * depth_scale  # Convert to meters
@@ -53,14 +61,31 @@
             else:
                 log.d(f"Frame {i} - {point_name} at ({x},{y}): {depth_value:.3f}m ≠ {expected_depth:.3f}m")
 
+        if DEBUG_MODE:
+            # display IR image along with transformed view of IR, get_roi_from_frame(infrared_frame)
+            infrared_np = np.asanyarray(infrared_frame.get_data())
+            w, h = infrared_np.shape
+            dbg_resized = cv2.resize(get_roi_from_frame(infrared_frame), (h, w))
+
+            dbg = np.hstack([infrared_np, dbg_resized])
+            cv2.imshow("Depth IQ - IR | Depth", dbg)
+            cv2.waitKey(1)
+
+    # wait for close
+    if DEBUG_MODE:
+        cv2.waitKey(0)
+
     # Check that each point passed the threshold
     min_passes = int(NUM_FRAMES * FRAMES_PASS_THRESHOLD)
     for point_name, count in depth_passes.items():
         log.i(f"{point_name.title()} passed in {count}/{NUM_FRAMES} frames")
         test.check(count >= min_passes)
 
 except Exception as e:
-    test.unexpected_exception(e)
+    test.fail()
+    raise e
+finally:
+    cv2.destroyAllWindows()
 
 pipeline.stop()
 test.finish()