improving the responses to match camara style (39b5cafd) · Commits · OOP / code / transformation-function-sdk

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

+160 −23

Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@ 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 ...core.common import OranHttpError, requires_capability
		from . import mappings as mappings_module
		from .mappings import flow_id_mapping, policy_mapping, qos_prio_to_oran_prio

		@@ -123,14 +123,39 @@ class OranManager(BaseOranClient):
		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.
		"""
		# Determine device IP candidates (public, then private)
		def create_qod_session(self, session_info: Dict, return_on_error: bool = False) -> Dict:
		"""Translate CAMARA QoD session into ORAN policy and submit it."""
		candidates = self._extract_device_ip_candidates(session_info)
		scope = self._resolve_scope_from_candidates(candidates)
		qos_profile, qos_prio, flow_id = self._normalize_qos_profile(session_info)
		scope_with_flow = {**scope, "flow_id": flow_id}
		expiry = self._parse_expiry(session_info)
		notification_uri = self._extract_notification_uri(session_info)
		policy = self._build_policy(scope_with_flow, qos_prio, expiry, notification_uri)

		resp = self._post_policy(policy, return_on_error)
		# If return_on_error and there was an error, _post_policy returns a camara-like UNAVAILABLE
		if resp.get("qosStatus") == "UNAVAILABLE" and "sessionId" not in resp:
		return resp

		policy_id = (
		(resp or {}).get("policy_id") or (resp or {}).get("policyId") or (resp or {}).get("id")
		)
		if not policy_id:
		raise ValueError("ORAN policy creation did not return an ID")

		return self._build_camara_create_response(
		session_info=session_info,
		policy_id=policy_id,
		qos_profile=qos_profile,
		notification_uri=notification_uri,
		expiry=expiry,
		)

		def _extract_device_ip_candidates(self, session_info: Dict) -> list[str]:
		device = session_info.get("device") or {}
		ipv4 = device.get("ipv4Address") if isinstance(device, dict) else None
		candidates = []
		candidates: list[str] = []
		if isinstance(ipv4, dict):
		pub_ip = ipv4.get("publicAddress")
		prv_ip = ipv4.get("privateAddress")
		@@ -140,19 +165,19 @@ class OranManager(BaseOranClient):
		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")
		return candidates

		def _resolve_scope_from_candidates(self, candidates: list[str]) -> Dict[str, Any]:
		ip_map = self._get_ip_mapping()
		scope = None
		for ip in candidates:
		scope = ip_map.get(ip)
		if scope:
		break
		if not scope:
		return scope
		raise ValueError(f"No PLMN/gNB/UE mapping found for device IPs {', '.join(candidates)}")

		def _normalize_qos_profile(self, session_info: Dict) -> tuple[str, int, int]:
		qos_profile = session_info.get("qosProfile")
		if isinstance(qos_profile, dict):
		qos_profile = qos_profile.get("root") or qos_profile.get("value")
		@@ -161,23 +186,30 @@ class OranManager(BaseOranClient):
		f"Unsupported qosProfile '{qos_profile}'. Allowed: {', '.join(qos_prio_to_oran_prio.keys())}"
		)
		qos_prio = qos_prio_to_oran_prio[qos_profile]
		# Flow ID from profile
		try:
		flow_id = flow_id_mapping[qos_profile]
		except KeyError:
		raise ValidationError(f"No flow_id mapping found for qosProfile '{qos_profile}'")
		scope_with_flow = {**scope, "flow_id": flow_id}
		return qos_profile, qos_prio, flow_id

		def _parse_expiry(self, session_info: Dict) -> int \| None:
		expiry = session_info.get("duration")
		try:
		expiry = int(expiry) if expiry is not None else None
		return int(expiry) if expiry is not None else None
		except Exception:
		expiry = None
		return None

		# Prefer explicit notificationDestination if present; fall back to sink
		notification_uri = session_info.get("notificationDestination") or None
		def _extract_notification_uri(self, session_info: Dict) -> str \| None:
		return session_info.get("notificationDestination") or None

		policy = schemas.OranPolicy(
		def _build_policy(
		self,
		scope_with_flow: Dict[str, Any],
		qos_prio: int,
		expiry: int \| None,
		notification_uri: str \| None,
		) -> schemas.OranPolicy:
		return schemas.OranPolicy(
		policyType=policy_mapping["oran-qod"],
		policyScope=scope_with_flow,
		policyStatement={"qos_prio": qos_prio},
		@@ -185,12 +217,117 @@ class OranManager(BaseOranClient):
		notificationUri=notification_uri,
		)

		def _post_policy(self, policy: schemas.OranPolicy, return_on_error: bool) -> Dict[str, Any]:
		try:
		return oran_common.oran_policy_post(self.base_url, self.scs_as_id, policy)
		except OranHttpError as e:
		if return_on_error:
		status_info = None
		if e.status_code is not None:
		if e.status_code >= 500:
		status_info = "NETWORK_TERMINATED"
		elif e.status_code == 410:
		status_info = "DELETE_REQUESTED"
		return {
		"qosStatus": "UNAVAILABLE",
		"statusInfo": status_info,
		"error": {"statusCode": e.status_code, "body": e.body},
		}
		raise

		def _build_camara_create_response(
		self,
		*,
		session_info: Dict,
		policy_id: Any,
		qos_profile: str,
		notification_uri: str \| None,
		expiry: int \| None,
		) -> Dict[str, Any]:
		return {
		"sessionId": str(policy_id),
		"qosStatus": "REQUESTED",
		"duration": expiry if isinstance(expiry, int) else session_info.get("duration"),
		"device": session_info.get("device"),
		"applicationServer": session_info.get("applicationServer"),
		"devicePorts": session_info.get("devicePorts"),
		"applicationServerPorts": session_info.get("applicationServerPorts"),
		"qosProfile": qos_profile,
		"sink": notification_uri,
		"policy_id": policy_id,
		"policyId": policy_id,
		}

		@requires_capability("oran-qod")
		def get_qod_session(self, session_id: str) -> Dict:
		"""Retrieve an ORAN policy by ID (maps to QoD session get)."""
		return oran_common.oran_policy_get(self.base_url, self.scs_as_id, session_id)
		def get_qod_session(
		self,
		session_id: str,
		original_session: Dict \| None = None,
		fallback_unavailable: bool = False,
		) -> Dict:
		"""Retrieve an ORAN policy by ID and map to a CAMARA-like response.

		Adds qosStatus=AVAILABLE and harmonizes keys while preserving original fields.
		"""
		try:
		resp: Dict[str, Any] = oran_common.oran_policy_get(
		self.base_url, self.scs_as_id, session_id
		)
		except OranHttpError:
		if fallback_unavailable:
		# Return a minimal CAMARA-like UNAVAILABLE response instead of raising
		return {
		"sessionId": str(session_id),
		"qosStatus": "UNAVAILABLE",
		# Backward-compatible identifiers
		"policy_id": session_id,
		"policyId": session_id,
		}
		raise

		# Determine policy/session identifier
		policy_id = (
		(resp or {}).get("policy_id")
		or (resp or {}).get("policyId")
		or (resp or {}).get("id")
		or session_id
		)

		mapped: Dict[str, Any] = dict(resp or {})
		mapped.update(
		{
		"sessionId": str(policy_id),
		"qosStatus": "AVAILABLE",
		}
		)

		# Best-effort mapping of optional fields
		expiry_val = mapped.get("expiry")
		if isinstance(expiry_val, int) and "duration" not in mapped:
		mapped["duration"] = expiry_val

		if mapped.get("sink") is None and mapped.get("notificationUri"):
		mapped["sink"] = mapped.get("notificationUri")

		# Enrich with original requested session fields when available
		if isinstance(original_session, dict):
		for key in (
		"device",
		"applicationServer",
		"devicePorts",
		"applicationServerPorts",
		"qosProfile",
		):
		if mapped.get(key) is None and original_session.get(key) is not None:
		mapped[key] = original_session.get(key)
		if mapped.get("sink") is None and original_session.get("notificationDestination"):
		mapped["sink"] = original_session.get("notificationDestination")

		# Preserve compatibility identifiers
		mapped.setdefault("policy_id", policy_id)
		mapped.setdefault("policyId", policy_id)

		return mapped

		@requires_capability("oran-qod")
		def delete_qod_session(self, session_id: str) -> None:

src/sunrise6g_opensdk/oran/core/common.py

+19 −7

Original line number	Diff line number	Diff line
		@@ -25,13 +25,20 @@ 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
		status = None
		body = None
		try:
		if e.response is not None:
		status = e.response.status_code
		try:
		body = e.response.json()
		except Exception:
		body = e.response.text
		except Exception:
		pass
		raise OranHttpError(str(e), status_code=status, body=body) from e
		except requests.exceptions.ConnectionError as e:
		if response.content:
		print(response.json())
		raise OranHttpError("connection error") from e
		raise OranHttpError("connection error", status_code=None, body=None) from e


		class CapabilityNotSupported(Exception):
		@@ -62,7 +69,12 @@ def requires_capability(feature: str):


		class OranHttpError(Exception):
		pass
		def __init__(
		self, message: str, status_code: int \| None = None, body: dict \| str \| None = None
		):
		super().__init__(message)
		self.status_code = status_code
		self.body = body


		# Subscription Event Methods

Admin message