added oai location retrieval through monitoring events, ip to external id... (!196) · Merge requests · OOP / code / transformation-function-sdk

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

+374 −1

Changes for src/sunrise6g_opensdk/network/adapters/oai/client.py: 374 added lines, 1 removed line.
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 schemas
		from sunrise6g_opensdk.network.core.base_network_client import BaseNetworkClient
		from sunrise6g_opensdk.network.core.schemas import (
		AsSessionWithQoSSubscription,
		@@ -27,16 +33,18 @@ 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):
		try:
		super().__init__()
		self.base_url = base_url
		self.scs_as_id = scs_as_id
		self.identity_service_url = f"{base_url.rstrip('/')}/3gpp-ueid/v1"
		log.info(
		f"Initialized OaiNefClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}"
		)
		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 +117,371 @@ 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 only supports networkAccessIdentifier (pre-resolved ExternalId) or
		ipv4Address (resolved via the NEF UE-ID API at /3gpp-ueid/v1/retrieve).
		phoneNumber and ipv6Address are not supported by the OAI southbound and will
		be rejected early here rather than failing later in the build step.

		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_supported_identifier = (
		device.networkAccessIdentifier is not None or device.ipv4Address is not None
		)
		if not has_supported_identifier:
		raise OaiValidationError(
		"OAI requires a supported 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

src/sunrise6g_opensdk/network/core/base_network_client.py

+41 −16

Changes for src/sunrise6g_opensdk/network/core/base_network_client.py: 41 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -237,11 +237,18 @@ class BaseNetworkClient:
		self.core_specific_monitoring_event_validation(retrieve_location_request)
		subscription_3gpp = self.add_core_specific_location_parameters(retrieve_location_request)
		device = retrieve_location_request.device
		subscription_3gpp.externalId = device.networkAccessIdentifier
		subscription_3gpp.ipv4Addr = device.ipv4Address
		subscription_3gpp.ipv6Addr = device.ipv6Address
		# subscription.msisdn = device.phoneNumber.root.lstrip('+')
		# subscription.notificationDestination = "http://127.0.0.1:8001"

		# Extract plain values from CAMARA RootModel types for 3GPP subscription fields
		subscription_3gpp.externalId = (
		device.networkAccessIdentifier.root if device.networkAccessIdentifier else None
		)
		subscription_3gpp.msisdn = (
		device.phoneNumber.root.lstrip("+") if device.phoneNumber else None
		)
		subscription_3gpp.ipv4Addr = (
		device.ipv4Address.root.publicAddress.root if device.ipv4Address else None
		)
		subscription_3gpp.ipv6Addr = device.ipv6Address.root if device.ipv6Address else None

		return subscription_3gpp

		@@ -265,23 +272,21 @@ class BaseNetworkClient:
		else:
		return event_time.replace(tzinfo=timezone.utc)

		@requires_capability("location_retrieval")
		def create_monitoring_event_subscription(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.Location:
		def _parse_location_response(self, response: dict) -> schemas.Location:
		"""
		Creates a Monitoring Event subscription based on CAMARA Location API input.
		Parse NEF monitoring event response into CAMARA Location format.

		Default implementation handles responses where the raw response is a direct
		MonitoringEventReport with polygon-based geographic areas (e.g. Open5GS).

		Override this method for NEF implementations with different response formats.

		args:
		retrieve_location_request: Dictionary containing location retrieval details conforming to
		the CAMARA Location API parameters.
		response: Raw JSON response dict from the NEF monitoring event POST.

		returns:
		dictionary containing the created subscription details, including its ID.
		CAMARA Location object.
		"""
		subscription = self._build_monitoring_event_subscription(retrieve_location_request)
		response = common.monitoring_event_post(self.base_url, self.scs_as_id, subscription)

		monitoring_event_report = schemas.MonitoringEventReport(**response)
		if monitoring_event_report.locationInfo is None:
		log.error("Failed to retrieve location information from monitoring event report")
		@@ -307,6 +312,26 @@ class BaseNetworkClient:

		return camara_location

		@requires_capability("location_retrieval")
		def create_monitoring_event_subscription(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.Location:
		"""
		Creates a one-shot Monitoring Event subscription to retrieve the current
		location of a device, based on CAMARA Location API input.

		args:
		retrieve_location_request: RetrievalLocationRequest containing location
		retrieval details conforming to the CAMARA
		Location API parameters.

		returns:
		CAMARA Location object containing the device location.
		"""
		subscription = self._build_monitoring_event_subscription(retrieve_location_request)
		response = common.monitoring_event_post(self.base_url, self.scs_as_id, subscription)
		return self._parse_location_response(response)

		@requires_capability("qod")
		def create_qod_session(self, session_info: Dict) -> Dict:
		"""

tests/network/aoi_location_retrieval_test.py

0 → 100644

+422 −0

Changes for tests/network/aoi_location_retrieval_test.py: 422 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		##
		# Tests for OAI NEF location retrieval via monitoring event subscriptions.
		#

		##

		import os

		import pytest

		from sunrise6g_opensdk.common.sdk import Sdk as sdkclient
		from sunrise6g_opensdk.network.core.base_network_client import BaseNetworkClient
		from sunrise6g_opensdk.network.core.common import CoreHttpError
		from sunrise6g_opensdk.network.core.schemas import (
		Device,
		DeviceIpv4Addr1,
		MonitoringEventSubscriptionRequest,
		MonitoringType,
		RetrievalLocationRequest,
		)

		# --- Test config (no identity_service_url / NEF UE-ID API for unit tests) ---
		client_specs_unit = {
		"network": {
		"client_name": "oai",
		"base_url": "http://localhost:80",
		"scs_as_id": "camara-test",
		}
		}
		clients_unit = sdkclient.create_adapters_from(client_specs_unit)
		network_client_unit: BaseNetworkClient = clients_unit.get("network")


		# --- CAMARA input payloads (using registered UEs) ---

		# Identify by pre-resolved ExternalId (hash from identity service)
		camara_payload_by_nai = RetrievalLocationRequest(
		device=Device(networkAccessIdentifier="imsi-001010000000001")
		)

		# Identify by IPv4 address (requires identity_service_url pointing at NEF UE-ID API)
		camara_payload_by_ip = RetrievalLocationRequest(
		device=Device(
		ipv4Address=DeviceIpv4Addr1(
		publicAddress="12.1.0.1",
		privateAddress="12.1.0.1",
		)
		)
		)

		# Identify by phone number (valid E.164 MSISDN format)
		camara_payload_by_phone = RetrievalLocationRequest(device=Device(phoneNumber="+10010100002"))


		# ============================================================
		# UNIT TESTS – CAMARA → 3GPP transformation (no network)
		# ============================================================


		class TestCamaraToOai3gppTransformation:
		"""Verify CAMARA → 3GPP monitoring event subscription transformation."""

		def test_transform_with_network_access_identifier(self):
		"""networkAccessIdentifier should map to externalId."""
		result = network_client_unit._build_monitoring_event_subscription(
		retrieve_location_request=camara_payload_by_nai
		)
		assert isinstance(result, MonitoringEventSubscriptionRequest)
		assert result.externalId == "imsi-001010000000001"
		assert result.monitoringType == MonitoringType.LOCATION_REPORTING
		assert result.maximumNumberOfReports == 1
		# OAI adapter must NOT set locationType (NEF rejects it)
		assert result.locationType is None
		# msisdn should not be set (OAI uses externalId)
		assert result.msisdn is None

		def test_serialized_json_has_correct_external_id(self):
		"""The serialized JSON sent to NEF must have externalId as a plain string."""
		result = network_client_unit._build_monitoring_event_subscription(
		retrieve_location_request=camara_payload_by_nai
		)
		json_str = result.model_dump_json(exclude_none=True, by_alias=True)
		assert '"externalId":"imsi-001010000000001"' in json_str.replace(" ", "")
		# locationType must NOT appear in the JSON
		assert "locationType" not in json_str

		def test_ip_without_identity_service_raises(self, monkeypatch):
		"""Using ipv4Address without identity_service_url (NEF UE-ID API) must raise an error."""
		from sunrise6g_opensdk.network.adapters.oai.client import OaiValidationError

		monkeypatch.setattr(network_client_unit, "identity_service_url", None)
		with pytest.raises(OaiValidationError, match="identity_service_url"):
		network_client_unit._build_monitoring_event_subscription(
		retrieve_location_request=camara_payload_by_ip
		)

		def test_validation_rejects_empty_device(self):
		"""OAI validation must reject a request with no device."""
		from sunrise6g_opensdk.network.adapters.oai.client import OaiValidationError

		bad_request = RetrievalLocationRequest(device=None)
		with pytest.raises(OaiValidationError):
		network_client_unit._build_monitoring_event_subscription(
		retrieve_location_request=bad_request
		)

		def test_validation_rejects_device_without_identifiers(self):
		"""OAI validation must reject a device with no identifiers at all."""
		from sunrise6g_opensdk.network.adapters.oai.client import OaiValidationError

		bad_request = RetrievalLocationRequest(device=Device())
		with pytest.raises(OaiValidationError):
		network_client_unit._build_monitoring_event_subscription(
		retrieve_location_request=bad_request
		)


		# ============================================================
		# UNIT TESTS – OAI _parse_location_response with real coordinates
		# ============================================================


		class TestParseLocationResponseRealCoordinates:
		"""Verify that _parse_location_response handles 3GPP geographic area formats."""

		def test_point_uncertainty_circle_3gpp_keys(self):
		"""Standard PointUncertaintyCircle with 3GPP lat/lon keys."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POINT_UNCERTAINTY_CIRCLE",
		"point": {"lat": 48.8566, "lon": 2.3522},
		"uncertainty": 150.0,
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.areaType.value == "CIRCLE"
		assert location.area.center.latitude == 48.8566
		assert location.area.center.longitude == 2.3522
		assert location.area.radius == 150.0

		def test_point_uncertainty_circle_verbose_keys(self):
		"""PointUncertaintyCircle with latitude/longitude keys (backward compat)."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POINT_UNCERTAINTY_CIRCLE",
		"point": {"latitude": 40.4168, "longitude": -3.7038},
		"uncertainty": 200.0,
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.center.latitude == 40.4168
		assert location.area.center.longitude == -3.7038
		assert location.area.radius == 200.0

		def test_plain_point_shape(self):
		"""POINT shape (no uncertainty) → Circle with radius 1m."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POINT",
		"point": {"lat": 52.5200, "lon": 13.4050},
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.center.latitude == 52.5200
		assert location.area.center.longitude == 13.4050
		assert location.area.radius == 1.0 # default minimum

		def test_point_uncertainty_ellipse(self):
		"""POINT_UNCERTAINTY_ELLIPSE → Circle using semiMajor axis."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POINT_UNCERTAINTY_ELLIPSE",
		"point": {"lat": 37.7749, "lon": -122.4194},
		"uncertaintyEllipse": {
		"semiMajor": 300.0,
		"semiMinor": 100.0,
		"orientationMajor": 45,
		},
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.center.latitude == 37.7749
		assert location.area.center.longitude == -122.4194
		assert location.area.radius == 300.0

		def test_point_altitude(self):
		"""POINT_ALTITUDE → Circle with radius 1m (altitude ignored)."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POINT_ALTITUDE",
		"point": {"lat": 35.6762, "lon": 139.6503},
		"altitude": 40.0,
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.center.latitude == 35.6762
		assert location.area.center.longitude == 139.6503
		assert location.area.radius == 1.0

		def test_ellipsoid_arc(self):
		"""ELLIPSOID_ARC → Circle approximation using inner+uncertainty radius."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "ELLIPSOID_ARC",
		"point": {"lat": 51.5074, "lon": -0.1278},
		"innerRadius": 500,
		"uncertaintyRadius": 100,
		"offsetAngle": 0,
		"includedAngle": 120,
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.center.latitude == 51.5074
		assert location.area.center.longitude == -0.1278
		assert location.area.radius == 600.0

		def test_3gpp_standard_polygon(self):
		"""3GPP standard POLYGON with top-level pointList array."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POLYGON",
		"pointList": [
		{"lat": 48.85, "lon": 2.35},
		{"lat": 48.86, "lon": 2.36},
		{"lat": 48.87, "lon": 2.34},
		],
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.areaType.value == "POLYGON"
		boundary = location.area.boundary.root
		assert len(boundary) == 3
		assert boundary[0].latitude == 48.85
		assert boundary[0].longitude == 2.35
		assert boundary[2].latitude == 48.87

		def test_internal_sdk_polygon_format(self):
		"""Internal SDK polygon format (polygon.point_list.geographical_coords)."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"polygon": {
		"point_list": {
		"geographical_coords": [
		{"lat": 10.0, "lon": 20.0},
		{"lat": 11.0, "lon": 21.0},
		{"lat": 12.0, "lon": 22.0},
		]
		}
		}
		}
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.areaType.value == "POLYGON"
		boundary = location.area.boundary.root
		assert len(boundary) == 3
		assert boundary[0].latitude == 10.0

		def test_cell_level_fallback(self):
		"""Cell-level data without geographicArea → Circle at (0,0)."""
		response = {
		"monitoringEventReports": [
		{
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"cellId": "cell-001",
		"userLocation": {
		"nrLocation": {
		"ncgi": {
		"nrCellId": "000000001",
		"plmnId": {"mcc": "001", "mnc": "01"},
		},
		"tai": {"tac": "0001", "plmnId": {"mcc": "001", "mnc": "01"}},
		}
		},
		},
		}
		]
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.areaType.value == "CIRCLE"
		assert location.area.center.latitude == 0.0
		assert location.area.center.longitude == 0.0
		assert location.area.radius == 50000.0

		def test_direct_report_format(self):
		"""Direct MonitoringEventReport (not wrapped in monitoringEventReports)."""
		response = {
		"monitoringType": "LOCATION_REPORTING",
		"eventTime": "2026-02-09T10:00:00Z",
		"locationInfo": {
		"geographicArea": {
		"shape": "POINT_UNCERTAINTY_CIRCLE",
		"point": {"lat": 41.3851, "lon": 2.1734},
		"uncertainty": 50.0,
		}
		},
		}
		location = network_client_unit._parse_location_response(response)
		assert location.area.center.latitude == 41.3851
		assert location.area.center.longitude == 2.1734
		assert location.area.radius == 50.0


		# ============================================================
		# INTEGRATION TESTS – against running OAI NEF at localhost:80
		#
		# Prerequisites:
		# - NEF stack running (nef-coresim-compose)
		# - UE Identity Service reachable (provide its URL)
		#
		# Run with:
		# RUN_INTEGRATION_TESTS=1 pytest tests/network/aoi_location_retrieval_test.py -m integration -v -s
		# ============================================================

		client_specs_integ = {
		"network": {
		"client_name": "oai",
		"base_url": "http://localhost:80",
		"scs_as_id": "camara-test",
		}
		}
		clients_integ = sdkclient.create_adapters_from(client_specs_integ)
		network_client_integ: BaseNetworkClient = clients_integ.get("network")


		@pytest.mark.integration
		@pytest.mark.skipif(
		os.getenv("RUN_INTEGRATION_TESTS") != "1",
		reason="Set RUN_INTEGRATION_TESTS=1 to run integration tests against a live NEF",
		)
		class TestOaiNefIntegration:
		"""
		Integration tests against a running OAI NEF at localhost:80.

		These tests require the NEF stack to be up
		(~/oop_nef-coresim-setup/deployment/nef-coresim-compose).
		"""

		def test_create_location_subscription_by_ip(self):
		"""
		Create a monitoring event subscription using UE IPv4 address.
		The adapter resolves IP → ExternalId via the NEF UE-ID API
		(POST /3gpp-ueid/v1/retrieve), then creates the monitoring event
		subscription on the NEF.
		"""
		try:
		location = network_client_integ.create_monitoring_event_subscription(
		retrieve_location_request=camara_payload_by_ip
		)
		assert location is not None
		assert location.area is not None
		assert location.lastLocationTime is not None
		print(f"\nLocation result: {location.model_dump_json(indent=2)}")

		except CoreHttpError as e:
		pytest.fail(f"NEF request failed (is the NEF running at localhost:80?): {e}")

tests/network/test_location_retrieval.py

deleted100644 → 0

+0 −1

Changes for tests/network/test_location_retrieval.py: 0 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		# PLACEHOLDER

.gitlab-ci.yml

+6 −3

Changes for .gitlab-ci.yml: 6 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,12 +3,17 @@ stages:
		- build
		- publish

		default:
		image: python:3.12-slim
		tags:
		- docker
		- vim

		variables:
		PACKAGE_NAME: sunrise6g-opensdk

		validate-mr:
		stage: validate
		image: python:3.12-slim
		before_script:
		- echo "Running merge request validation..."
		- pip install -r requirements.txt
		@@ -27,7 +32,6 @@ validate-mr:

		build-package:
		stage: build
		image: python:3.12-slim
		before_script:
		- pip install build twine
		- pip install -r requirements.txt
		@@ -48,7 +52,6 @@ build-package:

		publish-gitlab:
		stage: publish
		image: python:3.12-slim
		dependencies:
		- build-package
		before_script:

added oai location retrieval through monitoring events, ip to external id...

Merge request reports