Yamcs pv

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()

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()