1 # ============LICENSE_START=======================================================
2 # Copyright (C) 2021 Orange
3 # ================================================================================
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 # ============LICENSE_END=========================================================
17 from typing import Dict, Union, List
20 from jinja2 import Template
24 env_variable = dict(os.environ)
28 def __init__(self, filename: str = "service_config.yaml", env_dict=None, print_final_file=False):
31 self.filepath = os.path.join(os.path.dirname(os.path.dirname(
32 os.path.realpath(__file__))), filename)
33 self.content_env: Dict = env_dict
34 self.service_instance: Union[None, Dict] = None
35 self.service_model: Union[None, Dict] = None
36 self.user_params: Union[None, Dict] = None
37 self.cloud_regions: Union[None, List] = None
38 self.so_input: Union[None, Dict] = None
39 self.render(print_final_file)
40 self.so_input = self.create_so_input()
42 def _load_file(self) -> dict:
43 with open(self.filepath) as file:
44 file_content = yaml.safe_load(file)
48 def templating(rend_dict: dict, render_keys: dict):
49 for k, v in rend_dict.items():
50 if isinstance(v, str):
52 rend_dict[k] = t.render(**render_keys).strip()
53 elif isinstance(v, dict):
54 Config.templating(rend_dict=v, render_keys=render_keys)
55 elif isinstance(v, list):
57 Config.templating(rend_dict=i, render_keys=render_keys)
62 def render(self, print_final_file=False):
63 raw_file = self._load_file()
64 config_dict = self._render(templated_file=raw_file)
66 while not self._completed(templated_file=config_dict):
67 config_dict = self._render(templated_file=config_dict)
69 self.__dict__.update(**config_dict)
71 print(yaml.dump(config_dict, sort_keys=False))
73 def _render(self, templated_file: dict) -> dict:
74 config_dict = self.templating(
75 rend_dict=templated_file,
76 render_keys={**self.content_env, **templated_file})
80 def _completed(self, templated_file: dict) -> bool:
81 for v in templated_file.values():
82 if isinstance(v, str):
83 if "{{" in v and "}}" in v:
85 elif isinstance(v, dict):
86 return self._completed(templated_file=v)
87 elif isinstance(v, list):
89 return self._completed(templated_file=i)
94 def create_so_input(self, other_cluster=False) -> dict:
95 so_input_dict = dict()
96 so_input_dict["subscription_service_type"] = self.service_instance.get("model_name")
97 _vnfs = self.service_instance.get("vnfs")
102 # filter vnfs with cloud_region, code block not required with ONAP Jakrta+
103 if vnf.get("cloud_region") and vnf.get("tenant_name"):
104 if not other_cluster:
106 _vnf_raw["cloud_region"] = vnf.get("cloud_region")
107 _vnf_raw["tenant_name"] = vnf.get("tenant_name")
111 # end of filter, end of code block
112 _vnf_raw["model_name"] = vnf.get("model_name")
113 if vnf.get("vnf_name_suffix"):
114 _vnf_raw["instance_name"] = "Instance_" + vnf.get("model_name") + "_" + vnf.get("vnf_name_suffix")
116 _vnf_raw["instance_name"] = "Instance_" + vnf.get("model_name") + "_" + str(_vnfs.index(vnf))
117 if vnf.get("processing_priority"):
118 _vnf_raw["processing_priority"] = vnf.get("processing_priority")
119 _vnf_raw["parameters"] = vnf.get("parameters")
120 _vnf_raw["vf_modules"] = list()
121 _vf_modules = vnf.get("vf_modules")
122 for vf_module in _vf_modules:
123 _vf_module_raw = dict()
124 _vf_module_raw["model_name"] = vf_module.get("model_name")
125 if vf_module.get("vf_module_name_suffix"):
126 _vf_module_raw["instance_name"] = \
127 "Instance_" + vf_module.get("model_name") + "_" + vf_module.get("vf_module_name_suffix")
129 _vf_module_raw["instance_name"] = \
130 "Instance_" + vf_module.get("model_name") + "_" + str(_vnfs.index(vnf)) + \
131 "_" + str(_vf_modules.index(vf_module))
132 if vf_module.get("processing_priority"):
133 _vf_module_raw["processing_priority"] = vf_module["processing_priority"]
134 _vf_module_raw["parameters"] = vf_module.get("parameters")
135 _vnf_raw["vf_modules"].append(_vf_module_raw)
136 vnfs.append(_vnf_raw)
137 so_input_dict["vnfs"] = vnfs
141 if __name__ == "__main__":
142 config = Config(env_dict=VariablesDict.env_variable, print_final_file=True)