Ci: testing frame order and fps

test/groovy/LRS_libci_pipeline.groovy

@@ -0,0 +1,62 @@
+import org.jenkinsci.plugins.workflow.steps.FlowInterruptedException
+
+pipeline {
+	agent { label 'rs-orin-01.realsenseai.com' }
+
+	options {
+		timestamps()
+			timeout(time: 30, unit: 'MINUTES')
+	}
+
+	parameters {
+		booleanParam(name: 'REBOOT', defaultValue: false)
+	}
+
+	stages {
+		stage('Get artifacts') {
+			when {
+				expression { params.REBOOT == true }
+			}
+			steps {
+				script {
+					copyArtifacts filter: '**/*.tar.bz2',
+						      projectName: 'D4xx_Kernel_Module_Jetson_JP6',
+						      flatten: true,
+						      target: 'artifacts/'
+				}
+			}
+		}
+		stage('Install artifacts') {
+			when {
+				expression { params.REBOOT == true }
+			}
+			steps {
+				sh """#!/bin/sh
+					tar -xf artifacts/rootfs.tar.bz2
+					# external script on agent to install artifacts
+					sudo install.tegra.artifacts.sh
+				"""
+				script {
+					build job: env.JOB_NAME,
+					      wait: false,
+					      parameters: [ booleanParam(name: 'REBOOT', value: false) ]
+				}
+			}
+			post {
+				success {
+					sh 'nohup sudo reboot &'
+				}
+			}
+		}
+		stage('Pytest') {
+			when {
+				expression { params.REBOOT == false }
+			}
+			steps {
+				sh """#!/bin/sh
+					pytest --tb=no -s test
+				"""
+			}
+		}
+	}
+}

test/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+markers =
+        d457: realsense 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,97 @@
+import subprocess
+import pytest
+import re
+
+@pytest.mark.d457
+@pytest.mark.parametrize("frames", {150})
+@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 = 4.0 * frames / FPS
+                output = subprocess.run(cmd,
+                                        check=True,
+                                        text=True,
+                                        capture_output=True,
+                                        timeout=timeout).stderr.splitlines()
+                last = None
+                skip = True    # skip first FPS measurement
+                kpi = 5 # [%]
+                count = 0
+                for line in output:
+                    m = re.search(r"cap dqbuf:.*seq:\s*(\d*) bytesused:", line)
+                    if m:
+                        count += 1
+                        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"delta:\s*(\d+\.\d+) ms", line)
+                        if m:
+                            fps = 1000 / float(m.group(1))
+                            # print(f"/{fps:.2f}", end='')
+                            if not skip:
+                                assert fps > FPS * (1 - kpi/100), f"FPS too low: {fps:.2f}/{FPS}"
+                                assert fps < FPS * (1 + kpi/100), f"FPS too high: {fps:.2f}/{FPS}"
+                            else:
+                                # print('?', end='')
+                                skip = False
+                        # print(end=',')
+                        last = frame
+                print()
+                assert last, "No frames arrived"
+                assert count == frames, f"Missing frames: {count} < {frames}"
+    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