New O1 tests for python SDK
[it/dep.git] / smo-install / test / topology-generator / network-topology-converter.py
diff --git a/smo-install/test/topology-generator/network-topology-converter.py b/smo-install/test/topology-generator/network-topology-converter.py
new file mode 100755 (executable)
index 0000000..e2a4c7b
--- /dev/null
@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+
+import json
+import sys
+import yaml
+
+def read_json_topology_file (filename):
+       return json.loads(open(filename, "r").read())
+
+def save_helm_override_file (helm_file, filename):
+       with open(filename, 'w') as file:
+               return yaml.dump(helm_file, file)
+
+def get_name_of_node (node):
+       for name in node["name"]:
+               if name["value-name"] == "topology-node-name":
+                       return name["value"]
+
+def search_all_topology_nodes (topology_json):
+       ru_nodes = {}
+       du_nodes = {}
+       ue_nodes = {}
+       cu_nodes = {}
+       near_rt_ric_nodes = {}
+       smo_nodes = {}
+
+       for node in topology_json["tapi-common:context"]["tapi-topology:topology-context"]["topology"][0]["node"]:
+
+               if node["o-ran-topology:function"] == "o-ran-common-identity-refs:o-ru-function":
+                       ru_nodes[get_name_of_node(node)]=node
+
+               elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:o-du-function":
+                       du_nodes[get_name_of_node(node)]=node
+               elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:user-equipment-function":
+                       ue_nodes[get_name_of_node(node)]=node
+               elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:o-cu-up-function":
+                       cu_nodes[get_name_of_node(node)]=node
+               elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:near-rt-ric-function":
+                       near_rt_ric_nodes[get_name_of_node(node)]=node
+               elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:smo-function":
+                       smo_nodes[get_name_of_node(node)]=node
+       return {"ru_nodes":ru_nodes, "du_nodes": du_nodes, "ue_nodes": ue_nodes, "cu_nodes": cu_nodes, "near_rt_ric_nodes":near_rt_ric_nodes, "smo_nodes":smo_nodes}
+
+def generate_ru_node (ru_nodes):
+       rus=[]
+       for ru_key in ru_nodes:
+               rus.append({"name":ru_key, "simulatedFaults":[]})
+       return {"ru_simulator":{"rus":rus}}
+
+def generate_ru_faults ():
+       return {}
+
+all_nodes=search_all_topology_nodes(read_json_topology_file (sys.argv[1]))
+helm_override={}
+helm_override = {**helm_override, **generate_ru_node(all_nodes["ru_nodes"])}
+
+
+print ("#RU:"+str(len(all_nodes["ru_nodes"])))
+print ("#DU:"+str(len(all_nodes["du_nodes"])))
+print ("#UE:"+str(len(all_nodes["ue_nodes"])))
+print ("#CU:"+str(len(all_nodes["cu_nodes"])))
+print ("#NEAR_RTRIC:"+str(len(all_nodes["near_rt_ric_nodes"])))
+print ("#SMO:"+str(len(all_nodes["smo_nodes"])))
+number_of_nodes=len(all_nodes["ru_nodes"])+len(all_nodes["du_nodes"])+len(all_nodes["ue_nodes"])+len(all_nodes["cu_nodes"])+len(all_nodes["near_rt_ric_nodes"])+len(all_nodes["smo_nodes"])
+print ("#Nodes(total):"+str(number_of_nodes))
+
+print(save_helm_override_file(helm_override, sys.argv[2]))