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'])
69 if 'capacity' in d and d['capacity'] != '':
70 d['capacity'] = json.loads(d['capacity'])
73 def get_notification_dict(self):
74 return self.get_fields_as_dict(
75 ['deploymentManagerId', 'name', 'oCloudId', 'serviceUri',
79 class ResourcePool(InfrastructureInventoryObject, AgRoot, Serializer):
80 def __init__(self, id: str, name: str, location: str,
81 ocloudid: str, gLocationId: str = '',
82 description: str = '') -> None:
84 self.resourcePoolId = id
85 self.globalLocationId = gLocationId
87 self.description = description
88 self.oCloudId = ocloudid
89 self.location = location
93 self.version_number = 0
95 def get_notification_dict(self):
96 return self.get_fields_as_dict(
97 ['resourcePoolId', 'oCloudId', 'globalLocationId', 'name',
101 class ResourceType(InfrastructureInventoryObject, AgRoot, Serializer):
102 def __init__(self, typeid: str, name: str, typeEnum: ResourceTypeEnum,
103 vendor: str = '', model: str = '',
105 description: str = '') -> None:
107 self.resourceTypeId = typeid
108 self.resourceTypeEnum = typeEnum
110 self.description = description
113 self.version = version
114 self.alarmDictionary = None
115 self.resourceKind = ResourceKindEnum.UNDEFINED
116 self.resourceClass = ResourceTypeEnum.UNDEFINED
119 self.version_number = 0
122 d = Serializer.serialize(self)
123 if 'alarmDictionary' in d and \
124 type(d['alarmDictionary']) is AlarmDictionary:
125 d['alarmDictionary'] = d['alarmDictionary'].serialize()
128 def get_notification_dict(self):
129 return self.get_fields_as_dict(
130 ['resourceTypeId', 'name', 'description', 'vendor', 'model',
134 class Resource(InfrastructureInventoryObject, AgRoot, Serializer):
135 def __init__(self, resourceId: str, resourceTypeId: str,
136 resourcePoolId: str, parentId: str = '',
137 gAssetId: str = '', elements: str = '',
138 description: str = '', extensions: str = '') -> None:
140 self.resourceId = resourceId
141 self.description = description
142 self.resourceTypeId = resourceTypeId
143 self.globalAssetId = gAssetId
144 self.resourcePoolId = resourcePoolId
145 self.elements = elements
146 self.extensions = extensions
148 self.parentId = parentId
151 self.version_number = 0
153 def set_children(self, children: list):
154 self.children = children
156 def set_resource_type_name(self, resource_type_name: str):
157 self.resourceTypeName = resource_type_name
160 d = Serializer.serialize(self)
162 if 'elements' in d and d['elements'] != '':
163 d['elements'] = json.loads(d['elements'])
165 if not hasattr(self, 'children') or len(self.children) == 0:
170 for child in self.children:
171 d['children'].append(child.serialize())
174 def get_notification_dict(self):
175 return self.get_fields_as_dict(
176 ['resourceId', 'resourceTypeId', 'resourcePoolId', 'globalAssetId',
180 class Ocloud(InfrastructureInventoryObject, AgRoot, Serializer):
181 def __init__(self, ocloudid: str, name: str, imsendpoint: str,
182 globalcloudId: str = '',
183 description: str = '', version_number: int = 0) -> None:
185 self.oCloudId = ocloudid
186 self.globalCloudId = globalcloudId
188 self.description = description
189 self.serviceUri = imsendpoint
190 self.resourceTypes = []
191 self.resourcePools = []
192 self.deploymentManagers = []
193 self.smoRegistrationService = ''
196 self.version_number = version_number
198 def get_notification_dict(self):
199 return self.get_fields_as_dict(
200 ['oCloudId', 'globalcloudId', 'globalCloudId', 'name',
201 'description', 'serviceUri'])
202 # def addDeploymentManager(self,
203 # deploymentManager: DeploymentManager):
205 # deploymentManager.oCloudId = self.oCloudId
207 # lambda x: x.deploymentManagerId ==
208 # deploymentManager.deploymentManagerId,
209 # self.deploymentManagers)
210 # for o in old or []:
211 # self.deploymentManagers.remove(o)
212 # self.deploymentManagers.append(deploymentManager)