mirror of
https://github.com/ARM-software/workload-automation.git
synced 2024-10-06 02:41:11 +01:00
aa4df95a69
PEP8 has switched its guidance [1] for where a line break should occur in relation to a binary operator, so don't raise this warning for new code and update the code base to follow the new style. [1] https://www.python.org/dev/peps/pep-0008/#should-a-line-break-before-or-after-a-binary-operator
319 lines
14 KiB
Python
319 lines
14 KiB
Python
# Copyright 2016-2018 ARM Limited
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from copy import copy
|
|
from collections import defaultdict
|
|
from itertools import chain
|
|
|
|
from devlib.utils.misc import memoized
|
|
|
|
from wa.framework import pluginloader
|
|
from wa.framework.configuration.core import get_config_point_map
|
|
from wa.framework.exception import ConfigError, NotFoundError
|
|
from wa.framework.target.descriptor import list_target_descriptions
|
|
from wa.utils.types import obj_dict, caseless_string
|
|
|
|
GENERIC_CONFIGS = ["device_config", "workload_parameters",
|
|
"boot_parameters", "runtime_parameters"]
|
|
|
|
|
|
class PluginCache(object):
|
|
"""
|
|
The plugin cache is used to store configuration that cannot be processed at
|
|
this stage, whether thats because it is unknown if its needed
|
|
(in the case of disabled plug-ins) or it is not know what it belongs to (in
|
|
the case of "device-config" ect.). It also maintains where configuration came
|
|
from, and the priority order of said sources.
|
|
"""
|
|
|
|
def __init__(self, loader=pluginloader):
|
|
self.loader = loader
|
|
self.sources = []
|
|
self.plugin_configs = defaultdict(lambda: defaultdict(dict))
|
|
self.global_alias_values = defaultdict(dict)
|
|
self.targets = {td.name: td for td in list_target_descriptions()}
|
|
|
|
# Generate a mapping of what global aliases belong to
|
|
self._global_alias_map = defaultdict(dict)
|
|
self._list_of_global_aliases = set()
|
|
for plugin in self.loader.list_plugins():
|
|
for param in plugin.parameters:
|
|
if param.global_alias:
|
|
self._global_alias_map[plugin.name][param.global_alias] = param
|
|
self._list_of_global_aliases.add(param.global_alias)
|
|
|
|
def add_source(self, source):
|
|
if source in self.sources:
|
|
msg = "Source '{}' has already been added."
|
|
raise Exception(msg.format(source))
|
|
self.sources.append(source)
|
|
|
|
def add_global_alias(self, alias, value, source):
|
|
if source not in self.sources:
|
|
msg = "Source '{}' has not been added to the plugin cache."
|
|
raise RuntimeError(msg.format(source))
|
|
|
|
if not self.is_global_alias(alias):
|
|
msg = "'{} is not a valid global alias'"
|
|
raise RuntimeError(msg.format(alias))
|
|
|
|
self.global_alias_values[alias][source] = value
|
|
|
|
def add_configs(self, plugin_name, values, source):
|
|
if self.is_global_alias(plugin_name):
|
|
self.add_global_alias(plugin_name, values, source)
|
|
return
|
|
|
|
if source not in self.sources:
|
|
msg = "Source '{}' has not been added to the plugin cache."
|
|
raise RuntimeError(msg.format(source))
|
|
|
|
if caseless_string(plugin_name) in ['global', 'config']:
|
|
msg = '"{}" entry specified inside config/global section; If this is ' \
|
|
'defined in a config file, move the entry content into the top level'
|
|
raise ConfigError(msg.format((plugin_name)))
|
|
|
|
if (not self.loader.has_plugin(plugin_name)
|
|
and plugin_name not in self.targets
|
|
and plugin_name not in GENERIC_CONFIGS):
|
|
msg = 'configuration provided for unknown plugin "{}"'
|
|
raise ConfigError(msg.format(plugin_name))
|
|
|
|
if not hasattr(values, 'items'):
|
|
msg = 'Plugin configuration for "{}" not a dictionary ({} is {})'
|
|
raise ConfigError(msg.format(plugin_name, repr(values), type(values)))
|
|
|
|
for name, value in values.items():
|
|
if (plugin_name not in GENERIC_CONFIGS
|
|
and name not in self.get_plugin_parameters(plugin_name)):
|
|
msg = "'{}' is not a valid parameter for '{}'"
|
|
raise ConfigError(msg.format(name, plugin_name))
|
|
|
|
self.plugin_configs[plugin_name][source][name] = value
|
|
|
|
def is_global_alias(self, name):
|
|
return name in self._list_of_global_aliases
|
|
|
|
def list_plugins(self, kind=None):
|
|
return self.loader.list_plugins(kind)
|
|
|
|
def get_plugin_config(self, plugin_name, generic_name=None, is_final=True):
|
|
config = obj_dict(not_in_dict=['name'])
|
|
config.name = plugin_name
|
|
|
|
if plugin_name not in GENERIC_CONFIGS:
|
|
self._set_plugin_defaults(plugin_name, config)
|
|
self._set_from_global_aliases(plugin_name, config)
|
|
|
|
if generic_name is None:
|
|
# Perform a simple merge with the order of sources representing
|
|
# priority
|
|
plugin_config = self.plugin_configs[plugin_name]
|
|
cfg_points = self.get_plugin_parameters(plugin_name)
|
|
for source in self.sources:
|
|
if source not in plugin_config:
|
|
continue
|
|
for name, value in plugin_config[source].items():
|
|
cfg_points[name].set_value(config, value=value)
|
|
else:
|
|
# A more complicated merge that involves priority of sources and
|
|
# specificity
|
|
self._merge_using_priority_specificity(plugin_name, generic_name,
|
|
config, is_final)
|
|
|
|
return config
|
|
|
|
def get_plugin(self, name, kind=None, *args, **kwargs):
|
|
config = self.get_plugin_config(name)
|
|
kwargs = dict(list(config.items()) + list(kwargs.items()))
|
|
return self.loader.get_plugin(name, kind=kind, *args, **kwargs)
|
|
|
|
def get_plugin_class(self, name, kind=None):
|
|
return self.loader.get_plugin_class(name, kind)
|
|
|
|
@memoized
|
|
def get_plugin_parameters(self, name):
|
|
if name in self.targets:
|
|
return self._get_target_params(name)
|
|
params = self.loader.get_plugin_class(name).parameters
|
|
return get_config_point_map(params)
|
|
|
|
def resolve_alias(self, name):
|
|
return self.loader.resolve_alias(name)
|
|
|
|
def _set_plugin_defaults(self, plugin_name, config):
|
|
cfg_points = self.get_plugin_parameters(plugin_name)
|
|
for cfg_point in cfg_points.values():
|
|
cfg_point.set_value(config, check_mandatory=False)
|
|
|
|
try:
|
|
_, alias_params = self.resolve_alias(plugin_name)
|
|
for name, value in alias_params.items():
|
|
cfg_points[name].set_value(config, value)
|
|
except NotFoundError:
|
|
pass
|
|
|
|
def _set_from_global_aliases(self, plugin_name, config):
|
|
for alias, param in self._global_alias_map[plugin_name].items():
|
|
if alias in self.global_alias_values:
|
|
for source in self.sources:
|
|
if source not in self.global_alias_values[alias]:
|
|
continue
|
|
val = self.global_alias_values[alias][source]
|
|
param.set_value(config, value=val)
|
|
|
|
def _get_target_params(self, name):
|
|
td = self.targets[name]
|
|
return get_config_point_map(chain(td.target_params, td.platform_params, td.conn_params, td.assistant_params))
|
|
|
|
# pylint: disable=too-many-nested-blocks, too-many-branches
|
|
def _merge_using_priority_specificity(self, specific_name,
|
|
generic_name, merged_config, is_final=True):
|
|
"""
|
|
WA configuration can come from various sources of increasing priority,
|
|
as well as being specified in a generic and specific manner (e.g
|
|
``device_config`` and ``nexus10`` respectivly). WA has two rules for
|
|
the priority of configuration:
|
|
|
|
- Configuration from higher priority sources overrides
|
|
configuration from lower priority sources.
|
|
- More specific configuration overrides less specific configuration.
|
|
|
|
There is a situation where these two rules come into conflict. When a
|
|
generic configuration is given in config source of high priority and a
|
|
specific configuration is given in a config source of lower priority.
|
|
In this situation it is not possible to know the end users intention
|
|
and WA will error.
|
|
|
|
:param specific_name: The name of the specific configuration used
|
|
e.g ``nexus10``
|
|
:param generic_name: The name of the generic configuration
|
|
e.g ``device_config``
|
|
:param merge_config: A dict of ``ConfigurationPoint``s to be used when
|
|
merging configuration. keys=config point name,
|
|
values=config point
|
|
:param is_final: if ``True`` (the default) make sure that mandatory
|
|
parameters are set.
|
|
|
|
:rtype: A fully merged and validated configuration in the form of a
|
|
obj_dict.
|
|
"""
|
|
ms = MergeState()
|
|
ms.generic_name = generic_name
|
|
ms.specific_name = specific_name
|
|
ms.generic_config = copy(self.plugin_configs[generic_name])
|
|
ms.specific_config = copy(self.plugin_configs[specific_name])
|
|
ms.cfg_points = self.get_plugin_parameters(specific_name)
|
|
sources = self.sources
|
|
|
|
# set_value uses the 'name' attribute of the passed object in it error
|
|
# messages, to ensure these messages make sense the name will have to be
|
|
# changed several times during this function.
|
|
merged_config.name = specific_name
|
|
|
|
for source in sources:
|
|
try:
|
|
update_config_from_source(merged_config, source, ms)
|
|
except ConfigError as e:
|
|
raise ConfigError('Error in "{}":\n\t{}'.format(source, str(e)))
|
|
|
|
# Validate final configuration
|
|
merged_config.name = specific_name
|
|
for cfg_point in ms.cfg_points.values():
|
|
cfg_point.validate(merged_config, check_mandatory=is_final)
|
|
|
|
def __getattr__(self, name):
|
|
"""
|
|
This resolves methods for specific plugins types based on corresponding
|
|
generic plugin methods. So it's possible to say things like ::
|
|
|
|
loader.get_device('foo')
|
|
|
|
instead of ::
|
|
|
|
loader.get_plugin('foo', kind='device')
|
|
|
|
"""
|
|
error_msg = 'No plugins of type "{}" discovered'
|
|
if name.startswith('get_'):
|
|
name = name.replace('get_', '', 1)
|
|
if name in self.loader.kind_map:
|
|
def __wrapper(pname, *args, **kwargs):
|
|
return self.get_plugin(pname, name, *args, **kwargs)
|
|
return __wrapper
|
|
raise NotFoundError(error_msg.format(name))
|
|
if name.startswith('list_'):
|
|
name = name.replace('list_', '', 1).rstrip('s')
|
|
if name in self.loader.kind_map:
|
|
def __wrapper(*args, **kwargs): # pylint: disable=E0102
|
|
return self.list_plugins(name, *args, **kwargs)
|
|
return __wrapper
|
|
raise NotFoundError(error_msg.format(name))
|
|
if name.startswith('has_'):
|
|
name = name.replace('has_', '', 1)
|
|
if name in self.loader.kind_map:
|
|
def __wrapper(pname, *args, **kwargs): # pylint: disable=E0102
|
|
return self.loader.has_plugin(pname, name, *args, **kwargs)
|
|
return __wrapper
|
|
raise NotFoundError(error_msg.format(name))
|
|
raise AttributeError(name)
|
|
|
|
|
|
class MergeState(object):
|
|
|
|
def __init__(self):
|
|
self.generic_name = None
|
|
self.specific_name = None
|
|
self.generic_config = None
|
|
self.specific_config = None
|
|
self.cfg_points = None
|
|
self.seen_specific_config = defaultdict(list)
|
|
|
|
|
|
def update_config_from_source(final_config, source, state):
|
|
if source in state.generic_config:
|
|
final_config.name = state.generic_name
|
|
for name, cfg_point in state.cfg_points.items():
|
|
if name in state.generic_config[source]:
|
|
if name in state.seen_specific_config:
|
|
msg = ('"{generic_name}" configuration "{config_name}" has '
|
|
'already been specified more specifically for '
|
|
'{specific_name} in:\n\t\t{sources}')
|
|
seen_sources = state.seen_specific_config[name]
|
|
msg = msg.format(generic_name=state.generic_name,
|
|
config_name=name,
|
|
specific_name=state.specific_name,
|
|
sources=", ".join(seen_sources))
|
|
raise ConfigError(msg)
|
|
value = state.generic_config[source].pop(name)
|
|
cfg_point.set_value(final_config, value, check_mandatory=False)
|
|
|
|
if state.generic_config[source]:
|
|
msg = 'Unexpected values for {}: {}'
|
|
raise ConfigError(msg.format(state.generic_name,
|
|
state.generic_config[source]))
|
|
|
|
if source in state.specific_config:
|
|
final_config.name = state.specific_name
|
|
for name, cfg_point in state.cfg_points.items():
|
|
if name in state.specific_config[source]:
|
|
state.seen_specific_config[name].append(str(source))
|
|
value = state.specific_config[source].pop(name)
|
|
cfg_point.set_value(final_config, value, check_mandatory=False)
|
|
|
|
if state.specific_config[source]:
|
|
msg = 'Unexpected values for {}: {}'
|
|
raise ConfigError(msg.format(state.specific_name,
|
|
state.specific_config[source]))
|