Merge pull request #74 from SunriseOpenOperatorPlatform/feature/universal-client-invocation

from src.common.universal_client_catalog import UniversalClientCatalog

def main():

    # Hardcoded client configuration; here we would expect to load the config

    client_specs = {

        "edgecloud": {

            "client_name": "i2edge",

            "base_url": "http://192.168.123.237:30769/",

        }

        # ,

        # "network": {

        #     "client_name": "open5gs",

        #     "base_url": "http://IP:PORT"

        # }

    }

    clients = UniversalClientCatalog.create_clients(client_specs)

    edgecloud_client = clients.get("edgecloud")

    # network_client = clients.get("network")

    # Example usage

    print("Running test endpoint: get_edge_cloud_zones:")

    zones = edgecloud_client.get_edge_cloud_zones()

    print(zones)

if __name__ == "__main__":

    main()

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