code/network-topology-instance-generator/model/python/link_config.py

   1 # Copyright 2022 highstreet technologies GmbH
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 #!/usr/bin/python
  16 """
  17 Module to construct the input configuration for a TAPI link object.
  18 """
  19 from typing import Dict
  20 from model.python.tapi_node import TapiNode
  21 from model.python.tapi_node_edge_point import TapiNodeEdgePoint
  22 from model.python.top import Top
  23
  24
  25 class LinkConfig(Top):
  26     """
  27     Class containing  methods creating an link configuration object.
  28     """
  29
  30     __topology_reference = "unknown"
  31     __name_prefix: str = "unknown"
  32     __consumer: TapiNode = None
  33     __provider: TapiNode = None
  34
  35     __data: dict = {"link": {
  36         "name": "noName",
  37         "a": {},
  38         "z": {}
  39     }}
  40
  41     # constructor
  42     def __init__(self, topology_reference: str, name_prefix: str,
  43                  provider: TapiNode, consumer: TapiNode):
  44         super().__init__({})
  45         self.__topology_reference = topology_reference
  46         self.__name_prefix = name_prefix
  47         self.__consumer = consumer
  48         self.__provider = provider
  49
  50         self.data = {"link": {
  51             "name": self.name(),
  52             "a": {
  53                 "topology-uuid": topology_reference,
  54                 "node-uuid": consumer.data()["uuid"],
  55                 "node-edge-point-uuid":
  56                     self.consumer_node_edge_point().identifier()
  57
  58             },
  59             "z": {
  60                 "topology-uuid": topology_reference,
  61                 "node-uuid": provider.data()["uuid"],
  62                 "node-edge-point-uuid":
  63                     self.provider_node_edge_point().identifier()
  64             }
  65         }}
  66
  67     def configuration(self) -> Dict[str, Dict[str, Dict]]:
  68         """
  69         Getter returning the configuration.
  70         :return Link identifier as string
  71         """
  72         return {"configuration": {
  73             "topology-reference:": self.__topology_reference,
  74             "name_prefix": self.__name_prefix,
  75             "provider": self.__provider.json(),
  76             "consumer": self.__consumer.json()
  77         }}
  78
  79     def data(self) -> Dict[str, Dict]:
  80         """
  81         Getter returning the link data of the link.
  82         :return Link confguation data as json object
  83         """
  84         return self.__data
  85
  86     def identifier(self) -> str:
  87         """
  88         Getter returning the link configuration identifier of the link.
  89         :return Link identifier as string
  90         """
  91         return "--".join([
  92             self.__consumer.identifier(),
  93             self.__provider.identifier(),
  94         ])
  95
  96     def name(self) -> str:
  97         """
  98         Getter returning the name of the link.
  99         :return Link name as string
 100         """
 101         if self.__consumer:
 102             return "|".join([
 103                 self.__name_prefix.upper(),
 104                 self.__consumer.name(),
 105                 "->",
 106                 self.__provider.name(),
 107                 ""
 108             ])
 109         return ""
 110
 111     def json(self) -> dict:
 112         """
 113         Getter for the json represention of this object.
 114         :return JSON object of link configuration
 115         """
 116         return self.data
 117
 118     def consumer_node_edge_point(self) -> TapiNodeEdgePoint:
 119         name_prefix = self.__name_prefix
 120
 121         # exception for O-RAN Fronthaul Management plane to SMO
 122         if self.__consumer.function() == "o-ran-sc-topology-common:smo" and \
 123            self.__provider.function() == "o-ran-sc-topology-common:o-ru" and \
 124                 name_prefix == "ofh-netconf":  # "open-fronthaul-m-plane-netconf":
 125             name_prefix = "oam-netconf"
 126
 127         # exception for O-RAN Gateway plane to SMO
 128         if self.__consumer.function() == "o-ran-sc-topology-common:smo" and \
 129                 name_prefix == "o1-netconf":  # "open-fronthaul-m-plane-netconf":
 130             name_prefix = "oam-netconf"
 131
 132         # exception for O-RU to FHGW
 133         if self.__provider.function() == "o-ran-sc-topology-common:fronthaul-gateway" and \
 134                 name_prefix == "eth-ofh":  # "open-fronthaul-m-plane-netconf":
 135             name_prefix = "ofh-netconf"
 136
 137         # exception for O-RU to FHGW
 138         if self.__consumer.function() == "o-ran-sc-topology-common:fronthaul-gateway" and \
 139                 name_prefix == "ofh-netconf":  # "open-fronthaul-m-plane-netconf":
 140             name_prefix = "eth-ofh"
 141
 142         cep_name = name_prefix.lower() + "-consumer"
 143         return self.__consumer.node_edge_point_by_cep_name(cep_name,  self.__provider.local_id())
 144
 145     def provider_node_edge_point(self) -> TapiNodeEdgePoint:
 146         name_prefix = self.__name_prefix
 147
 148         cep_name = name_prefix.lower() + "-provider"
 149         # exception for f1-c and f1-u
 150         split = name_prefix.lower().split("-")
 151         if len(split) == 3:
 152             cep_name = "-".join([split[0], split[2], "provider"])
 153
 154         return self.__provider.node_edge_point_by_cep_name(cep_name, self.__consumer.local_id())