diff options
Diffstat (limited to 'tutorials/ApacheCNF/automation/config.py')
| -rw-r--r-- | tutorials/ApacheCNF/automation/config.py | 142 |
1 files changed, 142 insertions, 0 deletions
diff --git a/tutorials/ApacheCNF/automation/config.py b/tutorials/ApacheCNF/automation/config.py new file mode 100644 index 00000000..24983671 --- /dev/null +++ b/tutorials/ApacheCNF/automation/config.py @@ -0,0 +1,142 @@ +# ============LICENSE_START======================================================= +# Copyright (C) 2021 Orange +# ================================================================================ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# ============LICENSE_END========================================================= +from typing import Dict, Union, List +import oyaml as yaml +import os +from jinja2 import Template + + +class VariablesDict: + env_variable = dict(os.environ) + + +class Config: + def __init__(self, filename: str = "service_config.yaml", env_dict=None, print_final_file=False): + if env_dict is None: + env_dict = {} + self.filepath = os.path.join(os.path.dirname(os.path.dirname( + os.path.realpath(__file__))), filename) + self.content_env: Dict = env_dict + self.service_instance: Union[None, Dict] = None + self.service_model: Union[None, Dict] = None + self.user_params: Union[None, Dict] = None + self.cloud_regions: Union[None, List] = None + self.so_input: Union[None, Dict] = None + self.render(print_final_file) + self.so_input = self.create_so_input() + + def _load_file(self) -> dict: + with open(self.filepath) as file: + file_content = yaml.safe_load(file) + return file_content + + @staticmethod + def templating(rend_dict: dict, render_keys: dict): + for k, v in rend_dict.items(): + if isinstance(v, str): + t = Template(v) + rend_dict[k] = t.render(**render_keys).strip() + elif isinstance(v, dict): + Config.templating(rend_dict=v, render_keys=render_keys) + elif isinstance(v, list): + for i in v: + Config.templating(rend_dict=i, render_keys=render_keys) + else: + pass + return rend_dict + + def render(self, print_final_file=False): + raw_file = self._load_file() + config_dict = self._render(templated_file=raw_file) + + while not self._completed(templated_file=config_dict): + config_dict = self._render(templated_file=config_dict) + + self.__dict__.update(**config_dict) + if print_final_file: + print(yaml.dump(config_dict, sort_keys=False)) + + def _render(self, templated_file: dict) -> dict: + config_dict = self.templating( + rend_dict=templated_file, + render_keys={**self.content_env, **templated_file}) + + return config_dict + + def _completed(self, templated_file: dict) -> bool: + for v in templated_file.values(): + if isinstance(v, str): + if "{{" in v and "}}" in v: + return False + elif isinstance(v, dict): + return self._completed(templated_file=v) + elif isinstance(v, list): + for i in v: + return self._completed(templated_file=i) + else: + pass + return True + + def create_so_input(self, other_cluster=False) -> dict: + so_input_dict = dict() + so_input_dict["subscription_service_type"] = self.service_instance.get("model_name") + _vnfs = self.service_instance.get("vnfs") + vnfs = list() + + for vnf in _vnfs: + _vnf_raw = dict() + # filter vnfs with cloud_region, code block not required with ONAP Jakrta+ + if vnf.get("cloud_region") and vnf.get("tenant_name"): + if not other_cluster: + continue + _vnf_raw["cloud_region"] = vnf.get("cloud_region") + _vnf_raw["tenant_name"] = vnf.get("tenant_name") + else: + if other_cluster: + continue + # end of filter, end of code block + _vnf_raw["model_name"] = vnf.get("model_name") + if vnf.get("vnf_name_suffix"): + _vnf_raw["instance_name"] = "Instance_" + vnf.get("model_name") + "_" + vnf.get("vnf_name_suffix") + else: + _vnf_raw["instance_name"] = "Instance_" + vnf.get("model_name") + "_" + str(_vnfs.index(vnf)) + if vnf.get("processing_priority"): + _vnf_raw["processing_priority"] = vnf.get("processing_priority") + _vnf_raw["parameters"] = vnf.get("parameters") + _vnf_raw["vf_modules"] = list() + _vf_modules = vnf.get("vf_modules") + for vf_module in _vf_modules: + _vf_module_raw = dict() + _vf_module_raw["model_name"] = vf_module.get("model_name") + if vf_module.get("vf_module_name_suffix"): + _vf_module_raw["instance_name"] = \ + "Instance_" + vf_module.get("model_name") + "_" + vf_module.get("vf_module_name_suffix") + else: + _vf_module_raw["instance_name"] = \ + "Instance_" + vf_module.get("model_name") + "_" + str(_vnfs.index(vnf)) + \ + "_" + str(_vf_modules.index(vf_module)) + if vf_module.get("processing_priority"): + _vf_module_raw["processing_priority"] = vf_module["processing_priority"] + _vf_module_raw["parameters"] = vf_module.get("parameters") + _vnf_raw["vf_modules"].append(_vf_module_raw) + vnfs.append(_vnf_raw) + so_input_dict["vnfs"] = vnfs + + return so_input_dict + +if __name__ == "__main__": + config = Config(env_dict=VariablesDict.env_variable, print_final_file=True) |