added oai location retrieval through monitoring events, ip to external id... (217d0ef0) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/network/adapters/oai/client.py

+394 −2

Original line number	Diff line number	Diff line
		@@ -6,7 +6,13 @@
		# Contributors:
		# - Giulio Carota (giulio.carota@eurecom.fr)
		##
		from datetime import datetime, timedelta, timezone

		import requests

		from sunrise6g_opensdk import logger
		from sunrise6g_opensdk.network.adapters.errors import NetworkPlatformError
		from sunrise6g_opensdk.network.core import common, schemas
		from sunrise6g_opensdk.network.core.base_network_client import BaseNetworkClient
		from sunrise6g_opensdk.network.core.schemas import (
		AsSessionWithQoSSubscription,
		@@ -27,16 +33,19 @@ class NetworkManager(BaseNetworkClient):
		CAMARA APIs into specific HTTP requests understandable by the OAI NEF API.
		"""

		capabilities = {"qod", "traffic_influence"}
		capabilities = {"qod", "traffic_influence", "location_retrieval"}

		def __init__(self, base_url: str, scs_as_id: str = None):
		def __init__(self, base_url: str, scs_as_id: str = None, identity_service_url: str = None):
		try:
		super().__init__()
		self.base_url = base_url
		self.scs_as_id = scs_as_id
		self.identity_service_url = identity_service_url
		log.info(
		f"Initialized OaiNefClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}"
		)
		if self.identity_service_url:
		log.info(f"NEF UE-ID API URL: {self.identity_service_url}")

		except Exception as e:
		log.error(f"Failed to initialize OaiNefClient: {e}")
		@@ -109,6 +118,389 @@ class NetworkManager(BaseNetworkClient):
		):
		raise OaiValidationError("OAI requires UE IPv4 Address to activate Traffic Influence")

		# ---- Location Retrieval (Monitoring Event) ----

		def _resolve_ue_external_id(self, ip_address: str) -> str:
		"""
		Resolve a UE's IP address to the NEF ExternalId via the 3GPP
		UE Identifier API (TS 29.522).

		Calls ``POST {identity_service_url}/retrieve`` on the NEF's
		``/3gpp-ueid/v1`` northbound endpoint with the UE's IPv4 address
		and retrieves the ``externalId`` from the response.

		args:
		ip_address: IPv4 address of the UE (e.g. '12.1.0.1').

		returns:
		The NEF ExternalId string.

		raises:
		OaiValidationError: If the UE-ID API URL is not configured.
		NetworkPlatformError: If the lookup fails.
		"""
		if not self.identity_service_url:
		raise OaiValidationError(
		"UE-ID API URL (identity_service_url) is required for OAI "
		"location retrieval. Set it to the NEF /3gpp-ueid/v1 base URL."
		)
		url = f"{self.identity_service_url.rstrip('/')}/retrieve"
		body = {
		"afId": self.scs_as_id,
		"ueIpAddr": {"ipv4Addr": ip_address},
		"snssai": {"sst": 1, "sd": "FFFFFF"},
		}
		log.debug(f"Resolving UE ExternalId via NEF UE-ID API: POST {url}")
		try:
		resp = requests.post(
		url,
		json=body,
		headers={"accept": "application/json"},
		timeout=10,
		)
		resp.raise_for_status()
		data = resp.json()
		except requests.exceptions.RequestException as e:
		raise NetworkPlatformError(
		f"Failed to resolve UE ExternalId for IP {ip_address}: {e}"
		) from e

		external_id = data.get("externalId")
		if not external_id:
		raise NetworkPlatformError(
		f"NEF UE-ID API did not return externalId for IP {ip_address}: {data}"
		)
		log.info(f"Resolved UE IP {ip_address} → externalId {external_id}")
		return external_id

		def core_specific_monitoring_event_validation(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> None:
		"""
		Validates OAI-specific parameters for location retrieval via NEF monitoring events.

		OAI NEF requires a device identifier that can be resolved to a NEF
		ExternalId. The recommended approach is to provide the device's IPv4 address
		(which the adapter resolves via the NEF UE-ID API at /3gpp-ueid/v1/retrieve),
		or alternatively a pre-resolved ExternalId in the networkAccessIdentifier field.

		args:
		retrieve_location_request: The CAMARA location retrieval request to validate.

		raises:
		OaiValidationError: If the request does not meet OAI-specific requirements.
		"""
		if retrieve_location_request.device is None:
		raise OaiValidationError(
		"OAI requires a device to be specified for location retrieval."
		)
		device = retrieve_location_request.device
		has_identifier = (
		device.networkAccessIdentifier is not None
		or device.ipv4Address is not None
		or device.ipv6Address is not None
		or device.phoneNumber is not None
		)
		if not has_identifier:
		raise OaiValidationError(
		"OAI requires at least one device identifier for location retrieval. "
		"Provide ipv4Address (recommended) or a pre-resolved "
		"networkAccessIdentifier (ExternalId)."
		)

		def add_core_specific_location_parameters(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.MonitoringEventSubscriptionRequest:
		"""
		Build OAI-specific monitoring event subscription request for location retrieval.

		OAI NEF does not accept the locationType field. The subscription is configured
		for a single immediate report.

		args:
		retrieve_location_request: The CAMARA location retrieval request.

		returns:
		MonitoringEventSubscriptionRequest populated with OAI-specific parameters.
		"""
		expire_time = datetime.now(timezone.utc) + timedelta(hours=1)
		return schemas.MonitoringEventSubscriptionRequest(
		# notification destination is harded coded because it is not used currently but is needed as a placeholder, could later be added as a variable
		notificationDestination="http://localhost:8080/callback",
		monitoringType=schemas.MonitoringType.LOCATION_REPORTING,
		maximumNumberOfReports=1,
		monitorExpireTime=expire_time,
		)

		def _build_monitoring_event_subscription(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.MonitoringEventSubscriptionRequest:
		"""
		Override base class to handle OAI-specific identity resolution.

		The OAI NEF monitoring-event service requires an opaque ExternalId.
		If the CAMARA request provides an IPv4 address, this method resolves
		it to the ExternalId via the NEF UE-ID API (/3gpp-ueid/v1/retrieve).
		Alternatively, a pre-resolved ExternalId can be passed directly via the
		networkAccessIdentifier field.

		args:
		retrieve_location_request: The CAMARA location retrieval request.

		returns:
		MonitoringEventSubscriptionRequest ready for the OAI NEF.
		"""
		self.core_specific_monitoring_event_validation(retrieve_location_request)
		subscription = self.add_core_specific_location_parameters(retrieve_location_request)

		device = retrieve_location_request.device

		# Resolve ExternalId: prefer pre-resolved networkAccessIdentifier,
		# otherwise resolve from IPv4 via the NEF UE-ID API
		if device.networkAccessIdentifier is not None:
		subscription.externalId = device.networkAccessIdentifier.root
		log.debug(
		f"Using pre-resolved ExternalId from networkAccessIdentifier: "
		f"{subscription.externalId}"
		)
		elif device.ipv4Address is not None:
		ip_str = str(device.ipv4Address.root.publicAddress.root)
		subscription.externalId = self._resolve_ue_external_id(ip_str)
		else:
		raise OaiValidationError(
		"OAI location retrieval requires either a pre-resolved "
		"networkAccessIdentifier (ExternalId) or an ipv4Address "
		"that can be resolved via the NEF UE-ID API."
		)

		return subscription

		def _parse_location_response(self, response: dict) -> schemas.Location:
		"""
		Parse OAI NEF monitoring event response into CAMARA Location format.

		The OAI NEF may return either:
		- A direct MonitoringEventReport (with locationInfo at top level)
		- A subscription response wrapping monitoringEventReports

		Location data may be:
		- Geographic area (Point with uncertainty → CAMARA Circle, or Polygon)
		- Cell-level only (cellId, nrLocation with NCGI/TAI → CAMARA Circle
		centred at 0,0 with cellId in metadata – best-effort)

		args:
		response: Raw JSON response dict from the OAI NEF.

		returns:
		CAMARA Location object.
		"""
		# Extract the report from the response (may be wrapped or direct)
		reports = response.get("monitoringEventReports", [])
		if reports:
		report_data = reports[0]
		elif "locationInfo" in response or "locationInformation" in response:
		report_data = response
		else:
		log.error(f"No location data in OAI NEF response: {response}")
		raise NetworkPlatformError(
		"No location data found in OAI NEF monitoring event response"
		)

		# Extract location info (handle both key names)
		location_info = report_data.get("locationInfo") or report_data.get(
		"locationInformation"
		)
		if not location_info:
		raise NetworkPlatformError(
		"Location information not found in OAI NEF monitoring event report"
		)

		# Parse event time
		event_time_raw = report_data.get("eventTime") or report_data.get("timeStamp")
		if event_time_raw:
		if isinstance(event_time_raw, str):
		event_time = datetime.fromisoformat(
		event_time_raw.replace("Z", "+00:00")
		)
		else:
		event_time = event_time_raw
		else:
		event_time = datetime.now(timezone.utc)

		# Parse age of location info – the OAI NEF returns it as a plain integer
		# (minutes) inside userLocation.nrLocation.ageOfLocationInformation,
		# or as ageOfLocationInfo at the locationInfo level.
		age_minutes = None
		age_info = location_info.get("ageOfLocationInfo")
		if age_info is not None:
		if isinstance(age_info, dict):
		age_minutes = age_info.get("duration")
		elif isinstance(age_info, (int, float)):
		age_minutes = int(age_info)

		# Also check the nrLocation-level age
		user_loc = location_info.get("userLocation", {})
		nr_loc = user_loc.get("nrLocation", {})
		if age_minutes is None and nr_loc:
		nr_age = nr_loc.get("ageOfLocationInformation")
		if isinstance(nr_age, (int, float)):
		age_minutes = int(nr_age)

		last_location_time = self._compute_camara_last_location_time(
		event_time, age_minutes
		)
		log.debug(f"OAI Last Location time is {last_location_time}")

		# --- Build CAMARA area from available location data ---

		# Option 1: geographicArea is present (Point or Polygon)
		geo_area = location_info.get("geographicArea")
		if geo_area:
		area = self._parse_geographic_area(geo_area)
		return schemas.Location(area=area, lastLocationTime=last_location_time)

		# Option 2: Cell-level location only (cellId / nrLocation)
		# The core simulator (and some real deployments) may only return cell
		# identifiers without geographic coordinates. We return a CAMARA Circle
		# at (0, 0) with a large radius to signal cell-level accuracy. In a
		# production deployment with a real core the geographicArea path above
		# would be taken instead.
		cell_id = location_info.get("cellId")
		if nr_loc or cell_id:
		log.warning(
		"OAI NEF returned cell-level location only (no geographic coordinates). "
		f"cellId={cell_id}, ncgi={nr_loc.get('ncgi')}, tai={nr_loc.get('tai')}. "
		"Returning CAMARA Circle with cell-level accuracy."
		)
		area = schemas.Circle(
		areaType=schemas.AreaType.circle,
		center=schemas.Point(latitude=0.0, longitude=0.0),
		radius=50000.0, # ~50 km – coarse cell-level accuracy
		)
		return schemas.Location(area=area, lastLocationTime=last_location_time)

		raise NetworkPlatformError(
		f"No usable location data in OAI NEF response. "
		f"locationInfo: {location_info}"
		)

		@staticmethod
		def _extract_lat_lon(point_data: dict) -> tuple[float, float]:
		"""
		Extract latitude and longitude from a point dict.

		Supports both 3GPP standard keys (lat/lon) and verbose keys
		(latitude/longitude) for forward-compatibility.
		"""
		lat = float(
		point_data.get("lat", point_data.get("latitude", 0))
		)
		lon = float(
		point_data.get("lon", point_data.get("longitude", 0))
		)
		return lat, lon

		@staticmethod
		def _parse_geographic_area(geo_area: dict):
		"""
		Parse a NEF geographicArea dict into a CAMARA area (Circle or Polygon).

		Handles all standard 3GPP shape types from TS 29.572 (Nlmf_Location):
		- POINT → Circle (radius 1m)
		- POINT_UNCERTAINTY_CIRCLE → Circle
		- POINT_UNCERTAINTY_ELLIPSE → Circle (uses semi-major axis as radius)
		- POINT_ALTITUDE → Circle (radius 1m, altitude ignored)
		- POINT_ALTITUDE_UNCERTAINTY → Circle (uses uncertainty as radius)
		- POLYGON → Polygon
		- ELLIPSOID_ARC → Circle (centre of arc, arc radius as radius)

		Also supports the internal SDK polygon format (polygon.point_list.
		geographical_coords) for backward-compatibility with Open5GS responses.
		"""
		shape = (geo_area.get("shape") or "").upper()

		# --- ELLIPSOID_ARC → CAMARA Circle (best-effort approximation) ---
		# Must be checked before the generic point check, because arcs also
		# contain a "point" key but need different radius calculation.
		if shape == "ELLIPSOID_ARC":
		point_data = geo_area.get("point", {})
		lat, lon = NetworkManager._extract_lat_lon(point_data)
		inner_radius = float(geo_area.get("innerRadius", 0))
		uncertainty_radius = float(geo_area.get("uncertaintyRadius", 0))
		radius = inner_radius + uncertainty_radius
		center = schemas.Point(latitude=lat, longitude=lon)
		return schemas.Circle(
		areaType=schemas.AreaType.circle,
		center=center,
		radius=max(radius, 1.0),
		)

		# --- Point-based shapes → CAMARA Circle ---
		point_shapes = {
		"POINT",
		"POINT_UNCERTAINTY_CIRCLE",
		"POINT_UNCERTAINTY_ELLIPSE",
		"POINT_ALTITUDE",
		"POINT_ALTITUDE_UNCERTAINTY",
		}

		if shape in point_shapes or "point" in geo_area:
		point_data = geo_area.get("point", {})
		lat, lon = NetworkManager._extract_lat_lon(point_data)

		# Determine radius from the best available uncertainty field
		uncertainty = 0.0
		if "uncertainty" in geo_area:
		# POINT_UNCERTAINTY_CIRCLE
		uncertainty = float(geo_area["uncertainty"])
		elif "uncertaintyEllipse" in geo_area:
		# POINT_UNCERTAINTY_ELLIPSE – use the semi-major axis
		ellipse = geo_area["uncertaintyEllipse"]
		uncertainty = float(
		ellipse.get("semiMajor", ellipse.get("uncertainty", 0))
		)
		elif "uncertaintyAltitude" in geo_area:
		# POINT_ALTITUDE_UNCERTAINTY – horizontal uncertainty
		uncertainty = float(geo_area.get("uncertainty", 0))

		center = schemas.Point(latitude=lat, longitude=lon)
		return schemas.Circle(
		areaType=schemas.AreaType.circle,
		center=center,
		radius=max(uncertainty, 1.0),
		)

		# --- Polygon shapes → CAMARA Polygon ---
		if shape == "POLYGON" or "polygon" in geo_area or "pointList" in geo_area:
		# 3GPP standard format: top-level pointList array of {lat, lon}
		coords = geo_area.get("pointList", [])

		if not coords:
		# Internal SDK / Open5GS format: polygon.point_list.geographical_coords
		polygon_data = geo_area.get("polygon", {})
		point_list_data = polygon_data.get("point_list", {})
		coords = point_list_data.get("geographical_coords", [])

		camara_points = [
		schemas.Point(
		latitude=NetworkManager._extract_lat_lon(c)[0],
		longitude=NetworkManager._extract_lat_lon(c)[1],
		)
		for c in coords
		]
		if len(camara_points) < 3:
		raise NetworkPlatformError(
		"Polygon geographic area requires at least 3 points"
		)
		return schemas.Polygon(
		areaType=schemas.AreaType.polygon,
		boundary=schemas.PointList(camara_points),
		)

		raise NetworkPlatformError(
		f"Unsupported geographic area shape in OAI NEF response: {geo_area}"
		)


		def _retrieve_ue_ipv4(session_info: CreateSession):
		return session_info.device.ipv4Address.root.privateAddress

tests/network/aoi_location_retrieval_test.py

0 → 100644

+419 −0

File added.

Preview size limit exceeded, changes collapsed.

Admin message