Commit c4e1345b authored by Anastasios Pandis's avatar Anastasios Pandis
Browse files

added MSISDN mapping and extract parse_location_response as adaptable method... 

added MSISDN mapping and extract parse_location_response as adaptable method for adapter flexibility, fixed identifier fields to unwrap pydantic root model types correctly
parent 217d0ef0

src/sunrise6g_opensdk/network/core/base_network_client.py

+43 −16
Original line number Diff line number Diff line
@@ -237,11 +237,20 @@ 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 +274,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 +314,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:

        """