testing qod code (2c9931f0) · Commits · OOP / code / transformation-function-sdk

examples/example.py

+11 −13

Original line number	Diff line number	Diff line
		@@ -5,23 +5,21 @@ from sunrise6g_opensdk.common.sdk import Sdk as sdkclient # For developers
		def main():
		# The module that imports the SDK package, must specify which adapters will be used:
		adapter_specs = {
		"edgecloud": {
		"client_name": "kubernetes",
		"base_url": "http://IP:PORT",
		},
		"network": {
		"client_name": "open5gs",
		"base_url": "http://IP:PORT",
		"scs_as_id": "id_example",
		},
		"oran": {
		"client_name": "i2cat_ric",
		"base_url": "http://127.0.0.1:30000",
		"scs_as_id": "scs-test",
		}
		}

		adapters = sdkclient.create_adapters_from(adapter_specs)
		edgecloud_client = adapters.get("edgecloud")
		network_client = adapters.get("network")
		# edgecloud_client = adapters.get("edgecloud")
		# network_client = adapters.get("network")
		oran_client = adapters.get("oran")

		print("EdgeCloud client ready to be used:", edgecloud_client)
		print("Network client ready to be used:", network_client)
		# print("EdgeCloud client ready to be used:", edgecloud_client)
		# print("Network client ready to be used:", network_client)
		print("Oran client ready to be used:", oran_client)

		# Examples:
		# EdgeCloud

src/sunrise6g_opensdk/oran/adapters/i2cat_ric/client.py

+45 −20

Original line number	Diff line number	Diff line
		@@ -10,8 +10,6 @@ import json
		from pathlib import Path
		from typing import Any, Dict

		import mappings as mappings_module
		from mappings import flow_id_mapping, policy_mapping, qos_prio_to_oran_prio
		from pydantic import ValidationError

		from sunrise6g_opensdk import logger
		@@ -20,6 +18,8 @@ from sunrise6g_opensdk.oran.core.base_oran_client import BaseOranClient
		from ...core import common as oran_common
		from ...core import schemas
		from ...core.common import requires_capability
		from . import mappings as mappings_module
		from .mappings import flow_id_mapping, policy_mapping, qos_prio_to_oran_prio

		log = logger.get_logger(__name__)

		@@ -40,8 +40,7 @@ class OranManager(BaseOranClient):
		# Set required attributes without invoking BaseOranClient.__init__
		self.base_url = base_url
		self.scs_as_id = scs_as_id
		# Start periodic refresh of file-based mappings (every 60s, run immediately)
		self.start_periodic_refresh(interval_seconds=10, run_immediately=True)
		# Read mappings on-demand; no background thread
		log.info(
		f"Initialized OranNEFClient with base_url: {self.base_url} "
		f"and scs_as_id: {self.scs_as_id}"
		@@ -83,8 +82,15 @@ class OranManager(BaseOranClient):
		mnc = entry.get("mnc")
		gnb_length = entry.get("gnb_length")
		gnb_id = entry.get("gnb_id")

		if mcc is None or mnc is None or gnb_length is None or gnb_id is None:
		ran_ue_id = entry.get("ran_ue_id")

		if (
		mcc is None
		or mnc is None
		or gnb_length is None
		or gnb_id is None
		or ran_ue_id is None
		):
		raise ValueError("missing required keys")

		# Coerce types
		@@ -92,41 +98,60 @@ class OranManager(BaseOranClient):
		mnc_str = str(mnc)
		gnb_length_int = int(gnb_length)
		gnb_id_int = int(gnb_id)
		ran_ue_id_str = str(ran_ue_id)

		parsed[ip] = {
		"mcc": mcc_str,
		"mnc": mnc_str,
		"gnb_length": gnb_length_int,
		"gnb_id": gnb_id_int,
		"ran_ue_id": ran_ue_id_str,
		}
		except Exception:
		# Skip invalid entries
		continue
		return parsed

		def refresh_dynamic_mappings(self) -> None:
		"""Read IP→PLMN/gNB mapping from local JSON/YAML and apply atomically."""
		def _get_ip_mapping(self) -> Dict[str, Dict[str, Any]]:
		"""Return latest IP->PLMN/gNB mapping, reading file each time if present.

		Falls back to in-module defaults when the external file is missing or invalid.
		"""
		new_map = self._load_ip_mapping_from_file()
		if new_map:
		mappings_module.ip_to_plmn_gnb_mapping = new_map
		log.debug(f"Loaded {len(new_map)} IP mapping entries from file into runtime mapping")
		else:
		# If file missing or invalid, keep existing in-memory map
		pass
		return new_map
		return mappings_module.ip_to_plmn_gnb_mapping

		@requires_capability("oran-qod")
		def create_qod_session(self, session_info: Dict) -> Dict:
		"""
		Translate a CAMARA QoD session dict into an ORAN policy and submit it.
		"""
		ue = session_info.get("device") or {}
		server_ip = ue.get("ipv4Address")
		if not server_ip:
		raise ValueError("device.ipv4Address must be provided")

		scope = mappings_module.ip_to_plmn_gnb_mapping.get(server_ip)
		# Determine device IP candidates (public, then private)
		device = session_info.get("device") or {}
		ipv4 = device.get("ipv4Address") if isinstance(device, dict) else None
		candidates = []
		if isinstance(ipv4, dict):
		pub_ip = ipv4.get("publicAddress")
		prv_ip = ipv4.get("privateAddress")
		if isinstance(pub_ip, str) and pub_ip:
		candidates.append(pub_ip)
		if isinstance(prv_ip, str) and prv_ip and prv_ip not in candidates:
		candidates.append(prv_ip)
		elif isinstance(ipv4, str) and ipv4:
		candidates.append(ipv4)

		if not candidates:
		raise ValueError("device.ipv4Address (public/private) must be provided")

		ip_map = self._get_ip_mapping()
		scope = None
		for ip in candidates:
		scope = ip_map.get(ip)
		if scope:
		break
		if not scope:
		raise ValueError(f"No PLMN/gNB mapping found for server IP {server_ip}")
		raise ValueError(f"No PLMN/gNB/UE mapping found for device IPs {', '.join(candidates)}")

		qos_profile = session_info.get("qosProfile")
		if isinstance(qos_profile, dict):

src/sunrise6g_opensdk/oran/adapters/i2cat_ric/ip_to_plmn_gnb_mapping.json

+2 −2

Original line number	Diff line number	Diff line
		{
		"192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345 },
		"10.0.0.5": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890 }
		"192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id":"000008"},
		"10.45.0.10": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id":"000003"}
		}

src/sunrise6g_opensdk/oran/adapters/i2cat_ric/mappings.py

+21 −6

Original line number	Diff line number	Diff line
		@@ -7,22 +7,37 @@
		# - Miguel Catalan Cid (miguel.catalan@i2cat.net)
		##

		# mapping from QoD to O-RAN policy
		qos_prio_to_oran_prio = {
		"qos-e": "qos-e",
		"qos-s": "qos-s",
		"qos-m": "qos-m",
		"qos-l": "qos-l",
		} # mapping from QoD to O-RAN policy
		}
		# mapping from QoD to 5qi
		flow_id_mapping = {"qos-e": 3, "qos-s": 4, "qos-m": 5, "qos-l": 6}
		policy_mapping = {
		"oran-qod": "ue_policy_priority"
		} # maps a CAMARA policy request to the information type in the oran NEF


		# Maps an IP address to PLMN and gNB identifiers
		# Keys: IP (str)
		# Values: {"mcc": str, "mnc": str, "gnb_length": int, "gnb_id": int}
		ip_to_plmn_gnb_mapping = {
		# Example entries:
		"192.168.1.10": {"mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345},
		"10.10.45.1": {"mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890},
		"192.168.1.10": {
		"mcc": "001",
		"mnc": "01",
		"gnb_length": 28,
		"gnb_id": 12345,
		"ran_ue_id": "000001",
		},
		"10.10.45.1": {
		"mcc": "214",
		"mnc": "07",
		"gnb_length": 28,
		"gnb_id": 67890,
		"ran_ue_id": "000002",
		},
		}

		# maps a CAMARA policy request to the information type in the oran NEF
		policy_mapping = {"oran-qod": "qod_prb_prio"}

src/sunrise6g_opensdk/oran/core/common.py

+6 −1

Original line number	Diff line number	Diff line
		@@ -25,8 +25,12 @@ def _make_request(method: str, url: str, data=None):
		if response.content:
		return response.json()
		except requests.exceptions.HTTPError as e:
		if response.content:
		print(response.json())
		raise OranHttpError(e) from e
		except requests.exceptions.ConnectionError as e:
		if response.content:
		print(response.json())
		raise OranHttpError("connection error") from e


		@@ -98,4 +102,5 @@ def oran_policy_build_url(base_url: str, scs_as_id: str, session_id: str = None)
		if session_id is not None and len(session_id) > 0:
		return f"{url}/{session_id}"
		else:
		return url
		# Collection URL should end with a trailing slash to match FastAPI route
		return f"{url}/"

Admin message