### Start A1 Policy Manager - Python Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Initializes and starts the A1 policy manager, which queries the A1 mediator for active policies on xApp startup. It uses RMR for communication. ```python from ricxappframe.xapp_frame import RMRXapp from src.manager.A1PolicyManager import A1PolicyManager from src.utils.constants import Constants # Initialize RMR xApp rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, rmr_port=4560 ) # Create and start A1 policy manager a1_mgr = A1PolicyManager(rmr_xapp) a1_mgr.startup() # Sends policy query via RMR message type A1_POLICY_QUERY (20012): # {"policy_type_id": "2"} # This queries the A1 mediator for policies of type HELLOWORLD_POLICY_ID (2) # Response handling is done by registered A1PolicyHandler ``` -------------------------------- ### Initialize and Start HWXapp - Python Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Initializes the RIC xApp framework with RMR messaging, SDL storage, and configuration management. Supports both blocking and non-blocking startup for testing. ```python from src.hwxapp import HWXapp # Basic xApp initialization and startup hwxapp = HWXapp() # The xApp automatically initializes: # - RMR messaging on port 4560 # - SDL manager for data persistence # - A1 policy manager # - Subscription manager for eNB/gNB # - Metric collection # - Health check handlers # Start the xApp (blocking call) hwxapp.start() # For unit testing with fake SDL import os os.environ["USE_FAKE_SDL"] = "True" hwxapp = HWXapp() hwxapp.start(thread=True) # Non-blocking for tests # ... perform tests ... hwxapp.stop() ``` -------------------------------- ### Handle Configuration with Configuration Management Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Handle xApp configuration loading and dynamic updates from the RIC platform. This involves defining a configuration change handler, initializing the RMR xApp with this handler, and accessing the configuration at runtime. The configuration file location and an example structure are provided. ```python from ricxappframe.xapp_frame import RMRXapp import json def config_change_handler(rmr_xapp, config): """Called on startup and when config file changes""" print(f"Configuration updated: {json.dumps(config, indent=2)}") # Access configuration values app_name = config.get("xapp_name", "hw-python") log_level = config.get("log_level", "INFO") # Update application behavior based on config rmr_xapp.config = config # Initialize xApp with config handler rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, config_handler=config_change_handler, rmr_port=4560, post_init=lambda x: print("xApp initialized") ) # Configuration file location: config/config-file.json # Example configuration structure: config_example = { "name": "hw-python", "version": "1.0.0", "containers": [ { "name": "hw-python", "image": { "registry": "nexus3.o-ran-sc.org:10004", "name": "o-ran-sc/ric-app-hw-python", "tag": "1.1.0" } } ], "messaging": { "ports": [ {"name": "http", "port": 8080}, {"name": "rmrdata", "port": 4560} ] }, "rmr": { "protPort": "tcp:4560", "maxSize": 2072, "numWorkers": 1, "txMessages": ["RIC_SUB_REQ", "A1_POLICY_RESP"], "rxMessages": ["RIC_SUB_RESP", "A1_POLICY_REQ"] } } # Access config at runtime current_config = rmr_xapp.config ``` -------------------------------- ### RMR Message Routing Setup Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Demonstrates the setup for RMR (Ric Message Routing) message sending and receiving, including proper message type routing. This requires initializing the RMRXapp and importing necessary components from the ricxappframe library. ```python from ricxappframe.xapp_frame import RMRXapp, rmr from src.utils.constants import Constants import json ``` -------------------------------- ### Get Deployed xApps List - curl and jq Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst This command queries the App Manager service to retrieve a list of all deployed xApps. The output is then parsed using 'jq' to format it as JSON, making it easier to read and process. This is crucial for understanding the application's deployment status and configuration. ```bash curl http://service-ricplt-appmgr-http.ricplt:8080/ric/v1/xapps | jq . ``` -------------------------------- ### Get Deployed XApps Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst Retrieves a list of all deployed xApps managed by the application manager. This is useful for monitoring and understanding the current state of deployed applications. ```APIDOC ## GET /ric/v1/xapps ### Description Retrieves a list of all deployed xApps. ### Method GET ### Endpoint http://service-ricplt-appmgr-http.ricplt:8080/ric/v1/xapps ### Parameters #### Query Parameters None #### Request Body None ### Request Example ```bash curl http://service-ricplt-appmgr-http.ricplt:8080/ric/v1/xapps | jq . ``` ### Response #### Success Response (200) - **instances** (array) - List of instances for the xApp. - **ip** (string) - IP address of the instance. - **name** (string) - Name of the instance. - **policies** (array) - List of policies associated with the instance. - **port** (integer) - Port number for the instance. - **rxMessages** (array) - List of messages the instance receives. - **status** (string) - Current status of the instance. - **txMessages** (array) - List of messages the instance transmits. - **name** (string) - Name of the xApp. - **status** (string) - Deployment status of the xApp. - **version** (string) - Version of the xApp. #### Response Example ```json [ { "instances": [ { "ip": "service-ricxapp-hw-python-rmr.ricxapp", "name": "hw-python-55ff7549df-kpj6k", "policies": [ 1 ], "port": 4560, "rxMessages": [ "RIC_SUB_RESP", "A1_POLICY_REQ", "RIC_HEALTH_CHECK_REQ" ], "status": "running", "txMessages": [ "RIC_SUB_REQ", "A1_POLICY_RESP", "A1_POLICY_QUERY", "RIC_HEALTH_CHECK_RESP" ] } ], "name": "hw-python", "status": "deployed", "version": "1.0" } ] ``` ``` -------------------------------- ### Check Onboarded Charts Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst Retrieves a list of onboarded charts from the app manager to confirm successful onboarding of hw-python. This uses a GET request to the app manager API. ```bash $ curl --location --request GET "http://:32080/onboard/api/v1/charts" --header 'Content-Type: application/json' { "hw-python": [ { "name": "hw-python", "version": "1.0.0", "description": "Standard xApp Helm Chart", "apiVersion": "v1", "appVersion": "1.0", "urls": [ "charts/hw-python-1.0.0.tgz" ], "created": "2021-07-05T15:07:34.518377486Z", "digest": "e9db874d35154643a2c6f26dd52929c9dcf143f165683c03d07518bb0c2d768d" } ], "hw-python": [ { "name": "hw-python", "version": "1.0.0", "description": "Standard xApp Helm Chart", "apiVersion": "v1", "appVersion": "1.0", "urls": [ "charts/hw-python-1.0.0.tgz" ], "created": "2021-07-05T15:20:13.965653743Z", "digest": "975b1da1f8669e8ed1b1e5be809e7cf4841ef33abcb88207bc3a735e9b543a9a" } ] } ``` -------------------------------- ### Check dms_cli Health Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Verifies if the dms_cli tool is functioning correctly by running the 'health' command. This ensures that the tool is properly installed and configured before proceeding with onboarding. ```bash $ dms_cli health True ``` -------------------------------- ### Onboarding hw-python Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst This section details the process of onboarding the hw-python xApp using the dms_cli tool. It includes prerequisites, configuration, and the command to initiate onboarding. ```APIDOC ## Onboarding hw-python using dms_cli ### Description Onboard the hw-python xApp to chartmuseme using the `dms_cli` tool. Ensure `dms_cli` and helm3 are installed and configured. ### Method POST ### Endpoint `dms_cli onboard --config_file_path= --shcema_file_path=` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body This command does not directly use a request body, but relies on configuration files. - **config_file_path** (string) - Required - Path to the xApp deployment descriptor (e.g., `config-file.json`). - **shcema_file_path** (string) - Required - Path to the schema for validating the config file (e.g., `schema.json`). ### Request Example ```bash $ cd init $ dms_cli onboard --config_file_path=config-file.json --shcema_file_path=schema.json ``` ### Response #### Success Response (200) Returns a status indicating successful initiation of the onboarding process. ```json { "status": "Created" } ``` #### Response Example ```json { "status": "Created" } ``` ``` -------------------------------- ### HWXapp Initialization and Startup Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Initializes the RIC xApp framework with RMR messaging, SDL storage, and configuration management capabilities. It automatically sets up RMR messaging, SDL manager, A1 policy manager, subscription manager, metric collection, and health check handlers. ```APIDOC ## HWXapp Initialization and Startup ### Description Main xApp class that initializes the RIC xApp framework with RMR messaging, SDL storage, and configuration management capabilities. ### Method Instantiation and execution of the `HWXapp` class. ### Endpoint N/A (Internal xApp initialization) ### Parameters None ### Request Example ```python from src.hwxapp import HWXapp # Basic xApp initialization and startup hwxapp = HWXapp() # Start the xApp (blocking call) hwxapp.start() # For unit testing with fake SDL import os os.environ["USE_FAKE_SDL"] = "True" hwxapp = HWXapp() hwxapp.start(thread=True) # Non-blocking for tests # ... perform tests ... hwxapp.stop() ``` ### Response #### Success Response (Initialization) - The xApp initializes RMR messaging on port 4560. - Initializes SDL manager for data persistence. - Initializes A1 policy manager. - Initializes Subscription manager for eNB/gNB. - Initializes Metric collection. - Initializes Health check handlers. #### Response Example N/A (Execution is typically blocking or asynchronous) ``` -------------------------------- ### Build and Run Docker Container for HW Python xApp Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt This section provides bash commands to build a Docker image for the HW Python xApp and run it as a container. It includes environment variable configuration for database services and port mappings. ```bash # Build Docker image docker build --no-cache -t hw-python:latest . # Run the xApp container docker run \ -e USE_FAKE_SDL=False \ -e DBAAS_SERVICE_HOST=service-ricplt-dbaas-tcp.ricplt.svc.cluster.local \ -e DBAAS_SERVICE_PORT=6379 \ -p 8080:8080 \ -p 4560:4560 \ hw-python:latest # Check xApp logs docker logs # Expected output: # {"ts":1624562552123,"crit":"INFO","msg":"Using config file: config/config-file.json"} # {"ts":1624562552140,"crit":"INFO","msg":"Xapp started, listening on: :8080"} # {"ts":1624562553141,"crit":"INFO","msg":"xApp ready call back received"} # RMR is ready now ... # Health check endpoint curl http://localhost:8080/ric/v1/health/ready curl http://localhost:8080/ric/v1/health/alive # Metrics endpoint curl http://localhost:8080/ric/v1/metrics ``` -------------------------------- ### Deploy HW Python xApp on Kubernetes via App Manager Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt This section outlines the Kubernetes deployment process for the HW Python xApp using the dms_cli tool and App Manager. It covers setting the chart repository, onboarding the xApp, deploying it, and verifying the deployment status. ```bash # Set chart repository URL export CHART_REPO_URL=http://service-ricplt-xapp-onboarder-http.ricplt:8080 # Verify dms_cli connectivity dms_cli health # Output: True # Onboard xApp to chartmuseum cd init dms_cli onboard --config_file_path=config-file.json --shcema_file_path=schema.json # Verify onboarding curl --location --request GET \ "http://service-ricplt-appmgr-http.ricplt:32080/onboard/api/v1/charts" \ --header 'Content-Type: application/json' # Deploy xApp curl --location --request POST \ "http://service-ricplt-appmgr-http.ricplt:32080/appmgr/ric/v1/xapps" \ --header 'Content-Type: application/json' \ --data-raw '{"xappName": "hw-python", "helmVersion": "1.0.0"}' # Expected response: # {"instances":null,"name":"hw-python","status":"deployed","version":"1.0"} # Check deployment status kubectl get pods -n ricxapp # NAME READY STATUS RESTARTS AGE # ricxapp-hw-python-64b5447dcc-mbt5w 1/1 Running 0 5m45s kubectl get svc -n ricxapp # service-ricxapp-hw-python-http ClusterIP 10.104.223.245 8080/TCP # service-ricxapp-hw-python-rmr ClusterIP 10.103.243.21 4560/TCP,4561/TCP # Query deployed xApps curl http://service-ricplt-appmgr-http.ricplt:8080/ric/v1/xapps | jq . # View xApp logs kubectl logs ricxapp-hw-python-64b5447dcc-mbt5w -n ricxapp ``` -------------------------------- ### Onboard hw-python xApp using dms_cli Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Initiates the onboarding process for the hw-python xApp by running the dms_cli onboard command. This command requires the configuration file and schema file paths as arguments. The output indicates the status of the onboarding process. ```bash $ cd init $ dms_cli onboard --config_file_path=config-file.json --shcema_file_path=schema.json { "status": "Created" } ``` -------------------------------- ### Download Onboarded Chart Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst Download the Helm chart for the hw-python xApp after it has been onboarded. ```APIDOC ## Download Onboarded Chart ### Description Download the specific Helm chart package for the hw-python xApp. ### Method GET ### Endpoint `http://:32080/onboard/api/v1/charts/xapp/hw-python/ver/1.0.0` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```bash curl --location --request GET "http://:32080/onboard/api/v1/charts/xapp/hw-python/ver/1.0.0" --header 'Content-Type: application/json' --output hw-python.tgz ``` ### Response #### Success Response (200) Returns the Helm chart package (`.tgz` file) for the specified xApp version. The output is saved to `hw-python.tgz`. #### Response Example (Binary stream of the `.tgz` file) ``` tar -xvzf hw-python.tgz ``` ``` -------------------------------- ### A1 Policy Manager Startup Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Sends policy queries to the A1 mediator to discover active policies on xApp startup. This helps the xApp understand the current policy landscape it needs to interact with. ```APIDOC ## A1 Policy Manager Startup ### Description Sends policy queries to the A1 mediator to discover active policies on xApp startup. ### Method Initiates policy discovery via RMR message. ### Endpoint N/A (Internal RMR message sending) ### Parameters None ### Request Example ```json { "policy_type_id": 2 } ``` ### Response #### Success Response (Implicit) - The A1 Policy Manager sends an A1_POLICY_QUERY message. - The response is handled by the registered A1PolicyHandler, which processes policy details received from the A1 mediator. #### Response Example N/A (Response is handled internally by A1PolicyHandler ``` -------------------------------- ### Manage Data Persistence with SDL - Python Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Manages read and write operations to the Shared Data Layer (SDL) for persistent storage of network data. Supports using a fake SDL for testing. ```python from ricxappframe.xapp_frame import RMRXapp from src.manager.SdlManager import SdlManager import json # Initialize RMR xApp with SDL rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, rmr_port=4560, use_fake_sdl=False # Set to True for testing without SDL ) # Create SDL manager sdl_mgr = SdlManager(rmr_xapp) # Retrieve gNB list from e2Manager namespace gnb_list = sdl_mgr.sdlGetGnbList() ``` -------------------------------- ### Deploy hw-python xApp Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst Deploy the onboarded hw-python xApp to the RIC platform using the appmgr API. ```APIDOC ## Deploy hw-python xApp ### Description Deploy the hw-python xApp to the RIC platform using the application manager. The deployment occurs in the `ricxapp` namespace. ### Method POST ### Endpoint `http://:32080/appmgr/ric/v1/xapps` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **xappName** (string) - Required - The name of the xApp to deploy (e.g., `hw-python`). - **helmVersion** (string) - Required - The version of the Helm chart to deploy (e.g., `1.0.0`). ### Request Example ```bash $ curl --location --request POST "http://:32080/appmgr/ric/v1/xapps" --header 'Content-Type: application/json' --data-raw '{"xappName": "hw-python", "helmVersion": "1.0.0"}' ``` ### Response #### Success Response (200) Returns the deployment status of the xApp. - **name** (string) - The name of the deployed xApp. - **version** (string) - The version of the deployed xApp. - **status** (string) - The deployment status (e.g., `deployed`). - **instances** (null) - Placeholder for instance details, typically null on initial deployment. #### Response Example ```json {"instances":null,"name":"hw-python","status":"deployed","version":"1.0"} ``` ``` -------------------------------- ### Manage Subscriptions using SubscriptionManager Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Manage E2 subscriptions to network nodes (eNBs and gNBs) using the SubscriptionManager. This involves initializing the RMR xApp, creating a SubscriptionManager instance, retrieving lists of eNBs and gNBs, and sending subscription requests. ```python from ricxappframe.xapp_frame import RMRXapp from src.manager.SubscriptionManager import SubscriptionManager from src.utils.constants import Constants import json # Initialize RMR xApp rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, rmr_port=4560 ) # Create subscription manager sub_mgr = SubscriptionManager(rmr_xapp) # Get list of eNBs from RIC enb_list = sub_mgr.get_enb_list() # Returns: [{'inventory_name': 'enb_001', ...}, {'inventory_name': 'enb_002', ...}] # Get list of gNBs from RIC gnb_list = sub_mgr.get_gnb_list() # Returns: [{'inventory_name': 'gnb_001', ...}, {'inventory_name': 'gnb_002', ...}] # Send subscription request for each node for enb in enb_list: result = sub_mgr.send_subscription_request(enb) if result: print(f"Error subscribing to {enb}: {result}") # Subscription request format sent to submgr service: subscription_payload = { "xnb_id": "enb_001", "action_type": "REPORT" # Constants.ACTION_TYPE } # POST to: http://service-ricplt-submgr-http:3800 # With proper error handling for HTTP, connection, and timeout errors ``` -------------------------------- ### Check Onboarding Status Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst Verify that the hw-python xApp has been successfully onboarded by querying the available charts. ```APIDOC ## Check Onboarding Status ### Description Retrieve a list of all onboarded xApp charts, including hw-python, to confirm successful onboarding. ### Method GET ### Endpoint `http://:32080/onboard/api/v1/charts` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```bash $ curl --location --request GET "http://:32080/onboard/api/v1/charts" --header 'Content-Type: application/json' ``` ### Response #### Success Response (200) Returns a JSON object containing details of the onboarded charts. - **hw-python** (array) - Details of the hw-python chart. - **name** (string) - The name of the chart. - **version** (string) - The version of the chart. - **description** (string) - A description of the chart. - **apiVersion** (string) - The API version of the chart. - **appVersion** (string) - The application version. - **urls** (array) - List of URLs for the chart package. - **created** (string) - Timestamp of chart creation. - **digest** (string) - SHA256 hash of the chart package. #### Response Example ```json { "hw-python": [ { "name": "hw-python", "version": "1.0.0", "description": "Standard xApp Helm Chart", "apiVersion": "v1", "appVersion": "1.0", "urls": [ "charts/hw-python-1.0.0.tgz" ], "created": "2021-07-05T15:07:34.518377486Z", "digest": "e9db874d35154643a2c6f26dd52929c9dcf143f165683c03d07518bb0c2d768d" } ] } ``` ``` -------------------------------- ### Extract hw-python Chart Files Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/docs/onboard-and-deploy.rst Extracts the contents of the downloaded hw-python.tgz file. This reveals the Chart.yaml, values.yaml, and template files necessary for deployment. ```bash tar -xvzf hw-python.tgz hw-python/Chart.yaml hw-python/values.yaml hw-python/templates/_helpers.tpl hw-python/templates/appconfig.yaml hw-python/templates/appenv.yaml hw-python/templates/deployment.yaml hw-python/templates/service-http.yaml hw-python/templates/service-rmr.yaml hw-python/config/config-file.json hw-python/descriptors/schema.json ``` -------------------------------- ### Initialize and Handle RMR Messages in Python Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt This Python code initializes an RMR xApp, defines a default handler for unregistered message types, and sends an A1 policy query. It also includes a custom handler for RMR responses. ```python import rmr import json import time from ric.constants import Constants # Placeholder for RMRXapp and Constants if not imported from a library class RMRXapp: def __init__(self, default_handler, rmr_port): pass def healthcheck(self): return True def rmr_free(self, sbuf): pass def rmr_send(self, payload, mtype): pass def rmr_rts(self, sbuf, new_payload, new_mtype): pass class Constants: A1_POLICY_QUERY = 20012 A1_POLICY_RESP = 20011 # Initialize RMR xApp def default_handler(rmr_xapp, summary, sbuf): """Handle unregistered message types""" msg_type = summary[rmr.RMR_MS_MSG_TYPE] payload = summary[rmr.RMR_MS_PAYLOAD] print(f"Received message type {msg_type}: {payload}") rmr_xapp.rmr_free(sbuf) rmr_xapp = RMRXapp( default_handler=default_handler, rmr_port=4560 ) # Wait for RMR to be ready while not rmr_xapp.healthcheck(): print("Waiting for RMR to be ready...") time.sleep(1) # Send RMR message message_payload = { "policy_type_id": "2", "status": "active" } rmr_xapp.rmr_send( json.dumps(message_payload).encode(), Constants.A1_POLICY_QUERY # Message type 20012 ) # Send RMR response (return to sender) def custom_handler(rmr_xapp, summary, sbuf): request_data = json.loads(summary[rmr.RMR_MS_PAYLOAD]) response = {"status": "OK", "request_id": request_data.get("id")} rmr_xapp.rmr_rts( sbuf, new_payload=json.dumps(response).encode(), new_mtype=Constants.A1_POLICY_RESP # Message type 20011 ) rmr_xapp.rmr_free(sbuf) # Message type constants # A1_POLICY_QUERY = 20012 # A1_POLICY_REQ = 20010 # A1_POLICY_RESP = 20011 # RIC_HEALTH_CHECK_REQ = 100 # RIC_HEALTH_CHECK_RESP = 101 # SUBSCRIPTION_REQ = 12011 ``` -------------------------------- ### SDL Manager - Data Persistence Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Manages read and write operations to the Shared Data Layer (SDL) for persistent storage of network data. This allows the xApp to store and retrieve information reliably. ```APIDOC ## SDL Manager - Data Persistence ### Description Manages read/write operations to the Shared Data Layer for persistent storage of network data. ### Method Provides an interface for interacting with the SDL. ### Endpoint N/A (Internal SDL operations) ### Parameters None ### Request Example ```python from src.manager.SdlManager import SdlManager # Assume rmr_xapp is an initialized RMRXapp instance # rmr_xapp = RMRXapp(...) sdl_mgr = SdlManager(rmr_xapp) # Retrieve gNB list from e2Manager namespace gnb_list = sdl_mgr.sdlGetGnbList() ``` ### Response #### Success Response (Data Retrieval) - Returns data stored in the SDL, such as a list of gNBs. #### Response Example ```json [ "gnb-1", "gnb-2", "gnb-3" ] ``` ``` -------------------------------- ### Handle A1 Policy Requests - Python Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Processes A1 policy requests received via RMR, validates their structure, and sends acknowledgment responses back to the RIC platform. It utilizes the ricxappframe library. ```python from ricxappframe.xapp_frame import RMRXapp from src.handler.A1PolicyHandler import A1PolicyHandler from src.utils.constants import Constants import json # Initialize RMR xApp rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, rmr_port=4560 ) # Register A1 policy handler a1_handler = A1PolicyHandler(rmr_xapp, Constants.A1_POLICY_REQ) # Example A1 policy request (received via RMR): policy_request = { "policy_type_id": 2, "operation": "CREATE", "policy_instance_id": "policy-12345" } # Handler automatically: # 1. Validates required fields (policy_type_id, operation, policy_instance_id) # 2. Builds response with handler_id and status # 3. Sends response via RMR message type A1_POLICY_RESP (20011) # Expected response: policy_response = { "policy_type_id": 2, "policy_instance_id": "policy-12345", "handler_id": "hw-python", "status": "OK" } ``` -------------------------------- ### Download hw-python xApp Chart Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Downloads the Helm chart for the hw-python xApp from the application manager. This command retrieves the deployment files for the xApp. ```bash curl --location --request GET "http://:32080/onboard/api/v1/charts/xapp/hw-python/ver/1.0.0" --header 'Content-Type: application/json' --output hw-python.tgz ``` -------------------------------- ### A1 Policy Handler Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Processes A1 policy requests from the RIC platform, validates the policy structure, and sends acknowledgment responses. It handles policy creation, modification, and deletion requests. ```APIDOC ## A1 Policy Handler ### Description Processes A1 policy requests from the RIC platform, validates policy structure, and sends acknowledgment responses. ### Method Handles incoming A1 Policy Request messages via RMR. ### Endpoint N/A (Internal RMR message handling) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **policy_type_id** (integer) - Required - Identifier for the policy type. - **operation** (string) - Required - The operation to perform (e.g., "CREATE", "UPDATE", "DELETE"). - **policy_instance_id** (string) - Required - Unique identifier for the policy instance. ### Request Example ```json { "policy_type_id": 2, "operation": "CREATE", "policy_instance_id": "policy-12345" } ``` ### Response #### Success Response (A1_POLICY_RESP) - **policy_type_id** (integer) - The policy type ID. - **policy_instance_id** (string) - The policy instance ID. - **handler_id** (string) - Identifier of the handler processing the request (e.g., "hw-python"). - **status** (string) - Status of the operation (e.g., "OK", "FAILED"). #### Response Example ```json { "policy_type_id": 2, "policy_instance_id": "policy-12345", "handler_id": "hw-python", "status": "OK" } ``` ``` -------------------------------- ### SDL Operations with rmr_xapp Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Perform direct operations on the SDL (Service Data Layer) using the rmr_xapp. This includes searching for keys with patterns, writing data, reading data, and finding data based on a namespace and a pattern. It relies on the rmr_xapp object for interactions. ```python # Searches SDL for keys matching "GNB*" pattern # Returns: {'GNB_001': {...}, 'GNB_002': {...}} # Retrieve eNB list from e2Manager namespace enb_list = sdl_mgr.sdlGetEnbList() # Searches SDL for keys matching "ENB*" pattern # Returns: {'ENB_001': {...}, 'ENB_002': {...}} # Direct SDL operations using rmr_xapp namespace = "e2Manager" # Write data to SDL rmr_xapp.sdl_set(namespace, {"key1": "value1", "key2": "value2"}) # Read data from SDL data = rmr_xapp.sdl_get(namespace, ["key1", "key2"]) # Returns: {"key1": "value1", "key2": "value2"} # Find and get with pattern matching results = rmr_xapp.sdl_find_and_get(namespace, "GNB") print(json.dumps(results, indent=2)) ``` -------------------------------- ### Clone hw-python Repository Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Clones the hw-python xApp repository from the Gerrit repository using git. This is the first step to obtain the necessary files for onboarding. ```bash git clone "https://gerrit.o-ran-sc.org/r/ric-app/hw-python" ``` -------------------------------- ### Deploy hw-python xApp using Helm Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Deploys the hw-python xApp to the RIC platform using a curl command to the application manager's deployment API. This command specifies the xApp name and its Helm version. ```bash $ curl --location --request POST "http://:32080/appmgr/ric/v1/xapps" --header 'Content-Type: application/json' --data-raw '{"xappName": "hw-python", "helmVersion": "1.0.0"}' {"instances":null,"name":"hw-python","status":"deployed","version":"1.0"} ``` -------------------------------- ### Configure CHART_REPO_URL for Chartmuseme Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Sets the CHART_REPO_URL environment variable to point to the chartmuseme service. This is a prerequisite for the dms_cli tool to interact with the chart repository. ```bash $export CHART_REPO_URL=http://:8080 ``` -------------------------------- ### Verify hw-python Onboarding Status Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Checks if the hw-python xApp has been successfully onboarded by querying the application manager's API for onboarded charts. The response lists the details of the onboarded chart. ```bash $ curl --location --request GET "http://:32080/onboard/api/v1/charts" --header 'Content-Type: application/json' { "hw-python": [ { "name": "hw-python", "version": "1.0.0", "description": "Standard xApp Helm Chart", "apiVersion": "v1", "appVersion": "1.0", "urls": [ "charts/hw-python-1.0.0.tgz" ], "created": "2021-07-05T15:07:34.518377486Z", "digest": "e9db874d35154643a2c6f26dd52929c9dcf143f165683c03d07518bb0c2d768d" } ], "hw-python": [ { "name": "hw-python", "version": "1.0.0", "description": "Standard xApp Helm Chart", "apiVersion": "v1", "appVersion": "1.0", "urls": [ "charts/hw-python-1.0.0.tgz" ], "created": "2021-07-05T15:20:13.965653743Z", "digest": "975b1da1f8669e8ed1b1e5be809e7cf4841ef33abcb88207bc3a735e9b543a9a" } ] } ``` -------------------------------- ### Handle Health Checks - Python Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Responds to RIC platform health check requests by verifying RMR and SDL connectivity. It sends back a status indicating 'OK' or 'ERROR'. ```python from ricxappframe.xapp_frame import RMRXapp from src.handler.HealthCheckHandler import HealthCheckHandler from src.utils.constants import Constants # Initialize RMR xApp rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, rmr_port=4560, use_fake_sdl=False ) # Register health check handler health_handler = HealthCheckHandler(rmr_xapp, Constants.RIC_HEALTH_CHECK_REQ) # Handler receives RIC_HEALTH_CHECK_REQ (100) messages # Performs healthcheck via rmr_xapp.healthcheck() # Sends RIC_HEALTH_CHECK_RESP (101) with: # - "OK\n" if RMR and SDL are healthy # - "ERROR [RMR or SDL is unhealthy]\n" if checks fail # Test health check programmatically is_healthy = rmr_xapp.healthcheck() print(f"xApp health status: {'OK' if is_healthy else 'ERROR'}") ``` -------------------------------- ### List Deployed xApps Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Retrieves a list of all deployed xApps within the RIC platform, including details about their versions, statuses, and running instances. ```APIDOC ## GET /ric/v1/xapps ### Description Retrieves a list of all deployed xApps, including details about their versions, statuses, and running instances. ### Method GET ### Endpoint `/ric/v1/xapps` ### Parameters #### Query Parameters None ### Request Example ``` GET /ric/v1/xapps HTTP/1.1 Host: service-ricplt-appmgr-http.ricplt:8080 ``` ### Response #### Success Response (200) - **instances** (array) - List of running instances for the xApp. - **ip** (string) - The IP address of the instance. - **name** (string) - The name of the instance. - **policies** (array) - List of policies associated with the instance. - **port** (integer) - The port the instance is listening on. - **rxMessages** (array) - List of messages the instance receives. - **status** (string) - The current status of the instance (e.g., "running"). - **txMessages** (array) - List of messages the instance transmits. - **name** (string) - The name of the xApp. - **status** (string) - The deployment status of the xApp (e.g., "deployed"). - **version** (string) - The version of the xApp. #### Response Example ```json [ { "instances": [ { "ip": "service-ricxapp-hw-python-rmr.ricxapp", "name": "hw-python-55ff7549df-kpj6k", "policies": [ 1 ], "port": 4560, "rxMessages": [ "RIC_SUB_RESP", "A1_POLICY_REQ", "RIC_HEALTH_CHECK_REQ" ], "status": "running", "txMessages": [ "RIC_SUB_REQ", "A1_POLICY_RESP", "A1_POLICY_QUERY", "RIC_HEALTH_CHECK_RESP" ] } ], "name": "hw-python", "status": "deployed", "version": "1.0" } ] ``` ``` -------------------------------- ### Send Metrics using MetricManager Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Collect and send metrics to the RIC platform metrics collector service using the MetricManager. This includes sending a timestamp metric with a predefined name and sending custom metric data. Metrics are exposed on an HTTP endpoint and registered with Prometheus. ```python from ricxappframe.xapp_frame import RMRXapp from src.manager.MetricManager import MetricManager from datetime import datetime # Initialize RMR xApp rmr_xapp = RMRXapp( default_handler=lambda x, y, z: None, rmr_port=4560 ) # Create metric manager with custom metric name and app name metric_mgr = MetricManager(rmr_xapp) # Send timestamp metric metric_mgr.send_metric() # Sends current timestamp in format: "13/01/2026 04:17:23" # Metric name: "system-time" # App name: "hw-python" # Custom metric collection example from ricxappframe.metric import metric # Create custom metrics manager custom_metrics = metric.MetricsManager(rmr_xapp._mrc, "custom-metric", "hw-python") # Send custom metric data metric_list = ["value1", "value2", "value3"] custom_metrics.send_metrics(metric_list) # Metrics are exposed on HTTP endpoint: /ric/v1/metrics # And registered with Prometheus-style counters for SDL, RMR operations ``` -------------------------------- ### Check hw-python Pod Status Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Verifies the deployment status of the hw-python xApp by checking the status of its pods in the 'ricxapp' namespace using kubectl. This confirms that the xApp instances are running. ```bash # kubectl get pods -n ricxapp NAME READY STATUS RESTARTS AGE ricxapp-hw-python-64b5447dcc-mbt5w 1/1 Running 0 5m45s ``` -------------------------------- ### Health Check Handler Source: https://context7.com/o-ran-sc/ric-app-hw-python/llms.txt Responds to RIC platform health check requests by verifying the connectivity and status of RMR (RIC Message Router) and SDL (Shared Data Layer). It provides a mechanism for the RIC to monitor the health of the xApp. ```APIDOC ## Health Check Handler ### Description Responds to RIC platform health check requests by verifying RMR and SDL connectivity. ### Method Handles incoming RIC Health Check Request messages (type 100) via RMR. ### Endpoint N/A (Internal RMR message handling) ### Parameters None ### Request Example N/A (Triggered by RIC platform) ### Response #### Success Response (RIC_HEALTH_CHECK_RESP - type 101) - **Status**: "OK\n" if RMR and SDL are healthy. #### Error Response (RIC_HEALTH_CHECK_RESP - type 101) - **Status**: "ERROR [RMR or SDL is unhealthy]\n" if health checks fail. #### Response Example ``` OK ``` OR ``` ERROR [RMR or SDL is unhealthy] ``` ``` -------------------------------- ### Check hw-python Service Status Source: https://github.com/o-ran-sc/ric-app-hw-python/blob/master/README.md Displays the cluster IP and port information for the services associated with the hw-python xApp, including HTTP and RMR communication ports. This helps in verifying network accessibility for the deployed xApp. ```bash CLUSTER-IP EXTERNAL-IP PORT(S) AGE aux-entry ClusterIP 10.111.35.76 80/TCP,443/TCP 9d service-ricxapp-hw-python-http ClusterIP 10.104.223.245 8080/TCP 6m23s service-ricxapp-hw-python-rmr ClusterIP 10.103.243.21 4560/TCP,4561/TCP 6m23s ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.