device_config/enodeb_configuration.py - enodebd - Gitiles

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

 import json
 from typing import Any, List

 from data_models.data_model import DataModel
 from data_models.data_model_parameters import ParameterName

 from collections import namedtuple

 from lte_utils import DuplexMode, map_earfcndl_to_band_earfcnul_mode

 from exceptions import ConfigurationError
 from logger import EnodebdLogger as logger


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

 class EnodebConfiguration():
     """
     This represents the data model configuration for a single
     eNodeB device. This can correspond to either the current configuration
     of the device, or what configuration we desire to have for the device.
     """

     def __init__(self, data_model: DataModel) -> None:
         """
         The fields initialized in the constructor here should be enough to
         track state across any data model configuration.

         Most objects for eNodeB data models cannot be added or deleted.
         For those objects, we just track state with a simple mapping from
         parameter name to value.

         For objects which can be added/deleted, we track them separately.
         """

         # DataModel
         self._data_model = data_model

         # Dict[ParameterName, Any]
         self._param_to_value = {}

         # Dict[ParameterName, Dict[ParameterName, Any]]
         self._numbered_objects = {}
         # If adding a PLMN object, then you would set something like
         # self._numbered_objects['PLMN_1'] = {'PLMN_1_ENABLED': True}

     @property
     def data_model(self) -> DataModel:
         """
         The data model configuration is tied to a single data model
         """
         return self._data_model

     def get_parameter_names(self) -> List[ParameterName]:
         """
         Returns: list of ParameterName
         """
         return list(self._param_to_value.keys())

     def has_parameter(self, param_name: ParameterName) -> bool:
         return param_name in self._param_to_value

     def get_parameter(self, param_name: ParameterName) -> Any:
         """
         Args:
             param_name: ParameterName
         Returns:
             Any, value of the parameter, formatted to be understood by enodebd
         """
         self._assert_param_in_model(param_name)
         return self._param_to_value[param_name]

     def set_parameter(
         self,
         param_name: ParameterName,
         value: Any,
     ) -> None:
         """
         Args:
             param_name: the parameter name to configure
             value: the value to set, formatted to be understood by enodebd
         """
         self._assert_param_in_model(param_name)
         self._param_to_value[param_name] = value

     def set_parameter_if_present(self, parameter_name: ParameterName, value: Any) -> None:
         """
         Args:
             param_name: the parameter name to configure
             value: the value to set, formatted to be understood by enodebd
         """

         trparam_model = self.data_model
         tr_param = trparam_model.get_parameter(parameter_name)
         if tr_param is not None:
             self._param_to_value[parameter_name] = value

     def delete_parameter(self, param_name: ParameterName) -> None:
         del self._param_to_value[param_name]

     def get_object_names(self) -> List[ParameterName]:
         return list(self._numbered_objects.keys())

     def has_object(self, param_name: ParameterName) -> bool:
         """
         Args:
             param_name: The ParameterName of the object
         Returns: True if set in configuration
         """
         self._assert_param_in_model(param_name)
         return param_name in self._numbered_objects

     def add_object(self, param_name: ParameterName) -> None:
         if param_name in self._numbered_objects:
             raise ConfigurationError("Configuration already has object")
         self._numbered_objects[param_name] = {}

     def delete_object(self, param_name: ParameterName) -> None:
         if param_name not in self._numbered_objects:
             raise ConfigurationError("Configuration does not have object")
         del self._numbered_objects[param_name]

     def get_parameter_for_object(
         self,
         param_name: ParameterName,
         object_name: ParameterName,
     ) -> Any:
         return self._numbered_objects[object_name].get(param_name)

     def set_parameter_for_object(
         self,
         param_name: ParameterName,
         value: Any,
         object_name: ParameterName,
     ) -> None:
         """
         Args:
             param_name: the parameter name to configure
             value: the value to set, formatted to be understood by enodebd
             object_name: ParameterName of object
         """
         self._assert_param_in_model(object_name)
         self._assert_param_in_model(param_name)
         self._numbered_objects[object_name][param_name] = value

     def get_parameter_names_for_object(
         self,
         object_name: ParameterName,
     ) -> List[ParameterName]:
         return list(self._numbered_objects[object_name].keys())

     def get_debug_info(self) -> str:
         debug_info = 'Param values: {}, \n Object values: {}'
         return debug_info.format(
             json.dumps(self._param_to_value, indent=2),
             json.dumps(
                 self._numbered_objects,
                 indent=2,
             ),
         )

     def _assert_param_in_model(self, param_name: ParameterName) -> None:
         trparam_model = self.data_model
         tr_param = trparam_model.get_parameter(param_name)
         if tr_param is None:
             logger.warning('Parameter <%s> not defined in model', param_name)
             raise ConfigurationError("Parameter %s not defined in model." % param_name)

     def check_desired_configuration(self, current_config, desired_config: dict) -> bool:
         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)

         # _set_earfcn_freq_band_mode
         # Originally:
         # mconfig: loaded from proto definiation
         # service_config: loaded from mconfig
         # device_config: retrieved configuration from enodeb
         # data_model: defined in device parameter

         # device_config = config loaded from enodeb, desired_config = config loaded from file

         # _check_earfcn_freqw_band_mode
         try:
             band, duplex_mode, _ = map_earfcndl_to_band_earfcnul_mode(desired_config["earfcn_downlink1"])
         except ValueError as err:
             raise ConfigurationError(err)

         if current_config.has_parameter(ParameterName.DUPLEX_MODE_CAPABILITY):
             duplex_capability = current_config.get_parameter(ParameterName.DUPLEX_MODE_CAPABILITY)
             if duplex_mode == DuplexMode.TDD and duplex_capability != "TDDMode":
                 raise ConfigurationError("Duplex mode TDD is not supported by eNodeB")
             elif duplex_mode == DuplexMode.FDD and duplex_capability != "FDDMode":
                 raise ConfigurationError("Duplex mode FDD is not supported by eNodeB")
             elif duplex_mode not in [DuplexMode.TDD, DuplexMode.FDD]:
                 raise ConfigurationError("Invalid duplex mode")

         if current_config.has_parameter(ParameterName.BAND_CAPABILITY):
             band_capability = current_config.get_parameter(ParameterName.BAND_CAPABILITY).split(',')
             if str(band) not in band_capability:
                 logger.warning("Band %d not in capability list %s", band, band_capability)

         # _check_tdd_subframe_config
         config_assert(
             desired_config["subframe_assignment"] in range(0, 7),
             "Invalid TDD special subframe assignment (%d)" % desired_config["subframe_assignment"],
         )
         config_assert(
             desired_config["special_subframe_pattern"] in range(0, 10),
             "Invalid TDD special subframe pattern (%d)" % desired_config["special_subframe_pattern"],
         )

         # _check_plmnids_tac
         for char in str(desired_config["plmn_list"]):
             config_assert(char in "0123456789, ", "Invalid PLMNID (%s)" % desired_config["plmn_list"])

         # TODO - add support for multiple PLMNIDs
         plmnid_list = str(desired_config["plmn_list"]).split(",")
         config_assert(len(plmnid_list) == 1, "Only 1 PLMNID is supported")
         config_assert(len(plmnid_list[0]) <= 6, "PLMNID must be length <= 6 (%s)" % plmnid_list[0])

         # _check_s1_connection_configuration
         config_assert(type(desired_config["mme_address"]) is str, "Invalid MME address")
         config_assert(type(desired_config["mme_port"]) is int, "Invalid MME port")

     def apply_desired_configuration(self, current_config, desired_config: SingleEnodebConfig) -> None:

         # _set_earfcn_freq_band_mode
         self.set_parameter(ParameterName.EARFCNDL, desired_config["earfcn_downlink1"])
         band, duplex_mode, _ = map_earfcndl_to_band_earfcnul_mode(desired_config["earfcn_downlink1"])
         if duplex_mode == DuplexMode.FDD:
             self.set_parameter(ParameterName.EARFCNUL, desired_config["earfcn_uplink1"])
         self.set_parameter_if_present(ParameterName.BAND, band)

         # _set_tdd_subframe_config
         if (current_config.has_parameter(ParameterName.DUPLEX_MODE_CAPABILITY)
             and current_config.get_parameter(ParameterName.DUPLEX_MODE_CAPABILITY) == "TDDMode"):
             self.set_parameter(ParameterName.SUBFRAME_ASSIGNMENT, desired_config["subframe_assignment"])
             self.set_parameter(ParameterName.SPECIAL_SUBFRAME_PATTERN, desired_config["special_subframe_pattern"])

         # _set_plmnids_tac
         plmnid_list = str(desired_config["plmn_list"]).split(",")
         for i in range(1, 2):
             object_name = ParameterName.PLMN_N % i
             enable_plmn = i == 1
             self.add_object(object_name)
             self.set_parameter_for_object(ParameterName.PLMN_N_ENABLE % i, enable_plmn, object_name)
             if enable_plmn:
                 self.set_parameter_for_object(ParameterName.PLMN_N_CELL_RESERVED % i, False, object_name)
                 self.set_parameter_for_object(ParameterName.PLMN_N_PRIMARY % i, enable_plmn, object_name)
                 self.set_parameter_for_object(ParameterName.PLMN_N_PLMNID % i, plmnid_list[i - 1], object_name)
         self.set_parameter(ParameterName.TAC1, desired_config["tac1"])

         # _set_bandwidth
         self.set_parameter(ParameterName.DL_BANDWIDTH, desired_config["downlink_bandwidth"])
         self.set_parameter(ParameterName.UL_BANDWIDTH, desired_config["uplink_bandwidth"])

         # _set_cell_id
         self.set_parameter(ParameterName.CELL_ID, desired_config["cell_id"])

         # _set_misc_static_params
         self.set_parameter_if_present(ParameterName.LOCAL_GATEWAY_ENABLE, 0)
         self.set_parameter_if_present(ParameterName.GPS_ENABLE, True)
         self.set_parameter_if_present(ParameterName.IP_SEC_ENABLE, False)
         self.set_parameter_if_present(ParameterName.CELL_RESERVED, False)
         self.set_parameter_if_present(ParameterName.MME_POOL_ENABLE, False)

         # _set_s1_connection_configuration
         self.set_parameter(ParameterName.MME_ADDRESS, desired_config["mme_address"])
         self.set_parameter(ParameterName.MME_PORT, desired_config["mme_port"])

         # enable LTE if we should
         self.set_parameter(ParameterName.ADMIN_STATE, desired_config["admin_state"])

         # These parameters are already configured at above
         exclude_list = [
             "earfcn_downlink1", "earfcn_uplink1", "subframe_assignment", "special_subframe_pattern",
             "plmnid", "tac1", "downlink_bandwidth", "uplink_bandwidth", "cell_id",
             "mme_address", "mme_port", "admin_state"
         ]

         # Configure the additional parameters which are set in enodeb config files
         for name, value in desired_config.items():
             if name not in exclude_list:
                 self.set_parameter_if_present(name, value)
	# SPDX-FileCopyrightText: 2020 The Magma Authors.
	# SPDX-FileCopyrightText: 2022 Open Networking Foundation <support@opennetworking.org>
	#
	# SPDX-License-Identifier: BSD-3-Clause

	import json
	from typing import Any, List

	from data_models.data_model import DataModel
	from data_models.data_model_parameters import ParameterName

	from collections import namedtuple

	from lte_utils import DuplexMode, map_earfcndl_to_band_earfcnul_mode

	from exceptions import ConfigurationError
	from logger import EnodebdLogger as logger


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

	class EnodebConfiguration():
	"""
	This represents the data model configuration for a single
	eNodeB device. This can correspond to either the current configuration
	of the device, or what configuration we desire to have for the device.
	"""

	def __init__(self, data_model: DataModel) -> None:
	"""
	The fields initialized in the constructor here should be enough to
	track state across any data model configuration.

	Most objects for eNodeB data models cannot be added or deleted.
	For those objects, we just track state with a simple mapping from
	parameter name to value.

	For objects which can be added/deleted, we track them separately.
	"""

	# DataModel
	self._data_model = data_model

	# Dict[ParameterName, Any]
	self._param_to_value = {}

	# Dict[ParameterName, Dict[ParameterName, Any]]
	self._numbered_objects = {}
	# If adding a PLMN object, then you would set something like
	# self._numbered_objects['PLMN_1'] = {'PLMN_1_ENABLED': True}

	@property
	def data_model(self) -> DataModel:
	"""
	The data model configuration is tied to a single data model
	"""
	return self._data_model

	def get_parameter_names(self) -> List[ParameterName]:
	"""
	Returns: list of ParameterName
	"""
	return list(self._param_to_value.keys())

	def has_parameter(self, param_name: ParameterName) -> bool:
	return param_name in self._param_to_value

	def get_parameter(self, param_name: ParameterName) -> Any:
	"""
	Args:
	param_name: ParameterName
	Returns:
	Any, value of the parameter, formatted to be understood by enodebd
	"""
	self._assert_param_in_model(param_name)
	return self._param_to_value[param_name]

	def set_parameter(
	self,
	param_name: ParameterName,
	value: Any,
	) -> None:
	"""
	Args:
	param_name: the parameter name to configure
	value: the value to set, formatted to be understood by enodebd
	"""
	self._assert_param_in_model(param_name)
	self._param_to_value[param_name] = value

	def set_parameter_if_present(self, parameter_name: ParameterName, value: Any) -> None:
	"""
	Args:
	param_name: the parameter name to configure
	value: the value to set, formatted to be understood by enodebd
	"""

	trparam_model = self.data_model
	tr_param = trparam_model.get_parameter(parameter_name)
	if tr_param is not None:
	self._param_to_value[parameter_name] = value

	def delete_parameter(self, param_name: ParameterName) -> None:
	del self._param_to_value[param_name]

	def get_object_names(self) -> List[ParameterName]:
	return list(self._numbered_objects.keys())

	def has_object(self, param_name: ParameterName) -> bool:
	"""
	Args:
	param_name: The ParameterName of the object
	Returns: True if set in configuration
	"""
	self._assert_param_in_model(param_name)
	return param_name in self._numbered_objects

	def add_object(self, param_name: ParameterName) -> None:
	if param_name in self._numbered_objects:
	raise ConfigurationError("Configuration already has object")
	self._numbered_objects[param_name] = {}

	def delete_object(self, param_name: ParameterName) -> None:
	if param_name not in self._numbered_objects:
	raise ConfigurationError("Configuration does not have object")
	del self._numbered_objects[param_name]

	def get_parameter_for_object(
	self,
	param_name: ParameterName,
	object_name: ParameterName,
	) -> Any:
	return self._numbered_objects[object_name].get(param_name)

	def set_parameter_for_object(
	self,
	param_name: ParameterName,
	value: Any,
	object_name: ParameterName,
	) -> None:
	"""
	Args:
	param_name: the parameter name to configure
	value: the value to set, formatted to be understood by enodebd
	object_name: ParameterName of object
	"""
	self._assert_param_in_model(object_name)
	self._assert_param_in_model(param_name)
	self._numbered_objects[object_name][param_name] = value

	def get_parameter_names_for_object(
	self,
	object_name: ParameterName,
	) -> List[ParameterName]:
	return list(self._numbered_objects[object_name].keys())

	def get_debug_info(self) -> str:
	debug_info = 'Param values: {}, \n Object values: {}'
	return debug_info.format(
	json.dumps(self._param_to_value, indent=2),
	json.dumps(
	self._numbered_objects,
	indent=2,
	),
	)

	def _assert_param_in_model(self, param_name: ParameterName) -> None:
	trparam_model = self.data_model
	tr_param = trparam_model.get_parameter(param_name)
	if tr_param is None:
	logger.warning('Parameter <%s> not defined in model', param_name)
	raise ConfigurationError("Parameter %s not defined in model." % param_name)

	def check_desired_configuration(self, current_config, desired_config: dict) -> bool:
	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)

	# _set_earfcn_freq_band_mode
	# Originally:
	# mconfig: loaded from proto definiation
	# service_config: loaded from mconfig
	# device_config: retrieved configuration from enodeb
	# data_model: defined in device parameter

	# device_config = config loaded from enodeb, desired_config = config loaded from file

	# _check_earfcn_freqw_band_mode
	try:
	band, duplex_mode, _ = map_earfcndl_to_band_earfcnul_mode(desired_config["earfcn_downlink1"])
	except ValueError as err:
	raise ConfigurationError(err)

	if current_config.has_parameter(ParameterName.DUPLEX_MODE_CAPABILITY):
	duplex_capability = current_config.get_parameter(ParameterName.DUPLEX_MODE_CAPABILITY)
	if duplex_mode == DuplexMode.TDD and duplex_capability != "TDDMode":
	raise ConfigurationError("Duplex mode TDD is not supported by eNodeB")
	elif duplex_mode == DuplexMode.FDD and duplex_capability != "FDDMode":
	raise ConfigurationError("Duplex mode FDD is not supported by eNodeB")
	elif duplex_mode not in [DuplexMode.TDD, DuplexMode.FDD]:
	raise ConfigurationError("Invalid duplex mode")

	if current_config.has_parameter(ParameterName.BAND_CAPABILITY):
	band_capability = current_config.get_parameter(ParameterName.BAND_CAPABILITY).split(',')
	if str(band) not in band_capability:
	logger.warning("Band %d not in capability list %s", band, band_capability)

	# _check_tdd_subframe_config
	config_assert(
	desired_config["subframe_assignment"] in range(0, 7),
	"Invalid TDD special subframe assignment (%d)" % desired_config["subframe_assignment"],
	)
	config_assert(
	desired_config["special_subframe_pattern"] in range(0, 10),
	"Invalid TDD special subframe pattern (%d)" % desired_config["special_subframe_pattern"],
	)

	# _check_plmnids_tac
	for char in str(desired_config["plmn_list"]):
	config_assert(char in "0123456789, ", "Invalid PLMNID (%s)" % desired_config["plmn_list"])

	# TODO - add support for multiple PLMNIDs
	plmnid_list = str(desired_config["plmn_list"]).split(",")
	config_assert(len(plmnid_list) == 1, "Only 1 PLMNID is supported")
	config_assert(len(plmnid_list[0]) <= 6, "PLMNID must be length <= 6 (%s)" % plmnid_list[0])

	# _check_s1_connection_configuration
	config_assert(type(desired_config["mme_address"]) is str, "Invalid MME address")
	config_assert(type(desired_config["mme_port"]) is int, "Invalid MME port")

	def apply_desired_configuration(self, current_config, desired_config: SingleEnodebConfig) -> None:

	# _set_earfcn_freq_band_mode
	self.set_parameter(ParameterName.EARFCNDL, desired_config["earfcn_downlink1"])
	band, duplex_mode, _ = map_earfcndl_to_band_earfcnul_mode(desired_config["earfcn_downlink1"])
	if duplex_mode == DuplexMode.FDD:
	self.set_parameter(ParameterName.EARFCNUL, desired_config["earfcn_uplink1"])
	self.set_parameter_if_present(ParameterName.BAND, band)

	# _set_tdd_subframe_config
	if (current_config.has_parameter(ParameterName.DUPLEX_MODE_CAPABILITY)
	and current_config.get_parameter(ParameterName.DUPLEX_MODE_CAPABILITY) == "TDDMode"):
	self.set_parameter(ParameterName.SUBFRAME_ASSIGNMENT, desired_config["subframe_assignment"])
	self.set_parameter(ParameterName.SPECIAL_SUBFRAME_PATTERN, desired_config["special_subframe_pattern"])

	# _set_plmnids_tac
	plmnid_list = str(desired_config["plmn_list"]).split(",")
	for i in range(1, 2):
	object_name = ParameterName.PLMN_N % i
	enable_plmn = i == 1
	self.add_object(object_name)
	self.set_parameter_for_object(ParameterName.PLMN_N_ENABLE % i, enable_plmn, object_name)
	if enable_plmn:
	self.set_parameter_for_object(ParameterName.PLMN_N_CELL_RESERVED % i, False, object_name)
	self.set_parameter_for_object(ParameterName.PLMN_N_PRIMARY % i, enable_plmn, object_name)
	self.set_parameter_for_object(ParameterName.PLMN_N_PLMNID % i, plmnid_list[i - 1], object_name)
	self.set_parameter(ParameterName.TAC1, desired_config["tac1"])

	# _set_bandwidth
	self.set_parameter(ParameterName.DL_BANDWIDTH, desired_config["downlink_bandwidth"])
	self.set_parameter(ParameterName.UL_BANDWIDTH, desired_config["uplink_bandwidth"])

	# _set_cell_id
	self.set_parameter(ParameterName.CELL_ID, desired_config["cell_id"])

	# _set_misc_static_params
	self.set_parameter_if_present(ParameterName.LOCAL_GATEWAY_ENABLE, 0)
	self.set_parameter_if_present(ParameterName.GPS_ENABLE, True)
	self.set_parameter_if_present(ParameterName.IP_SEC_ENABLE, False)
	self.set_parameter_if_present(ParameterName.CELL_RESERVED, False)
	self.set_parameter_if_present(ParameterName.MME_POOL_ENABLE, False)

	# _set_s1_connection_configuration
	self.set_parameter(ParameterName.MME_ADDRESS, desired_config["mme_address"])
	self.set_parameter(ParameterName.MME_PORT, desired_config["mme_port"])

	# enable LTE if we should
	self.set_parameter(ParameterName.ADMIN_STATE, desired_config["admin_state"])

	# These parameters are already configured at above
	exclude_list = [
	"earfcn_downlink1", "earfcn_uplink1", "subframe_assignment", "special_subframe_pattern",
	"plmnid", "tac1", "downlink_bandwidth", "uplink_bandwidth", "cell_id",
	"mme_address", "mme_port", "admin_state"
	]

	# Configure the additional parameters which are set in enodeb config files
	for name, value in desired_config.items():
	if name not in exclude_list:
	self.set_parameter_if_present(name, value)