Merge pull request #68 from SunriseOpenOperatorPlatform/hotfix/network-refactor-issues (d36ee4ca) · Commits · OOP / code / transformation-function-sdk

src/network/clients/open5gs/client.py

+10 −7

Original line number	Diff line number	Diff line
		@@ -2,10 +2,11 @@
		from typing import Dict

		from pydantic import ValidationError

		from src import logger
		from src.network.core.network_interface import NetworkManagementInterface, build_flows
		from ...core import common
		from ...core import schemas

		from ...core import common, schemas

		log = logger.get_logger(__name__)

		@@ -40,17 +41,19 @@ class NetworkManager(NetworkManagementInterface):
		raise e

		def core_specific_validation(self, session_info: schemas.CreateSession):
		if session_info.qosProfile not in flow_id_mapping.keys():
		if session_info.qosProfile.root not in flow_id_mapping.keys():
		raise ValidationError(
		f"Open5Gs only supports these qos-profiles: {', '.join(flow_id_mapping.keys())}"
		)

		def add_core_specific_parameters(
		self, session_info: schemas.AsSessionWithQoSSubscription
		self,
		session_info: schemas.CreateSession,
		subscription: schemas.AsSessionWithQoSSubscription,
		) -> None:
		session_info.supportedFeatures = schemas.SupportedFeatures("003C")
		flow_id = flow_id_mapping[session_info.qosProfile]
		session_info.flowInfo = build_flows(flow_id, session_info)
		subscription.supportedFeatures = schemas.SupportedFeatures("003C")
		flow_id = flow_id_mapping[session_info.qosProfile.root]
		subscription.flowInfo = build_flows(flow_id, session_info)

		# --- Implementation of NetworkManagementInterface methods ---
		def create_qod_session(self, session_info: Dict) -> Dict:

src/network/core/common.py

+1 −1

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		# Common utilities (errors, HTTP helpers) used by the Open5GS client implementation (client.py).

		from pydantic import BaseModel
		import requests
		from pydantic import BaseModel

		from src import logger

src/network/core/network_factory.py

+14 −12

Original line number	Diff line number	Diff line
		@@ -13,9 +13,9 @@ from __future__ import annotations

		from typing import TYPE_CHECKING

		from src.network.clients.oai.client import OaiNefClient
		from src.network.clients.open5gcore.client import Open5GCoreClient
		from src.network.clients.open5gs.client import Open5GSClient
		from src.network.clients.oai.client import NetworkManager as OaiNefClient
		from src.network.clients.open5gcore.client import NetworkManager as Open5GCoreClient
		from src.network.clients.open5gs.client import NetworkManager as Open5GSClient

		if TYPE_CHECKING:
		from .network_interface import NetworkManagementInterface
		@@ -28,20 +28,18 @@ class NetworkClientFactory:

		@staticmethod
		def create_network_client(
		client_name: str, base_url: str
		client_name: str, base_url: str, scs_as_id: str
		) -> NetworkManagementInterface:
		"""
		Creates and returns an instance of the specified Network Client.
		"""
		try:
		constructor = NetworkClientFactory.network_client_constructors[client_name]
		network_client_instance = constructor(base_url)
		constructor = NetworkClientTypes.network_types[client_name]
		network_client_instance = constructor(base_url, scs_as_id)
		return network_client_instance
		except KeyError:
		# Get the list of supported client names
		supported_clients = list(
		NetworkClientFactory.network_client_constructors.keys()
		)
		supported_clients = list(NetworkClientTypes.network_types.keys())
		raise ValueError(
		f"Invalid network client name: '{client_name}'. "
		"Supported clients are: "
		@@ -60,7 +58,11 @@ class NetworkClientTypes:

		# --- Dictionary mapping type constants to constructors ---
		network_types = {
		OPEN5GS: lambda url: Open5GSClient(base_url=url),
		OAI: lambda url: OaiNefClient(base_url=url),
		OPEN5GCORE: lambda url: Open5GCoreClient(base_url=url),
		OPEN5GS: lambda url, scs_as_id: Open5GSClient(
		base_url=url, scs_as_id=scs_as_id
		),
		OAI: lambda url, scs_as_id: OaiNefClient(base_url=url, scs_as_id=scs_as_id),
		OPEN5GCORE: lambda url, scs_as_id: Open5GCoreClient(
		base_url=url, scs_as_id=scs_as_id
		),
		}

src/network/core/network_interface.py

+27 −13

Original line number	Diff line number	Diff line
		@@ -35,7 +35,8 @@ def flatten_port_spec(ports_spec: schemas.PortsSpec \| None) -> list[str]:


		def build_flows(
		flow_id: int, session_info: schemas.CreateSession
		flow_id: int,
		session_info: schemas.CreateSession,
		) -> list[schemas.FlowInfo]:
		device_ports = flatten_port_spec(session_info.devicePorts)
		server_ports = flatten_port_spec(session_info.applicationServerPorts)
		@@ -77,7 +78,11 @@ class NetworkManagementInterface(ABC):
		scs_as_id: str

		@abstractmethod
		def add_core_specific_parameters(x):
		def add_core_specific_parameters(
		self,
		session_info: schemas.CreateSession,
		subscription: schemas.AsSessionWithQoSSubscription,
		):
		"""
		Placeholder for adding core-specific parameters to the subscription.
		This method should be overridden by subclasses to implement specific logic.
		@@ -99,6 +104,23 @@ class NetworkManagementInterface(ABC):
		# This method should be overridden by subclasses if needed
		pass

		def _build_subscription(self, session_info: Dict) -> None:
		valid_session_info = schemas.CreateSession.model_validate(session_info)
		device_ipv4 = None
		if valid_session_info.device.ipv4Address:
		device_ipv4 = valid_session_info.device.ipv4Address.root.publicAddress.root

		self.core_specific_validation(valid_session_info)
		subscription = schemas.AsSessionWithQoSSubscription(
		notificationDestination=str(valid_session_info.sink),
		qosReference=valid_session_info.qosProfile.root,
		ueIpv4Addr=device_ipv4,
		ueIpv6Addr=valid_session_info.device.ipv6Address,
		usageThreshold=schemas.UsageThreshold(duration=valid_session_info.duration),
		)
		self.add_core_specific_parameters(valid_session_info, subscription)
		return subscription

		def create_qod_session(self, session_info: Dict) -> Dict:
		"""
		Creates a QoS session based on CAMARA QoD API input.
		@@ -110,18 +132,10 @@ class NetworkManagementInterface(ABC):
		returns:
		dictionary containing the created session details, including its ID.
		"""
		valid_session_info = schemas.CreateSession.model_validate(session_info)
		self.core_specific_validation(valid_session_info)
		subscription = schemas.AsSessionWithQoSSubscription(
		notificationDestination=valid_session_info.sink,
		qosReference=valid_session_info.qosProfile,
		ueIpv4Addr=valid_session_info.device.ipv4Address,
		ueIpv6Addr=valid_session_info.device.ipv6Address,
		usageThreshold=schemas.UsageThreshold(duration=valid_session_info.duration),
		subscription = self._build_subscription(session_info)
		return common.as_session_with_qos_post(
		self.base_url, self.scs_as_id, subscription
		)
		self.add_core_specific_parameters(subscription)
		url = f"{self.base_url}/{self.scs_as_id}/subscriptions"
		common.as_session_with_qos_post(self.base_url, self.scs_as_id, subscription)

		def get_qod_session(self, session_id: str) -> Dict:
		"""

src/network/core/schemas.py

+11 −11

Original line number	Diff line number	Diff line
		@@ -5,11 +5,11 @@

		import ipaddress
		from enum import Enum
		from ipaddress import IPv4Address, IPv6Address
		from typing import Annotated

		from pydantic import AnyUrl, BaseModel, ConfigDict, Field, NonNegativeInt, RootModel
		from pydantic_extra_types.mac_address import MacAddress
		from ipaddress import IPv4Address, IPv6Address


		class FlowDirection(Enum):
		@@ -86,7 +86,7 @@ class EthFlowDescription(BaseModel):
		fDesc: FlowDescriptionModel \| None = None
		fDir: FlowDirection \| None = None
		sourceMacAddr: MacAddress \| None = None
		vlanTags: list[str] \| None = Field(None, max_items=2, min_items=1)
		vlanTags: list[str] \| None = Field(None, max_length=2, min_length=1)
		srcMacAddrEnd: MacAddress \| None = None
		destMacAddrEnd: MacAddress \| None = None

		@@ -105,9 +105,9 @@ class SponsorInformation(BaseModel):

		class QosMonitoringInformationModel(BaseModel):
		reqQosMonParams: list[RequestedQosMonitoringParameter] \| None = Field(
		None, min_items=1
		None, min_length=1
		)
		repFreqs: list[ReportingFrequency] \| None = Field(None, min_items=1)
		repFreqs: list[ReportingFrequency] \| None = Field(None, min_length=1)
		repThreshDl: Uinteger \| None = None
		repThreshUl: Uinteger \| None = None
		repThreshRp: Uinteger \| None = None
		@@ -122,8 +122,8 @@ class FlowInfo(BaseModel):
		description="Indicates the packet filters of the IP flow. Refer to subclause \
		5.3.8 of 3GPP TS 29.214 for encoding. It shall contain UL and/or DL IP \
		flow description.",
		max_items=2,
		min_items=1,
		max_length=2,
		min_length=1,
		)


		@@ -133,10 +133,10 @@ class AsSessionWithQoSSubscription(BaseModel):
		supportedFeatures: SupportedFeatures \| None = None
		notificationDestination: Link
		flowInfo: list[FlowInfo] \| None = Field(
		None, description="Describe the data flow which requires QoS.", min_items=1
		None, description="Describe the data flow which requires QoS.", min_length=1
		)
		ethFlowInfo: list[EthFlowDescription] \| None = Field(
		None, description="Identifies Ethernet packet flows.", min_items=1
		None, description="Identifies Ethernet packet flows.", min_length=1
		)
		qosReference: str \| None = Field(
		None, description="Identifies a pre-defined QoS information"
		@@ -145,10 +145,10 @@ class AsSessionWithQoSSubscription(BaseModel):
		None,
		description="Identifies an ordered list of pre-defined QoS information. The \
		lower the index of the array for a given entry, the higher the priority.",
		min_items=1,
		min_length=1,
		)
		ueIpv4Addr: ipaddress.Ipv4Addr \| None = None
		ueIpv6Addr: ipaddress.Ipv6Addr \| None = None
		ueIpv4Addr: ipaddress.IPv4Address \| None = None
		ueIpv6Addr: ipaddress.IPv6Address \| None = None
		macAddr: MacAddress \| None = None
		usageThreshold: UsageThreshold \| None = None
		sponsorInfo: SponsorInformation \| None = None

Admin message