import inspect
from .exceptions import InvalidConfigError, RegistrationError
from .util.dict import filter_dict
[docs]class TargetFactory:
[docs] def __init__(self):
self.resources = {}
self.drivers = {}
self.all_classes = {}
[docs] def reg_resource(self, cls):
"""Register a resource with the factory.
Returns the class to allow using it as a decorator."""
cls_name = cls.__name__
if cls_name in self.all_classes:
raise RegistrationError("resource with name {} was already registered".format(cls_name))
self.resources[cls.__name__] = cls
self._insert_into_all(cls)
return cls
[docs] def reg_driver(self, cls):
"""Register a driver with the factory.
Returns the class to allow using it as a decorator."""
cls_name = cls.__name__
if cls_name in self.all_classes:
raise RegistrationError("driver with name {} was already registered".format(cls_name))
self.drivers[cls_name] = cls
self._insert_into_all(cls)
return cls
@staticmethod
def _convert_to_named_list(data):
"""Convert a tree of resources or drivers to a named list.
When using named resources or drivers, the config file uses a list of
dicts instead of simply nested dicts. This allows creating multiple
instances of the same class with different names.
resources: # or drivers
FooPort: {}
BarPort:
name: "bar"
or
resources: # or drivers
- FooPort: {}
- BarPort:
name: "bar"
should be transformed to
resources: # or drivers
- cls: "FooPort"
- cls: "BarPort"
name: "bar"
"""
# resolve syntactic sugar (list of dicts each containing a dict of key -> args)
result = []
if isinstance(data, list):
for item in data:
if not isinstance(item, dict):
raise InvalidConfigError(
"invalid list item type {} (should be dict)".format(type(item)))
if not item:
raise InvalidConfigError("invalid empty dict as list item")
if len(item) > 1:
if 'cls' in item:
item = item.copy()
else:
raise InvalidConfigError("missing 'cls' key in {}".format(item))
else:
# only one pair left
(key, value), = item.items()
if key == 'cls':
item = item.copy()
else:
item = {'cls': key}
if value is None:
raise InvalidConfigError("invalid list item, add empty dict for no arguments") # pylint: disable=line-too-long
item.update(value)
result.append(item)
elif isinstance(data, dict):
for cls, args in data.items():
args.setdefault('cls', cls)
result.append(args)
else:
raise InvalidConfigError("invalid type {} (should be dict or list)".format(type(data)))
for item in result:
item.setdefault('name', None)
assert 'cls' in item
return result
[docs] @staticmethod
def normalize_config(config):
resources = {}
drivers = {}
for item in TargetFactory._convert_to_named_list(config.get('resources', {})):
resource = item.pop('cls')
name = item.pop('name', None)
args = item # remaining args
resources.setdefault(resource, {})[name] = (args, )
for item in TargetFactory._convert_to_named_list(config.get('drivers', {})):
driver = item.pop('cls')
name = item.pop('name', None)
bindings = item.pop('bindings', {})
args = item # remaining args
drivers.setdefault(driver, {})[name] = (args, bindings)
return resources, drivers
[docs] def make_resource(self, target, resource, name, args):
assert isinstance(args, dict)
if not resource in self.resources:
raise InvalidConfigError("unknown resource class {}".format(resource))
try:
cls = self.resources[resource]
args = filter_dict(args, cls, warn=True)
r = cls(target, name, **args)
except TypeError as e:
raise InvalidConfigError(
"failed to create {} for target '{}' using {} ".format(
resource, target, args)) from e
return r
[docs] def make_driver(self, target, driver, name, args):
assert isinstance(args, dict)
if not driver in self.drivers:
raise InvalidConfigError("unknown driver class {}".format(driver))
try:
cls = self.drivers[driver]
args = filter_dict(args, cls, warn=True)
d = cls(target, name, **args)
except TypeError as e:
raise InvalidConfigError(
"failed to create {} for target '{}' using {} ".format(
driver, target, args)) from e
return d
[docs] def make_target(self, name, config, *, env=None):
from .target import Target
target = Target(name, env=env)
for item in TargetFactory._convert_to_named_list(config.get('resources', {})):
resource = item.pop('cls')
name = item.pop('name', None)
args = item # remaining args
self.make_resource(target, resource, name, args)
for item in TargetFactory._convert_to_named_list(config.get('drivers', {})):
driver = item.pop('cls')
name = item.pop('name', None)
bindings = item.pop('bindings', {})
args = item # remaining args
target.set_binding_map(bindings)
self.make_driver(target, driver, name, args)
return target
[docs] def class_from_string(self, string: str):
try:
return self.all_classes[string]
except KeyError:
raise KeyError("No driver/resource/protocol of type '{}' in factory, perhaps not registered?".format(string))
def _insert_into_all(self, cls):
classes = inspect.getmro(cls)
for cl in classes:
if not self.all_classes.get(cl.__name__):
self.all_classes[cl.__name__] = cl
#: Global TargetFactory instance
#:
#: This instance is used to register Resource and Driver classes so that
#: Targets can be created automatically from YAML files.
target_factory = TargetFactory()