### Configure and Start Response Handler

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Starts the response handler for incoming requests, optionally setting a custom callback function. If no callback is provided, a default handler is used. It returns a boolean indicating success or failure.

```python
from ricxappframe.xapp_rest import ThreadedHTTPServer

def custom_response_handler(data):
    print("Custom handler received:", data)

# Example usage with default handler:
success_default = xapp_instance.ResponseHandler()

# Example usage with custom handler:
success_custom = xapp_instance.ResponseHandler(responseCB=custom_response_handler)

print(f"Default handler started: {success_default}")
print(f"Custom handler started: {success_custom}")
```

--------------------------------

### rmr_ready

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Checks if a routing table has been received and installed.

```APIDOC
## rmr_ready

### Description
Checks if a routing table has been received and installed.

### Method
`rmr_ready(vctx: c_void_p) -> int`

### C Signature
```c
extern int rmr_ready(void* vctx)
```

### Parameters

#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters

*   **vctx** (c_void_p) - Pointer to the RMR context.

### Request Example
```python
# Example usage (assuming rmr_context is a valid pointer obtained from rmr_init)
# is_ready = rmr.rmr_ready(rmr_context)
```

### Response

#### Success Response (int)

*   **int**: Returns 1 if the routing table is ready, 0 otherwise.

#### Response Example
```json
{
  "is_ready": 1
}
```
```

--------------------------------

### Retrieve Data using SDLWrapper in Python

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Retrieves data from SDL using a specified namespace and key. The 'get' method can optionally deserialize the stored bytes using msgpack if they were originally stored with msgpack enabled.

```python
from ricxappframe.xapp_sdl import SDLWrapper

sdl_wrapper = SDLWrapper()

# Get a value, attempting msgpack deserialization (default)
value = sdl_wrapper.get("my_namespace", "my_key")

# Get raw bytes without deserialization
byte_value = sdl_wrapper.get("my_namespace", "my_bytes_key", usemsgpack=False)

# Handle cases where the key might not be found
non_existent_value = sdl_wrapper.get("my_namespace", "non_existent_key")
if non_existent_value is None:
    print("Key not found.")

```

--------------------------------

### Alarm Management with Python

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

Provides the setup for sending alarms to the Alarm Adapter via RMR for network fault management. This snippet initializes the AlarmManager and related classes for defining and transmitting alarm details. Requires the 'ricxappframe' library.

```python
from ricxappframe.xapp_frame import Xapp
from ricxappframe.alarm.alarm import AlarmManager, AlarmDetail, AlarmSeverity
import os

```

--------------------------------

### Check RMR Ready State (Python)

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Checks if the RMR routing table has been received and installed. It takes the RMR context as input and returns an integer indicating readiness.

```python
from ctypes import c_void_p

def rmr_ready(vctx: c_void_p) -> int:
    """
    Checks if a routing table has been received and installed.
    Refer to RMR C documentation for method:

    extern int rmr_ready(void* vctx)

    Parameters:
      vctx: ctypes c_void_p
        Pointer to RMR context
    Returns:
      1 for yes, 0 for no
    """
    # Actual implementation would involve calling the C library function
    pass # Replace with actual ctypes call
```

--------------------------------

### SDL Key-Value Operations

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Provides endpoints for finding, getting, deleting, and conditionally deleting key-value pairs within the SDL.

```APIDOC
## GET /o-ran-sc/ric-plt-xapp-frame-py/find_and_get

### Description
Gets all key-value pairs in the specified namespace with keys that start with the specified prefix, optionally deserializing stored bytes using msgpack.

### Method
GET

### Endpoint
/o-ran-sc/ric-plt-xapp-frame-py/find_and_get

### Parameters
#### Query Parameters
- **ns** (string) - Required - SDL namespace
- **prefix** (string) - Required - The key prefix
- **usemsgpack** (boolean) - Optional (default is True) - If True, deserializes stored bytes using msgpack. If False, returns byte arrays without further processing.

### Response
#### Success Response (200)
- **dict** - Dictionary of key-value pairs. Keys have the specified prefix. Value types depend on `usemsgpack`.

#### Response Example
```json
{
  "key1": "value1",
  "key2": "value2"
}
```

## DELETE /o-ran-sc/ric-plt-xapp-frame-py/delete

### Description
Deletes the key-value pair with the specified key in the specified namespace.

### Method
DELETE

### Endpoint
/o-ran-sc/ric-plt-xapp-frame-py/delete

### Parameters
#### Query Parameters
- **ns** (string) - Required - SDL namespace
- **key** (string) - Required - SDL key

### Response
#### Success Response (200)
- **bool** - Indicates if the deletion was successful.

#### Response Example
```json
true
```

## DELETE /o-ran-sc/ric-plt-xapp-frame-py/delete_if

### Description
Conditionally removes data from SDL storage if the current data value matches the user’s last known value.

### Method
DELETE

### Endpoint
/o-ran-sc/ric-plt-xapp-frame-py/delete_if

### Parameters
#### Query Parameters
- **ns** (string) - Required - SDL namespace
- **key** (string) - Required - SDL key
- **value** - Required - Object or byte array to compare against. See `usemsgpack` parameter.
- **usemsgpack** (boolean) - Optional (default is True) - Determines whether the value is serialized using msgpack before storing. If True, value can be anything serializable by msgpack. If False, value must be bytes.

### Response
#### Success Response (200)
- **bool** - True if successful removal, false if the user’s last known data did not match the current value.

#### Response Example
```json
true
```
```

--------------------------------

### Query RAN Topology Information from RNIB using Python

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

Shows how to access the RAN Information Base (RNIB) to retrieve various RAN node and cell details. This includes listing gNB and eNodeB identities, getting detailed node information by inventory name or global ID, and retrieving cell lists or specific cells by ID or PCI. It utilizes the 'ricxappframe' library and assumes an RMR context.

```Python
from ricxappframe.xapp_frame import RMRXapp
from ricxappframe.entities.rnib.nb_identity_pb2 import NbIdentity
from ricxappframe.entities.rnib import cell_pb2

def process_ran_topology(self, summary, sbuf):
    """Access RAN node and cell information"""

    # Get all gNodeB identities
    gnb_list = self.get_list_gnb_ids()
    for gnb_id in gnb_list:
        self.logger.info(f"gNodeB: {gnb_id.inventory_name}")
        self.logger.info(f"  Global NB ID: {gnb_id.global_nb_id.plmn_id}")

    # Get all eNodeB identities
    enb_list = self.get_list_enb_ids()
    for enb_id in enb_list:
        self.logger.info(f"eNodeB: {enb_id.inventory_name}")

    # Get all node identities (both ENB and GNB)
    all_nodes = self.GetListNodebIds()

    # Get detailed NodeB information
    inventory_name = "gnb_001_001_000001"
    nodeb_info = self.GetNodeb(inventory_name)
    if nodeb_info:
        self.logger.info(f"Node type: {nodeb_info.WhichOneof('node')}")
        self.logger.info(f"Connection status: {nodeb_info.connection_status}")
        self.logger.info(f"RAN functions: {len(nodeb_info.gnb.ran_functions)}")

    # Get NodeB by global identifier
    node = self.GetNodebByGlobalNbId(
        nodeType="GNB",
        plmnId="131014",
        nbId="000001"
    )
    if node:
        self.logger.info(f"Found node: {node.inventory_name}")

    # Get cell list for a node
    cells = self.GetCellList(inventory_name)
    if cells:
        for cell in cells:
            self.logger.info(f"Cell ID: {cell.cell_id}")
            self.logger.info(f"  PCI: {cell.served_nr_cell_information.cell_id}")

    # Get cell by ID
    cell = self.GetCellById(
        cell_type=cell_pb2.Cell.Type.Name(cell_pb2.Cell.NR_CELL),
        cell_id="ncgi:131014:000001:000001"
    )
    if cell:
        self.logger.info(f"Cell type: {cell.type}")

    # Get cell by PCI
    cell_by_pci = self.GetCell(inventory_name, pci=1)

    # Get RAN function definitions
    ran_functions = self.GetRanFunctionDefinition(
        inventory_name=inventory_name,
        ran_function_oid=1
    )
    for ran_func_def in ran_functions:
        self.logger.info(f"RAN Function: {ran_func_def[:50]}")

    self.rmr_free(sbuf)

xapp = RMRXapp(default_handler=process_ran_topology, rmr_port=4562)
xapp.run()
```

--------------------------------

### Xapp Initialization and Configuration

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Details on initializing the Xapp framework, including RMR port configuration, readiness checks, fake SDL usage, and config file watching.

```APIDOC
## _BaseXapp Class Initialization

### Description
Initializes the RMR connection, starts message listening threads, provisions an SDL object, and optionally sets up a config-file watcher.

### Method
`__init__` (constructor)

### Parameters
- **rmr_port** (int, optional, default=4562) - Port for RMR to listen on.
- **rmr_wait_for_ready** (bool, optional, default=True) - If True, waits for RMR to be ready to send. Set to False for receive-only clients.
- **use_fake_sdl** (bool, optional, default=False) - If True, uses a fake dictionary backend for SDL instead of Redis. Useful for development and testing.
- **post_init** (function, optional, default=None) - A user-provided function to be called after initialization. Signature: `post_init(self)`.

### Config File Watching
If the `CONFIG_FILE` environment variable is set to an existing file path, a watcher monitors the file for modifications using inotify. This watcher must be polled by calling the `config_check()` method.
```

--------------------------------

### Get MDCLogger Logging Level

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/mdclogger.md

API to retrieve the current logging level of an MDCLogger instance.

```python
def get_level(self) -> Level:
```

--------------------------------

### Xapp Class

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Represents a generic Xapp where the client provides a single entrypoint function for the framework to call at startup.

```APIDOC
## Class Xapp

### Description
Represents a generic Xapp where the client provides a single function for the framework to call at startup time (instead of providing callback functions by message type). The Xapp writer must implement and provide a function with a loop-forever construct similar to the `run` function in the `RMRXapp` class. That function should poll to retrieve RMR messages and dispatch them appropriately, poll for configuration changes, etc.

### Parameters

*   **entrypoint** (function) - This function is called when the `Xapp` class’s `run` method is invoked. The function signature must be just `function(self)`.
*   **rmr_port** (integer, optional, default is 4562) - Initialize RMR to listen on this port.
*   **rmr_wait_for_ready** (boolean, optional, default is True) - Wait for RMR to signal ready before starting the dispatch loop.
*   **use_fake_sdl** (boolean, optional, default is False) - Use an in-memory store instead of the real SDL service.

```

--------------------------------

### SDL Data Storage Operations with Msgpack in Python

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

Illustrates how to use the Shared Data Layer (SDL) wrapper for storing and retrieving data with msgpack serialization. Supports basic key-value operations, finding keys by prefix, conditional updates, and group operations (sets). Requires the 'ricxappframe' library and potentially a Redis instance if use_fake_sdl is False.

```python
from ricxappframe.xapp_sdl import SDLWrapper
import time

# Initialize SDL (use_fake_sdl=True for testing without Redis)
sdl = SDLWrapper(use_fake_sdl=False)

# Basic key-value operations
namespace = "traffic_steering"

# Store data with automatic msgpack serialization
metrics = {
    "throughput": 1500.5,
    "latency": 12.3,
    "packet_loss": 0.02,
    "timestamp": time.time()
}
sdl.set(namespace, "cell_123_metrics", metrics, usemsgpack=True)

# Retrieve data
stored_metrics = sdl.get(namespace, "cell_123_metrics", usemsgpack=True)
print(f"Retrieved metrics: {stored_metrics}")

# Find keys by prefix
sdl.set(namespace, "cell_123_config", {"power": 20})
sdl.set(namespace, "cell_456_config", {"power": 25})
all_configs = sdl.find_and_get(namespace, "cell_", usemsgpack=True)
# Returns: {"cell_123_config": {...}, "cell_456_config": {...}}

# Conditional operations
old_value = {"counter": 10}
new_value = {"counter": 11}
updated = sdl.set_if(namespace, "counter_key", old_value, new_value, usemsgpack=True)
if updated:
    print("Conditional update succeeded")

# Set only if key doesn't exist
created = sdl.set_if_not_exists(namespace, "init_flag", {"initialized": True})

# Delete operations
sdl.delete(namespace, "temp_key")
deleted = sdl.delete_if(namespace, "counter_key", new_value, usemsgpack=True)

# Working with SDL groups (sets)
sdl.add_member(namespace, "active_cells", "cell_123", usemsgpack=True)
sdl.add_member(namespace, "active_cells", "cell_456", usemsgpack=True)
members = sdl.get_members(namespace, "active_cells", usemsgpack=True)
is_active = sdl.is_member(namespace, "active_cells", "cell_123", usemsgpack=True)
size = sdl.group_size(namespace, "active_cells")

# Pub/Sub pattern
sdl.set_and_publish(namespace, "events", "update", "status", {"state": "running"})
sdl.subscribe_channel(namespace, lambda ch, msg: print(f"{ch}: {msg}"), "events")
sdl.start_event_listener()

```

--------------------------------

### RMR Message Transaction ID Functions

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Provides functions for getting and setting transaction IDs for RMR messages.

```APIDOC
## GET /rmr/xaction

### Description
Retrieves the transaction ID from an RMR message buffer.

### Method
GET

### Endpoint
/rmr/xaction

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```json
{
  "ptr_mbuf": "<ctypes.c_void_p>"
}
```

### Response
#### Success Response (200)
- **transaction_id** (bytes) - The transaction ID of the message.

#### Response Example
```json
{
  "transaction_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef"
}
```

## POST /rmr/generate-transaction-id

### Description
Generates a new UUID and sets it as the transaction ID for an RMR message buffer.

### Method
POST

### Endpoint
/rmr/generate-transaction-id

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **ptr_mbuf** (ctypes.c_void_p) - Required - Pointer to an RMR message buffer.

### Request Example
```json
{
  "ptr_mbuf": "<ctypes.c_void_p>"
}
```

### Response
#### Success Response (200)
- **message** (string) - Confirmation message.

#### Response Example
```json
{
  "message": "Transaction ID generated and set."
}
```

## PUT /rmr/transaction-id

### Description
Sets a specific transaction ID for an RMR message buffer.

### Method
PUT

### Endpoint
/rmr/transaction-id

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **ptr_mbuf** (ctypes.c_void_p) - Required - Pointer to an RMR message buffer.
- **tid_bytes** (bytes) - Required - The transaction ID to set (as bytes).

### Request Example
```json
{
  "ptr_mbuf": "<ctypes.c_void_p>",
  "tid_bytes": "a1b2c3d4-e5f6-7890-1234-567890abcdef"
}
```

### Response
#### Success Response (200)
- **message** (string) - Confirmation message.

#### Response Example
```json
{
  "message": "Transaction ID set successfully."
}
```
```

--------------------------------

### Initialize SDLWrapper in Python

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Instantiates the SDLWrapper class to enable communication with the SDL service. It can optionally use a fake SDL implementation for testing and supports msgpack for serialization.

```python
from ricxappframe.xapp_sdl import SDLWrapper

# Initialize SDLWrapper with default settings (uses real SDL, msgpack enabled by default)
sdl_wrapper = SDLWrapper()

# Initialize SDLWrapper using a fake SDL implementation
fake_sdl_wrapper = SDLWrapper(use_fake_sdl=True)

# Initialize SDLWrapper without msgpack support (values must be bytes)
no_msgpack_wrapper = SDLWrapper(usemsgpack=False)

```

--------------------------------

### Compile .proto files for RNIB using protoc and sed (Bash)

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rnib.md

This script compiles .proto files for the RNIB using the protoc compiler. It specifies the output directory for Python files and the path to the .proto definitions. A subsequent `sed` command modifies the generated Python files to use relative imports, addressing a common protobuf issue.

```bash
PYTHON_OUT="ricxappframe/entities/rnib"
RNIB_PROTO="nodeb-rnib/entities"
protoc --python_out="${PYTHON_OUT}" \
         --proto_path="${RNIB_PROTO}" \
         $(find "${RNIB_PROTO}" -iname "*.proto" -printf "%f ")
sed -i -E 's/^import.*_pb2/from . \0/' ${PYTHON_OUT}/*_pb2.py
```

--------------------------------

### Get MDC Value from MDCLogger

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/mdclogger.md

Retrieves the MDC (Mapped Diagnostic Context) value associated with a given key from the logger instance. Returns None if the key is not found.

```python
def get_mdc(self, key: str) -> Value:
```

--------------------------------

### Subscribe to a Service using SubscriptionParams

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Initiates a subscription request to a service using structured subscription data. It takes a SubscriptionParams object as input and returns a JSON string representing the SubscriptionResponse.

```python
from ric.subsclient import SubscriptionParams, SubscriptionResponse

# Example usage:
subs_params = SubscriptionParams(...)
response = xapp_instance.Subscribe(subs_params=subs_params)
print(response)
```

--------------------------------

### Manage Alarms with Python Alarm Manager

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

Demonstrates setting up environment variables for the Alarm Manager and using the AlarmManager class to create, raise, clear, and re-raise alarms. It also shows how to create alarms directly using AlarmDetail objects and clear all alarms for a managed object. This requires the 'ricxappframe' library.

```Python
import os
import time
from ricxappframe.xapp_frame import Xapp
from ricxappframe.xapp_frame import AlarmManager, AlarmSeverity, AlarmDetail

# Set environment variables for Alarm Manager
os.environ["ALARM_MGR_SERVICE_NAME"] = "service-ricplt-alarmmanager-rmr"
os.environ["ALARM_MGR_SERVICE_PORT"] = "4560"

def run_with_alarms(self):
    # Initialize Alarm Manager with RMR context
    alarm_mgr = AlarmManager(
        vctx=self._mrc,
        managed_object_id="cell_site_001",
        application_id="traffic_steering_xapp"
    )

    # Create alarm using AlarmManager convenience method
    alarm = alarm_mgr.create_alarm(
        specific_problem=8004,
        perceived_severity=AlarmSeverity.MAJOR,
        identifying_info="High packet loss detected",
        additional_info="Cell 123: packet loss 15% over 5 minutes"
    )

    # Raise alarm
    success = alarm_mgr.raise_alarm(alarm)
    if success:
        self.logger.info("Alarm raised successfully")

    # Create alarm directly with AlarmDetail
    critical_alarm = AlarmDetail(
        managed_object_id="cell_site_001",
        application_id="traffic_steering_xapp",
        specific_problem=8001,
        perceived_severity=AlarmSeverity.CRITICAL,
        identifying_info="Connection lost to RAN node",
        additional_info="Unable to reach eNodeB for 60 seconds"
    )
    alarm_mgr.raise_alarm(critical_alarm)

    # Clear specific alarm (must match identifying info)
    time.sleep(10)
    cleared_alarm = alarm_mgr.create_alarm(
        specific_problem=8004,
        perceived_severity=AlarmSeverity.CLEARED,
        identifying_info="High packet loss detected",
        additional_info="Packet loss returned to normal"
    )
    alarm_mgr.clear_alarm(cleared_alarm)

    # Re-raise alarm (clear then raise again)
    alarm_mgr.reraise_alarm(alarm)

    # Clear all alarms from this managed object
    alarm_mgr.clear_all_alarms()

xapp = Xapp(entrypoint=run_with_alarms, rmr_port=4562)
xapp.run()
```

--------------------------------

### Get RMR Message Payload

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Extracts the binary payload from an RMR message buffer. This function takes a pointer to the message buffer and returns the payload as a bytes object.

```python
import ctypes

def get_payload(ptr_mbuf: ctypes.c_void_p) -> bytes:
    """
    Gets the binary payload from the rmr_buf_t*.
    
    Parameters:
        ptr_mbuf: ctypes c_void_p - Pointer to an rmr message buffer
    Returns:
        bytes: the message payload
    """
    pass

```

--------------------------------

### Configuration Management API

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

This API enables xApps to monitor and react to changes in configuration files. It utilizes inotify integration for real-time file change detection.

```APIDOC
## Configuration File Monitoring

### Description
Allows xApps to monitor configuration files for changes and react accordingly. The `CONFIG_FILE` environment variable must be set to the path of the configuration file.

### Usage
The `config_handler` function is called when the configuration file is modified. The `message_handler` should periodically check for config changes using `self.config_check()`.

### Example Configuration File (`config.json`)
```json
{
  "log_level": "INFO",
  "polling_interval": 10,
  "endpoints": [
    "http://example.com/api1",
    "http://example.com/api2"
  ]
}
```

### Handlers
- **`config_change_handler(self, config)`**: This function is automatically called by the framework when a configuration file change is detected. It receives the updated configuration as a dictionary.
- **`message_handler(self, summary, sbuf)`**: This is the default message handler. It should include a call to `self.config_check(timeout=0)` to non-blockingly check for configuration changes. Detected changes trigger the `config_change_handler`.

### Environment Variable
- **`CONFIG_FILE`**: Set this environment variable to the absolute path of the configuration file (e.g., `/opt/xapp/config/config.json`).

### Initialization Example
```python
import os
from ricxappframe.xapp_frame import RMRXapp

def config_change_handler(self, config):
    # Process updated configuration
    pass

def message_handler(self, summary, sbuf):
    self.config_check(timeout=0) # Check for config changes
    self.rmr_free(sbuf)

os.environ["CONFIG_FILE"] = "/path/to/your/config.json"
xapp = RMRXapp(
    default_handler=message_handler,
    config_handler=config_change_handler,
    rmr_port=4562
)
xapp.run()
```
```

--------------------------------

### Get RMR Wormhole State

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Retrieves the current connection state of a wormhole identified by its ID. This function requires the RMR context and the wormhole ID. It returns an integer representing the connection state.

```python
import ctypes

def rmr_wh_state(vctx: ctypes.c_void_p, whid: ctypes.c_int) -> ctypes.c_int:
    """
    Gets the state of the connection associated with the given wormhole (whid).
    Refer to RMR C documentation for method:
    int rmr_wh_state( void* vctx, rmr_whid_t whid )
    
    Parameters:
        vctx: ctypes c_void_p - Pointer to RMR context
        whid: c_int - Wormhole ID returned by rmr_wh_open
    Returns:
        c_int: State of the connection
    """
    pass

```

```c
int rmr_wh_state( void* vctx, rmr_whid_t whid )

```

--------------------------------

### rmr_init

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Initializes the RMR environment for sending and receiving messages.

```APIDOC
## rmr_init

### Description
Prepares the environment for sending and receiving messages.

### Method
`rmr_init(uproto_port: c_char_p, max_msg_size: int, flags: int) -> c_void_p`

### C Signature
```c
extern void* rmr_init(char* uproto_port, int max_msg_size, int flags)
```

### Parameters

#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters

*   **uproto_port** (c_char_p) - Pointer to bytes built from the port number as a string (e.g., `b'4550'`).
*   **max_msg_size** (int) - Maximum message size to receive.
*   **flags** (int) - RMR option flags.

### Request Example
```python
# Example usage (assuming context is obtained elsewhere)
# import ctypes
# rmr_context = rmr.rmr_init(b'4550', 2048, 0)
```

### Response

#### Success Response (c_void_p)

*   **c_void_p**: Pointer to the RMR context.

#### Response Example
```json
{
  "context_pointer": "<memory_address>"
}
```

**Note:** This function raises an exception if the returned context is None.
```

--------------------------------

### Get RMR Payload Size

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Retrieves the number of bytes currently available in the payload of an RMR message buffer. It requires a pointer to the RMR message structure and returns an integer representing the payload size.

```python
from ctypes import c_void_p

def rmr_payload_size(ptr_mbuf: c_void_p) -> int:
    """Gets the number of bytes available in the payload."""
    # Implementation details would be here, calling the underlying C function.
    return 0
```

```c
extern int rmr_payload_size(rmr_mbuf_t* mbuf);
```

--------------------------------

### Get Message Source from RMR Message (Python)

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Copies the message-source information from an RMR message buffer into a provided destination buffer. This Python function is a wrapper for the RMR C function `rmr_get_src`. It requires the message buffer and a character pointer for the destination.

```python
from ricxappframe.rmr.rmr import rmr_get_src

# Assuming ptr_mbuf is a valid RMR message buffer pointer and dest is a ctypes c_char_p buffer
# message_source = rmr_get_src(ptr_mbuf, dest)

```

--------------------------------

### Create General Xapp with Custom Event Loop using Python

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

Illustrates the creation of a general Xapp with a custom event loop using the RIC Xapp Framework. This pattern provides full control over the application's execution flow, including health checks, RMR message sending and receiving, and SDL operations. It depends on the 'ricxappframe' library.

```python
from ricxappframe.xapp_frame import Xapp
import json
import time

def entrypoint(self):
    """Main application loop"""
    my_namespace = "myxapp"
    counter = 0

    while True:
        # Check health status
        is_healthy = self.healthcheck()
        self.logger.info(f"Xapp health: {is_healthy}")

        # Send RMR message
        message = json.dumps({"counter": counter, "timestamp": time.time()})
        success = self.rmr_send(message.encode(), 60000, retries=100)
        if success:
            self.logger.info(f"Sent message {counter}")

        # Store data in SDL
        self.sdl_set(my_namespace, "last_counter", counter)
        stored_value = self.sdl_get(my_namespace, "last_counter")
        self.logger.info(f"Retrieved from SDL: {stored_value}")

        # Process incoming messages
        for (summary, sbuf) in self.rmr_get_messages():
            msg_type = summary[rmr.RMR_MS_MSG_TYPE]
            payload = summary[rmr.RMR_MS_PAYLOAD]
            self.logger.info(f"Received type {msg_type}: {payload}")
            self.rmr_free(sbuf)

        counter += 1
        time.sleep(2)

# Initialize general Xapp
xapp = Xapp(
    entrypoint=entrypoint,
    rmr_port=4564,
    rmr_wait_for_ready=True,
    use_fake_sdl=True
)

# Start Xapp (blocks until entrypoint returns)
xapp.run()
```

--------------------------------

### Get Managed Entity ID from RMR Message (Python)

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/rmr_api.md

Retrieves the Managed Entity ID (MEID) from an RMR message header. This Python function provides a convenient way to access the MEID, acting as a friendly wrapper for the RMR C method `rmr_get_meid`. It requires a pointer to the RMR message buffer.

```python
from ricxappframe.rmr.rmr import rmr_get_meid

# Assuming ptr_mbuf is a valid RMR message buffer pointer
# managed_entity_id = rmr_get_meid(ptr_mbuf)

```

--------------------------------

### Create Reactive RMR Xapp using Python

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

Demonstrates how to create a reactive Xapp using the RIC Xapp Framework. This Xapp handles incoming RMR messages through registered callback functions, allowing for specific message type processing. It requires the 'ricxappframe' library and integrates with RMR for communication.

```python
from ricxappframe.xapp_frame import RMRXapp, rmr
import json

def default_handler(self, summary, sbuf):
    """Called for messages without specific handlers"""
    self.logger.info(f"Received message type: {summary[rmr.RMR_MS_MSG_TYPE]}")
    payload = json.loads(summary[rmr.RMR_MS_PAYLOAD])
    self.logger.info(f"Payload: {payload}")
    self.rmr_free(sbuf)

def handler_60000(self, summary, sbuf):
    """Handler for message type 60000"""
    self.logger.info("Processing message type 60000")
    payload = json.loads(summary[rmr.RMR_MS_PAYLOAD])

    # Send response back to sender
    response = json.dumps({"ACK": payload.get("test_send")}).encode()
    self.rmr_rts(sbuf, new_payload=response, new_mtype=60001, retries=100)
    self.rmr_free(sbuf)

def config_handler(self, config):
    """Called on startup and config file changes"""
    self.logger.info(f"Config updated: {config}")

# Initialize reactive Xapp
xapp = RMRXapp(
    default_handler=default_handler,
    config_handler=config_handler,
    rmr_port=4562,
    rmr_wait_for_ready=True,
    use_fake_sdl=False
)

# Register callback for specific message type
xapp.register_callback(handler_60000, 60000)

# Start processing messages (blocks forever)
xapp.run()
```

--------------------------------

### Manage E2 Subscriptions with RIC xApp Frame Python

Source: https://context7.com/o-ran-sc/ric-plt-xapp-frame-py/llms.txt

This snippet demonstrates how to create, query, and unsubscribe from E2 events using the Subscription Manager REST API provided by the RIC xApp framework. It configures the subscription client, defines subscription parameters including client endpoints, event triggers, and actions, and handles subscription responses and potential errors. Dependencies include 'ricxappframe.subsclient' and 'ricxappframe.xapp_frame'.

```python
from ricxappframe.subsclient import *
from ricxappframe.xapp_frame import RMRXapp
import time

def setup_subscriptions(self):
    """Create E2 subscriptions via REST API"""

    # Configure subscription client
    config = Configuration()
    config.host = "http://service-ricplt-submgr-http.ricplt:8088/ric/v1"
    api_client = ApiClient(configuration=config)
    sub_api = CommonApi(api_client)

    # Define subscription parameters
    client_endpoint = SubscriptionParamsClientEndpoint(
        host="service-ricxapp-myxapp-http.ricxapp",
        http_port=8080,
        rmr_port=4562
    )

    # Define event trigger (E2SM byte array)
    event_trigger = EventTriggerDefinition([1, 2, 3, 4, 5])

    # Define action
    action = ActionToBeSetup(
        action_id=1,
        action_type="report",
        action_definition=ActionDefinition([10, 20, 30]),
        subsequent_action=SubsequentAction(
            subsequent_action_type="continue",
            time_to_wait="w10ms"
        )
    )

    # Create subscription detail
    subscription_detail = SubscriptionDetail(
        xapp_event_instance_id=1001,
        event_triggers=event_trigger,
        action_to_be_setup_list=ActionsToBeSetup([action])
    )

    # Define E2 subscription directives
    e2_directives = SubscriptionParamsE2SubscriptionDirectives(
        e2_timeout_timer_value=2,
        e2_retry_count=3,
        rmr_routing_needed=True
    )

    # Create subscription request
    subscription_params = SubscriptionParams(
        client_endpoint=client_endpoint,
        meid="gnb_001_001_000001",
        ran_function_id=1,
        e2_subscription_directives=e2_directives,
        subscription_details=SubscriptionDetailsList([subscription_detail])
    )

    try:
        # Subscribe
        response = sub_api.subscribe(subscription_params)
        subscription_id = response.subscription_id
        self.logger.info(f"Subscription created: {subscription_id}")

        for instance in response.subscription_instances:
            self.logger.info(f"  Instance {instance.xapp_event_instance_id}")
            self.logger.info(f"    E2 Instance: {instance.e2_event_instance_id}")
            if instance.error_cause:
                self.logger.error(f"    Error: {instance.error_cause}")

        # Query all subscriptions
        all_subs = sub_api.get_all_subscriptions()
        for sub in all_subs:
            self.logger.info(f"Subscription {sub.subscription_id}: {sub.meid}")

        # Wait for events
        time.sleep(60)

        # Unsubscribe
        sub_api.unsubscribe(subscription_id)
        self.logger.info(f"Unsubscribed: {subscription_id}")

    except Exception as e:
        self.logger.error(f"Subscription error: {e}")

xapp = RMRXapp(default_handler=lambda s, sum, buf: s.rmr_free(buf), rmr_port=4562)
xapp.run(thread=True)
setup_subscriptions(xapp)

```

--------------------------------

### Initialize and Use MDCLogger in Python

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/mdclogger.md

Demonstrates how to import, initialize, and use the MDCLogger for logging error messages in a Python application. The logger can be configured to monitor configuration maps for dynamic log level changes.

```python
from ricappframe.logger.mdclogger import MDCLogger
my_logger = MDCLogger()
my_logger.mdclog_format_init(configmap_monitor=True)
my_logger.error("This is an error log")
```

--------------------------------

### Find Keys using SDLWrapper in Python

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Finds all keys within a given namespace that match a specified prefix. This function is useful for iterating over related data stored in SDL.

```python
from ricxappframe.xapp_sdl import SDLWrapper

sdl_wrapper = SDLWrapper()

# Find all keys in 'my_namespace' that start with 'my_prefix_'
matching_keys = sdl_wrapper.find_keys("my_namespace", "my_prefix_")

print(f"Found keys: {matching_keys}")

```

--------------------------------

### Service Discovery

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Methods for discovering service URLs based on host and service name.

```APIDOC
## Service Discovery

### GET /service

### Description
Retrieves the URL for connecting to a specified service.

### Method
`get_service`

### Parameters
#### Path Parameters
- **host** (string) - The hostname of the service.
- **service** (string) - The name of the service.

### Returns
- **string** - The URL for the service.
```

--------------------------------

### Store Data using SDLWrapper in Python

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/user-guide.md

Stores a key-value pair in SDL. The 'set' method allows optional serialization of the value using msgpack. 'set_if' conditionally updates a key if its current value matches an expected old value. 'set_if_not_exists' writes data only if the key does not already exist.

```python
from ricxappframe.xapp_sdl import SDLWrapper

sdl_wrapper = SDLWrapper()

# Store a simple value (serializes using msgpack by default)
sdl_wrapper.set("my_namespace", "my_key", "my_value")

# Store raw bytes (msgpack disabled)
sdl_wrapper.set("my_namespace", "my_bytes_key", b"\x01\x02\x03", usemsgpack=False)

# Conditionally set a value if the current value matches 'old_value'
suc = sdl_wrapper.set_if("my_namespace", "conditional_key", "old_value", "new_value")

# Set a value only if the key does not exist
suc = sdl_wrapper.set_if_not_exists("my_namespace", "unique_key", "initial_value")

```

--------------------------------

### Alarm API Documentation

Source: https://github.com/o-ran-sc/ric-plt-xapp-frame-py/blob/master/docs/alarm_api.md

Documentation for the RIC Alarm API, including classes for managing alarms.

```APIDOC
## RIC Alarm API

Provides classes and methods to define, raise, reraise and clear alarms. All actions are implemented by sending RMR messages to the Alarm Adapter.

### Class: ricxappframe.alarm.alarm.AlarmAction

**Description:** Action to perform at the Alarm Adapter.

### Class: ricxappframe.alarm.alarm.AlarmSeverity

**Description:** Severity of an alarm.

### Class: ricxappframe.alarm.alarm.AlarmDetail

**Description:** An alarm that can be raised or cleared.

**Parameters:**
- **managed_object_id** (str) - Required - The name of the managed object that is the cause of the fault
- **application_id** (str) - Required - The name of the process that raised the alarm
- **specific_problem** (int) - Required - The problem that is the cause of the alarm
- **perceived_severity** (AlarmSeverity) - Required - The severity of the alarm, a value from the enum.
- **identifying_info** (str) - Required - Identifying additional information, which is part of alarm identity
- **additional_info** (str) - Optional - Additional information given by the application

### Class: ricxappframe.alarm.alarm.AlarmManager

**Description:** Provides an API for an Xapp to raise and clear alarms by sending messages via RMR directly to an Alarm Adapter. Requires environment variables ALARM_MGR_SERVICE_NAME and ALARM_MGR_SERVICE_PORT with the destination host (service) name and port number; raises an exception if not found.

**Parameters:**
- **vctx** (ctypes c_void_p) - Required - Pointer to RMR context obtained by initializing RMR. The context is used to allocate space and send messages.
- **managed_object_id** (str) - Required - The name of the managed object that raises alarms
- **application_id** (str) - Required - The name of the process that raises alarms
```