working on qod code

from sunrise6g_opensdk.network.adapters.open5gs.client import (

    NetworkManager as Open5GSClient,

)

from sunrise6g_opensdk.oran.adapters.i2cat_ric.client import (

    OranManager as OranManageri2CAT,

)

def _edgecloud_adapters_factory(client_name: str, base_url: str, **kwargs):

        )

# def _oran_adapters_factory(client_name: str, base_url: str):

#     # TODO

def _oran_adapters_factory(client_name: str, base_url: str, **kwargs):

    if "scs_as_id" not in kwargs:

        raise ValueError("Missing required 'scs_as_id' for network adapters.")

    scs_as_id = kwargs.pop("scs_as_id")

    oran_factory = {

        "i2cat_ric": lambda url, scs_id, **kw: OranManageri2CAT(

            base_url=url, scs_as_id=scs_id, **kw

        ),

    }

    try:

        return oran_factory[client_name](base_url, scs_as_id, **kwargs)

    except KeyError:

        raise ValueError(f"Invalid Oran client '{client_name}'. Available: {list(oran_factory)}")

class AdaptersFactory:

    _domain_factories = {

        "edgecloud": _edgecloud_adapters_factory,

        "network": _network_adapters_factory,

        # "oran": _oran_adapters_factory,

        "oran": _oran_adapters_factory,

    }

    @classmethod

# -*- coding: utf-8 -*-

class OranPlatformError(Exception):

    pass

# -*- coding: utf-8 -*-

##

#

# This file is part of the Open SDK, based on sunrise6g_opensdk.network.core.adapters.open5gs.client

#

# Contributors:

#   - Miguel Catalan Cid (miguel.catalan@i2cat.net)

##

import json

from pathlib import Path

from typing import Any, Dict

import mappings as mappings_module

from mappings import flow_id_mapping, policy_mapping, qos_prio_to_oran_prio

from pydantic import ValidationError

from sunrise6g_opensdk import logger

from sunrise6g_opensdk.oran.core.base_oran_client import BaseOranClient

from ...core import common as oran_common

from ...core import schemas

from ...core.common import requires_capability

log = logger.get_logger(__name__)

class OranManager(BaseOranClient):

    """

    This client implements the BaseOranClient and translates the

    CAMARA APIs into specific HTTP requests understandable by the i2CAT ORAN NEF API.

    """

    capabilities = {"oran-qod", "oran-performance"}

    def __init__(self, base_url: str, scs_as_id):

        """

        Initializes the OranNEFClient Client.

        """

        try:

            # Set required attributes without invoking BaseOranClient.__init__

            self.base_url = base_url

            self.scs_as_id = scs_as_id

            # Start periodic refresh of file-based mappings (every 60s, run immediately)

            self.start_periodic_refresh(interval_seconds=10, run_immediately=True)

            log.info(

                f"Initialized OranNEFClient 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 OranNEFClient: {e}")

            raise e

    def oran_specific_qod_validation(self, session_info: schemas.CreateSession):

        if session_info.qosProfile.root not in qos_prio_to_oran_prio.keys():

            raise ValidationError(

                f"OranNEFClient only supports these qos-profiles: {', '.join(qos_prio_to_oran_prio.keys())}"

            )

    def _load_ip_mapping_from_file(self) -> Dict[str, Dict[str, Any]]:

        base_dir = Path(__file__).parent

        cfg_path = base_dir / "ip_to_plmn_gnb_mapping.json"

        if not cfg_path.exists():

            # No file present; nothing to load

            return {}

        try:

            with cfg_path.open("r", encoding="utf-8") as f:

                raw = json.load(f)

        except Exception as exc:

            log.warning(f"Failed to parse mapping file {cfg_path.name}: {exc}")

            return {}

        if not isinstance(raw, dict):

            log.warning(f"Mapping file root must be an object/dict: {cfg_path.name}")

            return {}

        parsed: Dict[str, Dict[str, Any]] = {}

        for ip, entry in raw.items():

            if not isinstance(ip, str) or not isinstance(entry, dict):

                continue

            try:

                mcc = entry.get("mcc")

                mnc = entry.get("mnc")

                gnb_length = entry.get("gnb_length")

                gnb_id = entry.get("gnb_id")

                if mcc is None or mnc is None or gnb_length is None or gnb_id is None:

                    raise ValueError("missing required keys")

                # Coerce types

                mcc_str = str(mcc)

                mnc_str = str(mnc)

                gnb_length_int = int(gnb_length)

                gnb_id_int = int(gnb_id)

                parsed[ip] = {

                    "mcc": mcc_str,

                    "mnc": mnc_str,

                    "gnb_length": gnb_length_int,

                    "gnb_id": gnb_id_int,

                }

            except Exception:

                # Skip invalid entries

                continue

        return parsed

    def refresh_dynamic_mappings(self) -> None:

        """Read IP→PLMN/gNB mapping from local JSON/YAML and apply atomically."""

        new_map = self._load_ip_mapping_from_file()

        if new_map:

            mappings_module.ip_to_plmn_gnb_mapping = new_map

            log.debug(f"Loaded {len(new_map)} IP mapping entries from file into runtime mapping")

        else:

            # If file missing or invalid, keep existing in-memory map

            pass

    @requires_capability("oran-qod")

    def create_qod_session(self, session_info: Dict) -> Dict:

        """

        Translate a CAMARA QoD session dict into an ORAN policy and submit it.

        """

        ue = session_info.get("device") or {}

        server_ip = ue.get("ipv4Address")

        if not server_ip:

            raise ValueError("device.ipv4Address must be provided")

        scope = mappings_module.ip_to_plmn_gnb_mapping.get(server_ip)

        if not scope:

            raise ValueError(f"No PLMN/gNB mapping found for server IP {server_ip}")

        qos_profile = session_info.get("qosProfile")

        if isinstance(qos_profile, dict):

            qos_profile = qos_profile.get("root") or qos_profile.get("value")

        if qos_profile not in qos_prio_to_oran_prio:

            raise ValidationError(

                f"Unsupported qosProfile '{qos_profile}'. Allowed: {', '.join(qos_prio_to_oran_prio.keys())}"

            )

        qos_prio = qos_prio_to_oran_prio[qos_profile]

        # Flow ID from profile

        try:

            flow_id = flow_id_mapping[qos_profile]

        except KeyError:

            raise ValidationError(f"No flow_id mapping found for qosProfile '{qos_profile}'")

        scope_with_flow = {**scope, "flow_id": flow_id}

        expiry = session_info.get("duration")

        try:

            expiry = int(expiry) if expiry is not None else None

        except Exception:

            expiry = None

        # Prefer explicit notificationDestination if present; fall back to sink

        notification_uri = session_info.get("notificationDestination") or None

        policy = schemas.OranPolicy(

            policyType=policy_mapping["oran-qod"],

            policyScope=scope_with_flow,

            policyStatement={"qos_prio": qos_prio},

            expiry=expiry,

            notificationUri=notification_uri,

        )

        return oran_common.oran_policy_post(self.base_url, self.scs_as_id, policy)

    @requires_capability("oran-qod")

    def get_qod_session(self, session_id: str) -> Dict:

        """Retrieve an ORAN policy by ID (maps to QoD session get)."""

        return oran_common.oran_policy_get(self.base_url, self.scs_as_id, session_id)

    @requires_capability("oran-qod")

    def delete_qod_session(self, session_id: str) -> None:

        """Delete an ORAN policy by ID (maps to QoD session delete)."""

        oran_common.oran_policy_delete(self.base_url, self.scs_as_id, session_id)

{

"192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345 },

"10.0.0.5": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890 }

}