Commit 93edc8f8 authored by Adrian Pino's avatar Adrian Pino Committed by GitHub
Browse files

Merge pull request #75 from SunriseOpenOperatorPlatform/feature/polish-sdk-invocation 

Feature/polish sdk invocation
parents a0a21190 de161ef9

examples/basic_usage.py

+13 −6
Changes for examples/basic_usage.py: 13 added lines, 6 removed lines.
Original line number Diff line number Diff line 
from src.common.universal_client_catalog import UniversalClientCatalog

from src.common.sdk_catalog_client import SdkCatalogClient



# "src.common.universal_catalog_client" would be equivalent to "sunrise6g_opensdk"





def main():

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

    # The module importing the SDK, loads the config

    client_specs = {

        "edgecloud": {

            "client_name": "i2edge",
@@ -11,20 +13,25 @@ def main(): 
        # ,

        # "network": {

        #     "client_name": "open5gs",

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

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

        #     "scs_as_id": "id_example"

        # }

    }



    clients = UniversalClientCatalog.create_clients(client_specs)

    clients = SdkCatalogClient.create_clients(client_specs)



    edgecloud_client = clients.get("edgecloud")

    # network_client = clients.get("network")



    # Example usage

    print("Running test endpoint: get_edge_cloud_zones:")

    # Example of edgecloud client in action

    print("Testing edgecloud client function: get_edge_cloud_zones:")

    zones = edgecloud_client.get_edge_cloud_zones()

    print(zones)



    # # Example of network client in action

    # print("Testing network client function: EXAMPLE_FUNCTION:")

    # network_client.get_qod_session(session_id="example_session_id")





if __name__ == "__main__":

    main()

src/common/universal_catalog.py→src/common/sdk_catalog.py

+21 −10
Changes for src/common/sdk_catalog.py: 21 added lines, 10 removed lines.
Original line number Diff line number Diff line
@@ -11,6 +11,7 @@ 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),

        # TODO: uncomment when missing PiEdge's imports are added

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

    }

    try:
@@ -21,35 +22,45 @@ def _edgecloud_catalog(client_name: str, base_url: str): 
        )





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

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

    network_factory = {

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

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

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

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

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

        "open5gcore": lambda url, scs_id: Open5GCoreClient(

            base_url=url, scs_as_id=scs_id

        ),

    }

    try:

        return network_factory[client_name](base_url)

        return network_factory[client_name](base_url, scs_as_id)

    except KeyError:

        raise ValueError(

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

        )





class UniversalClientCatalog:

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

#     # TODO





class SdkClientCatalog:

    _domain_factories = {

        "edgecloud": _edgecloud_catalog,

        "network": _network_catalog,

        # "oran": _oran_catalog,

    }



    @classmethod

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

    def get_client(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)





universal_client_catalog = UniversalClientCatalog()
 
        if domain == "network":

            if "scs_as_id" not in kwargs:

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

            return catalog(client_name, base_url, kwargs["scs_as_id"])

        else:

            return catalog(client_name, base_url)

src/common/sdk_catalog_client.py

0 → 100644
+64 −0
Changes for src/common/sdk_catalog_client.py: 64 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from typing import Dict



from src.common.sdk_catalog import SdkClientCatalog





class SdkCatalogClient:

    @staticmethod

    def create_clients(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")

        """

        universal_client_catalog = SdkClientCatalog()

        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 = universal_client_catalog.get_client(

                domain, client_name, base_url, **kwargs

            )

            clients[domain] = client



        return clients

src/common/universal_client_catalog.py

deleted100644 → 0
+0 −55
Changes for src/common/universal_client_catalog.py: 0 added lines, 55 removed lines.
Original line number Diff line number Diff line 
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

src/network/clients/open5gs/client.py

+1 −2
Changes for src/network/clients/open5gs/client.py: 1 added line, 2 removed lines.
Original line number Diff line number Diff line
@@ -25,8 +25,7 @@ class NetworkManager(NetworkManagementInterface): 
             3GPP Monitoring Event API exposed Open5GS NEF.

    """



    # TODO: Warning! "scs_as_is" is defaulted to None.

    def __init__(self, base_url: str, scs_as_id: str = None):

    def __init__(self, base_url: str, scs_as_id):

        """

        Initializes the Open5GS Client.

        """