1 # Copyright (C) 2021 Wind River Systems, Inc.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
15 from __future__ import annotations
18 from o2common.config import config
19 from o2common.domain.base import AgRoot, Serializer, \
20 InfrastructureInventoryObject
21 # from dataclasses import dataclass
22 # from datetime import date
23 # from typing import Optional, List, Set
24 from .resource_type import ResourceKindEnum, ResourceTypeEnum
25 from .alarm_obj import AlarmDictionary
28 DeploymentManagerProfileDefault = 'native_k8sapi'
29 DeploymentManagerProfileSOL018 = 'sol018'
30 DeploymentManagerProfileSOL018HelmCLI = 'sol018_helmcli'
33 class DeploymentManager(InfrastructureInventoryObject, AgRoot, Serializer):
34 def __init__(self, id: str, name: str, ocloudid: str,
35 dmsendpoint: str, description: str = '',
36 supportedLocations: str = '', capabilities: str = '',
37 capacity: str = '', profile: str = '') -> None:
39 self.deploymentManagerId = id
41 self.description = description
42 self.oCloudId = ocloudid
43 self.serviceUri = dmsendpoint
44 self.supportedLocations = supportedLocations
45 self.capabilities = capabilities
46 self.capacity = capacity
47 self.profile = profile
50 self.version_number = 0
53 d = Serializer.serialize(self)
55 if 'profile' in d and d['profile'] != '':
56 d['profile'] = json.loads(d['profile'])
57 d['profileSupportList'] = [
58 DeploymentManagerProfileDefault,
60 profiles = config.get_dms_support_profiles()
61 for profile in profiles:
62 if profile == DeploymentManagerProfileSOL018:
63 d['profileSupportList'].append(profile)
64 elif profile == DeploymentManagerProfileSOL018HelmCLI:
65 d['profileSupportList'].append(profile)
67 if 'capabilities' in d and d['capabilities'] != '':
68 d['capabilities'] = json.loads(d['capabilities'])
71 def get_notification_dict(self):
72 return self.get_fields_as_dict(
73 ['deploymentManagerId', 'name', 'oCloudId', 'serviceUri',
77 class ResourcePool(InfrastructureInventoryObject, AgRoot, Serializer):
78 def __init__(self, id: str, name: str, location: str,
79 ocloudid: str, gLocationId: str = '',
80 description: str = '') -> None:
82 self.resourcePoolId = id
83 self.globalLocationId = gLocationId
85 self.description = description
86 self.oCloudId = ocloudid
87 self.location = location
91 self.version_number = 0
93 def get_notification_dict(self):
94 return self.get_fields_as_dict(
95 ['resourcePoolId', 'oCloudId', 'globalLocationId', 'name',
99 class ResourceType(InfrastructureInventoryObject, AgRoot, Serializer):
100 def __init__(self, typeid: str, name: str, typeEnum: ResourceTypeEnum,
101 vendor: str = '', model: str = '',
103 description: str = '') -> None:
105 self.resourceTypeId = typeid
106 self.resourceTypeEnum = typeEnum
108 self.description = description
111 self.version = version
112 self.alarmDictionary = None
113 self.resourceKind = ResourceKindEnum.UNDEFINED
114 self.resourceClass = ResourceTypeEnum.UNDEFINED
117 self.version_number = 0
120 d = Serializer.serialize(self)
121 if 'alarmDictionary' in d and \
122 type(d['alarmDictionary']) is AlarmDictionary:
123 d['alarmDictionary'] = d['alarmDictionary'].serialize()
126 def get_notification_dict(self):
127 return self.get_fields_as_dict(
128 ['resourceTypeId', 'name', 'description', 'vendor', 'model',
132 class Resource(InfrastructureInventoryObject, AgRoot, Serializer):
133 def __init__(self, resourceId: str, resourceTypeId: str,
134 resourcePoolId: str, parentId: str = '',
135 gAssetId: str = '', elements: str = '',
136 description: str = '', extensions: str = '') -> None:
138 self.resourceId = resourceId
139 self.description = description
140 self.resourceTypeId = resourceTypeId
141 self.globalAssetId = gAssetId
142 self.resourcePoolId = resourcePoolId
143 self.elements = elements
144 self.extensions = extensions
146 self.parentId = parentId
149 self.version_number = 0
151 def set_children(self, children: list):
152 self.children = children
154 def set_resource_type_name(self, resource_type_name: str):
155 self.resourceTypeName = resource_type_name
158 d = Serializer.serialize(self)
160 if 'elements' in d and d['elements'] != '':
161 d['elements'] = json.loads(d['elements'])
163 if not hasattr(self, 'children') or len(self.children) == 0:
168 for child in self.children:
169 d['children'].append(child.serialize())
172 def get_notification_dict(self):
173 return self.get_fields_as_dict(
174 ['resourceId', 'resourceTypeId', 'resourcePoolId', 'globalAssetId',
178 class Ocloud(InfrastructureInventoryObject, AgRoot, Serializer):
179 def __init__(self, ocloudid: str, name: str, imsendpoint: str,
180 globalcloudId: str = '',
181 description: str = '', version_number: int = 0) -> None:
183 self.oCloudId = ocloudid
184 self.globalCloudId = globalcloudId
186 self.description = description
187 self.serviceUri = imsendpoint
188 self.resourceTypes = []
189 self.resourcePools = []
190 self.deploymentManagers = []
191 self.smoRegistrationService = ''
194 self.version_number = version_number
196 def get_notification_dict(self):
197 return self.get_fields_as_dict(
198 ['oCloudId', 'globalcloudId', 'globalCloudId', 'name',
199 'description', 'serviceUri'])
200 # def addDeploymentManager(self,
201 # deploymentManager: DeploymentManager):
203 # deploymentManager.oCloudId = self.oCloudId
205 # lambda x: x.deploymentManagerId ==
206 # deploymentManager.deploymentManagerId,
207 # self.deploymentManagers)
208 # for o in old or []:
209 # self.deploymentManagers.remove(o)
210 # self.deploymentManagers.append(deploymentManager)