Add common client invocation. Delete old dedicated factory files

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

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

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

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

def _edgecloud_catalog(client_name: str, base_url: str):

    edge_cloud_factory = {

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

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

        # "piedge": lambda url: PiEdgeClient(base_url=url),

    }

    try:

        return edge_cloud_factory[client_name](base_url)

    except KeyError:

        raise ValueError(

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

        )

def _network_catalog(client_name: str, base_url: str):

    network_factory = {

        "open5gs": lambda url: Open5GSClient(base_url=url),

        "oai": lambda url: OaiCoreClient(base_url=url),

        "open5gcore": lambda url: Open5GCoreClient(base_url=url),

    }

    try:

        return network_factory[client_name](base_url)

    except KeyError:

        raise ValueError(

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

        )

class UniversalClientCatalog:

    _domain_factories = {

        "edgecloud": _edgecloud_catalog,

        "network": _network_catalog,

    }

    @classmethod

    def get_client(cls, domain: str, client_name: str, base_url: str):

        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)

universal_client_catalog = UniversalClientCatalog()

from typing import Dict

from src.common.universal_catalog import universal_client_catalog

class UniversalClientCatalog:

    @staticmethod

    def create_clients(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 (list[dict]): A list of dictionaries, where each dictionary

                                    specifies a client to be created.

                                    Each client dictionary must contain:

                                    - 'domain' (str): The client's domain (e.g., 'edgecloud', 'network').

                                    - 'client_name' (str): The specific name of the client

                                                            (e.g., 'i2edge', 'open5gs').

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

                                    Optional parameters (like 'scs_as_id') can also be included

                                    if required by the specific client's constructor.

        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 UniversalClientCatalog

            >>>

            >>> client_specs_example = [

            >>>     {

            >>>         'domain': 'edgecloud',

            >>>         'client_name': 'i2edge',

            >>>         'base_url': 'http://localhost:8081',

            >>>         'description': 'i2edge client example.'

            >>>     },

            >>>     {

            >>>         'domain': 'network',

            >>>         'client_name': 'open5gs',

            >>>         'base_url': 'http://localhost:8084',

            >>>         'scs_as_id': 'my_unique_scs_id_example' # Example of optional parameter

            >>>     }

            >>> ]

            >>>

            >>> clients = invoke_clients(client_specs_example)

            >>> # Access a client: i2edge_client = clients['i2edge']

        """

        clients = {}

        for domain, config in specs.items():

            client_name = config["client_name"]

            base_url = config["base_url"]

            client = universal_client_catalog.get_client(domain, client_name, base_url)

            clients[domain] = client

        return clients

#!/usr/bin/env python3

# -*- 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)

#   - Sergio Giménez (sergio.gimenez@i2cat.net)

##

from __future__ import annotations

from typing import TYPE_CHECKING

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

if TYPE_CHECKING:

    from .edgecloud_interface import EdgeCloudInterface

class EdgeCloudFactory:

    """

    Factory class for creating EdgeCloud Clients

    """

    @staticmethod

    def create_edgecloud_client(client_name: str, base_url: str) -> EdgeCloudInterface:

        try:

            return EdgeCloudTypes.edgecloud_types[client_name](base_url)

        except KeyError:

            # Get the list of supported client names

            supported_clients = list(EdgeCloudTypes.edgecloud_types.keys())

            raise ValueError(

                f"Invalid edgecloud client name: '{client_name}'. "

                f"Supported clients are: {', '.join(supported_clients)}"

            )

class EdgeCloudTypes:

    """

    Class dedicated for the different types of edgecloud clients.

    """

    I2EDGE = "i2edge"

    AEROS = "aeros"

    PIEDGE = "piedge"

    edgecloud_types = {

        I2EDGE: lambda url: I2EdgeClient(base_url=url),

        AEROS: lambda url: AerosClient(base_url=url),

        PIEDGE: lambda url: PiEdgeClient(base_url=url),

    }

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

from src import logger

from src.edgecloud.core.edgecloud_factory import EdgeCloudFactory

logger.setup_logger(is_debug=True, file_name="sdk.log")

def create_edgecloud_client(client_name: str, base_url: str):

    """

    Create and return an edgecloud client.

    Args:

        client_name (str): Name of the edge cloud platform. Must be one of:

                          'i2edge', 'aeros', 'piedge'

        base_url (str): The base URL for the client.

    Returns:

        The created edgecloud client.

    Example:

        >>> client = create_edgecloud_client('i2edge', 'http://localhost:8080')

    """

    return EdgeCloudFactory.create_edgecloud_client(client_name, base_url)

# ###########################################

# # Temporal code - Testing purposes

# ###########################################

# if __name__ == "__main__":

#     # Define the client name and base URL

#     client_name = "i2edge"

#     base_url = "http://192.168.123.237:30769/"

#     # Create the edgecloud client

#     sbi = EdgeCloudFactory.create_edgecloud_client(client_name, base_url)

#     # Print the edgecloud client being used and its URL

#     print(f"Using edgecloud client: {sbi}")

#     print(f"Edge Cloud Platform: {client_name}")

#     print(f"URL: {sbi.base_url}")

#     print("")

#     # Get all availability zones

#     print("Running test endpoint: get_edge_cloud_zones:")

#     zones = sbi.get_edge_cloud_zones()

#     print(zones)

# ###########################################

# # End of temporal code

# ###########################################

#!/usr/bin/env python3

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

##

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

# All rights reserved.

#

# This file is part of the Open SDK

#

# Contributors:

#   - Reza Mosahebfard (reza.mosahebfard@i2cat.net)

##

from __future__ import annotations

from typing import TYPE_CHECKING

from src.network.clients.oai.client import OaiNefClient

from src.network.clients.open5gcore.client import Open5GCoreClient

from src.network.clients.open5gs.client import Open5GSClient

if TYPE_CHECKING:

    from .network_interface import NetworkManagementInterface

class NetworkClientFactory:

    """

    Factory class for creating Network Management Clients.

    """

    @staticmethod

    def create_network_client(

        client_name: str, base_url: str

    ) -> NetworkManagementInterface:

        """

        Creates and returns an instance of the specified Network Client.

        """

        try:

            constructor = NetworkClientFactory.network_client_constructors[client_name]

            network_client_instance = constructor(base_url)

            return network_client_instance

        except KeyError:

            # Get the list of supported client names

            supported_clients = list(

                NetworkClientFactory.network_client_constructors.keys()

            )

            raise ValueError(

                f"Invalid network client name: '{client_name}'. "

                "Supported clients are: "

                f"{', '.join(supported_clients)}"

            )

class NetworkClientTypes:

    """

    Class for creating Network Clients.

    """

    OPEN5GS = "open5gs"

    OAI = "oai"

    OPEN5GCORE = "open5gcore"

    # --- Dictionary mapping type constants to constructors ---

    network_types = {

        OPEN5GS: lambda url: Open5GSClient(base_url=url),

        OAI: lambda url: OaiNefClient(base_url=url),

        OPEN5GCORE: lambda url: Open5GCoreClient(base_url=url),

    }