Dev

# -*- coding: utf-8 -*-

##

# Copyright 2025-present by Software Networks Area, i2CAT.

# All rights reserved.

#

# This file is part of the Open SDK

#

# Contributors:

#   - Adrián Pino Martínez (adrian.pino@i2cat.net)

##

from typing import Dict

from src.common.sdk_factory import SdkFactory

class Sdk:

    @staticmethod

    def create_clients_from(

        client_specs: Dict[str, Dict[str, str]],

    ) -> Dict[str, object]:

        """

        Create and return a dictionary of instantiated edgecloud/network/o-ran clients

        based on the provided specifications.

        Args:

            client_specs (dict): A dictionary where each key is the client's domain (e.g., 'edgecloud', 'network'),

                                 and each value is a dictionary containing:

                                 - 'client_name' (str): The specific name of the client (e.g., 'i2edge', 'open5gs').

                                 - 'base_url' (str): The base URL for the client's API.

                                 Additional parameters like 'scs_as_id' may also be included.

        Returns:

            dict: A dictionary where keys are the 'client_name' (str) and values are

                  the instantiated client objects.

        Example:

            >>> from src.common.universal_client_catalog import UniversalCatalogClient

            >>>

            >>> client_specs_example = {

            >>>     'edgecloud': {

            >>>         'client_name': 'i2edge',

            >>>         'base_url': 'http://ip_edge_cloud:port',

            >>>         'additionalEdgeCloudParamater1': 'example'

            >>>     },

            >>>     'network': {

            >>>         'client_name': 'open5gs',

            >>>         'base_url': 'http://ip_network:port',

            >>>         'additionalNetworkParamater1': 'example'

            >>>     }

            >>> }

            >>>

            >>> clients = UniversalCatalogClient.create_clients(client_specs_example)

            >>> edgecloud_client = clients.get("edgecloud")

            >>> network_client = clients.get("network")

            >>>

            >>> edgecloud_client.get_edge_cloud_zones()

            >>> network_client.get_qod_session(session_id="example_session_id")

        """

        sdk_client = SdkFactory()

        clients = {}

        for domain, config in client_specs.items():

            client_name = config["client_name"]

            base_url = config["base_url"]

            # Support of additional paramaters for specific clients

            kwargs = {

                k: v for k, v in config.items() if k not in ("client_name", "base_url")

            }

            client = sdk_client.instantiate_and_retrieve_clients(

                domain, client_name, base_url, **kwargs

            )

            clients[domain] = client

        return clients

# -*- coding: utf-8 -*-

##

# Copyright 2025-present by Software Networks Area, i2CAT.

# All rights reserved.

#

# This file is part of the Open SDK

#

# Contributors:

#   - Adrián Pino Martínez (adrian.pino@i2cat.net)

##

from src.edgecloud.clients.aeros.client import EdgeApplicationManager as AerosClient

from src.edgecloud.clients.i2edge.client import EdgeApplicationManager as I2EdgeClient

from src.edgecloud.clients.piedge.client import EdgeApplicationManager as PiEdgeClient

# from src.network.clients.oai.client import NetworkManager as OaiCoreClient

# from src.network.clients.open5gcore.client import NetworkManager as Open5GCoreClient

# from src.network.clients.open5gs.client import NetworkManager as Open5GSClient

def _edgecloud_factory(client_name: str, base_url: str, **kwargs):

    edge_cloud_factory = {

        "aeros": lambda url, **kw: AerosClient(base_url=url, **kw),

        "i2edge": lambda url: I2EdgeClient(base_url=url),

        "piedge": lambda url, **kw: PiEdgeClient(base_url=url, **kw)

    }

    try:

        return edge_cloud_factory[client_name](base_url, **kwargs)

    except KeyError:

        raise ValueError(

            f"Invalid edgecloud client '{client_name}'. Available: {list(edge_cloud_factory)}"

        )

# def _network_factory(client_name: str, base_url: str, **kwargs):

#     if "scs_as_id" not in kwargs:

#         raise ValueError("Missing required 'scs_as_id' for network clients.")

#     scs_as_id = kwargs.pop("scs_as_id")

#     network_factory = {

#         "open5gs": lambda url, scs_id, **kw: Open5GSClient(

#             base_url=url, scs_as_id=scs_id, **kw

#         ),

#         "oai": lambda url, scs_id, **kw: OaiCoreClient(

#             base_url=url, scs_as_id=scs_id, **kw

#         ),

#         "open5gcore": lambda url, scs_id, **kw: Open5GCoreClient(

#             base_url=url, scs_as_id=scs_id, **kw

#         ),

#     }

#     try:

#         return network_factory[client_name](base_url, scs_as_id, **kwargs)

#     except KeyError:

#         raise ValueError(

#             f"Invalid network client '{client_name}'. Available: {list(network_factory)}"

#         )

# def _oran_factory(client_name: str, base_url: str):

#     # TODO

class SdkFactory:

    _domain_factories = {

        "edgecloud": _edgecloud_factory

        # "network": _network_factory,

        # "oran": _oran_factory,

    }

    @classmethod

    def instantiate_and_retrieve_clients(

        cls, domain: str, client_name: str, base_url: str, **kwargs

    ):

        try:

            catalog = cls._domain_factories[domain]

        except KeyError:

            raise ValueError(

                f"Unsupported domain '{domain}'. Supported: {list(cls._domain_factories)}"

            )

        return catalog(client_name, base_url, **kwargs)

"""

aerOS client

  This module provides a client for interacting with the aerOS REST API.

  It includes methods for onboarding/deploying applications,

    and querying aerOS continuum entities

  aerOS domain is exposed as zones

  aerOS services and service components are exposed as applications

  Client is initialized with a base URL for the aerOS API

    and an access token for authentication.

"""

from src.edgecloud.clients.aeros import config

from src.logger import setup_logger

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

# TODO: The following should only appear in case aerOS client is used

#       Currently even if another client is used, the logs appear

# logger.info("aerOS client initialized")

# logger.debug("aerOS API URL: %s", config.aerOS_API_URL)

# logger.debug("aerOS access token: %s", config.aerOS_ACCESS_TOKEN)

# logger.debug("aerOS debug mode: %s", config.DEBUG)

# logger.debug("aerOS log file: %s", config.LOG_FILE)

##

# This file is part of the Open SDK

#

# Contributors:

#   - Vasilis Pitsilis (vpitsilis@dat.demokritos.gr, vpitsilis@iit.demokritos.gr)

#   - Andreas Sakellaropoulos (asakellaropoulos@iit.demokritos.gr)

##

from typing import Any, Dict, List, Optional

from src.edgecloud.clients.aeros import config

from src.edgecloud.clients.aeros.continuum_client import ContinuumClient

from src.edgecloud.core.edgecloud_interface import EdgeCloudManagementInterface

from src.logger import setup_logger

class EdgeApplicationManager(EdgeCloudManagementInterface):

    """

    aerOS Continuum Client

    FIXME: Handle None responses from continuum client

    """

    def __init__(self, base_url: str, **kwargs):

        self.base_url = base_url

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

        # Overwrite config values if provided via kwargs

        if "aerOS_API_URL" in kwargs:

            config.aerOS_API_URL = kwargs["aerOS_API_URL"]

        if "aerOS_ACCESS_TOKEN" in kwargs:

            config.aerOS_ACCESS_TOKEN = kwargs["aerOS_ACCESS_TOKEN"]

        if "aerOS_HLO_TOKEN" in kwargs:

            config.aerOS_HLO_TOKEN = kwargs["aerOS_HLO_TOKEN"]

        if not config.aerOS_API_URL:

            raise ValueError("Missing 'aerOS_API_URL'")

        if not config.aerOS_ACCESS_TOKEN:

            raise ValueError("Missing 'aerOS_ACCESS_TOKEN'")

        if not config.aerOS_HLO_TOKEN:

            raise ValueError("Missing 'aerOS_HLO_TOKEN'")

    def onboard_app(self, app_manifest: Dict) -> Dict:

        # HLO-FE POST with TOSCA and app_id (service_id)

        service_id = app_manifest.get("serviceId")

        tosca_str = app_manifest.get("tosca")

        aeros_client = ContinuumClient(self.base_url)

        onboard_response = aeros_client.onboard_service(

            service_id=service_id, tosca_str=tosca_str

        )

        return {"appId": onboard_response["serviceId"]}

    def get_all_onboarded_apps(self) -> List[Dict]:

        aeros_client = ContinuumClient(self.base_url)

        ngsild_params = "type=Service&format=simplified"

        aeros_apps = aeros_client.query_entities(ngsild_params)

        return [

            {"appId": service["id"], "name": service["name"]} for service in aeros_apps

        ]

        # return [{"appId": "1234-5678", "name": "TestApp"}]

    def get_onboarded_app(self, app_id: str) -> Dict:

        aeros_client = ContinuumClient(self.base_url)

        ngsild_params = "format=simplified"

        aeros_app = aeros_client.query_entity(app_id, ngsild_params)

        return {"appId": aeros_app["id"], "name": aeros_app["name"]}

    def delete_onboarded_app(self, app_id: str) -> None:

        print(f"Deleting application: {app_id}")

        # TBD: Purge from continuum (make all ngsil-ld calls for servieId connected entities)

        # Should check if undeployed first

    def deploy_app(self, app_id: str, app_zones: List[Dict]) -> Dict:

        # HLO-FE PUT with app_id (service_id)

        aeros_client = ContinuumClient(self.base_url)

        deploy_response = aeros_client.deploy_service(app_id)

        return {"appInstanceId": deploy_response["serviceId"]}

    def get_all_deployed_apps(

        self,

        app_id: Optional[str] = None,

        app_instance_id: Optional[str] = None,

        region: Optional[str] = None,

    ) -> List[Dict]:

        # FIXME: Get services in deployed state

        aeros_client = ContinuumClient(self.base_url)

        ngsild_params = 'type=Service&format=simplified&q=actionType=="DEPLOYED"'

        if app_id:

            ngsild_params += f'&q=service=="{app_id}"'

        aeros_apps = aeros_client.query_entities(ngsild_params)

        return [

            {

                "appInstanceId": service["id"],

                "status":

                # scomponent["serviceComponentStatus"].split(":")[-1].lower()

                service["actionType"],

            }

            for service in aeros_apps

        ]

        # return [{"appInstanceId": "abcd-efgh", "status": "ready"}]

    # def get_all_deployed_apps(self,

    #                           app_id: Optional[str] = None,

    #                           app_instance_id: Optional[str] = None,

    #                           region: Optional[str] = None) -> List[Dict]:

    #     # FIXME: Get services in deployed state

    #     aeros_client = ContinuumClient(self.base_url)

    #     ngsild_params = "type=ServiceComponent&format=simplified"

    #     if app_id:

    #         ngsild_params += f'&q=service=="{app_id}"'

    #     aeros_apps = aeros_client.query_entities(ngsild_params)

    #     return [{

    #         "appInstanceId":

    #         scomponent["id"],

    #         "status":

    #         scomponent["serviceComponentStatus"].split(":")[-1].lower()

    #     } for scomponent in aeros_apps]

    #     # return [{"appInstanceId": "abcd-efgh", "status": "ready"}]

    def undeploy_app(self, app_instance_id: str) -> None:

        # HLO-FE DELETE with app_id (service_id)

        aeros_client = ContinuumClient(self.base_url)

        _ = aeros_client.undeploy_service(app_instance_id)

    def get_edge_cloud_zones(

        self, region: Optional[str] = None, status: Optional[str] = None

    ) -> List[Dict]:

        aeros_client = ContinuumClient(self.base_url)

        ngsild_params = "type=Domain&format=simplified"

        aeros_domains = aeros_client.query_entities(ngsild_params)

        return [

            {

                "edgeCloudZoneId": domain["id"],

                "status": domain["domainStatus"].split(":")[-1].lower(),

            }

            for domain in aeros_domains

        ]

    # return [{"edgeCloudZoneId": "zone-1", "status": "active"}]

    def get_edge_cloud_zones_details(

        self, zone_id: str, flavour_id: Optional[str] = None

    ) -> Dict:

        """

        Get details of a specific edge cloud zone.

        :param zone_id: The ID of the edge cloud zone

        :param flavour_id: Optional flavour ID to filter the results

        :return: Details of the edge cloud zone

        """

        # Minimal mocked response based on required fields of 'ZoneRegisteredData' in GSMA OPG E/WBI API

        # return {

        #     "zoneId":

        #     zone_id,

        #     "reservedComputeResources": [{

        #         "cpuArchType": "ISA_X86_64",

        #         "numCPU": "4",

        #         "memory": 8192,

        #     }],

        #     "computeResourceQuotaLimits": [{

        #         "cpuArchType": "ISA_X86_64",

        #         "numCPU": "8",

        #         "memory": 16384,

        #     }],

        #     "flavoursSupported": [{

        #         "flavourId":

        #         "medium-x86",

        #         "cpuArchType":

        #         "ISA_X86_64",

        #         "supportedOSTypes": [{

        #             "architecture": "x86_64",

        #             "distribution": "UBUNTU",

        #             "version": "OS_VERSION_UBUNTU_2204_LTS",

        #             "license": "OS_LICENSE_TYPE_FREE",

        #         }],

        #         "numCPU":

        #         4,

        #         "memorySize":

        #         8192,

        #         "storageSize":

        #         100,

        #     }],

        #     #

        # }

        aeros_client = ContinuumClient(self.base_url)

        ngsild_params = (

            f'format=simplified&type=InfrastructureElement&q=domain=="{zone_id}"'

        )

        self.logger.debug(

            "Querying infrastructure elements for zone %s with params: %s",

            zone_id,

            ngsild_params,

        )

        # Query the infrastructure elements for the specified zonese

        aeros_domain_ies = aeros_client.query_entities(ngsild_params)

        # Transform the infrastructure elements into the required format

        # and return the details of the edge cloud zone

        response = self.transform_infrastructure_elements(

            domain_ies=aeros_domain_ies, domain=zone_id

        )

        self.logger.debug("Transformed response: %s", response)

        # Return the transformed response

        return response

    def transform_infrastructure_elements(

        self, domain_ies: List[Dict[str, Any]], domain: str

    ) -> Dict[str, Any]:

        """

        Transform the infrastructure elements into a format suitable for the

        edge cloud zone details.

        :param domain_ies: List of infrastructure elements

        :param domain: The ID of the edge cloud zone

        :return: Transformed details of the edge cloud zone

        """

        total_cpu = 0

        total_ram = 0

        total_disk = 0

        total_available_ram = 0

        total_available_disk = 0

        flavours_supported = []

        for element in domain_ies:

            total_cpu += element.get("cpuCores", 0)

            total_ram += element.get("ramCapacity", 0)

            total_available_ram += element.get("availableRam", 0)

            total_disk += element.get("diskCapacity", 0)

            total_available_disk += element.get("availableDisk", 0)

            # Create a flavour per machine

            flavour = {

                "flavourId": f"{element.get('hostname')}-{element.get('containerTechnology')}",

                "cpuArchType": f"{element.get('cpuArchitecture')}",

                "supportedOSTypes": [

                    {

                        "architecture": f"{element.get('cpuArchitecture')}",

                        "distribution": f"{element.get('operatingSystem')}",  # assume

                        "version": "OS_VERSION_UBUNTU_2204_LTS",

                        "license": "OS_LICENSE_TYPE_FREE",

                    }

                ],

                "numCPU": element.get("cpuCores", 0),

                "memorySize": element.get("ramCapacity", 0),

                "storageSize": element.get("diskCapacity", 0),

            }

            flavours_supported.append(flavour)

        result = {

            "zoneId": domain,

            "reservedComputeResources": [

                {

                    "cpuArchType": "ISA_X86_64",

                    "numCPU": str(total_cpu),

                    "memory": total_ram,

                }

            ],

            "computeResourceQuotaLimits": [

                {

                    "cpuArchType": "ISA_X86_64",

                    "numCPU": str(total_cpu * 2),  # Assume quota is 2x total?

                    "memory": total_ram * 2,

                }

            ],

            "flavoursSupported": flavours_supported,

        }

        return result