Commit 2b6d513e authored by Panagiotis Pavlidis's avatar Panagiotis Pavlidis
Browse files

Implement TF business logic

parent d832dafb

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

+9 −1
Original line number Diff line number Diff line
@@ -59,10 +59,18 @@ class NetworkManager(NetworkManagementInterface): 
    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."

                "Open5GS requires a device to be specified for location retrieval in NEF."

            )

    def add_core_specific_location_parameters(self, retrieve_location_request: schemas.RetrievalLocationRequest, subscription: schemas.MonitoringEventSubscriptionRequest) -> None:

        subscription.msisdn = retrieve_location_request.device.phoneNumber

        subscription.monitoringType = schemas.MonitoringType.LOCATION_REPORTING

        subscription.locationType = schemas.LocationType.CURRENT_LOCATION 

        # subscription.locationType = schemas.LocationType.LAST_KNOWN

        # subscription.maximumNumberOfReports = 1

        # subscription.repPeriod = schemas.DurationSec(root=20)



        return subscription







# Note:

src/sunrise6g_opensdk/network/core/network_interface.py

+11 −2
Original line number Diff line number Diff line
@@ -115,7 +115,7 @@ class NetworkManagementInterface(ABC): 
        self,

        retrieve_location_request: schemas.RetrievalLocationRequest,

        subscription: schemas.MonitoringEventSubscriptionRequest,

    ):

    )-> schemas.MonitoringEventSubscriptionRequest:

        """

        Placeholder for adding core-specific parameters to the location subscription.

        This method should be overridden by subclasses to implement specific logic.
@@ -213,7 +213,16 @@ class NetworkManagementInterface(ABC): 
        return subscription

    

    def _build_monitoring_event_subscription(self, retrieve_location_request: schemas.RetrievalLocationRequest) ->schemas.MonitoringEventSubscriptionRequest:

        pass

        self.core_specific_monitoring_event_validation(retrieve_location_request)

        device = retrieve_location_request.device

        subscription = schemas.MonitoringEventSubscriptionRequest(

            externalId=device.networkAccessIdentifier,

            ipv4Address=device.ipv4Address,

            ipv6Addr=device.ipv6Address,

            msisdn=device.phoneNumber,

            notificationDestination= "http://test_server:8001")

        mapped_3gpp_subscription = self.add_core_specific_location_parameters(retrieve_location_request,subscription)

        return mapped_3gpp_subscription



    def create_monitoring_event_subscription(self, retrieve_location_request: Dict) -> Dict:

        """