PR #14420 from ttkhuong: add test-fps-manual-exposure.py

Nir-Az · web-flow · commit 8efb86f4abdb · 2025-11-17T17:26:54.000+02:00
diff --git a/unit-tests/live/frames/test-fps-manual-exposure.py b/unit-tests/live/frames/test-fps-manual-exposure.py
@@ -0,0 +1,158 @@
+# License: Apache 2.0. See LICENSE file in root directory.
+# Copyright(c) 2025 RealSense, Inc. All Rights Reserved.
+
+# Currently, we exclude D555 as it's failing
+# test:device each(D400*)
+# test:device each(D500*) !D555
+# test:donotrun:!nightly
+
+import pyrealsense2 as rs
+from rspy.stopwatch import Stopwatch
+from rspy import test, log
+import time
+import platform
+import fps_helper
+
+# This test mirrors test-fps.py but forces manual exposure where supported
+# Start depth + color streams and measure frame frequency using sensor API.
+# Verify that actual fps is as requested
+
+def set_exposure_half_frame_time(sensor, requested_fps):
+    """Set sensor exposure to half the frame time for the given requested_fps.
+    Returns the exposure value that was set (or None if not supported/failed).
+    """
+    if not sensor.supports(rs.option.exposure):
+        return None
+    try:
+        r = sensor.get_option_range(rs.option.exposure)
+        # frame time in seconds
+        frame_time_s = 1.0 / float(requested_fps)
+        desired_s = frame_time_s / 2.0
+
+        # Decide units based on sensor type: depth -> microseconds; RGB -> 100 microseconds units
+        is_color = False
+        try:
+            for p in sensor.profiles:
+                if p.stream_type() == rs.stream.color:
+                    is_color = True
+                    break
+        except Exception:
+            # fallback: assume depth unless explicitly color
+            is_color = False
+
+        if is_color:
+            # RGB sensors: many backends treat exposure as units of 100 microseconds
+            desired_units = desired_s * 1e6 / 100.0  # convert seconds -> 100-microsecond units
+        else:
+            # Depth sensors: use microseconds directly
+            desired_units = desired_s * 1e6
+
+        # clamp to reported range
+        if desired_units < r.min:
+            desired_units = r.min
+        if desired_units > r.max:
+            desired_units = r.max
+
+        # Set the option (use float/int as appropriate)
+        try:
+            sensor.set_option(rs.option.exposure, desired_units)
+        except Exception:
+            sensor.set_option(rs.option.exposure, int(desired_units))
+        return desired_s * 1e6  # return in microseconds
+    except Exception:
+        return None
+
+
+delta_Hz = 1
+tested_fps = [5, 6, 15, 30, 60, 90]
+time_to_test_fps = [25, 20, 13, 10, 5, 4]
+test.check_equal( len(tested_fps), len(time_to_test_fps) )
+
+dev, _ = test.find_first_device_or_exit()
+product_line = dev.get_info(rs.camera_info.product_line)
+camera_name = dev.get_info(rs.camera_info.name)
+firmware_version = dev.get_info(rs.camera_info.firmware_version)
+serial = dev.get_info(rs.camera_info.serial_number)
+log.i(f"Device: {camera_name}, product_line: {product_line}, serial: {serial}, firmware: {firmware_version}")
+
+#####################################################################################################
+test.start("Testing depth fps (manual exposure) " + product_line + " device - "+ platform.system() + " OS")
+
+ds = dev.first_depth_sensor()
+# Prepare depth sensor for manual exposure if supported
+if product_line == "D400":
+    if ds.supports(rs.option.enable_auto_exposure):
+        # disable auto exposure to force manual exposure
+        ds.set_option(rs.option.enable_auto_exposure, 0)
+
+for i in range(len(tested_fps)):
+    requested_fps = tested_fps[i]
+    try:
+        dp = next(p for p in ds.profiles
+                  if p.fps() == requested_fps
+                  and p.stream_type() == rs.stream.depth
+                  and p.format() == rs.format.z16
+                  and ((p.as_video_stream_profile().height() == 720 and p.fps() != 60) if "D585S" in camera_name else True))
+
+    except StopIteration:
+        log.i("Requested fps: {:.1f} [Hz], not supported".format(requested_fps))
+    else:
+        # set exposure to half frame time for this requested fps if supported
+        exposure_val = set_exposure_half_frame_time(ds, requested_fps)
+        # use shared fps helper which expects a dict of {sensor: [profile]}
+        fps_helper.TIME_TO_COUNT_FRAMES = time_to_test_fps[i]
+        fps_dict = fps_helper.measure_fps({ds: [dp]})
+        fps = fps_dict.get(dp.stream_name(), 0)
+        log.i("Exposure: {:.1f} [msec], requested fps: {:.1f} [Hz], actual fps: {:.1f} [Hz]".format((exposure_val or 0)/1000, requested_fps, fps))
+        delta_Hz = requested_fps * 0.05 # Validation KPI is 5%
+        test.check(fps <= (requested_fps + delta_Hz) and fps >= (requested_fps - delta_Hz))
+test.finish()
+
+
+#####################################################################################################
+test.start("Testing color fps (manual exposure) " + product_line + " device - "+ platform.system() + " OS")
+
+product_name = dev.get_info(rs.camera_info.name)
+cs = None
+try:
+    cs = dev.first_color_sensor()
+except RuntimeError as rte:
+    if 'D421' not in product_name and 'D405' not in product_name: # Cameras with no color sensor may fail.
+        test.unexpected_exception()
+
+if cs:
+    # Try to force manual exposure on color sensor
+    if product_line == "D400":
+        if cs.supports(rs.option.enable_auto_exposure):
+            cs.set_option(rs.option.enable_auto_exposure, 0)
+        if cs.supports(rs.option.auto_exposure_priority):
+            cs.set_option(rs.option.auto_exposure_priority, 0) # AE priority should be 0 for constant FPS
+
+    for i in range(len(tested_fps)):
+        requested_fps = tested_fps[i]
+        try:
+            # collect matching color profiles and pick median resolution
+            candidates = [p for p in cs.profiles
+                          if p.fps() == requested_fps
+                          and p.stream_type() == rs.stream.color
+                          and p.format() == rs.format.rgb8]
+            if not candidates:
+                raise StopIteration
+            candidates.sort(key=lambda pr: pr.as_video_stream_profile().width() * pr.as_video_stream_profile().height())
+            cp = candidates[len(candidates)//2]
+
+        except StopIteration:
+            log.i("Requested fps: {:.1f} [Hz], not supported".format(requested_fps))
+        else:
+            # set exposure to half frame time for this requested fps if supported
+            exposure_val = set_exposure_half_frame_time(cs, requested_fps)
+            fps_dict = fps_helper.measure_fps({cs: [cp]})
+            fps = fps_dict.get(cp.stream_name(), 0)
+            log.i("Exposure: {:.1f} [msec], requested fps: {:.1f} [Hz], actual fps: {:.1f} [Hz]".format((exposure_val or 0)/1000, requested_fps, fps))
+            delta_Hz = requested_fps * (0.10 if requested_fps == 5 else 0.05) # Validation KPI is 5% for all non 5 FPS rate
+            test.check(fps <= (requested_fps + delta_Hz) and fps >= (requested_fps - delta_Hz))
+
+test.finish()
+
+#####################################################################################################
+test.print_results_and_exit()
diff --git a/unit-tests/live/frames/test-fps.py b/unit-tests/live/frames/test-fps.py
@@ -11,58 +11,11 @@
 from rspy import test, log
 import time
 import platform
+import fps_helper
 
 # Start depth + color streams and measure frame frequency using sensor API.
 # Verify that actual fps is as requested
 
-global first_frame_seconds
-
-def measure_fps(sensor, profile, seconds_to_count_frames = 10):
-    """
-    Wait a few seconds to be sure that frames are at steady state after start
-    Count number of received frames for seconds_to_count_frames seconds and compare actual fps to requested fps
-    """
-    seconds_till_steady_state = 4
-
-    steady_state = False
-    first_frame_received = False
-    frames_received = 0
-    first_frame_stopwatch = Stopwatch()
-    prev_frame_number = 0
-
-    def frame_cb(frame):
-        global first_frame_seconds
-        nonlocal steady_state, frames_received, first_frame_received, prev_frame_number
-        current_frame_number = frame.get_frame_number()
-        if not first_frame_received:
-            first_frame_seconds = first_frame_stopwatch.get_elapsed()
-            first_frame_received = True
-        else:
-            if current_frame_number > prev_frame_number + 1:
-                log.w( f'Frame drop detected. Current frame number {current_frame_number} previous was {prev_frame_number}' )
-        if steady_state:
-            frames_received += 1
-        prev_frame_number = current_frame_number
-
-    sensor.open(profile)
-    sensor.start(frame_cb)
-    first_frame_stopwatch.reset()
-
-    time.sleep(seconds_till_steady_state)
-
-    steady_state = True
-
-    time.sleep(seconds_to_count_frames) # Time to count frames
-
-    steady_state = False # Stop counting
-
-    sensor.stop()
-    sensor.close()
-
-    fps = frames_received / seconds_to_count_frames
-    return fps
-
-
 delta_Hz = 1
 tested_fps = [5, 6, 15, 30, 60, 90]
 time_to_test_fps = [25, 20, 13, 10, 5, 4]
@@ -96,8 +49,10 @@ def frame_cb(frame):
     except StopIteration:
         log.i("Requested fps: {:.1f} [Hz], not supported".format(requested_fps))
     else:
-        fps = measure_fps(ds, dp, time_to_test_fps[i])
-        log.i("Requested fps: {:.1f} [Hz], actual fps: {:.1f} [Hz]. Time to first frame {:.6f}".format(requested_fps, fps, first_frame_seconds))
+        fps_helper.TIME_TO_COUNT_FRAMES = time_to_test_fps[i]
+        fps_dict = fps_helper.measure_fps({ds: [dp]})
+        fps = fps_dict.get(dp.stream_name(), 0)
+        log.i("Requested fps: {:.1f} [Hz], actual fps: {:.1f} [Hz]".format(requested_fps, fps))
         delta_Hz = requested_fps * 0.05 # Validation KPI is 5%
         test.check(fps <= (requested_fps + delta_Hz) and fps >= (requested_fps - delta_Hz))
 test.finish()
@@ -133,8 +88,10 @@ def frame_cb(frame):
         except StopIteration:
             log.i("Requested fps: {:.1f} [Hz], not supported".format(requested_fps))
         else:
-            fps = measure_fps(cs, cp, time_to_test_fps[i])
-            log.i("Requested fps: {:.1f} [Hz], actual fps: {:.1f} [Hz]. Time to first frame {:.6f}".format(requested_fps, fps, first_frame_seconds))
+            fps_helper.TIME_TO_COUNT_FRAMES = time_to_test_fps[i]
+            fps_dict = fps_helper.measure_fps({cs: [cp]})
+            fps = fps_dict.get(cp.stream_name(), 0)
+            log.i("Requested fps: {:.1f} [Hz], actual fps: {:.1f} [Hz]".format(requested_fps, fps))
             delta_Hz = requested_fps * (0.10 if requested_fps == 5 else 0.05) # Validation KPI is 5% for all non 5 FPS rate
             test.check(fps <= (requested_fps + delta_Hz) and fps >= (requested_fps - delta_Hz))
 
