state_machines/acs_state_utils.py - enodebd - Gitiles

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

 from typing import Any, Dict, List, Optional

 from data_models.data_model import DataModel
 from data_models.data_model_parameters import ParameterName
 from device_config.enodeb_configuration import EnodebConfiguration
 from devices.device_utils import EnodebDeviceName, get_device_name
 from exceptions import ConfigurationError
 from logger import EnodebdLogger as logger
 from tr069 import models


 def process_inform_message(
     inform: Any,
     data_model: DataModel,
     device_cfg: EnodebConfiguration,
 ) -> None:
     """
     Modifies the device configuration based on what is received in the Inform
     message. Will raise an error if it turns out that the data model we are
     using is incorrect. This is decided based on the device OUI and
     software-version that is reported in the Inform message.

     Args:
         inform: Inform Tr069 message
         device_handler: The state machine we are using for our device
     """
     param_values_by_path = _get_param_values_by_path(inform)
     param_name_list = data_model.get_parameter_names()
     name_to_val = {}
     for name in param_name_list:
         path = data_model.get_parameter(name).path
         if path in param_values_by_path:
             value = param_values_by_path[path]
             name_to_val[name] = value

     for name, val in name_to_val.items():
         device_cfg.set_parameter(name, val)


 def get_device_name_from_inform(
     inform: models.Inform,
 ) -> EnodebDeviceName:
     def _get_param_value_from_path_suffix(
         suffix: str,
         path_list: List[str],
         param_values_by_path: Dict[str, Any],
     ) -> Any:
         for path in path_list:
             if path.endswith(suffix):
                 return param_values_by_path[path]
         raise ConfigurationError('Did not receive expected info in Inform')

     param_values_by_path = _get_param_values_by_path(inform)

     # Check the OUI and version number to see if the data model matches
     path_list = list(param_values_by_path.keys())
     if hasattr(inform, 'DeviceId') and \
             hasattr(inform.DeviceId, 'OUI'):
         device_oui = inform.DeviceId.OUI
     else:
         device_oui = _get_param_value_from_path_suffix(
             'DeviceInfo.ManufacturerOUI',
             path_list,
             param_values_by_path,
         )
     sw_version = _get_param_value_from_path_suffix(
         'DeviceInfo.SoftwareVersion',
         path_list,
         param_values_by_path,
     )
     return get_device_name(device_oui, sw_version)


 def does_inform_have_event(
     inform: models.Inform,
     event_code: str,
 ) -> bool:
     """ True if the Inform message contains the specified event code """
     for event in inform.Event.EventStruct:
         if event.EventCode == event_code:
             return True
     return False


 def _get_param_values_by_path(
     inform: models.Inform,
 ) -> Dict[str, Any]:
     if not hasattr(inform, 'ParameterList') or \
             not hasattr(inform.ParameterList, 'ParameterValueStruct'):
         raise ConfigurationError('Did not receive ParamterList in Inform')
     param_values_by_path = {}
     for param_value in inform.ParameterList.ParameterValueStruct:
         path = param_value.Name
         value = param_value.Value.Data
         logger.debug(
             '(Inform msg) Received parameter: %s = %s', path,
             value,
         )
         param_values_by_path[path] = value
     return param_values_by_path


 def are_tr069_params_equal(param_a: Any, param_b: Any, type_: str) -> bool:
     """
     Compare two parameters in TR-069 format.
     The following differences are ignored:
     - Leading and trailing whitespace, commas and quotes
     - Capitalization, for booleans (true, false)
     Returns:
         True if params are the same
     """
     # Cast booleans to integers
     cmp_a, cmp_b = param_a, param_b
     if type_ == 'boolean' and cmp_b in ('0', '1') or cmp_a in ('0', '1'):
         cmp_a, cmp_b = map(int, (cmp_a, cmp_b))
     cmp_a, cmp_b = map(str, (cmp_a, cmp_b))
     cmp_a, cmp_b = map(lambda s: s.strip(', \'"'), (cmp_a, cmp_b))
     if cmp_a.lower() in ['true', 'false']:
         cmp_a, cmp_b = map(lambda s: s.lower(), (cmp_a, cmp_b))
     return cmp_a == cmp_b


 def get_all_objects_to_add(
     desired_cfg: EnodebConfiguration,
     device_cfg: EnodebConfiguration,
 ) -> List[ParameterName]:
     """
     Find a ParameterName that needs to be added to the eNB configuration,
     if any

     Note: This is the expected name of the parameter once it is added
           but this is different than how to add it. For example,
           enumerated objects of the form XX.YY.N. should be added
           by calling AddObject to XX.YY. and having the CPE assign
           the index.
     """
     desired = desired_cfg.get_object_names()
     current = device_cfg.get_object_names()
     return list(set(desired).difference(set(current)))


 def get_all_objects_to_delete(
     desired_cfg: EnodebConfiguration,
     device_cfg: EnodebConfiguration,
 ) -> List[ParameterName]:
     """
     Find a ParameterName that needs to be deleted from the eNB configuration,
     if any
     """
     desired = desired_cfg.get_object_names()
     current = device_cfg.get_object_names()
     return list(set(current).difference(set(desired)))


 def get_params_to_get(
     device_cfg: EnodebConfiguration,
     data_model: DataModel,
     request_all_params: bool = False,
 ) -> List[ParameterName]:
     """
     Returns the names of params not belonging to objects that are added/removed
     """
     desired_names = data_model.get_present_params()
     if request_all_params:
         return desired_names
     known_names = device_cfg.get_parameter_names()
     names = list(set(desired_names) - set(known_names))
     return names


 def get_object_params_to_get(
     desired_cfg: Optional[EnodebConfiguration],
     device_cfg: EnodebConfiguration,
     data_model: DataModel,
 ) -> List[ParameterName]:
     """
     Returns a list of parameter names for object parameters we don't know the
     current value of
     """
     names = []
     # TODO: This might a string for some strange reason, investigate why
     num_plmns = \
         int(device_cfg.get_parameter(ParameterName.NUM_PLMNS))
     for i in range(1, num_plmns + 1):
         obj_name = ParameterName.PLMN_N % i
         if not device_cfg.has_object(obj_name):
             device_cfg.add_object(obj_name)
         obj_to_params = data_model.get_numbered_param_names()
         desired = obj_to_params[obj_name]
         current = []
         if desired_cfg is not None:
             current = desired_cfg.get_parameter_names_for_object(obj_name)
         names_to_add = list(set(desired) - set(current))
         names = names + names_to_add
     return names


 # We don't attempt to set these parameters on the eNB configuration
 READ_ONLY_PARAMETERS = [
     ParameterName.OP_STATE,
     ParameterName.RF_TX_STATUS,
     ParameterName.GPS_STATUS,
     ParameterName.PTP_STATUS,
     ParameterName.MME_STATUS,
     ParameterName.GPS_LAT,
     ParameterName.GPS_LONG,
 ]


 def get_param_values_to_set(
     desired_cfg: EnodebConfiguration,
     device_cfg: EnodebConfiguration,
     data_model: DataModel,
     exclude_admin: bool = False,
 ) -> Dict[ParameterName, Any]:
     """
     Get a map of param names to values for parameters that we will
     set on the eNB's configuration, excluding parameters for objects that can
     be added/removed.

     Also exclude special parameters like admin state, since it may be set at
     a different time in the provisioning process than most parameters.
     """
     param_values = {}
     # Get the parameters we might set
     params = set(desired_cfg.get_parameter_names()) - set(READ_ONLY_PARAMETERS)
     if exclude_admin:
         params = set(params) - {ParameterName.ADMIN_STATE}
     # Values of parameters
     for name in params:
         new = desired_cfg.get_parameter(name)
         old = device_cfg.get_parameter(name)
         _type = data_model.get_parameter(name).type
         if not are_tr069_params_equal(new, old, _type):
             param_values[name] = new

     return param_values


 def get_obj_param_values_to_set(
     desired_cfg: EnodebConfiguration,
     device_cfg: EnodebConfiguration,
     data_model: DataModel,
 ) -> Dict[ParameterName, Dict[ParameterName, Any]]:
     """ Returns a map from object name to (a map of param name to value) """
     param_values = {}
     objs = desired_cfg.get_object_names()
     for obj_name in objs:
         param_values[obj_name] = {}
         params = desired_cfg.get_parameter_names_for_object(obj_name)
         for name in params:
             new = desired_cfg.get_parameter_for_object(name, obj_name)
             old = device_cfg.get_parameter_for_object(name, obj_name)
             _type = data_model.get_parameter(name).type
             if not are_tr069_params_equal(new, old, _type):
                 param_values[obj_name][name] = new
     return param_values


 def get_all_param_values_to_set(
     desired_cfg: EnodebConfiguration,
     device_cfg: EnodebConfiguration,
     data_model: DataModel,
     exclude_admin: bool = False,
 ) -> Dict[ParameterName, Any]:

     """ Returns a map of param names to values that we need to set """
     param_values = get_param_values_to_set(
         desired_cfg, device_cfg,
         data_model, exclude_admin,
     )
     obj_param_values = get_obj_param_values_to_set(
         desired_cfg, device_cfg,
         data_model,
     )
     for _obj_name, param_map in obj_param_values.items():
         for name, val in param_map.items():
             param_values[name] = val
     return param_values


 def parse_get_parameter_values_response(
     data_model: DataModel,
     message: models.GetParameterValuesResponse,
 ) -> Dict[ParameterName, Any]:
     """ Returns a map of ParameterName to the value read from the response """
     param_values_by_path = {}
     for param_value_struct in message.ParameterList.ParameterValueStruct:
         param_values_by_path[param_value_struct.Name] = \
             param_value_struct.Value.Data

     param_name_list = data_model.get_parameter_names()
     name_to_val = {}
     for name in param_name_list:
         path = data_model.get_parameter(name).path
         if path in param_values_by_path:
             value = param_values_by_path[path]
             name_to_val[name] = value

     return name_to_val


 def get_optional_param_to_check(
     data_model: DataModel,
 ) -> Optional[ParameterName]:
     """
     If there is a parameter which is optional in the data model, and we do not
     know if it exists or not, then return it so we can check for its presence.
     """
     params = data_model.get_names_of_optional_params()
     for param in params:
         try:
             data_model.is_parameter_present(param)
         except KeyError:
             return param
     return None
	# SPDX-FileCopyrightText: 2020 The Magma Authors.
	# SPDX-FileCopyrightText: 2022 Open Networking Foundation <support@opennetworking.org>
	#
	# SPDX-License-Identifier: BSD-3-Clause

	from typing import Any, Dict, List, Optional

	from data_models.data_model import DataModel
	from data_models.data_model_parameters import ParameterName
	from device_config.enodeb_configuration import EnodebConfiguration
	from devices.device_utils import EnodebDeviceName, get_device_name
	from exceptions import ConfigurationError
	from logger import EnodebdLogger as logger
	from tr069 import models


	def process_inform_message(
	inform: Any,
	data_model: DataModel,
	device_cfg: EnodebConfiguration,
	) -> None:
	"""
	Modifies the device configuration based on what is received in the Inform
	message. Will raise an error if it turns out that the data model we are
	using is incorrect. This is decided based on the device OUI and
	software-version that is reported in the Inform message.

	Args:
	inform: Inform Tr069 message
	device_handler: The state machine we are using for our device
	"""
	param_values_by_path = _get_param_values_by_path(inform)
	param_name_list = data_model.get_parameter_names()
	name_to_val = {}
	for name in param_name_list:
	path = data_model.get_parameter(name).path
	if path in param_values_by_path:
	value = param_values_by_path[path]
	name_to_val[name] = value

	for name, val in name_to_val.items():
	device_cfg.set_parameter(name, val)


	def get_device_name_from_inform(
	inform: models.Inform,
	) -> EnodebDeviceName:
	def _get_param_value_from_path_suffix(
	suffix: str,
	path_list: List[str],
	param_values_by_path: Dict[str, Any],
	) -> Any:
	for path in path_list:
	if path.endswith(suffix):
	return param_values_by_path[path]
	raise ConfigurationError('Did not receive expected info in Inform')

	param_values_by_path = _get_param_values_by_path(inform)

	# Check the OUI and version number to see if the data model matches
	path_list = list(param_values_by_path.keys())
	if hasattr(inform, 'DeviceId') and \
	hasattr(inform.DeviceId, 'OUI'):
	device_oui = inform.DeviceId.OUI
	else:
	device_oui = _get_param_value_from_path_suffix(
	'DeviceInfo.ManufacturerOUI',
	path_list,
	param_values_by_path,
	)
	sw_version = _get_param_value_from_path_suffix(
	'DeviceInfo.SoftwareVersion',
	path_list,
	param_values_by_path,
	)
	return get_device_name(device_oui, sw_version)


	def does_inform_have_event(
	inform: models.Inform,
	event_code: str,
	) -> bool:
	""" True if the Inform message contains the specified event code """
	for event in inform.Event.EventStruct:
	if event.EventCode == event_code:
	return True
	return False


	def _get_param_values_by_path(
	inform: models.Inform,
	) -> Dict[str, Any]:
	if not hasattr(inform, 'ParameterList') or \
	not hasattr(inform.ParameterList, 'ParameterValueStruct'):
	raise ConfigurationError('Did not receive ParamterList in Inform')
	param_values_by_path = {}
	for param_value in inform.ParameterList.ParameterValueStruct:
	path = param_value.Name
	value = param_value.Value.Data
	logger.debug(
	'(Inform msg) Received parameter: %s = %s', path,
	value,
	)
	param_values_by_path[path] = value
	return param_values_by_path


	def are_tr069_params_equal(param_a: Any, param_b: Any, type_: str) -> bool:
	"""
	Compare two parameters in TR-069 format.
	The following differences are ignored:
	- Leading and trailing whitespace, commas and quotes
	- Capitalization, for booleans (true, false)
	Returns:
	True if params are the same
	"""
	# Cast booleans to integers
	cmp_a, cmp_b = param_a, param_b
	if type_ == 'boolean' and cmp_b in ('0', '1') or cmp_a in ('0', '1'):
	cmp_a, cmp_b = map(int, (cmp_a, cmp_b))
	cmp_a, cmp_b = map(str, (cmp_a, cmp_b))
	cmp_a, cmp_b = map(lambda s: s.strip(', \'"'), (cmp_a, cmp_b))
	if cmp_a.lower() in ['true', 'false']:
	cmp_a, cmp_b = map(lambda s: s.lower(), (cmp_a, cmp_b))
	return cmp_a == cmp_b


	def get_all_objects_to_add(
	desired_cfg: EnodebConfiguration,
	device_cfg: EnodebConfiguration,
	) -> List[ParameterName]:
	"""
	Find a ParameterName that needs to be added to the eNB configuration,
	if any

	Note: This is the expected name of the parameter once it is added
	but this is different than how to add it. For example,
	enumerated objects of the form XX.YY.N. should be added
	by calling AddObject to XX.YY. and having the CPE assign
	the index.
	"""
	desired = desired_cfg.get_object_names()
	current = device_cfg.get_object_names()
	return list(set(desired).difference(set(current)))


	def get_all_objects_to_delete(
	desired_cfg: EnodebConfiguration,
	device_cfg: EnodebConfiguration,
	) -> List[ParameterName]:
	"""
	Find a ParameterName that needs to be deleted from the eNB configuration,
	if any
	"""
	desired = desired_cfg.get_object_names()
	current = device_cfg.get_object_names()
	return list(set(current).difference(set(desired)))


	def get_params_to_get(
	device_cfg: EnodebConfiguration,
	data_model: DataModel,
	request_all_params: bool = False,
	) -> List[ParameterName]:
	"""
	Returns the names of params not belonging to objects that are added/removed
	"""
	desired_names = data_model.get_present_params()
	if request_all_params:
	return desired_names
	known_names = device_cfg.get_parameter_names()
	names = list(set(desired_names) - set(known_names))
	return names


	def get_object_params_to_get(
	desired_cfg: Optional[EnodebConfiguration],
	device_cfg: EnodebConfiguration,
	data_model: DataModel,
	) -> List[ParameterName]:
	"""
	Returns a list of parameter names for object parameters we don't know the
	current value of
	"""
	names = []
	# TODO: This might a string for some strange reason, investigate why
	num_plmns = \
	int(device_cfg.get_parameter(ParameterName.NUM_PLMNS))
	for i in range(1, num_plmns + 1):
	obj_name = ParameterName.PLMN_N % i
	if not device_cfg.has_object(obj_name):
	device_cfg.add_object(obj_name)
	obj_to_params = data_model.get_numbered_param_names()
	desired = obj_to_params[obj_name]
	current = []
	if desired_cfg is not None:
	current = desired_cfg.get_parameter_names_for_object(obj_name)
	names_to_add = list(set(desired) - set(current))
	names = names + names_to_add
	return names


	# We don't attempt to set these parameters on the eNB configuration
	READ_ONLY_PARAMETERS = [
	ParameterName.OP_STATE,
	ParameterName.RF_TX_STATUS,
	ParameterName.GPS_STATUS,
	ParameterName.PTP_STATUS,
	ParameterName.MME_STATUS,
	ParameterName.GPS_LAT,
	ParameterName.GPS_LONG,
	]


	def get_param_values_to_set(
	desired_cfg: EnodebConfiguration,
	device_cfg: EnodebConfiguration,
	data_model: DataModel,
	exclude_admin: bool = False,
	) -> Dict[ParameterName, Any]:
	"""
	Get a map of param names to values for parameters that we will
	set on the eNB's configuration, excluding parameters for objects that can
	be added/removed.

	Also exclude special parameters like admin state, since it may be set at
	a different time in the provisioning process than most parameters.
	"""
	param_values = {}
	# Get the parameters we might set
	params = set(desired_cfg.get_parameter_names()) - set(READ_ONLY_PARAMETERS)
	if exclude_admin:
	params = set(params) - {ParameterName.ADMIN_STATE}
	# Values of parameters
	for name in params:
	new = desired_cfg.get_parameter(name)
	old = device_cfg.get_parameter(name)
	_type = data_model.get_parameter(name).type
	if not are_tr069_params_equal(new, old, _type):
	param_values[name] = new

	return param_values


	def get_obj_param_values_to_set(
	desired_cfg: EnodebConfiguration,
	device_cfg: EnodebConfiguration,
	data_model: DataModel,
	) -> Dict[ParameterName, Dict[ParameterName, Any]]:
	""" Returns a map from object name to (a map of param name to value) """
	param_values = {}
	objs = desired_cfg.get_object_names()
	for obj_name in objs:
	param_values[obj_name] = {}
	params = desired_cfg.get_parameter_names_for_object(obj_name)
	for name in params:
	new = desired_cfg.get_parameter_for_object(name, obj_name)
	old = device_cfg.get_parameter_for_object(name, obj_name)
	_type = data_model.get_parameter(name).type
	if not are_tr069_params_equal(new, old, _type):
	param_values[obj_name][name] = new
	return param_values


	def get_all_param_values_to_set(
	desired_cfg: EnodebConfiguration,
	device_cfg: EnodebConfiguration,
	data_model: DataModel,
	exclude_admin: bool = False,
	) -> Dict[ParameterName, Any]:

	""" Returns a map of param names to values that we need to set """
	param_values = get_param_values_to_set(
	desired_cfg, device_cfg,
	data_model, exclude_admin,
	)
	obj_param_values = get_obj_param_values_to_set(
	desired_cfg, device_cfg,
	data_model,
	)
	for _obj_name, param_map in obj_param_values.items():
	for name, val in param_map.items():
	param_values[name] = val
	return param_values


	def parse_get_parameter_values_response(
	data_model: DataModel,
	message: models.GetParameterValuesResponse,
	) -> Dict[ParameterName, Any]:
	""" Returns a map of ParameterName to the value read from the response """
	param_values_by_path = {}
	for param_value_struct in message.ParameterList.ParameterValueStruct:
	param_values_by_path[param_value_struct.Name] = \
	param_value_struct.Value.Data

	param_name_list = data_model.get_parameter_names()
	name_to_val = {}
	for name in param_name_list:
	path = data_model.get_parameter(name).path
	if path in param_values_by_path:
	value = param_values_by_path[path]
	name_to_val[name] = value

	return name_to_val


	def get_optional_param_to_check(
	data_model: DataModel,
	) -> Optional[ParameterName]:
	"""
	If there is a parameter which is optional in the data model, and we do not
	know if it exists or not, then return it so we can check for its presence.
	"""
	params = data_model.get_names_of_optional_params()
	for param in params:
	try:
	data_model.is_parameter_present(param)
	except KeyError:
	return param
	return None