feat: extend fiveg_rfsim interface

lib/charms/oai_ran_du_k8s/v0/fiveg_rfsim.py

@@ -5,7 +5,7 @@
 
 This library contains the Requires and Provides classes for handling the `fiveg_rfsim` interface.
 
-The purpose of this library is to relate two charms to pass the RF SIM address.
+The purpose of this library is to relate two charms to pass the RF SIM address and network information (SST and SD).
 In the Telco world this will typically be charms implementing
 the DU (Distributed Unit) and the UE (User equipment).
 
@@ -21,33 +21,37 @@
 - pytest-interface-tester
 
 ### Provider charm
-The provider charm is the one providing the information about RF SIM address.
+The provider charm is the one providing the information about RF SIM address, Network Slice Type (SST) and Network Differentiator (SD).
 Typically, this will be the DU charm.
 
 Example:
 ```python
 
-from ops.charm import CharmBase, RelationJoinedEvent
-from ops.main import main
+from ops import main
+from ops.charm import CharmBase, RelationChangedEvent
 
 from charms.oai_ran_du_k8s.v0.fiveg_rfsim import RFSIMProvides
 
 
 class DummyFivegRFSIMProviderCharm(CharmBase):
 
     RFSIM_ADDRESS = "192.168.70.130"
+    SST = 1
+    SD = 1
 
     def __init__(self, *args):
         super().__init__(*args)
         self.rfsim_provider = RFSIMProvides(self, "fiveg_rfsim")
         self.framework.observe(
-            self.on.fiveg_rfsim_relation_joined, self._on_fiveg_rfsim_relation_joined
+            self.on.fiveg_rfsim_relation_changed, self._on_fiveg_rfsim_relation_changed
         )
 
-    def _on_fiveg_rfsim_relation_joined(self, event: RelationJoinedEvent):
+    def _on_fiveg_rfsim_relation_changed(self, event: RelationChangedEvent):
         if self.unit.is_leader():
             self.rfsim_provider.set_rfsim_information(
                 rfsim_address=self.RFSIM_ADDRESS
+                sst=self.SST,
+                sd=self.SD,
             )
 
 
@@ -62,10 +66,10 @@ def _on_fiveg_rfsim_relation_joined(self, event: RelationJoinedEvent):
 Example:
 ```python
 
-from ops.charm import CharmBase
-from ops.main import main
+from ops import main
+from ops.charm import CharmBase, RelationChangedEvent
 
-from charms.oai_ran_du_k8s.v0.fiveg_rfsim import FivegRFSIMInformationAvailableEvent, RFSIMRequires
+from charms.oai_ran_du_k8s.v0.fiveg_rfsim import RFSIMRequires
 
 logger = logging.getLogger(__name__)
 
@@ -76,12 +80,14 @@ def __init__(self, *args):
         super().__init__(*args)
         self.rfsim_requirer = RFSIMRequires(self, "fiveg_rfsim")
         self.framework.observe(
-            self.rfsim_requirer.on.fiveg_rfsim_provider_available, self._on_rfsim_information_available
+            self.on.fiveg_rfsim_relation_changed, self._on_fiveg_rfsim_relation_changed
         )
 
-    def _on_rfsim_information_available(self, event: FivegRFSIMInformationAvailableEvent):
+    def _on_fiveg_rfsim_relation_changed(self, event: RelationChangedEvent):
         provider_rfsim_address = event.rfsim_address
-        <do something with the rfsim address here>
+        provider_sst = event.sst
+        provider_st = event.sd
+        <do something with the rfsim address, SST and SD here>
 
 
 if __name__ == "__main__":
@@ -95,10 +101,10 @@ def _on_rfsim_information_available(self, event: FivegRFSIMInformationAvailableE
 from typing import Any, Dict, Optional
 
 from interface_tester.schema_base import DataBagSchema
-from ops.charm import CharmBase, CharmEvents, RelationChangedEvent
-from ops.framework import EventBase, EventSource, Handle, Object
+from ops.charm import CharmBase
+from ops.framework import Object
 from ops.model import Relation
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel, Field, IPvAnyAddress, ValidationError
 
 
 # The unique Charmhub library identifier, never change it
@@ -109,25 +115,55 @@ def _on_rfsim_information_available(self, event: FivegRFSIMInformationAvailableE
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 2
+LIBPATCH = 3
 
 
 logger = logging.getLogger(__name__)
 
+"""Schemas definition for the provider and requirer sides of the `fiveg_rfsim` interface.
+It exposes two interface_tester.schema_base.DataBagSchema subclasses called:
+- ProviderSchema
+- RequirerSchema
+Examples:
+    ProviderSchema:
+        unit: <empty>
+        app: {
+            "rfsim_address": "192.168.70.130",
+            "sst": 1,
+            "sd": 1,
+        }
+    RequirerSchema:
+        unit: <empty>
+        app:  <empty>
+"""
+
 
-class FivegRFSIMProviderAppData(BaseModel):
+class ProviderAppData(BaseModel):
     """Provider app data for fiveg_rfsim."""
 
-    rfsim_address: str = Field(
+    rfsim_address: IPvAnyAddress = Field(
         description="RF simulator service address which is equal to DU pod ip",
         examples=["192.168.70.130"],
     )
+    sst: int = Field(
+        description="Slice/Service Type",
+        examples=[1, 2, 3, 4],
+        ge=0,
+        le=255,
+    )
+    sd: Optional[int] = Field(
+        description="Slice Differentiator",
+        default=None,
+        examples=[1],
+        ge=0,
+        le=16777215,
+    )
 
 
 class ProviderSchema(DataBagSchema):
     """Provider schema for the fiveg_rfsim interface."""
 
-    app: FivegRFSIMProviderAppData
+    app_data: ProviderAppData
 
 
 def provider_data_is_valid(data: Dict[str, Any]) -> bool:
@@ -140,41 +176,13 @@ def provider_data_is_valid(data: Dict[str, Any]) -> bool:
         bool: True if data is valid, False otherwise.
     """
     try:
-        ProviderSchema(app=data)
+        ProviderSchema(app_data=ProviderAppData(**data))
         return True
     except ValidationError as e:
         logger.error("Invalid data: %s", e)
         return False
 
 
-class FivegRFSIMInformationAvailableEvent(EventBase):
-    """Charm event emitted when the RFSIM provider info is available.
-
-    The event carries the RFSIM provider's address.
-    """
-
-    def __init__(self, handle: Handle, rfsim_address: str):
-        """Init."""
-        super().__init__(handle)
-        self.rfsim_address = rfsim_address
-
-    def snapshot(self) -> dict:
-        """Return snapshot."""
-        return {
-            "rfsim_address": self.rfsim_address,
-        }
-
-    def restore(self, snapshot: dict) -> None:
-        """Restores snapshot."""
-        self.rfsim_address = snapshot["rfsim_address"]
-
-
-class FivegRFSIMRequirerCharmEvents(CharmEvents):
-    """The event that the RFSIM requirer charm can leverage."""
-
-    fiveg_rfsim_provider_available = EventSource(FivegRFSIMInformationAvailableEvent)
-
-
 class FivegRFSIMError(Exception):
     """Custom error class for the `fiveg_rfsim` library."""
 
@@ -192,72 +200,89 @@ def __init__(self, charm: CharmBase, relation_name: str):
         self.relation_name = relation_name
         self.charm = charm
 
-    def set_rfsim_information(self, rfsim_address: str) -> None:
+    def set_rfsim_information(self, rfsim_address: str, sst: int, sd: Optional[int]) -> None:
         """Push the information about the RFSIM interface in the application relation data.
 
         Args:
             rfsim_address (str): rfsim service address which is equal to DU pod ip.
+            sst (int): Slice/Service Type
+            sd (Optional[int]): Slice Differentiator
         """
         if not self.charm.unit.is_leader():
             raise FivegRFSIMError("Unit must be leader to set application relation data.")
         relations = self.model.relations[self.relation_name]
         if not relations:
             raise FivegRFSIMError(f"Relation {self.relation_name} not created yet.")
-        if not provider_data_is_valid({"rfsim_address": rfsim_address}):
+        if not provider_data_is_valid(
+            {
+                "rfsim_address": rfsim_address,
+                "sst": sst,
+                "sd": sd,
+            }
+        ):
             raise FivegRFSIMError("Invalid relation data")
         for relation in relations:
-            relation.data[self.charm.app].update(
-                {
-                    "rfsim_address": rfsim_address,
-                }
-            )
+            data = {
+                "rfsim_address": rfsim_address,
+                "sst": str(sst),
+            }
+            if sd is not None:
+                data["sd"] = str(sd)
+            relation.data[self.charm.app].update(data)
 
 
 class RFSIMRequires(Object):
     """Class to be instantiated by the charm requiring relation using the `fiveg_rfsim` interface."""
 
-    on = FivegRFSIMRequirerCharmEvents()
-
     def __init__(self, charm: CharmBase, relation_name: str):
         """Init."""
         super().__init__(charm, relation_name)
         self.charm = charm
         self.relation_name = relation_name
-        self.framework.observe(charm.on[relation_name].relation_changed, self._on_relation_changed)
 
-    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
-        """Handle relation changed event.
+    @property
+    def rfsim_address(self) -> Optional[IPvAnyAddress]:
+        """Return address of the RFSIM.
 
-        Args:
-            event (RelationChangedEvent): Juju event.
+        Returns:
+            Optional[IPvAnyAddress]: rfsim address which is equal to DU pod ip.
         """
-        if remote_app_relation_data := self._get_remote_app_relation_data(event.relation):
-            self.on.fiveg_rfsim_provider_available.emit(
-                rfsim_address=remote_app_relation_data["rfsim_address"],
-            )
+        if remote_app_relation_data := self.get_provider_rfsim_information():
+            return remote_app_relation_data.rfsim_address
+        return None
 
     @property
-    def rfsim_address(self) -> Optional[str]:
-        """Return address of the RFSIM.
+    def sst(self) -> Optional[int]:
+        """Return the Network Slice Service Type (SST).
 
         Returns:
-            str: rfsim address which is equal to DU pod ip.
+            Optional[int]: sst (Network Slice Service Type)
         """
-        if remote_app_relation_data := self._get_remote_app_relation_data():
-            return remote_app_relation_data.get("rfsim_address")
+        if remote_app_relation_data := self.get_provider_rfsim_information():
+            return remote_app_relation_data.sst
         return None
 
-    def _get_remote_app_relation_data(
-        self, relation: Optional[Relation] = None
-    ) -> Optional[Dict[str, str]]:
+    @property
+    def sd(self) -> Optional[int]:
+        """Return the Network Slice Differentiator (SD).
+
+        Returns:
+           Optional[int] : sd (Network Slice Differentiator)
+        """
+        if remote_app_relation_data := self.get_provider_rfsim_information():
+            return remote_app_relation_data.sd
+        return None
+
+    def get_provider_rfsim_information(self, relation: Optional[Relation] = None
+    ) -> Optional[ProviderAppData]:
         """Get relation data for the remote application.
 
         Args:
             relation: Juju relation object (optional).
 
         Returns:
-            Dict: Relation data for the remote application
-            or None if the relation data is invalid.
+            ProviderAppData: Relation data for the remote application if data is valid,
+            None otherwise.
         """
         relation = relation or self.model.get_relation(self.relation_name)
         if not relation:
@@ -266,8 +291,16 @@ def _get_remote_app_relation_data(
         if not relation.app:
             logger.warning("No remote application in relation: %s", self.relation_name)
             return None
-        remote_app_relation_data = dict(relation.data[relation.app])
-        if not provider_data_is_valid(remote_app_relation_data):
+        remote_app_relation_data: Dict[str, Any] = dict(relation.data[relation.app])
+        try:
+            remote_app_relation_data["sd"] = int(remote_app_relation_data.get("sd", ""))
+            remote_app_relation_data["sst"] = int(remote_app_relation_data.get("sst", ""))
+        except ValueError:
+            logger.error("Invalid relation data: %s", remote_app_relation_data)
+            return None
+        try:
+            provider_app_data = ProviderAppData(**remote_app_relation_data)
+        except ValidationError:
             logger.error("Invalid relation data: %s", remote_app_relation_data)
             return None
-        return remote_app_relation_data
+        return provider_app_data

src/charm.py

@@ -88,7 +88,7 @@ def __init__(self, *args):
         self.framework.observe(self.on.du_pebble_ready, self._configure)
         self.framework.observe(self.on[F1_RELATION_NAME].relation_created, self._configure)
         self.framework.observe(self.on[F1_RELATION_NAME].relation_changed, self._configure)
-        self.framework.observe(self.on.fiveg_rfsim_relation_joined, self._configure)
+        self.framework.observe(self.on[RFSIM_RELATION_NAME].relation_changed, self._configure)
         self.framework.observe(self.on.remove, self._on_remove)
 
     def _on_collect_unit_status(self, event: CollectStatusEvent):  # noqa C901
@@ -211,7 +211,20 @@ def _set_fiveg_rfsim_relation_data(self) -> None:
             return
         if not self._get_rfsim_address():
             return
-        self.rfsim_provider.set_rfsim_information(self._get_rfsim_address())
+        if not self._relation_created(F1_RELATION_NAME):
+            return
+        if not (remote_network_information := self._f1_requirer.get_provider_f1_information()):
+            return
+        # There could be multiple PLMNs but UE simulator always publishes the first PLMN content
+        # according to Spec TE126. For real UE's, the device group of a slice includes IMSI
+        # which can be associated with a UE.
+        if not remote_network_information.plmns:
+            return
+        self.rfsim_provider.set_rfsim_information(
+            self._get_rfsim_address(),
+            remote_network_information.plmns[0].sst,
+            remote_network_information.plmns[0].sd,
+        )
 
     @staticmethod
     def _get_rfsim_address() -> str:

tests/unit/fixtures.py

@@ -18,6 +18,13 @@
     plmns=[PLMNConfig(mcc="001", mnc="01", sst=1)],
 )
 
+F1_PROVIDER_DATA_WITH_SD = ProviderAppData(
+    f1_ip_address=IPv4Address("4.3.2.1"),
+    f1_port=2152,
+    tac=1,
+    plmns=[PLMNConfig(mcc="001", mnc="01", sst=1, sd=1)],
+)
+
 
 class DUFixtures:
     patcher_check_output = patch("charm.check_output")