d555 -> d500, fix recovery issues

Author: AviaAv

third-party/realdds/src/dds-device-watcher.cpp

@@ -68,9 +68,11 @@ dds_device_watcher::dds_device_watcher( std::shared_ptr< dds_participant > const
                             }
                             continue;
                         }
-                        else if( device.alive )
+                        else if( device.alive && guid == device.writer_guid)
                         {
                             // We already know about this device; likely this was a broadcast meant for someone else
+                            // If guid mismatches the device was reset without us getting a subscription 
+                            // lost callback in time - handle like old device coming back to life
                             continue;
                         }
                         else if( device.alive = device.in_use.lock() )

unit-tests/run-unit-tests.py

@@ -55,7 +55,7 @@ def usage():
     print( '        --no-reset           Do not try to reset any devices, with or without a hub' )
     print( '        --hub-reset          If a hub is available, reset the hub itself' )
     print( '        --custom-fw-d400          If custom fw provided flash it if its different that the current fw installed' )
-    print( '        --custom-fw-d555          If custom fw provided flash it if its different that the current fw installed' )
+    print( '        --custom-fw-d500          If custom fw provided flash it if its different that the current fw installed' )
     print( '        --rslog              Enable LibRS logging (LOG_DEBUG etc.) to console in each test' )
     print( '        --skip-disconnected  Skip live test if required device is disconnected (only applies w/o a hub)' )
     print( '        --test-dir <>        Path to test dir; default: librealsense/unit-tests' )
@@ -85,7 +85,7 @@ def usage():
     opts, args = getopt.getopt( sys.argv[1:], 'hvqr:st:',
                                 longopts=['help', 'verbose', 'debug', 'quiet', 'regex=', 'stdout', 'tag=', 'list-tags',
                                           'list-tests', 'no-exceptions', 'context=', 'repeat=', 'retry=', 'config=', 'no-reset', 'hub-reset',
-                                          'rslog', 'skip-disconnected', 'live', 'not-live', 'device=', 'exclude-device=', 'test-dir=','skip-regex=','custom-fw-d400=','custom-fw-d555='] )
+                                          'rslog', 'skip-disconnected', 'live', 'not-live', 'device=', 'exclude-device=', 'test-dir=','skip-regex=','custom-fw-d400=','custom-fw-d500='] )
 except getopt.GetoptError as err:
     log.e( err )  # something like "option -a not recognized"
     usage()
@@ -105,7 +105,7 @@ def usage():
 hub_reset = False
 skip_disconnected = False
 custom_fw_path=''
-custom_fw_d555_path=''
+custom_fw_d500_path=''
 rslog = False
 only_live = False
 only_not_live = False
@@ -180,10 +180,10 @@ def usage():
         skip_regex = arg
     elif opt == '--custom-fw-d400':
         custom_fw_path = arg  # Store the custom firmware path
-        log.i(f"custom firmware path was provided ${custom_fw_path}")
-    elif opt == '--custom-fw-d555':
-        custom_fw_d555_path = arg  # Store the custom D555 firmware path
-        log.i(f"custom D555 firmware path was provided ${custom_fw_d555_path}")
+        log.i(f"custom D400 firmware path was provided ${custom_fw_path}")
+    elif opt == '--custom-fw-d500':
+        custom_fw_d500_path = arg  # Store the custom D500 firmware path
+        log.i(f"custom D500 firmware path was provided ${custom_fw_d500_path}")
 
 def find_build_dir( dir ):
     """
@@ -474,9 +474,9 @@ def test_wrapper_( test, configuration=None, repetition=1, curr_retry=0, max_ret
     if test.name == "test-fw-update" and custom_fw_path:
         opts.append('--custom-fw-d400')
         opts.append(custom_fw_path)
-    if test.name == "test-fw-update" and custom_fw_d555_path:
-        opts.append('--custom-fw-d555')
-        opts.append(custom_fw_d555_path)
+    if test.name == "test-fw-update" and custom_fw_d500_path:
+        opts.append('--custom-fw-d500')
+        opts.append(custom_fw_d500_path)
     try:
         test.run_test( configuration = configuration, log_path = log_path, opts = opts )
     except FileNotFoundError as e:

unit-tests/test-fw-update.py

@@ -21,19 +21,10 @@
 
 # Parse command-line arguments
 parser = argparse.ArgumentParser(description="Test firmware update")
-parser.add_argument('--custom-fw-d400', type=str, help='Path to custom firmware file')
-parser.add_argument('--custom-fw-d555', type=str, help='Path to custom D555 firmware file')
+parser.add_argument('--custom-fw-d400', type=str, help='Path to custom D400 firmware file')
+parser.add_argument('--custom-fw-d500', type=str, help='Path to custom D500 firmware file')
 args = parser.parse_args()
 
-custom_fw_d400_path = args.custom_fw_d400
-custom_fw_d555_path = args.custom_fw_d555
-if custom_fw_d400_path:
-    log.i(f"Custom D400 firmware path provided: {custom_fw_d400_path}")
-elif custom_fw_d555_path:
-    log.i(f"Custom D555 firmware path provided: {custom_fw_d555_path}")
-else:
-    log.i(f"No Custom firmware path provided. using bundled firmware")
-
 
 def send_hardware_monitor_command(device, command):
     # byte_index = -1
@@ -61,7 +52,7 @@ def extract_version_from_filename(file_path):
         return None
 
     filename = os.path.basename(file_path)
-    match = re.search(r'(\d+)[._](\d+)[._](\d+)[._](\d+)', filename)
+    match = re.search(r'[-_](\d+)[._](\d+)[._](\d+)[._](\d+)', filename)
     if match:
         groups = match.groups()
         if groups[3] == '0':
@@ -77,15 +68,14 @@ def extract_version_from_filename(file_path):
 
 def get_update_counter(device):
     product_line = device.get_info(rs.camera_info.product_line)
-    product_name = device.get_info(rs.camera_info.name)
     opcode = 0x09
     start_index = 0x30
     size = None
 
     if product_line == "D400":
         size = 0x2
-    elif "D555" in product_name:
-        return 0 # D555 does not have update counter
+    elif product_line == "D500":
+        return 0 # D500 do not have update counter
     else:
         log.f( "Incompatible product line:", product_line )
 
@@ -157,11 +147,22 @@ def find_image_or_exit( product_name, fw_version_regex = r'(\d+\.){3}(\d+)' ):
 device, ctx = test.find_first_device_or_exit()
 product_line = device.get_info( rs.camera_info.product_line )
 product_name = device.get_info( rs.camera_info.name )
-serial_number = device.get_info(rs.camera_info.serial_number)
-log.d( 'product line:', product_line, 'serial number:', serial_number )
+log.d( 'product line:', product_line )
 ###############################################################################
 #
 
+current_fw_version = rsutils.version( device.get_info( rs.camera_info.firmware_version ))
+log.d( 'current FW version:', current_fw_version )
+
+# Determine which firmware to use based on product
+bundled_fw_version = rsutils.version("")
+custom_fw_path = None
+custom_fw_version = None
+if product_line == "D400" and args.custom_fw_d400:
+    custom_fw_path = args.custom_fw_d400
+elif product_line == "D500" and args.custom_fw_d500:
+    custom_fw_path = args.custom_fw_d500
+
 
 test.start( "Update FW" )
 # check if recovery. If so recover
@@ -170,8 +171,10 @@ def find_image_or_exit( product_name, fw_version_regex = r'(\d+\.){3}(\d+)' ):
     log.d( "recovering device ..." )
     try:
         # always flash signed fw when device on recovery before flashing anything else
-        image_file = find_image_or_exit(product_name)
-        cmd = [fw_updater_exe, '-r', '-f', image_file, '-s', serial_number]
+        # on D500 we currently do not have bundled FW
+        image_file = find_image_or_exit(product_name) if not product_line == "D500" else custom_fw_path
+        cmd = [fw_updater_exe, '-r', '-f', image_file]
+        del device, ctx
         log.d( 'running:', cmd )
         subprocess.run( cmd )
         recovered = True
@@ -186,41 +189,27 @@ def find_image_or_exit( product_name, fw_version_regex = r'(\d+\.){3}(\d+)' ):
         log.f( "Unexpected error while trying to recover device:", e )
     else:
         device, ctx = test.find_first_device_or_exit()
-        # check we got the same serial number
-        new_serial_number = device.get_info(rs.camera_info.serial_number)
-        if new_serial_number != serial_number:
-            log.f( "Recovered device has different serial number:", new_serial_number )
+        current_fw_version = rsutils.version(device.get_info(rs.camera_info.firmware_version))
+        log.d("FW version after recovery:", current_fw_version)
 
-current_fw_version = rsutils.version( device.get_info( rs.camera_info.firmware_version ))
-log.d( 'current FW version:', current_fw_version )
-bundled_fw_version = rsutils.version("")
-if device.supports( rs.camera_info.recommended_firmware_version ): # currently, D500 does not support recommended FW
-    bundled_fw_version = rsutils.version( device.get_info( rs.camera_info.recommended_firmware_version ) )
-    log.d( 'bundled FW version:', bundled_fw_version )
-custom_fw_d400_version = extract_version_from_filename(custom_fw_d400_path)
-log.d( 'custom FW D400 version:', custom_fw_d400_version )
-custom_fw_d555_version = extract_version_from_filename(custom_fw_d555_path)
-log.d( 'custom FW D555 version:', custom_fw_d555_version )
-
-# Determine which custom firmware to use based on product
-custom_fw_path = None
-custom_fw_version = None
-if product_line == "D400" and custom_fw_d400_path:
-    custom_fw_path = custom_fw_d400_path
-    custom_fw_version = custom_fw_d400_version
-elif "D555" in product_name and custom_fw_d555_path:
-    custom_fw_path = custom_fw_d555_path
-    custom_fw_version = custom_fw_d555_version
+
+if custom_fw_path:
+    custom_fw_version = extract_version_from_filename(custom_fw_path)
+    log.d('Using custom FW version: ', custom_fw_version)
+elif device.supports(rs.camera_info.recommended_firmware_version):  # currently, D500 does not support recommended FW
+    log.i(f"No Custom firmware path provided. using bundled firmware")
+    bundled_fw_version = rsutils.version(device.get_info(rs.camera_info.recommended_firmware_version))
+    log.d('bundled FW version:', bundled_fw_version)
+else:
+    log.w("No custom FW provided and no bundled FW version available; skipping FW update test")
+    exit(0)
 
 if (current_fw_version == bundled_fw_version and not custom_fw_path) or \
    (current_fw_version == custom_fw_version):
     if recovered or 'nightly' not in test.context:
         log.d('versions are same; skipping FW update')
         test.finish()
         test.print_results_and_exit()
-else:
-    # It is expected that, post-recovery, the FW versions will be the same
-    test.check(not recovered, on_fail=test.ABORT)
 
 update_counter = get_update_counter( device )
 log.d( 'update counter:', update_counter )
@@ -236,7 +225,7 @@ def find_image_or_exit( product_name, fw_version_regex = r'(\d+\.){3}(\d+)' ):
 image_file = find_image_or_exit(product_name, fw_version_regex) if not custom_fw_path else custom_fw_path
 # finding file containing image for FW update
 
-cmd = [fw_updater_exe, '-f', image_file, '-s', serial_number]
+cmd = [fw_updater_exe, '-f', image_file]
 if custom_fw_path:
     # Add '-u' only if the path doesn't include 'signed'
     if ('signed' not in custom_fw_path.lower()