smo-install/test/topology-generator/network-topology-converter.py

   1 #!/usr/bin/env python3
   2
   3 import json
   4 import sys
   5 import yaml
   6
   7 def read_json_topology_file (filename):
   8         return json.loads(open(filename, "r").read())
   9
  10 def save_helm_override_file (helm_file, filename):
  11         with open(filename, 'w') as file:
  12                 return yaml.dump(helm_file, file)
  13
  14 def get_name_of_node (node):
  15         for name in node["name"]:
  16                 if name["value-name"] == "topology-node-name":
  17                         return name["value"]
  18
  19 def search_all_topology_nodes (topology_json):
  20         ru_nodes = {}
  21         du_nodes = {}
  22         ue_nodes = {}
  23         cu_nodes = {}
  24         near_rt_ric_nodes = {}
  25         smo_nodes = {}
  26
  27         for node in topology_json["tapi-common:context"]["tapi-topology:topology-context"]["topology"][0]["node"]:
  28
  29                 if node["o-ran-topology:function"] == "o-ran-common-identity-refs:o-ru-function":
  30                         ru_nodes[get_name_of_node(node)]=node
  31
  32                 elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:o-du-function":
  33                         du_nodes[get_name_of_node(node)]=node
  34                 elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:user-equipment-function":
  35                         ue_nodes[get_name_of_node(node)]=node
  36                 elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:o-cu-up-function":
  37                         cu_nodes[get_name_of_node(node)]=node
  38                 elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:near-rt-ric-function":
  39                         near_rt_ric_nodes[get_name_of_node(node)]=node
  40                 elif node["o-ran-topology:function"] == "o-ran-common-identity-refs:smo-function":
  41                         smo_nodes[get_name_of_node(node)]=node
  42         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}
  43
  44 def generate_ru_node (ru_nodes):
  45         rus=[]
  46         for ru_key in ru_nodes:
  47                 rus.append({"name":ru_key, "simulatedFaults":[]})
  48         return {"ru_simulator":{"rus":rus}}
  49
  50 def generate_ru_faults ():
  51         return {}
  52
  53 all_nodes=search_all_topology_nodes(read_json_topology_file (sys.argv[1]))
  54 helm_override={}
  55 helm_override = {**helm_override, **generate_ru_node(all_nodes["ru_nodes"])}
  56
  57
  58 print ("#RU:"+str(len(all_nodes["ru_nodes"])))
  59 print ("#DU:"+str(len(all_nodes["du_nodes"])))
  60 print ("#UE:"+str(len(all_nodes["ue_nodes"])))
  61 print ("#CU:"+str(len(all_nodes["cu_nodes"])))
  62 print ("#NEAR_RTRIC:"+str(len(all_nodes["near_rt_ric_nodes"])))
  63 print ("#SMO:"+str(len(all_nodes["smo_nodes"])))
  64 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"])
  65 print ("#Nodes(total):"+str(number_of_nodes))
  66
  67 print(save_helm_override_file(helm_override, sys.argv[2]))