From 4a3d3c5acdfc06a8aebc100d2847763ee4610c4f Mon Sep 17 00:00:00 2001 From: "sunil.n" Date: Mon, 10 Nov 2025 17:29:26 +0530 Subject: [PATCH] Implement function to subscribe to RAN NSSMF for NSSI LCM, PM notifications Change-Id: I244226f4d84f4f814dbb80894cb48a311b8a154e Signed-off-by: sunil.n --- .../src/ran_nssmf_client.py | 54 ++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/sample-rapp-generator/rapp-slice-prb-prediction/src/ran_nssmf_client.py b/sample-rapp-generator/rapp-slice-prb-prediction/src/ran_nssmf_client.py index 9feccc0..39bee6f 100644 --- a/sample-rapp-generator/rapp-slice-prb-prediction/src/ran_nssmf_client.py +++ b/sample-rapp-generator/rapp-slice-prb-prediction/src/ran_nssmf_client.py @@ -1,6 +1,8 @@ import json import logging +import requests + from sme_client import SMEClient @@ -37,4 +39,56 @@ class RAN_NSSMF_CLIENT(object): logger.debug(f"InfluxDB Address: {self.address}") else: logger.error("Failed to discover RAN NSSMF URL from SME.") + + def subscribe_to_file_ready_notifications(self, callback_uri: str): + """ + Subscribes to file-ready notifications from the RAN NSSMF. + + Args: + callback_uri (str): The URI where the RAN NSSMF should send notifications. + + Returns: + requests.Response: The response object from the POST request. + """ + base_url = self.ran_nssmf_address + + if not base_url: + logger.error("RAN NSSMF address is not configured. Cannot subscribe.") + return None + + # Ensure base_url does not have a trailing slash + base_url = base_url.rstrip('/') + subscription_url = f"{base_url}/3GPPManagement/FileDataReportingMnS/v17.0.0/subscriptions" + + payload = { + "consumerReference": callback_uri + } + + headers = { + "Content-Type": "application/json" + } + + logger.info(f"Subscribing to file-ready notifications at: {subscription_url}") + logger.debug(f"Payload: {payload}") + + try: + response = requests.post(subscription_url, json=payload, headers=headers, timeout=10) + response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx) + + logger.info(f"Successfully subscribed to notifications. Status: {response.status_code}") + if response.headers.get("Location"): + logger.info(f"Subscription Location: {response.headers.get('Location')}") + + return response + + except requests.exceptions.HTTPError as http_err: + logger.error(f"HTTP error occurred while subscribing: {http_err} - Response: {http_err.response.text}") + except requests.exceptions.ConnectionError as conn_err: + logger.error(f"Connection error occurred while subscribing: {conn_err}") + except requests.exceptions.Timeout as timeout_err: + logger.error(f"Timeout error occurred while subscribing: {timeout_err}") + except requests.exceptions.RequestException as req_err: + logger.error(f"An unexpected error occurred while subscribing: {req_err}") + + return None \ No newline at end of file -- 2.16.6