Adaptations after pre-commit rules (fbf4e5d4) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/network/clients/open5gs/client.py

+10 −5

Original line number	Diff line number	Diff line
		@@ -56,17 +56,22 @@ class NetworkManager(NetworkManagementInterface):
		flow_id = flow_id_mapping[session_info.qosProfile.root]
		subscription.flowInfo = build_flows(flow_id, session_info)

		def core_specific_monitoring_event_validation(self, retrieve_location_request : schemas.RetrievalLocationRequest) -> None:
		def core_specific_monitoring_event_validation(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> None:
		if retrieve_location_request.device is None:
		raise ValidationError(
		"Open5GS requires a device to be specified for location retrieval in NEF."
		)
		def add_core_specific_location_parameters(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> schemas.MonitoringEventSubscriptionRequest:

		def add_core_specific_location_parameters(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.MonitoringEventSubscriptionRequest:
		return schemas.MonitoringEventSubscriptionRequest(
		msisdn=retrieve_location_request.device.phoneNumber.root.lstrip('+'),
		msisdn=retrieve_location_request.device.phoneNumber.root.lstrip("+"),
		notificationDestination="http://127.0.0.1:8001",
		monitoringType=schemas.MonitoringType.LOCATION_REPORTING,
		locationType=schemas.LocationType.LAST_KNOWN
		locationType=schemas.LocationType.LAST_KNOWN,
		)
		# subscription.msisdn = retrieve_location_request.device.phoneNumber.root.lstrip('+')
		# monitoringType = schemas.MonitoringType.LOCATION_REPORTING

src/sunrise6g_opensdk/network/core/common.py

+3 −2

Original line number	Diff line number	Diff line
		@@ -31,9 +31,10 @@ def _make_request(method: str, url: str, data=None):
		raise CoreHttpError("connection error") from e



		# Monitoring Event Methods
		def monitoring_event_post(base_url: str, scs_as_id: str, model_payload: BaseModel) -> dict:
		def monitoring_event_post(
		base_url: str, scs_as_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True, by_alias=True)
		url = monitoring_event_build_url(base_url, scs_as_id)
		return _make_request("POST", url, data=data)

src/sunrise6g_opensdk/network/core/network_interface.py

+52 −24

Original line number	Diff line number	Diff line
		@@ -12,9 +12,9 @@
		##
		import uuid
		from abc import ABC
		from datetime import datetime, timedelta, timezone
		from itertools import product
		from typing import Dict
		from datetime import datetime, timedelta, timezone

		from sunrise6g_opensdk import logger
		from sunrise6g_opensdk.network.clients.errors import NetworkPlatformError
		@@ -113,8 +113,7 @@ class NetworkManagementInterface(ABC):
		pass

		def add_core_specific_location_parameters(
		self,
		retrieve_location_request: schemas.RetrievalLocationRequest
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.MonitoringEventSubscriptionRequest:
		"""
		Placeholder for adding core-specific parameters to the location subscription.
		@@ -152,7 +151,9 @@ class NetworkManagementInterface(ABC):
		# This method should be overridden by subclasses if needed
		pass

		def core_specific_monitoring_event_validation(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> None:
		def core_specific_monitoring_event_validation(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> None:
		"""
		Validates core-specific parameters for the monitoring event subscription.

		@@ -212,9 +213,13 @@ class NetworkManagementInterface(ABC):
		self.add_core_specific_ti_parameters(traffic_influence_data, subscription)
		return subscription

		def _build_monitoring_event_subscription(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> schemas.MonitoringEventSubscriptionRequest:
		def _build_monitoring_event_subscription(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.MonitoringEventSubscriptionRequest:
		self.core_specific_monitoring_event_validation(retrieve_location_request)
		subscription_3gpp = self.add_core_specific_location_parameters(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
		@@ -224,7 +229,9 @@ class NetworkManagementInterface(ABC):

		return subscription_3gpp

		def _compute_camara_last_location_time(self, event_time: datetime, age_of_location_info_min : int = None) -> datetime:
		def _compute_camara_last_location_time(
		self, event_time: datetime, age_of_location_info_min: int = None
		) -> datetime:
		"""
		Computes the last location time based on the event time and age of location info.

		@@ -236,12 +243,16 @@ class NetworkManagementInterface(ABC):
		datetime object representing the last location time in UTC.
		"""
		if age_of_location_info_min is not None:
		last_location_time = event_time - timedelta(minutes=age_of_location_info_min)
		last_location_time = event_time - timedelta(
		minutes=age_of_location_info_min
		)
		return last_location_time.replace(tzinfo=timezone.utc)
		else:
		return event_time.replace(tzinfo=timezone.utc)

		def create_monitoring_event_subscription(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> schemas.Location:
		def create_monitoring_event_subscription(
		self, retrieve_location_request: schemas.RetrievalLocationRequest
		) -> schemas.Location:
		"""
		Creates a Monitoring Event subscription based on CAMARA Location API input.

		@@ -252,28 +263,45 @@ class NetworkManagementInterface(ABC):
		returns:
		dictionary containing the created subscription details, including its ID.
		"""
		subscription = self._build_monitoring_event_subscription(retrieve_location_request)
		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")
		raise NetworkPlatformError("Location information not found in monitoring event report")
		log.error(
		"Failed to retrieve location information from monitoring event report"
		)
		raise NetworkPlatformError(
		"Location information not found in monitoring event report"
		)
		geo_area = monitoring_event_report.locationInfo.geographicArea
		report_event_time = monitoring_event_report.eventTime
		age_of_location_info = None
		if monitoring_event_report.locationInfo.ageOfLocationInfo is not None:
		age_of_location_info = monitoring_event_report.locationInfo.ageOfLocationInfo.duration
		last_location_time = self._compute_camara_last_location_time(report_event_time,age_of_location_info)
		age_of_location_info = (
		monitoring_event_report.locationInfo.ageOfLocationInfo.duration
		)
		last_location_time = self._compute_camara_last_location_time(
		report_event_time, age_of_location_info
		)
		print(f"Last Location time is {last_location_time}")
		camara_point_list: list[schemas.Point] = []
		for point in geo_area.polygon.point_list.geographical_coords:
		camara_point_list.append(schemas.Point(latitude=point.lat,longitude=point.lon))
		camara_polygon = schemas.Polygon(areaType=schemas.AreaType.polygon,boundary=schemas.PointList(camara_point_list))
		camara_point_list.append(
		schemas.Point(latitude=point.lat, longitude=point.lon)
		)
		camara_polygon = schemas.Polygon(
		areaType=schemas.AreaType.polygon,
		boundary=schemas.PointList(camara_point_list),
		)

		camara_location = schemas.Location(area=camara_polygon,lastLocationTime=last_location_time)
		camara_location = schemas.Location(
		area=camara_polygon, lastLocationTime=last_location_time
		)

		return camara_location

src/sunrise6g_opensdk/network/core/schemas.py

+247 −70

Original line number	Diff line number	Diff line
		@@ -10,7 +10,15 @@ from ipaddress import IPv4Address, IPv6Address
		from typing import Annotated, Literal
		from uuid import UUID

		from pydantic import AnyUrl, BaseModel, ConfigDict, Field, NonNegativeInt, RootModel, AnyHttpUrl
		from pydantic import (
		AnyHttpUrl,
		AnyUrl,
		BaseModel,
		ConfigDict,
		Field,
		NonNegativeInt,
		RootModel,
		)
		from pydantic_extra_types.mac_address import MacAddress


		@@ -214,88 +222,193 @@ class TrafficInfluSub(BaseModel): # Replace with a meaningful name

		##Monitoring Event API


		class DurationMin(BaseModel):
		duration: int = Field(0,description="Unsigned integer identifying a period of time in units of minutes",ge=0)
		duration: int = Field(
		0,
		description="Unsigned integer identifying a period of time in units of minutes",
		ge=0,
		)


		class PlmnId(BaseModel):
		mcc: str = Field(...,description="String encoding a Mobile Country Code, comprising of 3 digits.")
		mnc: str = Field(...,description="String encoding a Mobile Network Code, comprising of 2 or 3 digits.")
		mcc: str = Field(
		...,
		description="String encoding a Mobile Country Code, comprising of 3 digits.",
		)
		mnc: str = Field(
		...,
		description="String encoding a Mobile Network Code, comprising of 2 or 3 digits.",
		)


		# The enumeration Accuracy represents a desired granularity of accuracy of the requested location information.
		class Accuracy(str, Enum):
		cgi_ecgi = "CGI_ECGI" # The AF requests to be notified using cell level location accuracy.
		ta_ra = "TA_RA" # The AF requests to be notified using TA/RA level location accuracy.
		cgi_ecgi = (
		"CGI_ECGI" # The AF requests to be notified using cell level location accuracy.
		)
		ta_ra = (
		"TA_RA" # The AF requests to be notified using TA/RA level location accuracy.
		)
		geo_area = "GEO_AREA" # The AF requests to be notified using the geographical area accuracy.
		civic_addr = "CIVIC_ADDR" # The AF requests to be notified using the civic address accuracy. #EDGEAPP


		# If locationType set to "LAST_KNOWN_LOCATION", the monitoring event request from AF shall be only for one-time monitoring request
		class LocationType(str, Enum):
		CURRENT_LOCATION = "CURRENT_LOCATION" # The AF requests to be notified for current location.
		LAST_KNOWN = "LAST_KNOWN_LOCATION" # The AF requests to be notified for last known location.
		CURRENT_LOCATION = (
		"CURRENT_LOCATION" # The AF requests to be notified for current location.
		)
		LAST_KNOWN = (
		"LAST_KNOWN_LOCATION" # The AF requests to be notified for last known location.
		)


		# This data type represents a monitoring event type.
		class MonitoringType(str, Enum):
		LOCATION_REPORTING = "LOCATION_REPORTING"


		class LocationFailureCause(str, Enum):
		position_denied = "POSITIONING_DENIED" # Positioning is denied.
		unsupported_by_ue = "UNSUPPORTED_BY_UE" # Positioning is not supported by UE.
		not_registered_ue = "NOT_REGISTERED_UE" # UE is not registered.
		unspecified = "UNSPECIFIED" # Unspecified cause.


		class GeographicalCoordinates(BaseModel):
		lon: float = Field(..., description="Longitude coordinate.")
		lat: float = Field(..., description="Latitude coordinate.")


		class PointListNef(BaseModel):
		geographical_coords: list[GeographicalCoordinates] = Field(..., description="List of geographical coordinates defining the points.",min_length=3,max_length=15)
		geographical_coords: list[GeographicalCoordinates] = Field(
		...,
		description="List of geographical coordinates defining the points.",
		min_length=3,
		max_length=15,
		)


		class NefPolygon(BaseModel):
		point_list: PointListNef = Field(..., description="List of points defining the polygon.")
		point_list: PointListNef = Field(
		..., description="List of points defining the polygon."
		)


		class GeographicArea(BaseModel):
		polygon: NefPolygon \| None = Field(None, description="Identifies a polygonal geographic area.")
		polygon: NefPolygon \| None = Field(
		None, description="Identifies a polygonal geographic area."
		)


		# This data type represents the user location information which is sent from the NEF to the AF.
		class LocationInfo(BaseModel):
		ageOfLocationInfo: DurationMin \| None = Field(None,description="Indicates the elapsed time since the last network contact of the UE.")
		ageOfLocationInfo: DurationMin \| None = Field(
		None,
		description="Indicates the elapsed time since the last network contact of the UE.",
		)
		cellId: str \| None = Field(None, description="Cell ID where the UE is located.")
		trackingAreaId: str \| None = Field(None, description="TrackingArea ID where the UE is located.")
		trackingAreaId: str \| None = Field(
		None, description="TrackingArea ID where the UE is located."
		)
		enodeBId: str \| None = Field(None, description="eNodeB ID where the UE is located.")
		routingAreaId: str \| None = Field(None, description="Routing Area ID where the UE is located")
		routingAreaId: str \| None = Field(
		None, description="Routing Area ID where the UE is located"
		)
		plmnId: PlmnId \| None = Field(None, description="PLMN ID where the UE is located.")
		twanId: str \| None = Field(None, description="TWAN ID where the UE is located.")
		geographicArea: GeographicArea \| None = Field(None,description="Identifies a geographic area of the user where the UE is located.")
		geographicArea: GeographicArea \| None = Field(
		None,
		description="Identifies a geographic area of the user where the UE is located.",
		)


		class MonitoringEventSubscriptionRequest(BaseModel):
		accuracy: Accuracy \| None = Field(None,description="Accuracy represents a desired granularity of accuracy of the requested location information.")
		externalId: str \| None = Field(None, description="Identifies a user clause 4.6.2 TS 23.682 (optional)")
		msisdn: str \| None = Field(None,description="Identifies the MS internal PSTN/ISDN number allocated for a UE.")
		ipv4Addr: IPv4Address \| None = Field(None,description="Identifies the Ipv4 address.")
		ipv6Addr: IPv6Address \| None = Field(None,description="Identifies the Ipv6 address.")
		notificationDestination: AnyHttpUrl = Field(..., description="URI of a notification destination that the T8 message shall be delivered to.")
		monitoringType: MonitoringType = Field(..., description="Enumeration of monitoring type. Refer to clause 5.3.2.4.3.")
		maximumNumberOfReports: int \| None = Field(None, description="Identifies the maximum number of event reports to be generated by the AMF to the NEF and then the AF.")
		monitorExpireTime: datetime \| None = Field(None, description="Identifies the absolute time at which the related monitoring event request is considered to expire.")
		locationType: LocationType \| None = Field(None, description="Indicates whether the request is for Current Location, Initial Location, or Last Known Location.")
		repPeriod: DurationSec \| None = Field(None,description="Identifies the periodic time for the event reports.")
		minimumReportInterval: DurationSec \| None = Field(None,description="identifies a minimum time interval between Location Reporting notifications")
		accuracy: Accuracy \| None = Field(
		None,
		description="Accuracy represents a desired granularity of accuracy of the requested location information.",
		)
		externalId: str \| None = Field(
		None, description="Identifies a user clause 4.6.2 TS 23.682 (optional)"
		)
		msisdn: str \| None = Field(
		None,
		description="Identifies the MS internal PSTN/ISDN number allocated for a UE.",
		)
		ipv4Addr: IPv4Address \| None = Field(
		None, description="Identifies the Ipv4 address."
		)
		ipv6Addr: IPv6Address \| None = Field(
		None, description="Identifies the Ipv6 address."
		)
		notificationDestination: AnyHttpUrl = Field(
		...,
		description="URI of a notification destination that the T8 message shall be delivered to.",
		)
		monitoringType: MonitoringType = Field(
		..., description="Enumeration of monitoring type. Refer to clause 5.3.2.4.3."
		)
		maximumNumberOfReports: int \| None = Field(
		None,
		description="Identifies the maximum number of event reports to be generated by the AMF to the NEF and then the AF.",
		)
		monitorExpireTime: datetime \| None = Field(
		None,
		description="Identifies the absolute time at which the related monitoring event request is considered to expire.",
		)
		locationType: LocationType \| None = Field(
		None,
		description="Indicates whether the request is for Current Location, Initial Location, or Last Known Location.",
		)
		repPeriod: DurationSec \| None = Field(
		None, description="Identifies the periodic time for the event reports."
		)
		minimumReportInterval: DurationSec \| None = Field(
		None,
		description="identifies a minimum time interval between Location Reporting notifications",
		)


		# This data type represents a monitoring event notification which is sent from the NEF to the AF.
		class MonitoringEventReport(BaseModel):
		externalId: str \| None = Field(None,description="Identifies a user, clause 4.6.2 TS 23.682")
		msisdn: str \| None = Field(None,description="Identifies the MS internal PSTN/ISDN number allocated for a UE.")
		locationInfo: LocationInfo \| None = Field(None, description="Indicates the user location related information.")
		locFailureCause: LocationFailureCause \| None = Field(None, description="Indicates the location positioning failure cause.")
		monitoringType: MonitoringType = Field(..., description="Identifies the type of monitoring as defined in clause 5.3.2.4.3.")
		eventTime: datetime \| None = Field(None, description="Identifies when the event is detected or received. Shall be included for each group of UEs.")
		externalId: str \| None = Field(
		None, description="Identifies a user, clause 4.6.2 TS 23.682"
		)
		msisdn: str \| None = Field(
		None,
		description="Identifies the MS internal PSTN/ISDN number allocated for a UE.",
		)
		locationInfo: LocationInfo \| None = Field(
		None, description="Indicates the user location related information."
		)
		locFailureCause: LocationFailureCause \| None = Field(
		None, description="Indicates the location positioning failure cause."
		)
		monitoringType: MonitoringType = Field(
		...,
		description="Identifies the type of monitoring as defined in clause 5.3.2.4.3.",
		)
		eventTime: datetime \| None = Field(
		None,
		description="Identifies when the event is detected or received. Shall be included for each group of UEs.",
		)


		# This data type represents a monitoring notification which is sent from the NEF to the AF.
		class MonitoringNotification(BaseModel):
		subscription: AnyHttpUrl = Field(..., description="Link to the subscription resource to which this notification is related.")
		monitoringEventReports: list[MonitoringEventReport] \| None = Field(None, description="Each element identifies a monitoring event report (optional).")
		cancelInd: bool \| None = Field(False,description="Indicates whether to request to cancel the corresponding monitoring subscription. Set to false or omitted otherwise.")
		subscription: AnyHttpUrl = Field(
		...,
		description="Link to the subscription resource to which this notification is related.",
		)
		monitoringEventReports: list[MonitoringEventReport] \| None = Field(
		None,
		description="Each element identifies a monitoring event report (optional).",
		)
		cancelInd: bool \| None = Field(
		False,
		description="Indicates whether to request to cancel the corresponding monitoring subscription. Set to false or omitted otherwise.",
		)


		###############################################################
		@@ -381,44 +494,108 @@ class RetrievalLocationRequest(BaseModel):
		"""
		Request to retrieve the location of a device. Device is not required when using a 3-legged access token.
		"""
		device: Annotated[Device \| None, Field(None,description="End-user device able to connect to a mobile network.")]
		maxAge: Annotated[int \| None, Field(None, description="Maximum age of the location information which is accepted for the location retrieval (in seconds).")]
		maxSurface: Annotated[int \| None, Field(None,description="Maximum surface in square meters which is accepted by the client for the location retrieval.",ge=1,examples=[1000000])]

		device: Annotated[
		Device \| None,
		Field(None, description="End-user device able to connect to a mobile network."),
		]
		maxAge: Annotated[
		int \| None,
		Field(
		None,
		description="Maximum age of the location information which is accepted for the location retrieval (in seconds).",
		),
		]
		maxSurface: Annotated[
		int \| None,
		Field(
		None,
		description="Maximum surface in square meters which is accepted by the client for the location retrieval.",
		ge=1,
		examples=[1000000],
		),
		]


		class AreaType(str, Enum):
		circle = "CIRCLE" # The area is defined as a circle.
		polygon = "POLYGON" # The area is defined as a polygon.


		class Point(BaseModel):
		latitude: Annotated[float,Field(description="Latitude component of a location.",examples=["50.735851"],ge=-90,le=90)]
		longitude: Annotated[float,Field(..., description="Longitude component of location.",examples=["7.10066"],ge=-180,le=180)]
		latitude: Annotated[
		float,
		Field(
		description="Latitude component of a location.",
		examples=["50.735851"],
		ge=-90,
		le=90,
		),
		]
		longitude: Annotated[
		float,
		Field(
		...,
		description="Longitude component of location.",
		examples=["7.10066"],
		ge=-180,
		le=180,
		),
		]


		class PointList(RootModel[Annotated[
		class PointList(
		RootModel[
		Annotated[
		list[Point],
		Field(min_length=3,max_length=15, description="List of points defining the area.")]]):
		Field(
		min_length=3,
		max_length=15,
		description="List of points defining the area.",
		),
		]
		]
		):
		pass


		class Circle(BaseModel):
		areaType: Literal[AreaType.circle]
		center: Annotated[Point, Field(description="Center point of the circle.")]
		radius: Annotated[float, Field(description="Radius of the circle.", ge=1)]


		class Polygon(BaseModel):
		areaType: Literal[AreaType.polygon]
		boundary: Annotated[PointList, Field(description="List of points defining the polygon.")]
		boundary: Annotated[
		PointList, Field(description="List of points defining the polygon.")
		]


		Area = Annotated[Circle \| Polygon, Field(discriminator="areaType")]

		class LastLocationTime(RootModel[Annotated[

		class LastLocationTime(
		RootModel[
		Annotated[
		datetime,
		Field( description="Last date and time when the device was localized.",examples="2023-09-07T10:40:52Z")]]):
		Field(
		description="Last date and time when the device was localized.",
		examples="2023-09-07T10:40:52Z",
		),
		]
		]
		):
		pass


		class Location(BaseModel):
		lastLocationTime: Annotated[LastLocationTime, Field(description="Last known location time.")]
		lastLocationTime: Annotated[
		LastLocationTime, Field(description="Last known location time.")
		]
		area: Annotated[Area, Field(description="Geographical area of the location.")]


		class ApplicationServerIpv4Address(RootModel[str]):
		root: Annotated[
		str,

tests/network/conftest.py

+24 −11

File changed.

Preview size limit exceeded, changes collapsed.

Admin message