Fix pre-commit errors (a7732ecd) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/edgecloud/adapters/aeros/client.py

+188 −216

File changed.

Preview size limit exceeded, changes collapsed.

src/sunrise6g_opensdk/edgecloud/adapters/aeros/continuum_models.py

+74 −54

Original line number	Diff line number	Diff line
		'''
		"""
		aerOS continuum models
		'''
		"""

		from enum import Enum
		from typing import List, Dict, Any, Union, Optional
		from typing import Any, Dict, List, Optional, Union

		from pydantic import BaseModel, Field


		class ServiceNotFound(BaseModel):
		'''
		"""
		Docstring
		'''
		"""

		detail: str = "Service not found"


		@@ -17,6 +20,7 @@ class CPUComparisonOperator(BaseModel):
		"""
		CPU requirment for now is that usage should be less than
		"""

		less_or_equal: Union[float, None] = None


		@@ -24,6 +28,7 @@ class CPUArchComparisonOperator(BaseModel):
		"""
		CPU arch requirment, equal to str
		"""

		equal: Union[str, None] = None


		@@ -31,6 +36,7 @@ class MEMComparisonOperator(BaseModel):
		"""
		RAM requirment for now is that available RAM should be more than
		"""

		greater_or_equal: Union[str, None] = None


		@@ -38,6 +44,7 @@ class EnergyEfficienyComparisonOperator(BaseModel):
		"""
		Energy Efficiency requirment for now is that IE should have energy efficiency more than a %
		"""

		greater_or_equal: Union[str, None] = None


		@@ -45,6 +52,7 @@ class GreenComparisonOperator(BaseModel):
		"""
		IE Green requirment for now is that IE should have green energy mix which us more than a %
		"""

		greater_or_equal: Union[str, None] = None


		@@ -52,22 +60,25 @@ class RTComparisonOperator(BaseModel):
		"""
		Real Time requirment T/F
		"""

		equal: Union[bool, None] = None


		class CpuArch(str, Enum):
		'''
		"""
		Enumeration with possible cpu types
		'''
		"""

		x86_64 = "x86_64"
		arm64 = "arm64"
		arm32 = "arm32"


		class Coordinates(BaseModel):
		'''
		"""
		IE coordinate requirements
		'''
		"""

		coordinates: List[List[float]]


		@@ -75,26 +86,24 @@ class DomainIdOperator(BaseModel):
		"""
		CPU arch requirment, equal to str
		"""

		equal: Union[str, None] = None


		class Property(BaseModel):
		'''
		"""
		IE capabilities
		'''
		cpu_usage: CPUComparisonOperator = Field(
		default_factory=CPUComparisonOperator)
		cpu_arch: CPUArchComparisonOperator = Field(
		default_factory=CPUArchComparisonOperator)
		mem_size: MEMComparisonOperator = Field(
		default_factory=MEMComparisonOperator)
		realtime: RTComparisonOperator = Field(
		default_factory=RTComparisonOperator)
		"""

		cpu_usage: CPUComparisonOperator = Field(default_factory=CPUComparisonOperator)
		cpu_arch: CPUArchComparisonOperator = Field(default_factory=CPUArchComparisonOperator)
		mem_size: MEMComparisonOperator = Field(default_factory=MEMComparisonOperator)
		realtime: RTComparisonOperator = Field(default_factory=RTComparisonOperator)
		area: Coordinates = None
		energy_efficiency: EnergyEfficienyComparisonOperator = Field(
		default_factory=EnergyEfficienyComparisonOperator)
		green: GreenComparisonOperator = Field(
		default_factory=GreenComparisonOperator)
		default_factory=EnergyEfficienyComparisonOperator
		)
		green: GreenComparisonOperator = Field(default_factory=GreenComparisonOperator)
		domain_id: DomainIdOperator = Field(default_factory=DomainIdOperator)

		# @field_validator('mem_size')
		@@ -108,71 +117,80 @@ class Property(BaseModel):


		class HostCapability(BaseModel):
		'''
		"""
		Host properties
		'''
		"""

		properties: Property


		class NodeFilter(BaseModel):
		'''
		"""
		Node filter,
		How to filter continuum IE and select canditate list
		'''
		"""

		properties: Optional[Dict[str, List[str]]] = None
		capabilities: Optional[List[Dict[str, HostCapability]]] = None


		class HostRequirement(BaseModel):
		'''
		"""
		capabilities of node
		'''
		"""

		# node_filter: Dict[str, List[Dict[str, HostCapability]]]
		node_filter: NodeFilter


		class PortProperties(BaseModel):
		'''
		"""
		Workload port description
		'''
		"""

		protocol: List[str] = Field(...)
		source: int = Field(...)


		class ExposedPort(BaseModel):
		'''
		"""
		Workload exposed network ports
		'''
		"""

		properties: PortProperties = Field(...)


		class NetworkProperties(BaseModel):
		'''
		"""
		Dict of network requirments, name of port and protperty = [protocol, port] mapping
		'''
		"""

		ports: Dict[str, ExposedPort] = Field(...)
		exposePorts: Optional[bool]


		class NetworkRequirement(BaseModel):
		'''
		"""
		Top level key of network requirments
		'''
		"""

		properties: NetworkProperties


		class CustomRequirement(BaseModel):
		'''
		"""
		Define a custom requirement type that can be either a host or a network requirement
		'''
		"""

		host: HostRequirement = None
		network: NetworkRequirement = None


		class ArtifactModel(BaseModel):
		'''
		"""
		Artifact has a useer defined id and then a dict with the following keys:
		'''
		"""

		file: str
		type: str
		repository: str
		@@ -182,9 +200,10 @@ class ArtifactModel(BaseModel):


		class NodeTemplate(BaseModel):
		'''
		"""
		Node template "tosca.nodes.Container.Application"
		'''
		"""

		type: str
		requirements: List[CustomRequirement]
		artifacts: Dict[str, ArtifactModel]
		@@ -193,9 +212,10 @@ class NodeTemplate(BaseModel):


		class TOSCA(BaseModel):
		'''
		"""
		The TOSCA structure
		'''
		"""

		tosca_definitions_version: str
		description: str
		serviceOverlay: Optional[bool] = False

src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/aeros2gsma_zone_details.py

+37 −39

Original line number	Diff line number	Diff line
		'''
		"""
		aeros2gsma_zone_details.py
		'''
		from typing import List, Dict, Any
		"""

		from typing import Any, Dict, List


		def transformer(domain_ies: List[Dict[str, Any]],
		domain: str) -> Dict[str, Any]:
		def transformer(domain_ies: List[Dict[str, Any]], domain: str) -> Dict[str, Any]: # noqa: C901
		"""
		Transform aerOS InfrastructureElements into GSMA ZoneRegisteredData structure.
		:param domain_ies: List of aerOS InfrastructureElement dicts
		@@ -78,8 +78,7 @@ def transformer(domain_ies: List[Dict[str, Any]],
		cpu_cores = int(element.get("cpuCores", 0) or 0)
		ram_cap = int(element.get("ramCapacity", 0) or 0) # MB?
		avail_ram = int(element.get("availableRam", 0) or 0) # MB?
		disk_cap = int(element.get("diskCapacity", 0)
		or 0) # MB/GB? (pass-through)
		disk_cap = int(element.get("diskCapacity", 0) or 0) # MB/GB? (pass-through)
		avail_disk = int(element.get("availableDisk", 0) or 0)

		total_cpu += cpu_cores
		@@ -99,22 +98,19 @@ def transformer(domain_ies: List[Dict[str, Any]],

		# Create a flavour per machine
		flavour = {
		"flavourId":
		f"{element.get('hostname', 'host')}-{element.get('containerTechnology', 'CT')}",
		"cpuArchType":
		isa, # Literal ISA_*
		"supportedOSTypes": [{
		"flavourId": f"{element.get('hostname', 'host')}-{element.get('containerTechnology', 'CT')}",
		"cpuArchType": isa, # Literal ISA_*
		"supportedOSTypes": [
		{
		"architecture": ost_arch, # 'x86_64' or 'x86'
		"distribution": dist, # 'UBUNTU' or 'OTHER'
		"version": ver, # 'OS_VERSION_UBUNTU_2204_LTS' or 'OTHER'
		"license": "OS_LICENSE_TYPE_FREE",
		}],
		"numCPU":
		cpu_cores,
		"memorySize":
		ram_cap,
		"storageSize":
		disk_cap,
		}
		],
		"numCPU": cpu_cores,
		"memorySize": ram_cap,
		"storageSize": disk_cap,
		}
		flavours_supported.append(flavour)

		@@ -133,22 +129,24 @@ def transformer(domain_ies: List[Dict[str, Any]],
		agg_isa = pick_aggregate_isa()

		result = {
		"zoneId":
		domain,
		"reservedComputeResources": [{
		"zoneId": domain,
		"reservedComputeResources": [
		{
		"cpuArchType": agg_isa,
		"numCPU": int(
		total_cpu
		), # Same as Quotas untill we have somem policy or data to differentiate
		"memory": total_ram, # ditto
		}],
		"computeResourceQuotaLimits": [{
		}
		],
		"computeResourceQuotaLimits": [
		{
		"cpuArchType": agg_isa,
		"numCPU": int(total_cpu),
		"memory": total_ram,
		}],
		"flavoursSupported":
		flavours_supported,
		}
		],
		"flavoursSupported": flavours_supported,
		}

		return result

src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/camara2aeros_converter.py

+62 −41

Original line number	Diff line number	Diff line
		'''
		"""
		Module: converter.py
		This module provides functions to convert application manifests into TOSCA models.
		It includes the `generate_tosca` function that constructs a TOSCA model based on
		the application manifest and associated app zones.
		'''
		from typing import List, Dict, Any
		"""

		from typing import List

		import yaml

		from sunrise6g_opensdk.edgecloud.adapters.aeros import config
		from sunrise6g_opensdk.logger import setup_logger
		from sunrise6g_opensdk.edgecloud.core.camara_schemas import AppManifest, VisibilityType
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		TOSCA, NodeTemplate, CustomRequirement, HostRequirement, HostCapability,
		Property as HostProperty, DomainIdOperator, NodeFilter, NetworkRequirement,
		NetworkProperties, ExposedPort, PortProperties, ArtifactModel)
		TOSCA,
		ArtifactModel,
		CustomRequirement,
		DomainIdOperator,
		ExposedPort,
		HostCapability,
		HostRequirement,
		NetworkProperties,
		NetworkRequirement,
		NodeFilter,
		NodeTemplate,
		PortProperties,
		)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		Property as HostProperty,
		)
		from sunrise6g_opensdk.edgecloud.core.camara_schemas import AppManifest, VisibilityType
		from sunrise6g_opensdk.logger import setup_logger

		logger = setup_logger(__name__, is_debug=True, file_name=config.LOG_FILE)


		def generate_tosca(app_manifest: AppManifest, app_zones: List[str]) -> str:
		'''
		"""
		Generate a TOSCA model from the application manifest and app zones.
		Args:
		app_manifest (AppManifest): The application manifest containing details about the app.
		app_zones (List[Dict[str, Any]]): List of app zones where the app will be deployed.
		Returns:
		TOSCA yaml as string which can be used in a POST request with applcation type yaml
		'''
		"""
		component = app_manifest.componentSpec[0]
		image_path = app_manifest.appRepo.imagePath.root
		image_file = image_path.split("/")[-1]
		repository_url = "/".join(
		image_path.split("/")[:-1]) if "/" in image_path else "docker_hub"
		repository_url = "/".join(image_path.split("/")[:-1]) if "/" in image_path else "docker_hub"

		zone_id = app_zones[0]
		logger.info("DEBUG : %s", app_manifest.requiredResources.root)
		@@ -38,22 +53,24 @@ def generate_tosca(app_manifest: AppManifest, app_zones: List[str]) -> str:

		res = app_manifest.requiredResources.root
		if hasattr(res, "applicationResources") and hasattr(
		res.applicationResources.cpuPool.topology, "minNodeMemory"):
		res.applicationResources.cpuPool.topology, "minNodeMemory"
		):
		min_node_memory = res.applicationResources.cpuPool.topology.minNodeMemory
		else:
		min_node_memory = 1024

		# Build exposed network ports
		ports = {
		iface.interfaceId:
		ExposedPort(properties=PortProperties(
		protocol=[iface.protocol.value.lower()], source=iface.port))
		iface.interfaceId: ExposedPort(
		properties=PortProperties(protocol=[iface.protocol.value.lower()], source=iface.port)
		)
		for iface in component.networkInterfaces
		}

		expose_ports = any(
		iface.visibilityType == VisibilityType.VISIBILITY_EXTERNAL
		for iface in component.networkInterfaces)
		for iface in component.networkInterfaces
		)

		# Define host property constraints
		host_props = HostProperty(
		@@ -68,14 +85,18 @@ def generate_tosca(app_manifest: AppManifest, app_zones: List[str]) -> str:

		# Create Node compute and network requirements
		requirements = [
		CustomRequirement(network=NetworkRequirement(
		properties=NetworkProperties(ports=ports,
		exposePorts=expose_ports))),
		CustomRequirement(host=HostRequirement(node_filter=NodeFilter(
		capabilities=[{
		"host": HostCapability(properties=host_props)
		}],
		properties=None)))
		CustomRequirement(
		network=NetworkRequirement(
		properties=NetworkProperties(ports=ports, exposePorts=expose_ports)
		)
		),
		CustomRequirement(
		host=HostRequirement(
		node_filter=NodeFilter(
		capabilities=[{"host": HostCapability(properties=host_props)}], properties=None
		)
		)
		),
		]
		# Define the NodeTemplate
		node_template = NodeTemplate(
		@@ -83,40 +104,40 @@ def generate_tosca(app_manifest: AppManifest, app_zones: List[str]) -> str:
		isJob=False,
		requirements=requirements,
		artifacts={
		"application_image":
		ArtifactModel(
		"application_image": ArtifactModel(
		file=image_file,
		type="tosca.artifacts.Deployment.Image.Container.Docker",
		repository=repository_url,
		is_private=app_manifest.appRepo.type == "PRIVATEREPO",
		username=app_manifest.appRepo.userName,
		password=app_manifest.appRepo.credentials)
		password=app_manifest.appRepo.credentials,
		)
		},
		interfaces={
		"Standard": {
		"create": {
		"implementation": "application_image",
		"inputs": {
		"cliArgs": [],
		"envVars": []
		"inputs": {"cliArgs": [], "envVars": []},
		}
		}
		}
		})
		},
		)

		# Assemble full TOSCA object
		tosca = TOSCA(tosca_definitions_version="tosca_simple_yaml_1_3",
		tosca = TOSCA(
		tosca_definitions_version="tosca_simple_yaml_1_3",
		description=f"TOSCA for {app_manifest.name}",
		serviceOverlay=False,
		node_templates={component.componentName: node_template})
		node_templates={component.componentName: node_template},
		)

		tosca_dict = tosca.model_dump(by_alias=True, exclude_none=True)

		for template in tosca_dict.get("node_templates", {}).values():
		template["requirements"] = [{
		k: v
		for k, v in req.items() if v is not None
		} for req in template.get("requirements", [])]
		template["requirements"] = [
		{k: v for k, v in req.items() if v is not None}
		for req in template.get("requirements", [])
		]

		yaml_str = yaml.dump(tosca_dict, sort_keys=False)
		return yaml_str

src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/gsma2aeros_converter.py

+46 −43

Original line number	Diff line number	Diff line
		@@ -10,35 +10,39 @@ Notes:
		- Network ports are omitted for now (exposePorts = False).
		"""

		from typing import Optional, Callable, Dict, Any, List
		from typing import Callable, Dict, List, Optional

		import yaml

		from sunrise6g_opensdk.edgecloud.adapters.aeros import config
		from sunrise6g_opensdk.edgecloud.adapters.aeros.errors import (
		ResourceNotFoundError,
		InvalidArgumentError,
		)
		from sunrise6g_opensdk.logger import setup_logger
		from sunrise6g_opensdk.edgecloud.core import gsma_schemas
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		TOSCA,
		NodeTemplate,
		ArtifactModel,
		CustomRequirement,
		HostRequirement,
		HostCapability,
		Property as HostProperty,
		DomainIdOperator,
		NodeFilter,
		NetworkRequirement,
		NetworkProperties,
		ExposedPort,
		HostCapability,
		HostRequirement,
		NetworkProperties,
		NetworkRequirement,
		NodeFilter,
		NodeTemplate,
		PortProperties,
		ArtifactModel,
		)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		Property as HostProperty,
		)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.errors import (
		InvalidArgumentError,
		ResourceNotFoundError,
		)
		from sunrise6g_opensdk.edgecloud.core import gsma_schemas
		from sunrise6g_opensdk.logger import setup_logger

		logger = setup_logger(__name__, is_debug=True, file_name=config.LOG_FILE)


		def generate_tosca_from_gsma_with_artefacts(
		def generate_tosca_from_gsma_with_artefacts( # noqa: C901
		app_model: gsma_schemas.ApplicationModel,
		zone_id: str,
		artefact_resolver: Callable[[str], Optional[gsma_schemas.Artefact]],
		@@ -70,8 +74,7 @@ def generate_tosca_from_gsma_with_artefacts(
		for comp in app_model.appComponentSpecs:
		artefact = artefact_resolver(comp.artefactId)
		if not artefact:
		raise ResourceNotFoundError(
		f"GSMA artefact '{comp.artefactId}' not found")
		raise ResourceNotFoundError(f"GSMA artefact '{comp.artefactId}' not found")

		# pick the componentSpec that matches componentName, else first
		comp_spec = None
		@@ -83,12 +86,10 @@ def generate_tosca_from_gsma_with_artefacts(
		if comp_spec is None:
		comp_spec = artefact.componentSpec[0]
		else:
		raise InvalidArgumentError(
		f"Artefact '{artefact.artefactId}' has no componentSpec")
		raise InvalidArgumentError(f"Artefact '{artefact.artefactId}' has no componentSpec")

		# Resolve container image
		image = comp_spec.images[
		0] if comp_spec.images else "docker.io/library/nginx:stable"
		image = comp_spec.images[0] if comp_spec.images else "docker.io/library/nginx:stable"
		if "/" in image:
		repository_url = "/".join(image.split("/")[:-1])
		image_file = image.split("/")[-1]
		@@ -103,8 +104,9 @@ def generate_tosca_from_gsma_with_artefacts(
		protocol = str(iface.get("protocol", "TCP")).lower()
		port = iface.get("port")
		if isinstance(port, int):
		ports[f"if{idx}"] = ExposedPort(properties=PortProperties(
		protocol=[protocol], source=port))
		ports[f"if{idx}"] = ExposedPort(
		properties=PortProperties(protocol=[protocol], source=port)
		)
		expose_ports = True

		# Build cliArgs as a list of dicts: [{"KEY": "VAL"}, {"FLAG": ""}, ...]
		@@ -135,8 +137,7 @@ def generate_tosca_from_gsma_with_artefacts(
		for item in comp_spec.compEnvParams:
		if isinstance(item, dict):
		if "name" in item and "value" in item:
		env_vars.append(
		{str(item["name"]): str(item["value"])})
		env_vars.append({str(item["name"]): str(item["value"])})
		elif len(item) == 1: # already mapping-like {"KEY": "VAL"}
		k, v = next(iter(item.items()))
		env_vars.append({str(k): str(v)})
		@@ -153,15 +154,19 @@ def generate_tosca_from_gsma_with_artefacts(
		)

		requirements = [
		CustomRequirement(network=NetworkRequirement(
		properties=NetworkProperties(ports=ports,
		exposePorts=expose_ports))),
		CustomRequirement(host=HostRequirement(node_filter=NodeFilter(
		capabilities=[{
		"host": HostCapability(properties=host_props)
		}],
		CustomRequirement(
		network=NetworkRequirement(
		properties=NetworkProperties(ports=ports, exposePorts=expose_ports)
		)
		),
		CustomRequirement(
		host=HostRequirement(
		node_filter=NodeFilter(
		capabilities=[{"host": HostCapability(properties=host_props)}],
		properties=None,
		))),
		)
		)
		),
		]

		# PUBLICREPO => is_private=False and omit credentials
		@@ -180,8 +185,7 @@ def generate_tosca_from_gsma_with_artefacts(
		isJob=False,
		requirements=requirements,
		artifacts={
		"application_image":
		ArtifactModel(
		"application_image": ArtifactModel(
		file=image_file,
		type="tosca.artifacts.Deployment.Image.Container.Docker",
		repository=repository_url,
		@@ -206,8 +210,7 @@ def generate_tosca_from_gsma_with_artefacts(
		# Assemble and dump TOSCA
		tosca = TOSCA(
		tosca_definitions_version="tosca_simple_yaml_1_3",
		description=
		f"GSMA->TOSCA for {app_model.appMetaData.appName} ({app_model.appId})",
		description=f"GSMA->TOSCA for {app_model.appMetaData.appName} ({app_model.appId})",
		serviceOverlay=False,
		node_templates=node_templates,
		)
		@@ -215,9 +218,9 @@ def generate_tosca_from_gsma_with_artefacts(
		tosca_dict = tosca.model_dump(by_alias=True, exclude_none=True)
		# Clean requirements lists from None entries
		for template in tosca_dict.get("node_templates", {}).values():
		template["requirements"] = [{
		k: v
		for k, v in req.items() if v is not None
		} for req in template.get("requirements", [])]
		template["requirements"] = [
		{k: v for k, v in req.items() if v is not None}
		for req in template.get("requirements", [])
		]

		return yaml.dump(tosca_dict, sort_keys=False)