device_config/configuration_init.py

# SPDX-FileCopyrightText: 2020 The Magma Authors.
# SPDX-FileCopyrightText: 2022 Open Networking Foundation <support@opennetworking.org>
#
# SPDX-License-Identifier: BSD-3-Clause

from distutils.sysconfig import customize_compiler
import json
from collections import namedtuple
from typing import Any, Optional, Union

from lte.protos.mconfig import mconfigs_pb2
from common.misc_utils import get_ip_from_if
from configuration.exceptions import LoadConfigError
from configuration.service_configs import load_enb_config, load_common_config
from configuration.mconfig_managers import load_service_mconfig_as_json
from data_models.data_model import DataModel
from data_models.data_model_parameters import ParameterName
from device_config.enodeb_config_postprocessor import (
    EnodebConfigurationPostProcessor,
)
from device_config.enodeb_configuration import EnodebConfiguration
from exceptions import ConfigurationError
from logger import EnodebdLogger as logger
from lte_utils import (
    DuplexMode,
    map_earfcndl_to_band_earfcnul_mode,
    map_earfcndl_to_duplex_mode,
)

# LTE constants
DEFAULT_S1_PORT = 36412
# This is a known working value for supported eNB devices.
# Cell Identity is a 28 bit number, but not all values are supported.
DEFAULT_CELL_IDENTITY = 138777000

SingleEnodebConfig = namedtuple(
    'SingleEnodebConfig',
    [
        'earfcndl', 'subframe_assignment',
        'special_subframe_pattern',
        'pci', 'plmnid_list', 'tac',
        'bandwidth_mhz', 'cell_id',
        'allow_enodeb_transmit',
        'mme_address', 'mme_port',
    ],
)


def config_assert(condition: bool, message: str = None) -> None:
    """ To be used in place of 'assert' so that ConfigurationError is raised
        for all config-related exceptions. """
    if not condition:
        raise ConfigurationError(message)


def build_desired_config(
        device_config: EnodebConfiguration,
        data_model: DataModel,
        post_processor: EnodebConfigurationPostProcessor,
) -> EnodebConfiguration:
    """
    Factory for initializing DESIRED data model configuration.

    When working with the configuration of an eNodeB, we track the
    current state of configuration for that device, as well as what
    configuration we want to set on the device.
    Args:
        device_config: The current configuration of the device.
        data_model: The data model for the device.
        post_processor: The post-processor to use for the device.
    Returns:
        Desired data model configuration for the device
    """

    # The configuration we want to push to the target device
    desired_configuration = EnodebConfiguration(data_model)
    # The configuration read from local configuration files by serial number specific
    customized_configuration = _get_enb_config(device_config)
    # device_config is the current configuration read from the target device

    # Check if the customized configuration is valid
    desired_configuration.check_desired_configuration(device_config, customized_configuration)
    # Apply the customized configuration to the target device
    desired_configuration.apply_desired_configuration(device_config, customized_configuration)

    post_processor.postprocess(desired_configuration)

    return desired_configuration


def _get_enb_config(
    device_config: EnodebConfiguration,
) -> dict:
    # The eNodeB parameters to be generated with default value,
    # It will load from eNB configs based on serial number or default value
    # The params is a nested list which contains 2 format of parameter names.
    # The first parameter is the name of eNB / ACS configuration in
    #   magma_configs/serial_number/ and magma_configs/acs_common.yml
    # The second parameter is the name of gateway configuration in 
    #   override_configs/gateway.mconfig
    def flatten_dictionary(dictionary: dict) -> dict:
        return {k: v for item in dictionary.values() for k, v in item.items()}

    params_dict = dict()

    # The common configuration file is loaded from magma_configs/acs_common.yml
    common_config = flatten_dictionary(load_common_config())
    
    # The serial number configuration file is loaded from magma_configs/serial_number/
    # Get the specific configuration with the serial number as the key
    enb_configs = load_enb_config()
    enb_serial = device_config.get_parameter(ParameterName.SERIAL_NUMBER)
    enb_config = enb_configs.get(enb_serial, dict())

    generated_config = dict()
    for parameter_name in ParameterName.all_paramters():
        parameter_value = enb_config.get(parameter_name, common_config.get(parameter_name))
        if parameter_value is not None:
            generated_config[parameter_name] = parameter_value

    return generated_config