Ci: testing frame order and fps

test/groovy/LRS_libci_pipeline.groovy

@@ -0,0 +1,20 @@
+import org.jenkinsci.plugins.workflow.steps.FlowInterruptedException
+
+pipeline {
+    agent { label 'jetson && jetpack && camera' }
+
+    options { 
+        timestamps()
+        timeout(time: 10, unit: 'MINUTES')
+    }
+
+    stages {
+        stage('pytest') {
+            steps {
+                script {
+					sh 'pytest --tb=no test'
+                }
+            }
+        }
+    }
+}

test/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+markers =
+        d457: realsesne D457 camera
+

test/run_ci.py

@@ -46,7 +46,7 @@ def run_test(cmd):
                 timeout=200,
                 check=True )
     except Exception as e:
-        print( "Exception occurred.")
+        print("Exception occurred: {}".format( e ))
 
 
 def run_tests_on_d457():     
@@ -73,6 +73,7 @@ def run_tests_on_d457():
     except getopt.GetoptError as err:
         print( err )
         usage()
+        sys.exit ( 1 )
 
     for opt, arg in opts:
         if opt in ('-h', '--help'):

test/test_fps.py

@@ -0,0 +1,90 @@
+import subprocess
+import pytest
+import re
+
+@pytest.mark.d457
+@pytest.mark.parametrize("frames", {20})
+@pytest.mark.parametrize("device", {'0', '2'})
+def test_fps(device, frames):
+    try:
+        print(f"\nDevice: {device}")
+        formats = get_formats(device)
+        for w, h in formats:
+            print(f"Format: {w}x{h}")
+            for FPS in formats[(w, h)]:
+                cmd = [ "v4l2-ctl",
+                       f"-d{device}",
+                       f"--set-fmt-video=width={w},height={h}",
+                       ]
+                subprocess.check_call(cmd)
+                print(f"FPS/{FPS}:", end='')
+                cmd = [ "v4l2-ctl",
+                       f"-d{device}",
+                       "-p",
+                       f"{FPS}",
+                       ]
+                subprocess.check_call(cmd, stdout=subprocess.DEVNULL)
+                cmd = [ "v4l2-ctl",
+                       f"-d{device}",
+                       "--stream-mmap",
+                       "--stream-count",
+                       f"{frames}",
+                       "--verbose",
+                       ]
+                timeout = 5.0 * frames / FPS
+                output = subprocess.run(cmd,
+                                        check=True,
+                                        text=True,
+                                        capture_output=True,
+                                        timeout=timeout).stderr.splitlines()
+                last = None
+                kpi = 5 # [%]
+                for line in output:
+                    m = re.search(r"cap dqbuf:.*seq:\s*(\d*)\s*bytesused:.*", line)
+                    if m:
+                        frame = int(m.group(1))
+                        print(f"{frame}", end='')
+                        if last:
+                            assert frame > last, f"Repeated frame: {frame}"
+                            assert (frame - last) < 3 , f"Frames dropped between: {last} and {frame}"
+                        m = re.search(r"cap dqbuf:.*bytesused:.*fps:\s*(\d+\.\d+)\s*.*", line)
+                        if m:
+                            fps = float(m.group(1))
+                            print(f"/{fps}", end=',')
+                            assert fps > FPS * (1 - kpi/100), f"FPS too low: {fps}/{FPS}"
+                            assert fps < FPS * (1 + kpi/100), f"FPS too high: {fps}/{FPS}"
+                        else:
+                            print(end=',')
+                        last = frame
+                print()
+                assert last, "No frames arrived"
+    except subprocess.TimeoutExpired:
+        assert False, "No frames arrived"
+
+def get_formats(device):
+    cmd = [ "v4l2-ctl",
+           "-d" + device,
+           "--list-formats-ext",
+           ]
+    output = subprocess.run(cmd,
+                            check=True,
+                            text=True,
+                            capture_output=True
+                            ).stdout.splitlines()
+    formats = {}
+    last = None
+    for line in output:
+        m = re.search(r"\s*Size: Discrete\s*(\d+)x(\d+)", line)
+        if m:
+            w = int(m.group(1))
+            h = int(m.group(2))
+            last = (w ,h)
+            if not last in formats:
+                formats[last] = set()
+            continue
+        m = re.search(r"\s*Interval: Discrete.*\((\d+\.\d+)\s+fps\)", line)
+        if m:
+            fps = float(m.group(1))
+            if last:
+                formats[last].add(fps)
+    return formats

test/test_fw_version.py

@@ -5,14 +5,16 @@
 @pytest.mark.parametrize("device", {'0'})
 def test_fw_version(device):
     try:
-        result = subprocess.check_call(["v4l2-ctl", "-d"+device, "-C", "fw_version"])
+        key = "fw_version"
+        result = subprocess.check_call(["v4l2-ctl", "-d"+device, "-C", key])
         assert result == 0
 
-        std_output = subprocess.check_output(["v4l2-ctl", "-d"+device, "-C", "fw_version"])
-        assert "fw version: " in std_output, "Couldn't fetch FW version"
+        std_output = subprocess.check_output(["v4l2-ctl", "-d"+device, "-C", key])
+        key += ": "
+        assert key in std_output.decode(), "Couldn't fetch FW version"
 
         # Remove the 'fw version: ' string from std output
-        fw_version = int(std_output.replace("fw version: ", ""))
+        fw_version = int(std_output.decode().replace(key, ""))
 
         fw_version_str = str(fw_version>>24 & 0xFF) + "." + str(fw_version>>16 & 0xFF) + "." + str(fw_version>>8 & 0xFF)  + "." + str(fw_version & 0xFF)
         print ("fw_version:", fw_version_str)
@@ -21,14 +23,14 @@ def test_fw_version(device):
         assert fw_version == (fw_version & 0x05FFFFFF), "Expected FW version is 5.x.x.x, but received {}".format(fw_version_str)
 
         # Get DFU device name
-        dfu_device = subprocess.check_output(["ls", "/sys/class/d4xx-class/"])
+        dfu_device = subprocess.check_output(["ls", "/sys/class/d4xx-class/"]).decode()
         assert "d4xx-dfu-" in dfu_device, "D4xx DFU device not found"
 
         # Get FW version from DFU device info
-        dfu_device_info = subprocess.check_output(["cat", "/dev/"+dfu_device.strip()])
+        dfu_device_info = subprocess.check_output(["cat", "/dev/"+dfu_device.strip()]).decode()
 
         # Check whether the DFU info also has same FW version
         assert fw_version_str in dfu_device_info, "FW versions read through v4l2-ctl utility and DFU device info doesn't match"
 
     except Exception as e:
-        assert False, e
+        assert False, "Exception caught during test: {}".format(e)

test/test_metadata/build.sh

@@ -4,6 +4,4 @@ set -x
 
 rm -rf test_metadata test_metadata.o
 
-gcc test_metadata.c framesextract.c -o test_metadata
-
-./test_metadata "$1" "$2" "$3" "$4" "$5" "$6" "$7" "$8"
+gcc -g -O0 test_metadata.c framesextract.c -o test_metadata