diff --git a/core/base/tools/commander/workspace_template/bin/pv_remap.py b/core/base/tools/commander/workspace_template/bin/pv_remap.py new file mode 100644 index 000000000..26a980c2a --- /dev/null +++ b/core/base/tools/commander/workspace_template/bin/pv_remap.py @@ -0,0 +1,73 @@ +""" +Usage: +python3 pv_remap.pv --csv_document /home/lgomez/Downloads/VIPER-to-Airliner_PV_Mapping_Sheet1.csv --old_col Airliner --new_col Viper --display_root_dir /home/lgomez/airliner/core/base/tools/commander/workspace_template/Displays +""" + +import argparse +import csv +import os + + +def parse_cli() -> argparse.Namespace: + """ + Parses cli arguments. + :return: The namespace that has all of the arguments that have been parsed. + """ + parser = argparse.ArgumentParser(description='Takes in path to sqlite database.') + + parser.add_argument('--csv_document', type=str, required=True, + help='The csv file to scan.') + parser.add_argument('--old_col', type=str, required=False, + help='The old column that has old pvs which will be replaced.') + parser.add_argument('--new_col', type=str, required=False, + help='The new column.') + + parser.add_argument('--display_root_dir', type=str, required=False, + help='The input directory to scan(recursively) for displays.') + + return parser.parse_args() + + +def replace_pv(old: str, new: str, display: str): + return display.replace(old, new) + + +def remap_display(csv_document, old_col, new_col, in_display, out_display): + rows = [] + old_display = '' + with open(csv_document, 'r+') as csvfile: + reader = csv.reader(csvfile, delimiter=",", dialect='excel') + rows = list(reader) + header = rows[0] + old_index = 0 + new_index = 1 + if header[1] == old_col: + old_index = 1 + new_index = 0 + with open(in_display, 'r') as old_display_file: + old_display = old_display_file.read() + with open(out_display, 'w+') as new_display_file: + new_display = str(old_display) + for row in rows[1:]: + if row[old_index] != '': + new_display = replace_pv(row[old_index], row[new_index], new_display) + + new_display_file.write(new_display) + + +def main(): + args = parse_cli() + opi_files = [] + + for (parent, subdirs, files) in os.walk(args.display_root_dir): + for name in files + subdirs: + filename, file_extension = os.path.splitext(os.path.join(parent, name)) + if file_extension == '.opi': + opi_files.append(os.path.join(parent, name)) + + for file in opi_files: + remap_display(args.csv_document, args.old_col, args.new_col, file, file) + + +if __name__ == '__main__': + main() diff --git a/core/base/tools/commander/workspace_template/bin/pv_to_csv.py b/core/base/tools/commander/workspace_template/bin/pv_to_csv.py new file mode 100644 index 000000000..287f1acec --- /dev/null +++ b/core/base/tools/commander/workspace_template/bin/pv_to_csv.py @@ -0,0 +1,79 @@ +""" +Gather all PV names from YAMCS server and dump them on a CSV file. +""" +import csv +import requests +from yamcs.client import YamcsClient + +INSTANCE = "viper" +SERVER = "http://192.168.2.96:8090" + +all_pvalues = f"{SERVER}/api/mdb/{INSTANCE}/parameters" + +# Could be useful, though beware that this returns not just PValue names but also the actual value as well. +pvalues_to_csv = f"{SERVER}/api/archive/{INSTANCE}:exportParameterValues" + +output_file = "PValues.csv" + +MEMBER_SEPARATOR = "." + + +def get_aggregate_qualified_names(parameter, param_qualified_name, out_names: [str]): + if 'member' in parameter['type']: + for member in parameter['type']['member']: + if member['type']['engType'] == 'aggregate': + get_aggregate_qualified_names(member, param_qualified_name + MEMBER_SEPARATOR + member['name'], + out_names) + else: + out_names.append(param_qualified_name + MEMBER_SEPARATOR + member['name']) + + +def get_params_from_aggregates(): + response = requests.get(all_pvalues) + parameters = response.json()['parameters'] + parameter_count = 0 + + for p in parameters: + # Avoid params that are not part of XTCE such YAMCS internal command counters + if 'type' in p: + if p['type']['engType'] == 'aggregate' and 'member' in p['type']: + out_list = [] + get_aggregate_qualified_names(p, p['qualifiedName'], out_list) + with open(output_file, 'a+') as csvfile: + parameter_count = len(out_list) + parameter_count + writer = csv.writer(csvfile, delimiter=",", dialect='excel') + writer.writerows((pv,) for pv in out_list) + else: + out_list = [p['qualifiedName']] + with open(output_file, 'a+') as csvfile: + writer = csv.writer(csvfile, delimiter=",", dialect='excel') + parameter_count = len(out_list) + parameter_count + writer.writerows((pv,) for pv in out_list) + + print(f'Loaded {parameter_count} parameters from YAMCS') + + +def get_params_from_api(): + response = requests.get(pvalues_to_csv) + with open(output_file, 'wb') as csvfile: + csvfile.write(response.content) + + +def get_params_from_python_api(): + client = YamcsClient(SERVER[7:]) + mdb = client.get_mdb(instance=INSTANCE) + parameter_count = 0 + for parameter in mdb.list_parameters(): + out_list = [parameter] + with open(output_file, 'a+') as csvfile: + writer = csv.writer(csvfile, delimiter=",", dialect='excel') + parameter_count = len(out_list) + parameter_count + writer.writerows((pv,) for pv in out_list) + print(f'Loaded {parameter_count} parameters from YAMCS') + +def main(): + get_params_from_python_api() + + +if __name__ == '__main__': + main()