Loading src/network/clients/open5gs/client.py +21 −12 Original line number Diff line number Diff line Loading @@ -6,14 +6,17 @@ from . import schemas log = logger.get_logger(__name__) class Open5GSClient(NetworkManagementInterface): class NetworkManager(NetworkManagementInterface): """ This client implements the NetworkManagementInterface and translates the CAMARA APIs into specific HTTP requests understandable by the Open5GS NEF API. Invloved partners and their roles in this implementation: - I2CAT: Responsible for the CAMARA QoD API and its mapping to the 3GPP AsSessionWithQoS API exposed by Open5GS NEF. - NCSRD: Responsible for the CAMARA Location API and its mapping to the 3GPP Monitoring Even API exposed Open5GS NEF. - I2CAT: Responsible for the CAMARA QoD API and its mapping to the 3GPP AsSessionWithQoS API exposed by Open5GS NEF. - NCSRD: Responsible for the CAMARA Location API and its mapping to the 3GPP Monitoring Event API exposed Open5GS NEF. """ def __init__(self, base_url: str, scs_as_id: str): Loading @@ -23,12 +26,14 @@ class Open5GSClient(NetworkManagementInterface): try: self.base_url = base_url self.scs_as_id = scs_as_id log.info(f"Initialized Open5GSClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}") log.info( f"Initialized Open5GSClient with base_url: {self.base_url} " f"and scs_as_id: {self.scs_as_id}" ) except Exception as e: log.error(f"Failed to initialize Open5GSClient: {e}") raise e # --- Implementation of NetworkManagementInterface methods --- def create_qod_session(self, session_info: Dict) -> Dict: """ Loading @@ -37,21 +42,25 @@ class Open5GSClient(NetworkManagementInterface): """ pass def get_qod_session(self, session_id: str) -> Dict: """ Retrieves a specific Open5GS QoS Subscription details. Maps CAMARA QoD GET /sessions/{sessionId} to Open5GS NEF GET /{scsAsId}/subscriptions/{subscriptionId}. Maps CAMARA QoD GET /sessions/{sessionId} to Open5GS NEF GET / {scsAsId}/subscriptions/{subscriptionId}. """ pass def delete_qod_session(self, session_id: str) -> None: """ Deletes a specific Open5GS QoS Subscription. Maps CAMARA QoD DELETE /sessions/{sessionId} to Open5GS NEF DELETE /{scsAsId}/subscriptions/{subscriptionId}. Maps CAMARA QoD DELETE /sessions/{sessionId} to Open5GS NEF DELETE / {scsAsId}/subscriptions/{subscriptionId}. """ pass # Note: # As this class is inheriting from NetworkManagementInterface, it is # expected to implement all the abstract methods defined in that interface. # # In case this network adapter doesn't support a specific method, it should # be marked as NotImplementedError. src/network/clients/open5gs/common.py +6 −4 Original line number Diff line number Diff line Loading @@ -13,26 +13,28 @@ log = logger.get_logger(__name__) class Open5GSError(NetworkPlatformError): pass class Open5GSErrorResponse(BaseModel): message: str detail: dict # --- HTTP Request Helper Functions --- def open5gs_post(url: str, model_payload: BaseModel) -> dict: """ Placeholder for the POST request function.""" pass def open5gs_get(url: str, params: Optional[dict] = None) -> dict: """ Placeholder for the GET request function. """ pass def open5gs_delete(url: str) -> None: """ Placeholder for the DELETE request function. """ pass src/network/clients/open5gs/schemas.py +5 −4 Original line number Diff line number Diff line Loading @@ -4,6 +4,7 @@ from pydantic import BaseModel # Dummy examples of Pydantic models for the Open5GS NEF API. class Open5GSQoSSubscription(BaseModel): """ Loading @@ -11,9 +12,9 @@ class Open5GSQoSSubscription(BaseModel): """ pass class CamaraQoDSessionInfo(BaseModel): """ Represents the input data for creating a QoD session. """ pass Loading
src/network/clients/open5gs/client.py +21 −12 Original line number Diff line number Diff line Loading @@ -6,14 +6,17 @@ from . import schemas log = logger.get_logger(__name__) class Open5GSClient(NetworkManagementInterface): class NetworkManager(NetworkManagementInterface): """ This client implements the NetworkManagementInterface and translates the CAMARA APIs into specific HTTP requests understandable by the Open5GS NEF API. Invloved partners and their roles in this implementation: - I2CAT: Responsible for the CAMARA QoD API and its mapping to the 3GPP AsSessionWithQoS API exposed by Open5GS NEF. - NCSRD: Responsible for the CAMARA Location API and its mapping to the 3GPP Monitoring Even API exposed Open5GS NEF. - I2CAT: Responsible for the CAMARA QoD API and its mapping to the 3GPP AsSessionWithQoS API exposed by Open5GS NEF. - NCSRD: Responsible for the CAMARA Location API and its mapping to the 3GPP Monitoring Event API exposed Open5GS NEF. """ def __init__(self, base_url: str, scs_as_id: str): Loading @@ -23,12 +26,14 @@ class Open5GSClient(NetworkManagementInterface): try: self.base_url = base_url self.scs_as_id = scs_as_id log.info(f"Initialized Open5GSClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}") log.info( f"Initialized Open5GSClient with base_url: {self.base_url} " f"and scs_as_id: {self.scs_as_id}" ) except Exception as e: log.error(f"Failed to initialize Open5GSClient: {e}") raise e # --- Implementation of NetworkManagementInterface methods --- def create_qod_session(self, session_info: Dict) -> Dict: """ Loading @@ -37,21 +42,25 @@ class Open5GSClient(NetworkManagementInterface): """ pass def get_qod_session(self, session_id: str) -> Dict: """ Retrieves a specific Open5GS QoS Subscription details. Maps CAMARA QoD GET /sessions/{sessionId} to Open5GS NEF GET /{scsAsId}/subscriptions/{subscriptionId}. Maps CAMARA QoD GET /sessions/{sessionId} to Open5GS NEF GET / {scsAsId}/subscriptions/{subscriptionId}. """ pass def delete_qod_session(self, session_id: str) -> None: """ Deletes a specific Open5GS QoS Subscription. Maps CAMARA QoD DELETE /sessions/{sessionId} to Open5GS NEF DELETE /{scsAsId}/subscriptions/{subscriptionId}. Maps CAMARA QoD DELETE /sessions/{sessionId} to Open5GS NEF DELETE / {scsAsId}/subscriptions/{subscriptionId}. """ pass # Note: # As this class is inheriting from NetworkManagementInterface, it is # expected to implement all the abstract methods defined in that interface. # # In case this network adapter doesn't support a specific method, it should # be marked as NotImplementedError.
src/network/clients/open5gs/common.py +6 −4 Original line number Diff line number Diff line Loading @@ -13,26 +13,28 @@ log = logger.get_logger(__name__) class Open5GSError(NetworkPlatformError): pass class Open5GSErrorResponse(BaseModel): message: str detail: dict # --- HTTP Request Helper Functions --- def open5gs_post(url: str, model_payload: BaseModel) -> dict: """ Placeholder for the POST request function.""" pass def open5gs_get(url: str, params: Optional[dict] = None) -> dict: """ Placeholder for the GET request function. """ pass def open5gs_delete(url: str) -> None: """ Placeholder for the DELETE request function. """ pass
src/network/clients/open5gs/schemas.py +5 −4 Original line number Diff line number Diff line Loading @@ -4,6 +4,7 @@ from pydantic import BaseModel # Dummy examples of Pydantic models for the Open5GS NEF API. class Open5GSQoSSubscription(BaseModel): """ Loading @@ -11,9 +12,9 @@ class Open5GSQoSSubscription(BaseModel): """ pass class CamaraQoDSessionInfo(BaseModel): """ Represents the input data for creating a QoD session. """ pass