import os
import copy
import six
import importlib
import contextlib
import weakref
from collections import OrderedDict
from yggdrasil.doctools import docs2args
_registry = {}
_registry_complete = False
[docs]class ComponentError(BaseException):
r"""Error raised when there is a problem import a component."""
pass
# class ClassRegistry(OrderedDict):
# r"""Class for registering classes."""
# def __init__(self, *args, import_function=None, **kwargs):
# module = inspect.getmodule(inspect.stack()[1][0])
# self._module = module.__name__
# self._directory = os.path.dirname(module.__file__)
# self._import_function = import_function
# self._imported = False
# super(ClassRegistry, self).__init__(*args, **kwargs)
# def import_classes(self):
# r"""Import all classes in the same directory."""
# if self._imported:
# return
# self._imported = True
# for x in sorted(glob.glob(os.path.join(self._directory, '*.py'))):
# mod = os.path.basename(x)[:-3]
# if not mod.startswith('__'):
# importlib.import_module(self._module + '.%s' % mod)
# if self._import_function is not None:
# self._import_function()
# def keys(self, *args, **kwargs):
# self.import_classes()
# return super(ClassRegistry, self).keys(*args, **kwargs)
# def values(self, *args, **kwargs):
# self.import_classes()
# return super(ClassRegistry, self).values(*args, **kwargs)
# def items(self, *args, **kwargs):
# self.import_classes()
# return super(ClassRegistry, self).items(*args, **kwargs)
# def __contains__(self, key):
# self.import_classes()
# return super(ClassRegistry, self).__contains__(key)
# def get(self, key, default=None):
# if (not self.has_entry(key)):
# self.import_classes()
# return super(ClassRegistry, self).get(key, default)
# def __getitem__(self, *args, **kwargs):
# try:
# return super(ClassRegistry, self).__getitem__(*args, **kwargs)
# except KeyError: # pragma: no cover
# # This will only be called during import
# if self._imported:
# raise
# self.import_classes()
# return super(ClassRegistry, self).__getitem__(*args, **kwargs)
# def has_entry(self, key):
# return super(ClassRegistry, self).__contains__(key)
[docs]def registration_in_progress():
r"""Determine if a registration is in progress."""
return bool(os.environ.get('YGGDRASIL_REGISTRATION_IN_PROGRESS', None))
[docs]@contextlib.contextmanager
def registering(recurse=False):
r"""Context for preforming registration."""
if not recurse:
assert not registration_in_progress()
try:
previous = os.environ.get('YGGDRASIL_REGISTRATION_IN_PROGRESS', None)
os.environ['YGGDRASIL_REGISTRATION_IN_PROGRESS'] = '1'
yield
finally:
if previous is None:
if 'YGGDRASIL_REGISTRATION_IN_PROGRESS' in os.environ:
del os.environ['YGGDRASIL_REGISTRATION_IN_PROGRESS']
else:
os.environ['YGGDRASIL_REGISTRATION_IN_PROGRESS'] = previous
[docs]def init_registry(recurse=False):
r"""Initialize the registries and schema."""
from yggdrasil.tools import import_all_modules
global _registry
global _registry_complete
with registering(recurse=recurse):
import_all_modules(exclude=['yggdrasil.examples',
'yggdrasil.languages',
'yggdrasil.interface',
'yggdrasil.timing'],
do_first=['yggdrasil.serialize'])
_registry_complete = True
return _registry
[docs]def get_registry(comptype=None):
r"""Get the registry that should be used for looking up components.
Args:
comptype (str, optional): The name of a component to get the
registry for. Defaults to None and the entire registry will be
returned.
"""
global _registry
if registration_in_progress():
out = _registry
else:
from yggdrasil import constants
out = constants.COMPONENT_REGISTRY
if comptype:
if comptype not in out: # pragma: debug
raise Exception(f"Importing a component type that has not yet "
f"been registered: {comptype}")
out = out[comptype]
return out
[docs]def suspend_registry():
r"""Suspend the registry by storing the global registries in a dictionary."""
global _registry
global _registry_complete
out = {'_registry': _registry, '_registry_complete': _registry_complete}
_registry = {}
_registry_complete = False
return out
[docs]def restore_registry(reg_dict):
r"""Restore the registry to values in the provided dictionary."""
global _registry
global _registry_complete
_registry = reg_dict['_registry']
_registry_complete = reg_dict['_registry_complete']
[docs]def import_component(comptype, subtype=None, **kwargs):
r"""Dynamically import a component by name.
Args:
comptype (str): Component type.
subtype (str, optional): Component subtype. If subtype is not one of
the registered subtypes for the specified comptype, subtype is
treated as the name of class. Defaults to None if not provided and
the default subtype defined in the schema for the specified
component will be used.
**kwargs: Additional keyword arguments are used to determine the
subtype if it is None.
Returns:
class: Component class.
Raises:
ComponentError: If comptype is not a registered component type.
ComponentError: If subtype is not a registered subtype or the name of
a registered subtype class for the specified comptype.
"""
registry = get_registry(comptype=comptype)
if subtype is None:
subtype = kwargs.get(registry["key"], None)
if (comptype == 'comm') and (subtype is None):
subtype = 'DefaultComm'
if subtype is None:
subtype = registry["default"]
rev_subtypes = {v: k for k, v in registry["subtypes"].items()}
if subtype in registry["subtypes"]:
class_name = registry["subtypes"][subtype]
elif subtype in rev_subtypes:
class_name = subtype
subtype = rev_subtypes[subtype]
else:
class_name = subtype
if subtype in registry.get("subtype_modules", {}):
module_name = registry["subtype_modules"][subtype]
else:
module_name = class_name
# Check registered components to prevent importing multiple times
if class_name not in registry.get("classes", {}):
registry.setdefault("classes", {})
try:
registry["classes"][class_name] = getattr(
importlib.import_module(f"{registry['module']}."
f"{module_name}"),
class_name)
except ImportError:
if comptype == 'comm':
try:
return import_component('file', subtype, **kwargs)
except ComponentError:
pass
raise ComponentError(f"Could not locate a {comptype} component "
f"{subtype}.")
out_cls = registry["classes"][class_name]
# Check for an aliased class
if hasattr(out_cls, '_get_alias'):
out_cls = out_cls._get_alias()
return out_cls
[docs]def create_component(comptype, subtype=None, **kwargs):
r"""Dynamically create an instance of a component with the specified options
as outlined in the component schema. This function requires loading the
component schemas and so should not be used at the module or class level to
prevent circular dependencies.
Args:
comptype (str): Component type.
subtype (str, optional): Component subtype. If subtype is not one of the
registered subtypes for the specified comptype, subtype is treated
as the name of the class. Defaults to None if not provided and the
default subtype defined in the schema for the specified component
will be used. If the subtype is specified by the component subtype
key in the remaining kwargs, that subtype will be used instead.
**kwargs: Additional keyword arguments are treated as options for the
component as outlined in the component schema.
Returns:
ComponentBase: Instance of the specified component type/subtype and
options.
Raises:
ComponentError: If comptype is not a registered component type.
"""
from yggdrasil.schema import get_schema
s = get_schema().get(comptype, None)
if s is None: # pragma: debug
raise ComponentError("Unrecognized component type: %s" % comptype)
if s.subtype_key in kwargs:
subtype = kwargs[s.subtype_key]
if subtype is None:
subtype = s.identify_subtype(kwargs)
cls = import_component(comptype, subtype=subtype, **kwargs)
return cls(**kwargs)
[docs]def get_component_base_class(comptype, subtype=None, **kwargs):
r"""Determine the base class for a component type.
Args:
comptype (str): The name of a component to test against.
subtype (str, optional): Subtype to use to determine the component
base class. Defaults to None.
**kwargs: Additional keyword arguments are used to determine the
subtype if it is None.
Returns:
ComponentBase: Component base class.
"""
registry = get_registry(comptype=comptype)
base_class_name = registry['base']
return import_component(comptype, subtype=base_class_name, **kwargs)
[docs]def isinstance_component(x, comptype, subtype=None, **kwargs):
r"""Determine if an object is an instance of a component type.
Args:
x (object): Object to test.
comptype (str, list): The name of one or more components to test
against.
subtype (str, optional): Subtype to use to determine the component
base class. Defaults to None.
**kwargs: Additional keyword arguments are used to determine the
subtype if it is None.
Returns:
bool: True if the object is an instance of the specified component(s).
"""
if isinstance(comptype, (list, tuple)):
for icomp in comptype:
if isinstance_component(x, icomp):
return True
else:
return False
base_class = get_component_base_class(comptype, subtype=subtype,
**kwargs)
return isinstance(x, base_class)
[docs]def inherit_schema(orig, new_properties=None, new_required=None,
remove_keys=[], **kwargs):
r"""Create an inherited schema, adding new value to accepted ones for
dependencies.
Args:
orig (dict): Schema that will be inherited.
new_properties (dict, optional): Dictionary of new properties to add.
Defaults to None and is ignored.
new_requried (list, optional): Properties that should be required by the
new class. Defaults to None and is ignored.
remove_keys (list, optional): Keys that should be removed form orig
before adding the new keys. Defaults to empty list.
**kwargs: Additional keyword arguments will be added to the schema
with dependency on the provided key/value pair.
Returns:
tuple(dict, list): New schema properties and a list of requried.
"""
remove_keys = copy.deepcopy(remove_keys)
# Get set of original properties
assert issubclass(orig, ComponentBase)
out_prp = copy.deepcopy(orig._schema_properties)
if orig._schema_excluded_from_inherit is not None:
remove_keys += orig._schema_excluded_from_inherit
out_req = copy.deepcopy(orig._schema_required)
# Don't add duplicates
if new_properties == out_prp:
new_properties = None
if new_required == out_req:
new_required = None
# Remove keys
for k in remove_keys:
if k in out_prp:
out_prp.pop(k)
if k in out_req:
out_req.remove(k)
# Add new values and keyword arguments
if new_properties is not None:
out_prp.update(new_properties)
if new_required is not None:
for k in new_required:
if k not in out_req:
out_req.append(k)
out_prp.update(kwargs)
return out_prp, out_req
[docs]class ComponentBaseUnregistered(object):
r"""Base class for schema components w/o schema and registration."""
__slots__ = []
_disconnect_attr = []
def __del__(self):
self.disconnect()
[docs] def disconnect(self):
r"""Disconnect attributes that are aliases."""
for k in self._disconnect_attr:
if hasattr(getattr(self, k, None), 'disconnect'):
getattr(self, k).disconnect()
[docs]@six.add_metaclass(ComponentMeta)
class ComponentBase(ComponentBaseUnregistered):
r"""Base class for schema components.
Args:
skip_component_schema_normalization (bool, optional): If True, the
schema will not be used to normalize/validate input keyword
arguments (e.g. in case they were already parsed). Defaults to
False.
**kwargs: Keyword arguments are added to the class as attributes
according to the class attributes _schema_properties and
_schema_excluded_from_class. Keyword arguments not added to the
class as attributes are assigned to the extra_kwargs dictionary.
Attributes:
extra_kwargs (dict): Keyword arguments that were not parsed.
Class Attributes:
_schema_type (str): Name of the component type the class represents.
_schema_subtype_key (str): Attribute that should be used to identify the
subtype associated with each class.
_schema_subtype_description (str): Description for the subtype represented
by the class that should be used in documentation tables.
_schema_required (list): Keys from _schema_properties that are required
to produce a valid component.
_schema_properties (dict): Schemas describing keyword arguments that
can be supplied to the class constructor and used to specify
component behavior in YAML/JSON files. At initialization, these
keywords are added to the class instance as attributes of the
same name unless they are in _schema_excluded_from_class. Unless
_schema_inherit is False, these properties will be added in addition
to the schema properties defined by the class base.
_schema_excluded_from_class (list): Keywords in _schema_properties that
should not be added to the class as attributes during initialization.
_schema_excluded_from_inherit (list): Keywords in _schema_properties that
should not be inherited either from the base class or by child
classes.
_schema_inherit (bool, ComponentBase): If False, the base schema will
not be inherited. If a Component subclass, the schema from that
class will be inherited instead of the base class. Defaults to True.
_dont_register (bool): If True, the component class will be be registered
and the before_registration class method will not be called. Defaults
to False.
_schema_no_default_subtype (bool): If True, the subtype schema
will not have a default subtype set. Defaults to False.
"""
_schema_type = None
_schema_subtype_key = None
_schema_subtype_description = None
_schema_subtype_default = None
_schema_base_class = None
_schema_required = []
_schema_properties = {}
_schema_excluded_from_class = []
_schema_excluded_from_inherit = []
_schema_excluded_from_class_validation = []
_schema_inherit = True
_schema_additional_kwargs = {}
_schema_additional_kwargs_base = {}
_schema_additional_kwargs_no_inherit = {}
_schema_no_default_subtype = False
_dont_register = False
def __new__(cls, *args, **kwargs):
obj = object.__new__(cls)
obj._input_args = []
for x in args:
try:
obj._input_args.append(weakref.ref(x))
except TypeError:
obj._input_args.append(x)
obj._input_kwargs = {}
for k, v in kwargs.items():
try:
obj._input_kwargs[k] = weakref.ref(v)
except TypeError:
obj._input_kwargs[k] = v
return obj
def __getstate__(self):
out = self.__dict__.copy()
del out['_input_args'], out['_input_kwargs']
return out
def __setstate__(self, state):
state['_input_args'] = []
state['_input_kwargs'] = {}
self.__dict__.update(state)
def _ygg_rapidjson(self):
r"""Method for getting a rapidjson-friendly version of a class.
Returns:
dict: Dictionary of properties obeying the schema.
"""
out = {self._schema_subtype_key:
getattr(self, f"_{self._schema_subtype_key}")}
for k in self._schema_properties.keys():
if getattr(self, k, None) is not None:
out[k] = getattr(self, k)
return out
def __init__(self, skip_component_schema_normalization=None,
additional_component_properties=None, **kwargs):
if skip_component_schema_normalization is None:
skip_component_schema_normalization = (
not (os.environ.get('YGG_VALIDATE_COMPONENTS', 'None').lower()
in ['true', '1']))
comptype = self._schema_type
if (comptype is None) and (not self._schema_properties):
self.extra_kwargs = kwargs
return
subtype = None
if self._schema_subtype_key is not None:
subtype = getattr(self, self._schema_subtype_key,
getattr(self, '_%s' % self._schema_subtype_key, None))
# Fall back to some simple parsing/normalization to save time on
# full rapidjson normalization
self._defaults_set = []
for k, v in self._schema_properties.items():
if k in self._schema_excluded_from_class:
continue
default = v.get('default', None)
if (k == self._schema_subtype_key) and (subtype is not None):
default = subtype
if default is not None:
if k not in kwargs:
self._defaults_set.append(k)
kwargs.setdefault(k, copy.deepcopy(default))
if v.get('type', None) == 'array':
if isinstance(kwargs.get(k, None), (bytes, str)):
kwargs[k] = kwargs[k].split()
# Parse keyword arguments using schema
if (((comptype is not None) and (subtype is not None)
and (not skip_component_schema_normalization)
and (not self._dont_register))):
from yggdrasil.schema import get_schema
s = get_schema().get_component_schema(
comptype, subtype, relaxed=True)
props = list(s['properties'].keys())
if not skip_component_schema_normalization:
kwargs.setdefault(self._schema_subtype_key, subtype)
if additional_component_properties:
kwargs.update(additional_component_properties)
# Remove properties that shouldn't ve validated in class
extra_kwargs = {}
for k in self._schema_excluded_from_class_validation:
if k in s['properties']:
del s['properties'][k]
if k in s['required']:
s['required'].remove(k)
if k in kwargs:
extra_kwargs[k] = kwargs.pop(k)
# Validate and normalize
from yggdrasil import rapidjson
# import pprint
# print(f'before: {self}\n{pprint.pformat(kwargs)}')
try:
kwargs = rapidjson.normalize(kwargs, s)
kwargs.update(extra_kwargs)
except BaseException: # pragma: debug
import pprint
pprint.pprint(kwargs)
raise
# print(f'after: {self}\n{pprint.pformat(kwargs)}')
else:
props = self._schema_properties.keys()
for k in props:
for x in self._schema_properties[k].get('aliases', []):
if x in kwargs:
kwargs.setdefault(k, kwargs.pop(x))
# Set attributes based on properties
for k in props:
if k in self._schema_excluded_from_class:
continue
v = kwargs.pop(k, None)
if getattr(self, k, None) is None:
setattr(self, k, v)
# elif (getattr(self, k) != v) and (v is not None):
# warnings.warn(("The schema property '%s' is provided as a "
# "keyword with a value of %s, but the class "
# "already has an attribute of the same name "
# "with the value %s.")
# % (k, v, getattr(self, k)))
self.extra_kwargs = kwargs
[docs] @staticmethod
def before_registration(cls):
r"""Operations that should be performed to modify class attributes prior
to registration. These actions will still be performed if the environment
variable YGGDRASIL_REGISTRATION_IN_PROGRESS is set."""
pass
[docs] @staticmethod
def after_registration(cls):
r"""Operations that should be preformed to modify class attributes after
registration. These actions will not be performed if the environment
variable YGGDRASIL_REGISTRATION_IN_PROGRESS is set."""
pass
[docs] @staticmethod
def finalize_registration(cls):
r"""Final operations to perform after a class has been fully initialized.
These actions will not be performed if the environment variable
YGGDRASIL_REGISTRATION_IN_PROGRESS is set."""
pass
[docs]def create_component_class(globals_dict, base, name, attr):
r"""Create a new component class.
Args:
globals_dict (dict): Globals dictionary that new class should be
added to.
base (type): Base class for new class. The new class will have
the same module as base.
name (str): Name for the new class.
attr (dict): Attributes that should be added to the new class.
"""
attr['__module__'] = base.__module__
cls = type(name, (base, ), attr)
globals_dict[cls.__name__] = cls
del cls