Import TF-SDK developer network and O-RAN adapters (2026-10-07) (d228aa1a) · Commits · OOP / code / transformation-function-sdk

.gitlab-ci.yml

+1 −0

Changes for .gitlab-ci.yml: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -27,6 +27,7 @@ validate-mr:
		- echo "Running tests..."
		- pytest -v tests/common/test_invoke_edgecloud_clients.py
		- pytest -v tests/common/test_invoke_network_clients.py
		- pytest -v tests/contract/
		only:
		- merge_requests

src/openop_tfsdk/contract/adapters/init.py

+14 −1

Changes for src/openop_tfsdk/contract/adapters/__init__.py: 14 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -11,10 +11,23 @@ from typing import Optional
		from openop_tfsdk.contract.adapters.kubernetes import (
		KubernetesCapabilityAdapter,
		)
		from openop_tfsdk.contract.adapters.network import (
		OaiCapabilityAdapter,
		Open5gcoreCapabilityAdapter,
		Open5gsCapabilityAdapter,
		)
		from openop_tfsdk.contract.adapters.oran import I2catRicCapabilityAdapter
		from openop_tfsdk.contract.errors import UnknownAdapterKind

		ADAPTERS_BY_TYPE = {
		KubernetesCapabilityAdapter.adapter_type: KubernetesCapabilityAdapter,
		adapter.adapter_type: adapter
		for adapter in (
		KubernetesCapabilityAdapter,
		OaiCapabilityAdapter,
		Open5gsCapabilityAdapter,
		Open5gcoreCapabilityAdapter,
		I2catRicCapabilityAdapter,
		)
		}

src/openop_tfsdk/contract/adapters/kubernetes.py

+2 −2

Changes for src/openop_tfsdk/contract/adapters/kubernetes.py: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ What a kubernetes control path can execute. Not topology, so not read from
		the bootstrap YAML.
		"""
		import json
		import logging
		import subprocess
		from typing import Optional

		@@ -20,9 +21,8 @@ from openop_tfsdk.contract.execution_models import (
		RuntimeState,
		)
		from openop_tfsdk.contract.models import SupportedAction
		from openop_tfsdk.logger import setup_logger

		log = setup_logger(__name__)
		log = logging.getLogger(__name__)

		LIFECYCLE_CAPABILITY_KINDS = frozenset({"workload"})

src/openop_tfsdk/contract/adapters/network.py

0 → 100644

+359 −0

Changes for src/openop_tfsdk/contract/adapters/network.py: 359 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		##
		# This file is part of the TF SDK
		##
		"""
		Contract adapters for the network cores of the previous codebase. They translate
		contract calls to the existing NetworkManager clients.
		"""
		import logging
		from typing import Optional

		import pydantic
		import requests

		from openop_tfsdk.contract.errors import (
		DomainExecutionFailed,
		ResourceNotFound,
		ValidationError,
		)
		from openop_tfsdk.contract.execution_models import (
		CapabilityLifecycleRequest,
		CapabilityLifecycleResult,
		CapabilityQueryRequest,
		CapabilityQueryResult,
		CapabilityRuntimeState,
		CapabilityRuntimeStateRequest,
		CircleArea,
		GeoPoint,
		LifecycleState,
		LocationQueryResult,
		PolygonArea,
		RuntimeState,
		)
		from openop_tfsdk.contract.models import SupportedAction
		from openop_tfsdk.network.adapters.oai.client import NetworkManager as OaiClient
		from openop_tfsdk.network.adapters.oai.client import OaiValidationError
		from openop_tfsdk.network.adapters.open5gcore.client import (
		NetworkManager as Open5gcoreClient,
		)
		from openop_tfsdk.network.adapters.open5gs.client import NetworkManager as Open5gsClient
		from openop_tfsdk.network.core import schemas
		from openop_tfsdk.network.core.common import CoreHttpError

		log = logging.getLogger(__name__)

		QOD_ACTIONS = [
		SupportedAction.ACTIVATE,
		SupportedAction.TERMINATE,
		SupportedAction.READ_RUNTIME_STATE,
		]

		LOCATION_ACTIONS = [SupportedAction.QUERY]

		TRAFFIC_INFLUENCE_ACTIONS = [
		SupportedAction.ACTIVATE,
		SupportedAction.UPDATE,
		SupportedAction.TERMINATE,
		SupportedAction.READ_RUNTIME_STATE,
		]

		# capability_kind -> (feature the old client must declare, actions we report)
		_SURFACES = {
		"qod_session": ("qod", QOD_ACTIONS),
		"location_context": ("location_retrieval", LOCATION_ACTIONS),
		"traffic_influence": ("traffic_influence", TRAFFIC_INFLUENCE_ACTIONS),
		}

		_LIFECYCLE_STATES = {
		"REQUESTED": LifecycleState.PROVISIONING,
		"AVAILABLE": LifecycleState.ACTIVE,
		"UNAVAILABLE": LifecycleState.FAILED,
		}

		_RUNTIME_STATES = {
		"AVAILABLE": RuntimeState.ACTIVE,
		"UNAVAILABLE": RuntimeState.FAILED,
		}


		def _surface(client_class, capability_kind: str):
		feature, actions = _SURFACES.get(capability_kind, (None, []))
		if feature in client_class.capabilities:
		return list(actions), []
		return [], []


		def _device(device) -> dict:
		camara = {}
		if device.phone_number:
		camara["phoneNumber"] = device.phone_number
		if device.network_access_id:
		camara["networkAccessIdentifier"] = device.network_access_id
		if device.ipv4:
		camara["ipv4Address"] = {
		"publicAddress": str(device.ipv4),
		"privateAddress": str(device.ipv4),
		}
		if device.ipv6:
		camara["ipv6Address"] = str(device.ipv6)
		return camara


		def _address(network) -> str:
		# The cores append the prefix themselves, so a single host goes without "/32".
		if network.prefixlen == network.max_prefixlen:
		return str(network.network_address)
		return str(network)


		def _qod_session(request: CapabilityLifecycleRequest) -> dict:
		target = request.parameters.target
		parameters = request.parameters.parameters
		if target.device is None or target.application_server is None:
		raise ValidationError("qod_session needs a target device and application server")

		session = {
		"device": _device(target.device),
		"applicationServer": {},
		"qosProfile": parameters.profile_ref,
		"duration": parameters.duration_seconds,
		}
		if target.application_server.ipv4:
		session["applicationServer"]["ipv4Address"] = _address(target.application_server.ipv4)
		if target.application_server.ipv6:
		session["applicationServer"]["ipv6Address"] = _address(target.application_server.ipv6)

		filters = parameters.traffic_filters
		if filters and filters.source and filters.source.port:
		session["devicePorts"] = {"ports": [filters.source.port]}
		if filters and filters.destination and filters.destination.port:
		session["applicationServerPorts"] = {"ports": [filters.destination.port]}
		if parameters.extra.get("sink"):
		session["sink"] = parameters.extra["sink"]
		return session


		def _traffic_influence(
		request: CapabilityLifecycleRequest, notification_url: Optional[str]
		) -> dict:
		target = request.parameters.target
		if (
		target.device is None
		or target.application_reference is None
		or target.application_server is None
		or target.application_server.ipv4 is None
		):
		raise ValidationError(
		"traffic_influence needs a target device, application reference and application server"
		)
		if not notification_url:
		raise ValidationError("traffic_influence needs notification_url in the control path")
		return {
		"appId": target.application_reference.external_app_id,
		# The old client takes the instance id as the application server IPv4.
		"appInstanceId": _address(target.application_server.ipv4),
		"edgeCloudZoneId": request.target.zone_ref,
		"device": _device(target.device),
		"notificationUri": notification_url,
		}


		def _resource_id(link: Optional[str]) -> str:
		# The old client returns the NEF "self" link; the id is its last segment.
		if not link:
		raise DomainExecutionFailed("traffic influence was created without an id")
		return link.rstrip("/").rsplit("/", 1)[-1]


		def _location(location) -> LocationQueryResult:
		area = location.area
		if isinstance(area, schemas.Polygon):
		canonical = PolygonArea(
		boundary=[
		GeoPoint(latitude=p.latitude, longitude=p.longitude) for p in area.boundary.root
		]
		)
		elif area.center.latitude == 0 and area.center.longitude == 0:
		# The OAI client reports a cell-only fix as a circle at (0, 0).
		raise DomainExecutionFailed("no geographic location for the device")
		else:
		canonical = CircleArea(
		center=GeoPoint(latitude=area.center.latitude, longitude=area.center.longitude),
		radius_m=area.radius,
		)
		return LocationQueryResult(last_location_time=location.lastLocationTime.root, area=canonical)


		class NetworkCapabilityAdapter:
		"""Executes network capabilities on a network core through its NetworkManager client."""

		adapter_type: str
		client_class: type
		# Device identifiers the core locates by, in the order the old client tries them.
		location_identifiers: tuple = ()

		def __init__(self, infrastructure: Optional[dict] = None):
		infrastructure = infrastructure or {}
		self._client = self.client_class(
		base_url=infrastructure.get("base_url"), scs_as_id=infrastructure.get("scs_as_id")
		)
		self._notification_url = infrastructure.get("notification_url")

		def activate(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		self._require(request.capability_kind, SupportedAction.ACTIVATE)
		if request.capability_kind == "traffic_influence":
		resource = self._call(
		"create_traffic_influence_resource",
		_traffic_influence(request, self._notification_url),
		)
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		state=LifecycleState.ACTIVE,
		external_id=_resource_id(resource.get("trafficInfluenceID")),
		)
		session = self._call("create_qod_session", _qod_session(request))
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		state=_LIFECYCLE_STATES.get(session["qosStatus"], LifecycleState.PROVISIONING),
		external_id=session["sessionId"],
		)

		def read_runtime_state(self, request: CapabilityRuntimeStateRequest) -> CapabilityRuntimeState:
		self._require(request.capability_kind, SupportedAction.READ_RUNTIME_STATE)
		if not request.external_id:
		raise ValidationError("read_runtime_state needs the external_id returned by activate")
		influence = request.capability_kind == "traffic_influence"
		method = "get_individual_traffic_influence_resource" if influence else "get_qod_session"
		try:
		resource = self._call(method, request.external_id)
		except ResourceNotFound:
		state = RuntimeState.UNKNOWN
		else:
		# A traffic influence subscription carries no status; it exists or it does not.
		state = (
		RuntimeState.ACTIVE
		if influence
		else _RUNTIME_STATES.get(resource["qosStatus"], RuntimeState.UNKNOWN)
		)
		return CapabilityRuntimeState(
		external_id=request.external_id,
		capability_kind=request.capability_kind,
		state=state,
		)

		def terminate(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		self._require(request.capability_kind, SupportedAction.TERMINATE)
		if not request.external_id:
		raise ValidationError("terminate needs the external_id returned by activate")
		if request.capability_kind == "traffic_influence":
		self._call("delete_traffic_influence_resource", request.external_id)
		else:
		self._call("delete_qod_session", request.external_id)
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		state=LifecycleState.DEACTIVATED,
		external_id=request.external_id,
		)

		def update(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		self._require(request.capability_kind, SupportedAction.UPDATE)
		if not request.external_id:
		raise ValidationError("update needs the external_id returned by activate")
		self._call(
		"put_traffic_influence_resource",
		request.external_id,
		_traffic_influence(request, self._notification_url),
		)
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		state=LifecycleState.ACTIVE,
		external_id=request.external_id,
		)

		def query(self, request: CapabilityQueryRequest) -> CapabilityQueryResult:
		if (
		request.capability_kind != "location_context"
		or "location_retrieval" not in self.client_class.capabilities
		):
		raise ValidationError(f"{self.adapter_type} does not query {request.capability_kind}")
		device = request.parameters.target.device
		identifier = next(
		(name for name in self.location_identifiers if device and getattr(device, name)), None
		)
		if identifier is None:
		raise ValidationError(
		f"{self.adapter_type} locates a device by {' or '.join(self.location_identifiers)}"
		)
		parameters = request.parameters.parameters
		try:
		location_request = schemas.RetrievalLocationRequest(
		device=_device(device),
		maxAge=parameters.max_age_seconds,
		maxSurface=parameters.max_surface_sqm,
		)
		except pydantic.ValidationError as error:
		log.error("location_context rejected the request: %s", error)
		raise ValidationError(
		f"location_context: invalid request for {self.adapter_type}"
		) from error
		location = self._call("create_monitoring_event_subscription", location_request)
		return CapabilityQueryResult(
		capability_kind=request.capability_kind,
		result=_location(location),
		srm_resolved_identifier=identifier,
		)

		def _require(self, capability_kind: str, action: SupportedAction):
		if action not in self.capability_surface(capability_kind)[0]:
		raise ValidationError(f"{self.adapter_type} does not {action.value} {capability_kind}")

		def _call(self, method: str, *args):
		try:
		return getattr(self._client, method)(*args)
		except (pydantic.ValidationError, OaiValidationError) as error:
		log.error("%s rejected the request: %s", method, error)
		raise ValidationError(f"{method}: invalid request for {self.adapter_type}") from error
		except CoreHttpError as error:
		cause = error.__cause__
		if isinstance(cause, requests.HTTPError) and cause.response is not None:
		if cause.response.status_code == 404:
		raise ResourceNotFound(f"{method}: not found") from error
		log.error("%s failed: %s", method, error)
		raise DomainExecutionFailed(f"{method} failed") from error
		except Exception as error:
		log.error("%s failed: %s", method, error)
		raise DomainExecutionFailed(f"{method} failed") from error


		class OaiCapabilityAdapter(NetworkCapabilityAdapter):
		adapter_type = "oai"
		client_class = OaiClient
		location_identifiers = ("network_access_id", "ipv4")

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		return _surface(OaiClient, capability_kind)


		class Open5gsCapabilityAdapter(NetworkCapabilityAdapter):
		adapter_type = "open5gs"
		client_class = Open5gsClient
		location_identifiers = ("phone_number",)

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		return _surface(Open5gsClient, capability_kind)


		class Open5gcoreCapabilityAdapter(NetworkCapabilityAdapter):
		adapter_type = "open5gcore"
		client_class = Open5gcoreClient

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		return _surface(Open5gcoreClient, capability_kind)

src/openop_tfsdk/contract/adapters/oran.py

0 → 100644

+34 −0

Changes for src/openop_tfsdk/contract/adapters/oran.py: 34 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		##
		# This file is part of the TF SDK
		##
		"""
		Contract adapter for the i2CAT O-RAN RIC of the previous codebase. Its client has the
		same QoD methods as the network cores, so it reuses their translation.
		"""
		from typing import Optional

		from openop_tfsdk.contract.adapters.network import QOD_ACTIONS, NetworkCapabilityAdapter
		from openop_tfsdk.contract.errors import DomainExecutionFailed, ResourceNotFound
		from openop_tfsdk.oran.adapters.i2cat_ric.client import OranManager
		from openop_tfsdk.oran.core.common import OranHttpError


		class I2catRicCapabilityAdapter(NetworkCapabilityAdapter):
		adapter_type = "i2cat_ric"
		client_class = OranManager

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		if capability_kind == "qod_session" and "oran-qod" in OranManager.capabilities:
		return list(QOD_ACTIONS), []
		return [], []

		def _call(self, method: str, *args):
		try:
		return super()._call(method, *args)
		except DomainExecutionFailed as error:
		cause = error.__cause__
		if isinstance(cause, OranHttpError) and cause.status_code == 404:
		raise ResourceNotFound(f"{method}: not found") from cause
		raise