Merge pull request #76 from SunriseOpenOperatorPlatform/feature/polish-sdk-invocation (b8727ac9) · Commits · OOP / code / transformation-function-sdk

.github/workflows/ci.yaml

+15 −12

Original line number	Diff line number	Diff line
		@@ -7,21 +7,24 @@ on:
		branches: [ main ]

		jobs:
		# tests:
		# name: Run tests
		# runs-on: ubuntu-latest
		# container:
		# image: python:3.12-slim
		tests:
		name: Run tests
		runs-on: ubuntu-latest
		container:
		image: python:3.12-slim

		# steps:
		# - name: Checkout code
		# uses: actions/checkout@v4
		steps:
		- name: Checkout code
		uses: actions/checkout@v4

		- name: Install dependencies
		run: pip install -r requirements.txt

		# - name: Install dependencies
		# run: pip install -r requirements.txt
		- name: "Run test: validate edge cloud clients instantiation"
		run: pytest -v tests/common/test_invoke_edgecloud_clients.py

		# - name: Run pytest
		# run: pytest tests/edgecloud
		# - name: "Run test: validate network clients instantiation"
		# run: pytest -v tests/common/test_invoke_network_clients.py

		lint:
		name: Run linters

examples/basic_usage.py

+5 −9

Original line number	Diff line number	Diff line
		from src.common.sdk_catalog_client import SdkCatalogClient

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


		def main():
		# The module importing the SDK, loads the config
		@@ -9,8 +7,7 @@ def main():
		"edgecloud": {
		"client_name": "i2edge",
		"base_url": "http://192.168.123.237:30769/",
		}
		# ,
		},
		# "network": {
		# "client_name": "open5gs",
		# "base_url": "http://IP:PORT",
		@@ -18,17 +15,16 @@ def main():
		# }
		}

		clients = SdkCatalogClient.create_clients(client_specs)
		clients = SdkCatalogClient.create_clients_from(client_specs)

		# EdgeCloud
		edgecloud_client = clients.get("edgecloud")
		# network_client = clients.get("network")

		# 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
		# # Network
		# network_client = clients.get("network")
		# print("Testing network client function: EXAMPLE_FUNCTION:")
		# network_client.get_qod_session(session_id="example_session_id")

src/common/sdk_catalog.py

+22 −19

Original line number	Diff line number	Diff line
		@@ -7,31 +7,38 @@ 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):
		def _edgecloud_catalog(client_name: str, base_url: str, **kwargs):
		edge_cloud_factory = {
		"aeros": lambda url: AerosClient(base_url=url),
		"aeros": lambda url, kw: AerosClient(base_url=url, kw),
		"i2edge": lambda url: I2EdgeClient(base_url=url),
		# TODO: uncomment when missing PiEdge's imports are added
		# "piedge": lambda url: PiEdgeClient(base_url=url),
		# "piedge": lambda url: PiEdgeClient(base_url=url), Uncomment when import issues are solved
		}
		try:
		return edge_cloud_factory[client_name](base_url)
		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_catalog(client_name: str, base_url: str, scs_as_id: str):
		def _network_catalog(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: 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
		"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)
		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)}"
		@@ -50,17 +57,13 @@ class SdkClientCatalog:
		}

		@classmethod
		def get_client(cls, domain: str, client_name: str, base_url: str, **kwargs):
		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)}"
		)

		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)
		return catalog(client_name, base_url, **kwargs)

src/common/sdk_catalog_client.py

+5 −3

Original line number	Diff line number	Diff line
		@@ -5,7 +5,9 @@ from src.common.sdk_catalog import SdkClientCatalog

		class SdkCatalogClient:
		@staticmethod
		def create_clients(client_specs: Dict[str, Dict[str, str]]) -> Dict[str, object]:
		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.
		@@ -44,7 +46,7 @@ class SdkCatalogClient:
		>>> edgecloud_client.get_edge_cloud_zones()
		>>> network_client.get_qod_session(session_id="example_session_id")
		"""
		universal_client_catalog = SdkClientCatalog()
		sdk_client_catalog = SdkClientCatalog()
		clients = {}

		for domain, config in client_specs.items():
		@@ -56,7 +58,7 @@ class SdkCatalogClient:
		k: v for k, v in config.items() if k not in ("client_name", "base_url")
		}

		client = universal_client_catalog.get_client(
		client = sdk_client_catalog.instantiate_and_retrieve_clients(
		domain, client_name, base_url, **kwargs
		)
		clients[domain] = client

src/edgecloud/clients/aeros/client.py

+16 −1

Original line number	Diff line number	Diff line
		@@ -19,10 +19,25 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		FIXME: Handle None responses from continuum client
		"""

		def __init__(self, base_url: str):
		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")

Admin message