ruff format

Author: LowriJenkins

SECI_Config_Analyser/Directory_Operations.py

@@ -17,7 +17,7 @@ def __init__(self, search_path):
         """
         create lists for filenames and initialise to empty lists
         call methods for reading directory contents and searching for config filenames
-        :param search_path: the search path for files 
+        :param search_path: the search path for files
         """
 
         self.full_path = search_path
@@ -34,19 +34,16 @@ def _extract_config_filenames(self):
         """
 
         for filename in self.directory_contents:
-
-            if fnmatch(filename, '*.conf'):
-
+            if fnmatch(filename, "*.conf"):
                 self.config_filenames.append(filename)
 
-            elif fnmatch(filename, '*.comp'):
-
+            elif fnmatch(filename, "*.comp"):
                 self.comp_filenames.append(filename)
 
     def _parse_vis_from_files(self, filenames):
         """
         extracts VI names and paths from config files
-        :param filenames: list of filenames to process 
+        :param filenames: list of filenames to process
         :return: vis_in_files: list of VIs with fullpaths
         """
 
@@ -70,7 +67,7 @@ def _parse_file(filename_and_path):
         vis_in_file = []
         tree = ET.parse(filename_and_path)
         root = tree.getroot()
-        for child in root.iter('FilePath'):
+        for child in root.iter("FilePath"):
             vi_path = ET.tostring(child, encoding=None, method="text")
             vis_in_file.append(vi_path.strip())
 
@@ -91,8 +88,7 @@ def _remove_duplicates(input_list):
         encountered = set()
 
         for item in input_list:
-
-            item = str(item, 'UTF-8')
+            item = str(item, "UTF-8")
             if item not in encountered:
                 output_list.append(item)
                 encountered.add(item)

SECI_Config_Analyser/SECI_Config_Analyser.py

@@ -7,65 +7,67 @@
 
 from Directory_Operations import ReadConfigFiles
 
-if __name__ == '__main__':
-
+if __name__ == "__main__":
     parser = argparse.ArgumentParser(
-        description='Analyse the SECI configuration files of specified instrument and output results to file')
+        description="Analyse the SECI configuration files of specified instrument and output results to file"
+    )
 
-    parser.add_argument('--instrument', '-i',
-                        type=str,
-                        default=None,
-                        help='instrument to be analysed')
+    parser.add_argument(
+        "--instrument", "-i", type=str, default=None, help="instrument to be analysed"
+    )
 
-    parser.add_argument('--output_file', '-o',
-                        type=str,
-                        default=None,
-                        help='file to write the results to')
+    parser.add_argument(
+        "--output_file", "-o", type=str, default=None, help="file to write the results to"
+    )
 
-    parser.add_argument('--adminaccount', '-a',
-                        type=str,
-                        default=None,
-                        help=r'admin account name (without host i.e. NOT \\NDXxxx\xxx, just xxx')
+    parser.add_argument(
+        "--adminaccount",
+        "-a",
+        type=str,
+        default=None,
+        help=r"admin account name (without host i.e. NOT \\NDXxxx\xxx, just xxx",
+    )
 
-    parser.add_argument('--adminpassword', '-p',
-                        type=str,
-                        default=None,
-                        help='admin account password')
+    parser.add_argument(
+        "--adminpassword", "-p", type=str, default=None, help="admin account password"
+    )
 
     args = parser.parse_args()
 
     if args.instrument is None:
         parser.print_help()
 
     if args.output_file is None:
-        args.output_file = f'{args.instrument.upper()}_-_VIs_from_SECI_configs.txt'
+        args.output_file = f"{args.instrument.upper()}_-_VIs_from_SECI_configs.txt"
 
     # open reference to appropriate network drive with supplied credentials
 
-    print('Connecting to remote directory...')
-    system(f"net use \\\\ndx{args.instrument}\\c$ /user:ndx{args.instrument}\\{args.adminaccount} {args.adminpassword}")
+    print("Connecting to remote directory...")
+    system(
+        f"net use \\\\ndx{args.instrument}\\c$ /user:ndx{args.instrument}\\{args.adminaccount} {args.adminpassword}"
+    )
 
     # create instance of 'ReadConfigFiles' as 'filelist' and supply directory to be read
 
-    print('Reading configuration files...')
-    filelist = ReadConfigFiles(f"//ndx{args.instrument}/c$/Program Files (x86)/STFC ISIS Facility/SECI/Configurations/")
+    print("Reading configuration files...")
+    filelist = ReadConfigFiles(
+        f"//ndx{args.instrument}/c$/Program Files (x86)/STFC ISIS Facility/SECI/Configurations/"
+    )
 
-    print('Analysing files...')
+    print("Analysing files...")
     xml_data = filelist.analyse_config_files()
 
     # delete reference to network drive
 
-    print('Removing network connection...')
+    print("Removing network connection...")
     system(f"net use \\\\ndx{args.instrument}\\c$ /delete")
 
     # concatenate each element in list with newline character, then print whole string
 
-    print('Writing output file...')
-    with open(args.output_file, 'w') as outputfile:
-
+    print("Writing output file...")
+    with open(args.output_file, "w") as outputfile:
         outputfile.write(f"SECI configuration analysis for {args.instrument.upper()}:\n\n")
 
         for item in xml_data:
-
-            outputfile.write('\n'.join(x for x in item))
+            outputfile.write("\n".join(x for x in item))
             outputfile.write(f"\nNumber of VIs in configuration/component: {str(len(item))}\n\n")

build_tools/check_builds_are_recent.py

@@ -6,40 +6,53 @@
 
 kits_root = r"\\isis\inst$\kits$\CompGroup\ICP"
 
+
 def format_build_num_with_dash(build_num):
     return f"BUILD-{build_num}"
 
+
 def get_latest_build(dir, directory_formatter) -> Union[None, str]:
     latest_build = None
     with open(os.path.join(dir, "LATEST_BUILD.txt")) as latest_build_file:
         latest_build = latest_build_file.readline().strip()
-    latest_build_dir = os.path.join(dir, directory_formatter(latest_build)) if latest_build is not None else None
+    latest_build_dir = (
+        os.path.join(dir, directory_formatter(latest_build)) if latest_build is not None else None
+    )
     return latest_build_dir
 
+
 def get_last_modified_datetime(dir):
     time_in_seconds_of_last_modification = os.path.getmtime(dir)
     datetime_of_last_modification = datetime.fromtimestamp(time_in_seconds_of_last_modification)
     return datetime_of_last_modification
 
+
 def get_formatted_last_modified_datetime(dir):
     last_modified_date = get_last_modified_datetime(dir)
     return last_modified_date.strftime("%d/%m/%Y, %H:%M:%S")
 
+
 def modified_in_last_x_days(dir, x_days):
     datetime_of_last_modification = get_last_modified_datetime(dir)
     datetime_now = datetime.now()
     timedelta_since_last_modification = datetime_now - datetime_of_last_modification
     return timedelta_since_last_modification.days < x_days
 
+
 def was_modified_recently(latest_build, num_days):
     last_modified_datetime = get_formatted_last_modified_datetime(latest_build)
     modified_recently = modified_in_last_x_days(latest_build, num_days)
     if modified_recently:
-        print(f"SUCCESS: {latest_build} has been modified in the last {num_days} days. Last modified: {last_modified_datetime}")
+        print(
+            f"SUCCESS: {latest_build} has been modified in the last {num_days} days. Last modified: {last_modified_datetime}"
+        )
     else:
-        print(f"WARNING: {latest_build} modified longer than {num_days} ago. Last modified: {last_modified_datetime}")
+        print(
+            f"WARNING: {latest_build} modified longer than {num_days} ago. Last modified: {last_modified_datetime}"
+        )
     return modified_recently
 
+
 def check_build_dir(build_dir, num_days, directory_formatter):
     latest_build = get_latest_build(build_dir, directory_formatter)
     modified_recently = False
@@ -49,33 +62,39 @@ def check_build_dir(build_dir, num_days, directory_formatter):
         modified_recently = was_modified_recently(latest_build, num_days)
     return modified_recently
 
+
 def check_build_dirs(build_dirs):
     build_dirs_not_modified_recently = []
     for build_dir in build_dirs:
         directory = build_dir[0]
         stale_days_limit = int(build_dir[1])
         directory_formatter = build_dir[2]
         build_dir_full_path = os.path.join(kits_root, directory)
-        modified_recently = check_build_dir(build_dir_full_path, stale_days_limit, directory_formatter)
+        modified_recently = check_build_dir(
+            build_dir_full_path, stale_days_limit, directory_formatter
+        )
         if not modified_recently:
             build_dirs_not_modified_recently.append(build_dir_full_path)
     return build_dirs_not_modified_recently
 
+
 if __name__ == "__main__":
     builds_to_check = os.getenv("BUILDS_TO_CHECK")
     if builds_to_check is None:
-        print ("ERROR: BUILDS_TO_CHECK enviroment variable not set")
+        print("ERROR: BUILDS_TO_CHECK enviroment variable not set")
         sys.exit(1)
     try:
         builds_by_stale_times = json.loads(builds_to_check)
     except ValueError as e:
-        print(f"ERROR: parameter is not valid JSON.\nParameter:{builds_to_check}\nSpecific Error: {e.args}")
+        print(
+            f"ERROR: parameter is not valid JSON.\nParameter:{builds_to_check}\nSpecific Error: {e.args}"
+        )
         sys.exit(1)
     build_dirs = []
     for time, builds_by_stale_time in builds_by_stale_times.items():
         for build in builds_by_stale_time:
             directory_formatter = format_build_num_with_dash
-            build_dirs.append((build,time,directory_formatter))    
+            build_dirs.append((build, time, directory_formatter))
     build_dirs_not_modified_recently = check_build_dirs(build_dirs)
     if build_dirs_not_modified_recently:
         sys.exit(1)

build_tools/purge_archive.py

@@ -37,6 +37,7 @@ def delete_dir(directory):
     except Exception as e:
         print(f"Unable to delete directory {directory}: {e}")
 
+
 def delete_file(file):
     print(f"Deleting file: {file}")
     try:
@@ -46,7 +47,9 @@ def delete_file(file):
 
 
 def old_enough_to_delete(f):
-    return datetime.now() - datetime.fromtimestamp(os.path.getmtime(f)) > timedelta(days=max_build_age_in_days)
+    return datetime.now() - datetime.fromtimestamp(os.path.getmtime(f)) > timedelta(
+        days=max_build_age_in_days
+    )
 
 
 def is_build_dir(d):
@@ -57,35 +60,57 @@ def deletion_directories(project_areas):
     dirs = []
     for project_area in project_areas:
         project_dirs = [os.path.join(project_area, sub_dir) for sub_dir in os.listdir(project_area)]
-        build_dirs_by_age = sorted(filter(is_build_dir, project_dirs), key=os.path.getmtime, reverse=True)
+        build_dirs_by_age = sorted(
+            filter(is_build_dir, project_dirs), key=os.path.getmtime, reverse=True
+        )
         build_dirs_we_could_delete = build_dirs_by_age[minimum_number_of_builds_to_keep:]
         build_dirs_we_will_delete = filter(old_enough_to_delete, build_dirs_we_could_delete)
         dirs_to_add = list(build_dirs_we_will_delete)
         dirs += dirs_to_add
         print(f"Identifying {len(dirs_to_add)} old builds for deletion in: {project_area}")
     return dirs
 
+
 def deletion_files(project_areas, pattern):
     files = []
     for project_area in project_areas:
-        files_by_age = sorted(glob.glob(os.path.join(project_area, pattern)), key=os.path.getmtime, reverse=True)
+        files_by_age = sorted(
+            glob.glob(os.path.join(project_area, pattern)), key=os.path.getmtime, reverse=True
+        )
         files_we_could_delete = files_by_age[minimum_number_of_builds_to_keep:]
         files_we_will_delete = filter(old_enough_to_delete, files_we_could_delete)
         files_to_add = list(files_we_will_delete)
         files += files_to_add
-        print(f"Identifying {len(files_to_add)} old {pattern} files for deletion in: {project_area}")
+        print(
+            f"Identifying {len(files_to_add)} old {pattern} files for deletion in: {project_area}"
+        )
     return files
 
 
 def purge(dry_run=False):
     print("Beginning archive purge...")
-    project_areas = [os.path.join(build_area, proj) for proj in ("Client_E4", "script_generator", "genie_python_3", "VHDS")] + \
-        [os.path.join(build_area, "EPICS", proj) for proj in os.listdir(os.path.join(build_area, "EPICS"))
-         if proj.startswith("EPICS")]
-    project_areas.extend([os.path.join(build_area, "developer", "EPICS", proj) for proj in os.listdir(os.path.join(build_area, "developer", "EPICS"))
-                          if proj.startswith("windows")])
-    project_areas.extend([os.path.join(build_area, "developer", "EPICS32", proj) for proj in os.listdir(os.path.join(build_area, "developer", "EPICS32"))
-                          if proj.startswith("win32")])
+    project_areas = [
+        os.path.join(build_area, proj)
+        for proj in ("Client_E4", "script_generator", "genie_python_3", "VHDS")
+    ] + [
+        os.path.join(build_area, "EPICS", proj)
+        for proj in os.listdir(os.path.join(build_area, "EPICS"))
+        if proj.startswith("EPICS")
+    ]
+    project_areas.extend(
+        [
+            os.path.join(build_area, "developer", "EPICS", proj)
+            for proj in os.listdir(os.path.join(build_area, "developer", "EPICS"))
+            if proj.startswith("windows")
+        ]
+    )
+    project_areas.extend(
+        [
+            os.path.join(build_area, "developer", "EPICS32", proj)
+            for proj in os.listdir(os.path.join(build_area, "developer", "EPICS32"))
+            if proj.startswith("win32")
+        ]
+    )
     for d in deletion_directories(project_areas):
         try:
             if dry_run:
@@ -97,7 +122,7 @@ def purge(dry_run=False):
         except BaseException as be:
             print(f"BaseException - unable to delete directory {d}: {be}")
             break
-    for f in deletion_files(project_areas, 'ZBUILD-*.7z'):
+    for f in deletion_files(project_areas, "ZBUILD-*.7z"):
         try:
             if dry_run:
                 print(f"{f}")
@@ -110,4 +135,5 @@ def purge(dry_run=False):
             break
     print("Archive purge complete.")
 
+
 purge()