br_rapp_sdk.a1_services.a1_policy_types

PlmnId Objects

class PlmnId(SnakeModel)

Represents a Public Land Mobile Network Identifier (PLMN ID), as defined in 3GPP TS 23.003.

Attributes:

mcc - Mobile Country Code (MCC), must be exactly 3 digits.
mnc - Mobile Network Code (MNC), must be 2 or 3 digits.

SliceId Objects

class SliceId(SnakeModel)

Represents a network slice identifier (S-NSSAI) as defined in 3GPP TS 23.003.

Attributes:

sst - Slice/Service Type (SST), 0–255. Applicable to 5G RAN.
sd - Slice Differentiator (SD), exactly 6 hexadecimal characters. Required if used.
plmn_id - Public Land Mobile Network Identifier (PLMN ID). Applicable to both 4G and 5G RAN.

QosId Objects

class QosId(SnakeModel)

Represents a QoS Identifier used in 4G or 5G RANs.

Exactly one of qci (4G) or field_5ql (5G) must be set.

Attributes:

qci - QoS Class Identifier for 4G (3GPP TS 23.203), range 1-256.
field_5ql - 5G QoS Identifier (5QI) as per 3GPP TS 23.501, range 1-256.

CId Objects

class CId(SnakeModel)

Represents a Cell Identifier, either for 4G (E-UTRAN) or 5G (NR), as defined in 3GPP TS 23.003.

Exactly one of ec_i (for 4G) or nc_i (for 5G) must be set.

Attributes:

ec_i - E-UTRAN Cell Identifier (28-bit integer, 0-268435455).
nc_i - NR Cell Identifier (36-bit integer, 0-68719476735).

CellId Objects

class CellId(SnakeModel)

Represents a global Cell Identifier based on ECGI (for 4G) or NCGI (for 5G), as defined in 3GPP TS 23.003.

Attributes:

plmn_id - Public Land Mobile Network Identifier (PLMN ID) as per 3GPP TS 23.003.
c_id - Cell Identifier component of the global cell identity.

GroupId Objects

class GroupId(SnakeModel)

Represents a group identifier used to implicitly reference a dynamic set of UEs sharing a common radio resource or subscriber profile.

Only one of the two attributes may be set, depending on the RAN type:

sp_id for 4G RAN (subscriber profile ID, 3GPP TS 36.300)
rfsp_index for 5G RAN (RF selection priority index, 3GPP TS 23.501)

Value range for both is 1-256.

ScopeIdentifier Objects

class ScopeIdentifier(SnakeModel)

Identifies the scope in which a policy statement is applicable.

Attributes:

slice_id - Identifies the network slice the policy applies to.
ue_id - Identifies a specific UE the policy applies to.
group_id - Identifies a group of UEs to which the policy applies.
qos_id - Identifies the QoS flow the policy applies to.
cell_id - Identifies the cell the policy applies to.
targets - (Optional) List of custom target entities, if applicable.
scopes - (Optional) List of named scopes (e.g., for orchestration grouping or policy context).

ReliabilityType Objects

class ReliabilityType(SnakeModel)

Represents the reliability requirement for successful transmission of a data packet of a given size within a user-plane latency constraint.

Attributes:

packet_size - Size of the data packet in bytes.
user_plane_latency - Maximum allowed latency in ms for delivering the packet across the radio interface.
success_probability - Probability (0–1) that the packet is successfully delivered within the latency budget.

QosObjectives Objects

class QosObjectives(SnakeModel)

Represents QoS (Quality of Service) objectives as defined in 3GPP TS 23.501.

Attributes:

gfbr - Guaranteed Flow Bit Rate (GFBR) in kbps. Ensures minimum bandwidth is available.
mfbr - Maximum Flow Bit Rate (MFBR) in kbps. Upper limit of allowed bandwidth.
priority_level - QoS priority level (lower value means higher priority).
pdb - Packet Delay Budget (PDB) in milliseconds. Max tolerated delay for packet delivery.

QoeObjectives Objects

class QoeObjectives(SnakeModel)

Represents QoE (Quality of Experience) objectives for media services.

Attributes:

qoe_score - Mean Opinion Score (MOS) between 1 and 5. Can represent video MOS (e.g., per ITU-T P.1203.3) or a custom-defined score.
initial_buffering - Initial buffering time in seconds (between user action and playback start).
re_buff_freq - Rebuffering frequency (stalling events per media duration or time window).
stall_ratio - Ratio of total stall duration to total media length.

UeLevelObjectives Objects

class UeLevelObjectives(SnakeModel)

Represents UE-level performance targets or RAN optimization constraints.

Attributes:

ul_throughput - Average uplink RAN UE throughput in kbps (3GPP TS 28.552).
dl_throughput - Average downlink RAN UE throughput in kbps (3GPP TS 28.552).
ul_packet_delay - Uplink packet delay in milliseconds (typically 0-1 ms) (3GPP TS 38.314, 28.552).
dl_packet_delay - Downlink packet delay in milliseconds (typically 0-1 ms) (3GPP TS 38.314, 28.552).
ul_pdcp_sdu_packet_loss_rate - Uplink PDCP SDU packet loss rate as reliability target (3GPP TS 28.552).
dl_rlc_sdu_packet_loss_rate - Downlink RLC SDU packet loss rate as reliability target (3GPP TS 38.314).
dl_reliability - Downlink reliability as performance target (3GPP TS 28.552).
ul_reliability - Uplink reliability as performance target (3GPP TS 28.552).

SliceSlaObjectives Objects

class SliceSlaObjectives(SnakeModel)

Represents the Slice Service Level Agreement (SLA) objectives for a network slice.

Attributes:

max_number_of_ues - Maximum number of RRC-connected UEs supported concurrently by the slice (see NG.116 §3.4.17).
max_number_of_pdu_sessions - Maximum number of PDU sessions supported concurrently by the slice (NG.116 §3.4.16).
gua_dl_thpt_per_slice - Guaranteed downlink throughput in kbps for the entire slice (NG.116 §3.4.5).
max_dl_thpt_per_slice - Maximum supported downlink throughput in kbps for all UEs in the slice (NG.116 §3.4.5).
max_dl_thpt_per_ue - Maximum supported downlink throughput in kbps per UE (NG.116 §3.4.6).
gua_ul_thpt_per_slice - Guaranteed uplink throughput in kbps for the entire slice (NG.116 §3.4.31).
max_ul_thpt_per_slice - Maximum supported uplink throughput in kbps for all UEs in the slice (NG.116 §3.4.31).
max_ul_thpt_per_ue - Maximum supported uplink throughput in kbps per UE (NG.116 §3.4.32).
max_dl_packet_delay_per_ue - Maximum downlink packet delay in milliseconds.
max_ul_packet_delay_per_ue - Maximum uplink packet delay in milliseconds.
max_dl_pdcp_sdu_packet_loss_rate_per_ue - Max DL PDCP SDU packet loss rate (0-1 range).
max_ul_rlc_sdu_packet_loss_rate_per_ue - Max UL RLC SDU packet loss rate (0-1 range).
min_dl_reliability_per_ue - Minimum downlink reliability requirement.
min_ul_reliability_per_ue - Minimum uplink reliability requirement.
max_dl_jitter_per_ue - Maximum downlink jitter in milliseconds.
max_ul_jitter_per_ue - Maximum uplink jitter in milliseconds.
dl_slice_priority - Downlink slice priority (1 = highest).
ul_slice_priority - Uplink slice priority (1 = highest).
slice_enforce - Custom slice enforcement policy (FlexRIC-specific).

LbObjectives Objects

class LbObjectives(SnakeModel)

Represents load balancing objectives related to PRB (Physical Resource Block) usage.

Attributes:

target_prb_usg (Optional[int]): The target PRB usage in percent. The denominator is the total number of PRBs in the cell, and the numerator is the number of PRBs specified by prb_usg_type. Value range: 0-100 [%].

prb_usg_type (Optional[int]): Specifies the PRB usage type used in the calculation of target_prb_usg.

Valid values (from 3GPP TS 28.552):

1: Mean DL PRB used for data traffic (5.1.1.2.5)
2: Mean UL PRB used for data traffic (5.1.1.2.7)
3: Peak DL PRB used for data traffic (5.1.1.2.9)
4: Peak UL PRB used for data traffic (5.1.1.2.10)
5: Mean DL PRB used for data traffic per S-NSSAI (5.1.1.2.5)
6: Mean UL PRB used for data traffic per S-NSSAI (5.1.1.2.7)
7: Peak DL PRB used for data traffic per S-NSSAI (5.1.1.2.9)
8: Peak UL PRB used for data traffic per S-NSSAI (5.1.1.2.10)

Applicability:

If only cellId is included in the scope: valid values are 1-4.
If both cellId and sliceId are included in the scope: valid values are 5-8.

PolicyObjectives Objects

class PolicyObjectives(SnakeModel)

Represents the policy objectives defining various objectives based on A1TD spec.

Attributes:

qos_objectives Optional[QosObjectives] - Objectives related to Quality of Service.
qoe_objectives Optional[QoeObjectives] - Objectives related to Quality of Experience.
ue_level_objectives Optional[UeLevelObjectives] - Objectives at the User Equipment level.
slice_sla_objectives Optional[SliceSlaObjectives] - Objectives related to Slice Service Level Agreements.
lb_objectives Optional[LbObjectives] - Objectives related to Load Balancing.

PolicyResources Objects

class PolicyResources(SnakeModel)

Represents the policy resources defining various resources based on A1TD spec.

Attributes:

tsp_resources Optional[dict] - Resources related to Traffic Steering Policies.
slice_sla_resources Optional[dict] - Resources related to Slice Service Level Agreements.
lb_resources Optional[dict] - Resources related to Load Balancing.

PolicyStatements Objects

class PolicyStatements(SnakeModel)

Represents the policy statements defining objectives and resources based on A1TD spec.

Attributes:

policy_objectives PolicyObjectives - A statement for policy objectives expresses the goal for the policy.
policy_resources PolicyResources - A statement for policy resources expresses the conditions for resource usage for the policy.

PolicyObject Objects

class PolicyObject(SnakeModel)

Represents a policy object containing scope and policy statements based on A1TD spec. A PolicyObject contains a scope identifier and at least one policy statement (e.g., one or more policy objective statements and/or one or more policy resource statements)

Attributes:

scope_identifier ScopeIdentifier - The scope of the policy; Identifier of what the statements in the policy applies to (UE, group of UEs, slice, QoS flow, network resource or combinations thereof).
policy_statements PolicyStatements - The statements defining the policy objectives and resources.

Endpoint Objects

class Endpoint(SnakeModel)

full_url

@property
def full_url() -> HttpUrl

Constructs the full URL from scheme, host, port, and apiPath.

PolicyFeedbackDestination Objects

class PolicyFeedbackDestination(Endpoint)

Represents the endpoint for policy feedback. Inherits from Endpoint and includes additional fields specific to policy feedback.

Attributes:

api_path str - The API path for the feedback endpoint.
element_name str - The name of the element for the feedback endpoint.
host str - The host for the feedback endpoint.
ip IPvAnyAddress - The IP address for the feedback endpoint.
method Literal["GET", "POST", "PUT", "DELETE", "PATCH"] - The HTTP method for the feedback endpoint.
port int - The port for the feedback endpoint.
scheme Literal["http", "https"] - The scheme for the feedback endpoint.
service str - The service name for the feedback endpoint.

PolicyTypeInformation Objects

class PolicyTypeInformation(SnakeModel)

Represents the information of a policy type, including its ID and Near Real-Time RIC ID.

Attributes:

policy_type_id PolicyTypeId - Identifier for the policy type.
near_rt_ric_id NearRtRicId - Identifier for the Near Real-Time RIC.

PolicyInformation Objects

class PolicyInformation(SnakeModel)

Represents the information of a policy, including its ID and Near Real-Time RIC ID.

Attributes:

policy_id PolicyId - Identifier for the policy.
near_rt_ric_id NearRtRicId - Identifier for the Near Real-Time RIC.

PolicyObjectInformation Objects

class PolicyObjectInformation(SnakeModel)

Represents the information of a policy object based on R1 A1-Related Services, including its target, type, and the actual policy object.

Attributes:

near_rt_ric_id NearRtRicId - Identifier for the Near Real-Time RIC that is to be used for the policy deployment. format: .
policy_type_id PolicyTypeId - Identifier for the policy type that is to be used for the policy deployment. format: /
policy_object PolicyObject - Policy Object is a JSON representation of an A1 policy; the A1 policies are specified in A1TD.

br_rapp_sdk.a1_services.a1_services

A1Services Objects

class A1Services()

This class provides methods to interact with the A1 Services API in BubbleRAN environment.

A1Services is a client for managing A1 policies in the BubbleRAN environment.

Attributes:

kubeconfig_path Optional[str] - Path to the kubeconfig file.

namespace str - Kubernetes namespace for A1 policy jobs.

Examples :

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyObjectInformation, PolicyId
a1_services = A1Services()
# List all policies
result = a1_services.list_policies()
if result.status == "success":
    for policy_id, policy_info in result.data:
        print(f"Policy ID: {policy_id}, Info: {policy_info}")
else:
    print(f"Error: {result.error}")
# Get a specific policy
policy_id = PolicyId("example-policy")
result = a1_services.get_policy(policy_id)
if result.status == "success":
    policy_info = result.data.get('item')
    print(f"Policy Info: {policy_info}")
else:
    print(f"Error: {result.error}")

init

def __init__(kubeconfig_path: Optional[str] = None,
             namespace: str = "trirematics")

Initialize the A1Services client by loading the Kubernetes configuration and setting up defaults.

Arguments:

kubeconfig_path Optional[str] - Path to the kubeconfig file. If None, uses the default kubeconfig.
namespace str - Kubernetes namespace for A1 policy jobs. Default is "trirematics".

Raises:

RuntimeError - If kubeconfig cannot be loaded or if there are issues with the API.

list_policies

def list_policies(
        policy_id: Optional[PolicyId] = None) -> KubectlOperationResult

This method lists all policies in the A1 Services API.

Arguments:

Policy_id Optional[PolicyId] - If provided, filters policies by the given ID.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing a list of PolicyId and PolicyObjectInformation tuples if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
a1_services = A1Services()
result = a1_services.list_policies()
# Check if the operation was successful
if result.status == "success":
    for policy_id, policy_info in result.data:
        print(f"Policy ID: {policy_id}, Info: {policy_info}")
else:
    print(f"Error: {result.error}")

get_policy

def get_policy(policy_id: PolicyId) -> KubectlOperationResult

Get a specific policy from the A1 Services API.

Arguments:

policy_id PolicyId - The ID of the policy to retrieve.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing the PolicyObjectInformation if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyId
a1_services = A1Services()
policy_id = PolicyId("example-policy")
result = a1_services.get_policy(policy_id)
if result.status == "success":
    policy_info = result.data.get('item')
    print(f"Policy Info: {policy_info}")
else:
    print(f"Error: {result.error}")

get_policy_feedback_api_urls

def get_policy_feedback_api_urls(
        policy_id: PolicyId) -> KubectlOperationResult

Get the feedback API URLs for dynamic xApps in a specific policy.

Arguments:

policy_id PolicyId - The ID of the policy to retrieve feedback URLs for.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing a list of tuples with DynamicXappId and their corresponding feedback URLs if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyId, DynamicXappId
a1_services = A1Services()
policy_id = PolicyId("example-policy")
result = a1_services.get_policy_feedback_api_urls(policy_id)
if result.status == "success":
    for dynamic_xapp, feedback_urls in result.data:
        print(f"Dynamic xApp: {dynamic_xapp}, Feedback URLs: {[url.ip for url in feedback_urls]}")
else:
    print(f"Error getting policy feedback URLs: {result.error}")

get_policy_feedback

def get_policy_feedback(policy_feedback_dest: PolicyFeedbackDestination,
                        time_out: int = 5) -> KubectlOperationResult

Get the policy feedback from a specific feedback destination.

Arguments:

policy_feedback PolicyFeedbackDestination - The feedback destination containing the full URL to retrieve feedback from.
time_out int - The timeout for the HTTP request in seconds. Default is 5 seconds.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing the feedback data if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyFeedbackDestination
a1_services = A1Services()
policy_feedback = PolicyFeedbackDestination(...)
result = a1_services.get_policy_feedback(policy_feedback)
if result.status == "success":
    print(f"Policy Feedback: {result.data.get('item', {})}")
else:
    print(f"Error getting policy feedback: {result.error}")

apply_policy

def apply_policy(
        policy_name: str,
        policy_object: PolicyObjectInformation) -> KubectlOperationResult

Apply a policy to the A1 Services API.

Arguments:

policy_name str - The name of the policy to apply.
policy_object PolicyObjectInformation - The policy object containing the specifications.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing the PolicyId if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyObjectInformation, PolicyId, NearRtRicId, PolicyTypeId, PolicyObject, ScopeIdentifier, PolicyStatements
a1_services = A1Services()
policy_object = PolicyObjectInformation(
    near_rt_ric_id=NearRtRicId("ric-name.network-name"),
    policy_type_id=PolicyTypeId("cm/example"),
    policy_object=PolicyObject(
        scope_identifier=ScopeIdentifier(...),
        policy_statements=PolicyStatements(...)
    )
)
result = a1_services.apply_policy(policy_name="example-policy", policy_object=policy_object)
if result.status == 'success':
    policy_id = result.data.get('policy_id')
    print(f"Policy applied successfully: {policy_id}")
else:
    print(f"Error applying policy: {result.error}")

delete_policy

def delete_policy(policy_id: PolicyId) -> KubectlOperationResult

Delete a policy from the A1 Services API.

Arguments:

policy_id PolicyId - The ID of the policy to delete.

Returns:

KubectlOperationResult - An object representing the result of the operation, indicating success or failure.

Example:

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyId
a1_services = A1Services()
policy_id = PolicyId("example-policy")
result = a1_services.delete_policy(policy_id)
if result.status == "success":
    print(f"Policy {policy_id} deleted successfully.")
else:
    print(f"Error deleting policy: {result.error}")

get_policy_status

def get_policy_status(policy_id: PolicyId) -> KubectlOperationResult

Get the status of a specific policy in the A1 Services API.

Arguments:

policy_id PolicyId - The ID of the policy to check.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing the policy status if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
from br_rapp_sdk.a1_services.a1_policy_types import PolicyId
a1_services = A1Services()
policy_id = PolicyId("example-policy")
result = a1_services.get_policy_status(policy_id)
if result.status == "success":
    print(f"Policy Status: {result.data.get('status')}")
else:
    print(f"Error getting policy status: {result.error}")

get_rics

def get_rics(network_id: Optional[NetworkId] = None) -> KubectlOperationResult

Get the list of Near RT RICs from the OAM services. This method retrieves the Near RT RICs from the OAM services and returns their IDs.

Arguments:

Optional[NetworkId] - If provided, filters Near RT RICs by the given NetworkId.

Returns:

KubectlOperationResult - An object representing the result of the operation, containing a list of NearRtRicId if successful, or an error message if not.

Example:

from br_rapp_sdk import A1Services
a1_services = A1Services()
result = a1_services.get_rics()
if result.status == "success":
    near_rt_rics = result.data.get('items', [])
    for near_rt_ric in near_rt_rics:
        print(f"Near RT RIC ID: {near_rt_ric}")
else:
    print(f"Error getting Near RT RICs: {result.error}")

PlmnId Objects​

SliceId Objects​

QosId Objects​

CId Objects​

CellId Objects​

GroupId Objects​

ScopeIdentifier Objects​

ReliabilityType Objects​

QosObjectives Objects​

QoeObjectives Objects​

UeLevelObjectives Objects​

SliceSlaObjectives Objects​

LbObjectives Objects​

PolicyObjectives Objects​

PolicyResources Objects​

PolicyStatements Objects​

PolicyObject Objects​

Endpoint Objects​

full_url​

PolicyFeedbackDestination Objects​

PolicyTypeInformation Objects​

PolicyInformation Objects​

PolicyObjectInformation Objects​

br_rapp_sdk.a1_services.a1_services

A1Services Objects​

__init__​

list_policies​

get_policy​

get_policy_feedback_api_urls​

get_policy_feedback​

apply_policy​

delete_policy​

get_policy_status​

get_rics​

PlmnId Objects

SliceId Objects

QosId Objects

CId Objects

CellId Objects

GroupId Objects

ScopeIdentifier Objects

ReliabilityType Objects

QosObjectives Objects

QoeObjectives Objects

UeLevelObjectives Objects

SliceSlaObjectives Objects

LbObjectives Objects

PolicyObjectives Objects

PolicyResources Objects

PolicyStatements Objects

PolicyObject Objects

Endpoint Objects

full_url

PolicyFeedbackDestination Objects

PolicyTypeInformation Objects

PolicyInformation Objects

PolicyObjectInformation Objects

A1Services Objects

init

list_policies

get_policy

get_policy_feedback_api_urls

get_policy_feedback

apply_policy

delete_policy

get_policy_status

get_rics