Added flashing and reading design name to CICD pipeline.

Closes L2SDP-571

test/py/control/sdp_firmware.py

@@ -39,6 +39,7 @@ import time
 import pprint
 import math
 import struct
+import numpy as np
 
 from base import *
 
@@ -145,7 +146,7 @@ class SDPfirmware(object):
                 sectors = range(self.user_sector_start, self.user_sector_end + 1)
                 addr = self.user_sector_start * self.sector_size_bytes
 
-            print('flash read {} inmage'.format(self.bank))
+            print('flash read {} image'.format(self.bank))
             n_sectors = len(sectors)
             n_pages = n_sectors * self.pages_per_sector
             if self.n_pages_in_image > 0:
@@ -193,7 +194,7 @@ class SDPfirmware(object):
                 logger.error("exception while opening file {}".format(filename))
                 return
 
-            print('flash write {} inmage'.format(self.bank))
+            print('flash write {} image'.format(self.bank))
 
             # need to be word aligned
             n_pad_bytes = (self.word_size_bytes - (len(rbf) % self.word_size_bytes)) if (len(rbf) % self.word_size_bytes) > 0 else 0
@@ -239,40 +240,64 @@ class SDPfirmware(object):
         return False
 
     def flash_verify(self, filename):
-        # check only first selected fpga
-        read_file = 'read_{}_{}.rbf'.format(self.bank, self.active_nodes[0])
-        print('compare image "{}" with "{}"'.format(read_file, filename))
-        same = True
-        try:
-            rbf_size_bytes = os.path.getsize(filename)
-            fd1 = open(filename, 'rb')
-            fd2 = open(read_file, 'rb')
-            byte_cnt = 0
-            while same:
-                if byte_cnt == rbf_size_bytes:
-                    break
-                w1 = fd1.read(4)
-                w2 = fd2.read(4)
-                if w1 != w2:
-                    logger.error('byte {} not same: w1={}, w2={}'.format(cnt, w1, w2))
-                    same = False
-                byte_cnt += 4
-            print('checked {} bytes'.format(byte_cnt))
-        except:
-            logger.error("exception in flash_verify")
-            logger.error('Caught %s', str(sys.exc_info()[0]))
-            logger.error(str(sys.exc_info()[1]))
-            logger.error('TRACEBACK:\n%s', traceback.format_exc())
-            logger.error('Aborting NOW')
-
-        fd1.close()
-        fd2.close()
-
-        if not same:
-            print('NOT SAME')
-            return False
-        print('SAME')
-        return True
+        for node in node_list:
+            # check only first selected fpga
+            read_file = 'read_{}_{}.rbf'.format(self.bank, node)
+            print('compare image "{}" with "{}"'.format(read_file, filename))
+            same = True
+            try:
+                rbf_size_bytes = os.path.getsize(filename)
+                fd1 = open(filename, 'rb')
+                fd2 = open(read_file, 'rb')
+                byte_cnt = 0
+                while same:
+                    if byte_cnt == rbf_size_bytes:
+                        break
+                    w1 = fd1.read(4)
+                    w2 = fd2.read(4)
+                    if w1 != w2:
+                        logger.error('byte {} not same: w1={}, w2={}'.format(byte_cnt, w1, w2))
+                        same = False
+                    byte_cnt += 4
+                print('checked {} bytes'.format(byte_cnt))
+            except:
+                logger.error("exception in flash_verify")
+                logger.error('Caught %s', str(sys.exc_info()[0]))
+                logger.error(str(sys.exc_info()[1]))
+                logger.error('TRACEBACK:\n%s', traceback.format_exc())
+                logger.error('Aborting NOW')
+    
+            fd1.close()
+            fd2.close()
+    
+            if not same:
+                print('NOT SAME')
+                return False
+            print('SAME')
+            return True
+
+    def read_fw_version(self, version_test=None):
+        """
+        Read firmware version, return list of requested nodes or True when version==version_test, else False.
+        """
+        version = np.array(self.client.read('firmware_version'))[node_list]
+
+        if version_test is None:
+            return version
+        else:
+            result_list = []
+
+            # Issue: we have a mismatch between name formats:
+            # . RBF name (version_test) : 'lofar2_unb2b_sdp_full-r9ff51058a'
+            # . Version  (version)      : '2022-01-12T13.14.20_9ff51058a_lofar2_unb2b_sdp_station_full'
+            # So we're checking whether both the name and the hash from version_test are in the version - in that case the test passes.
+            name, githash = version_test.split('-r')
+            for n in version:
+                result = True
+                if not name in n or not githash in n:
+                    result = False 
+                result_list.append(result)
+            return result
 
 
 def main():
@@ -312,6 +337,16 @@ def main():
             logger.info(sdp_firmware.reboot())
             if args.image == 'USER':
                 logger.info("It takes 8 seconds before user image is active")
+
+        if args.version:
+            print(sdp_firmware.read_fw_version())
+
+        if args.version_test:
+            result = sdp_firmware.read_fw_version(args.version_test)
+            client.disconnect()
+            # Convert False->1 and True->0
+            return(int(not(result)))
+
         break
 
     client.disconnect()
@@ -333,6 +368,10 @@ if __name__ == "__main__":
     parser.add_argument('--read', action='store_true', help="read image from flash and save it to a file, also use --image and --file")
     parser.add_argument('--verify', action='store_true', help="verify read image, also use --image and --file")
 
+    parser.add_argument('--version', action="store_true", help="Read firmware version")
+    parser.add_argument('--version_test', dest='version_test', type=str, help="Read firmware version and compare to VERSION_TEST")
+
+
     args = parser.parse_args()
 
     node_list = arg_str_to_list(args.nodes) if args.nodes else None
@@ -354,8 +393,9 @@ if __name__ == "__main__":
         sys.exit(1)
 
     clientRunning = True
+    exitcode = 1
     try:
-        main()
+        exitcode=main()
     except KeyboardInterrupt:
         logger.info(" user hit ctrl-c")
         clientRunning = False
@@ -368,4 +408,4 @@ if __name__ == "__main__":
         clientRunning = False
         time.sleep(1)
 
-sys.exit(1)
+    sys.exit(exitcode)