Source code for ignite_framework.framework.framework
# ### DEBUGGING/STEPPING TIPPS ###
# - `from ignite_framework.franework.framework import MetaState, MetaStateContainer, MetaStateObjectContainer for
# for inspecting/debugging the hidden infrastructure variables
# - turn on/off debugging # prints with un/commenting `# print` in the whole project
# - clean up `utils.namespace_decorator` and change `only_for_ctrs` to `only_under_condition`
# - Optimizer ConditionalTrnasition Numth...
from collections import OrderedDict
from inspect import isclass, Parameter, Signature, signature
# CONSTANTS
from ignite_framework.framework.framework_utils import CALLABLE_TYPES, FSO_OVERLOAD_SUFFIXES
from ignite_framework.framework.state_component_attrs_integration import INTEGRATED_STATE_COMPONENT_ATTRS
# CONTAINER IDS
from ignite_framework.framework.container_ids import ContainerID, BaseTransitionsContainerID, ConfigurationsContainerID, \
DataloadersContainerID, EnginesContainerID, LoggersContainerID, MaintenanceContainerID, MetricID, \
MetricsContainerID, ModulesContainerID, OptimizersContainerID, OutputhandlersContainerID, ParametersContainerID, \
PipelinesContainerID, StateComponentID, TensorboardContainerID, TransitionsContainerID
# FRAME STATE OBJECTS
from ignite_framework.framework.frame_state_objects import FrameCallbackoverloadsStateObject, FrameFunctionStateObject, \
FrameParameterStateObject, FrameStateObject, FrameVariableStateObject
# STATE OBJECTS
from ignite_framework.framework.modular_state_objects import StateObject, StateVariable, \
StateParameter, StateFunction, StateDevice, StateConfigurationVariable, StateBooleanVariable, \
StateBooleanConfigurationVariable, StateIntCounter, StateIterationCounter
# FRAME STATE OBJECTS REFERENCE
from ignite_framework.framework.frame_state_objects_reference import FrameStateObjectsReference
# UTILS FOR STATE COMPONENT INTEGRATION
from ignite_framework.framework.framework_utils import get_component_attr, set_component_attr
# UTILS CLASSES
from ignite_framework.framework.framework_utils import DecoratorAndAttributeToItemDictionary, \
SingleKeyAssignmentDict, CallbacksList
# UTILS METHODS
from ignite_framework.framework.framework_utils import bases_have_attr, check_if_tuple_is_callback_and_reformat, \
get_argument_name_from_frame_inspection, namespace_decorator, \
raise_or_warn_abstract_method_not_implemented_error, raise_run_not_implemented_and_appendable_to_callback_error
# FRAME & FEATURE EXCEPTIONS
from ignite_framework.exceptions import FrameworkAttributeError, FrameworkNameError, FrameworkNotImplementedError, \
FrameworkImplementationLimitError, FrameworkRestrictionError, FrameworkTypeError, FrameworkValueError
from ignite_framework.exceptions import FeatureTypeError, FeatureNotImplementedError
import torch
from torch.utils.data.dataloader import DataLoader
from torch.nn import Module
from torch.optim.optimizer import Optimizer
from torch.optim.lr_scheduler import _LRScheduler, ReduceLROnPlateau
from torch.utils.tensorboard import SummaryWriter
from warnings import warn
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### S T A T E ### GLOBAL STATE VARIABLE, TRANSITION & COMPONENT CONTAINER
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### META STATES ###
# ==================================================================================================================
class MetaSingleton(type):
# Hide `__singleton` attribute from user
__singleton = dict()
def __prepare__(cls, name, **kwds):
return OrderedDict()
# Metaclass '__init__' is called when class is created (metaclass initialization), NOT when class is initialized
@classmethod
def __init__(cls, name, bases, namespace, **kwargs):
# cls.__singleton = None
super().__init__(cls, name, bases, namespace)
def __call__(cls, **kwargs):
if cls.__name__ in type(cls).__singleton.keys():
# # print('Retrieving existing singleton State.')
return type(cls).__singleton[cls.__name__]
# print('Constructing new singleton State.')
# singleton = cls.__new__(cls)
singleton = cls.__new__(cls)
singleton.__init__(**kwargs)
type(cls).__singleton[cls.__name__] = singleton
return singleton
[docs]class MetaState(MetaSingleton):
"""
Singleton `Metaclass` from which all meta state containers inherit.
"""
_variable_overload_suffixes = FSO_OVERLOAD_SUFFIXES # == ['ref', 'callbacks', 'once', 'every']
_state_shortcuts = dict(ctr_attrs=SingleKeyAssignmentDict(), fsos=SingleKeyAssignmentDict())
_blocked_state_shortcuts = dict(
ctrs=set(['configs', 'params', 'dataloaders', 'modules', 'optimizers', 'engines',
'output_handlers', 'metrics', 'transitions', 'tensorboard', 'maintenance']),
redundant_variables=set())
_non_state_shortcuts = set(['arguments', 'helpers'])
# TODO: - name checking to avoid ambiguity and use of overload suffixes
[docs] def __new__(typ, name, bases, namespace):
namespace_item = namespace_decorator(name, bases, namespace)
@namespace_item
def __getattr__(self, name):
try:
# Get fso-containers directly from state, e.g. `state.trainer` instead of `state.engines.trainer`
return type(self.__class__)._state_shortcuts['ctr_attrs'][name]
except KeyError:
try:
# Get fso value directly from state with `__get__(name)` including all overloads with 1 suffix,
# e.g. `state.state_run_started` instead of `state.engines.state_status.state_run_started`
return getattr(type(self.__class__)._state_shortcuts['fsos'][name], name)
except KeyError:
try:
# Case: fso with 3 suffices, e.g. `<fso_name>_every_33_ref` or `<fso_name>_once_100_callbacks`
# or `<fso_name>_every_33` or `<fso_name>_once_100`
# Note: Removing the last 2 suffixes returns either the pure fso-name or including 1 suffix,
# e.g. `_ref`, `_every`, but the shortcuts with 1 suffix return the same fso-ctr & name
# as without a suffix
# Get fso-ctr from shortcuts without 3 suffixes and then get fso with suffix from fso-ctr
fso_name_without_2_or_3_suffixes = '_'.join(name.split(sep='_')[:-2])
return getattr(type(self.__class__)._state_shortcuts['fsos'][fso_name_without_2_or_3_suffixes],
name)
except KeyError:
# Provoke regular `KeyError`
return object.__getattribute__(self, name)
@namespace_item
def __setattr__(self, name, value):
"""
Restrictive `__settattr__` only allowing setting for existing `state objects`.
Existing or new state attributes and shortcuts of container attributes (transitions, components, references)
are read only and cannot be (re-)set after initialization.
Args:
self:
name:
value:
Returns:
"""
try:
setattr(type(self.__class__)._state_shortcuts['fsos'][name], name, value)
except KeyError:
try:
fso_name_without_2_suffixes = '_'.join(name.split(sep='_')[:-1])
setattr(type(self.__class__)._state_shortcuts['fsos'][fso_name_without_2_suffixes], name, value)
except KeyError:
if name in type(self.__class__)._state_shortcuts['ctr_attrs']:
raise FrameworkRestrictionError(
'Container attributes cannot be (re-)set after initialization. Therefore, the state shortcut '
'`state.{}` of a container attributes (e.g. of a transition `state.trainer` or component '
'`state.model`) is also read-only after initialization.'.format(name))
else:
raise FrameworkRestrictionError(
'The `state` is read-only and restricted to its predefined state containers, '
'e.g. `state.engines`, `state.configs`, etc.. The new attributes `{}` cannot be '
'set.'.format(name))
@namespace_item
def __delattr__(self, name):
# If `name` is a state variable/parameter shortcut then `__del__` the actual state variable/parameter
if name in type(self.__class__)._state_shortcuts['fsos'].keys():
delattr(type(self.__class__)._state_shortcuts['fsos'][name], name)
# if it's a regular attribute - which is a container - then delete it... no good idead
elif hasattr(self, name):
# object.__delattr__(self, name)
raise FrameworkRestrictionError(
'`state` is read-only. The state container `{}` cannot be deleted.'.format(name))
# Provoke regular `KeyError`
else:
object.__delattr__(self, name)
@namespace_item
def get(self, *name_or_parts):
"""
Get method for parameterized attribute names, especially for `state objects`,
but not only.
Main reason is to simply avoid unhandy commands e.g. `getattr(state.trainer, '{}_{}'.format(name,
suffix)` commands
and simply write `state.trainer.get(name, suffix)`.
"""
try:
return getattr(self, *name_or_parts, FrameworkAttributeError(
'`{c}` object has no attribute `a_value`. Note: You may find this error message unappropriately '
'assigned to a variable or raised within another error. This error message is a hacky workaround '
'to avoid the `default` value of the internal `gettattr(self, *name_ort_parts [, default])` call '
'in the method `{c}.get()`. It hinders `{c}.get()` from ignoring the last of two arguments due to '
'the optional `default` arguemnt of `getattr`. Sorry for the inconveniences, it may be fixed '
'soon.'.format(c=self.__class__.__name__)))
except TypeError:
return getattr(self, '_'.join([str(name_part) for name_part in name_or_parts]))
@namespace_item
def set(self, *name_or_parts_and_value):
try:
setattr(self, *name_or_parts_value[:-1], name_ort_parts_value[-1])
except TypeError:
setattr(self, '_'.join([str(name_part) for name_part in name_or_parts_and_value[:-1]]),
name_or_parts_and_value[-1])
@namespace_item
def get_state_ctr_names(self):
return [state_ctr_name for state_ctr_name, state_ctr in self.__class__.__dict__.items()
if self.__class__.__class__ in [meta_cls for meta_cls in type.mro(state_ctr.__class__.__class__)]]
return super().__new__(typ, name, bases, namespace)
# TODO:
# - Whu is Metastate being deleted during class creation? Possibly this should go to __new__() in namespace
# def __del__(self):
# raise RuntimeError('Objects stored in a `State` are not supposed to be deleted. '
# 'If `{}` is a class you implemented yourself then it possibly '
# 'should have been implemented as a `Transition` '
# 'subclass.'.format(self.name))
# ==================================================================================================================
### META STATE CONTAINERS ### CATEGORIZATION OF STATE COMPONENTS AND TRANSITIONS: Overview improvement
# ==================================================================================================================
[docs]class MetaStateContainer(MetaState):
# - Hide infrastruture from user in metaclass
_ctr_shortcuts_for_fsos = {}
_blocked_ctr_shortcuts_for_fsos = {}
# Pytorch class instance attrs that are integrated in state as frame state object, ordered from
# sub- so superclasses. The items of `_integrated_state_compoennt_attrs` are initialized after the
# required subclasses of `BaseStateObject`s are defined.
_integrated_state_component_attrs = {}
_user_defined_ctrID_classes = {}
[docs] def __new__(typ, cls_name, bases, namespace, containing_name: str, containing_types):
typ._ctr_shortcuts_for_fsos[cls_name] = SingleKeyAssignmentDict()
typ._blocked_ctr_shortcuts_for_fsos[cls_name] = set()
typ._user_defined_ctrID_classes[cls_name] = set()
# Decorator for adding methods to namespace ONLY IF they are not overriden by any baseclasses
namespace_item = namespace_decorator(cls_name, bases, namespace)
# Argument checking
if not isinstance(containing_types, list):
containing_types = list(containing_types)
if not (all([isinstance(containing_type, type) for containing_type in containing_types])):
raise TypeError('All elements in `containing_types` must be of type `type`, but given types: {}'.format(
[type(containing_type) for containing_type in containing_types]))
@namespace_item
def __getattr__(self, name):
"""
A smart getter for getting with shortcuts.
.. code-block:: python
from ignite_framework.states import state
>>> state.engines.state_status.state_run_started
False
>>> state.state_run_started
False
>>> id(state.engines.state_status.state_run_started) == id(state.state_run_started)
Args:
name (str): `state container` name or `state object` name
Returns:
"""
# ========================
### SHORTCUT GETTING ###
# ========================
try:
return getattr(type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__][name], name)
except KeyError:
# If no shortcut available, provoke standard `KeyError`
return object.__getattribute__(self, name)
@namespace_item
def __setattr__(self, name, value):
"""
A smart setter for setting with shortcuts.
.. code-block:: python
from ignite_framework.states import state
>>> state.engines.state_status.state_run_completed = True
>>> state.engines.state_status.state_run_completed
True
>>> state.state_run_compelted = False
>>> state.engines.state_status.state_run_completed
False
.. Note:: Performance
The ``__setattr__`` of a `state container` is only used during `state` initialization, \
not during training. Therefore the setting-performance is of lower relevance.
"""
# ========================
### SHORTCUT SETTING ###
# ========================
try:
# If `name` is a shortcutted state variables/parameters try setting it
self._set_ctr_shortcut_of_fso(name, value)
except KeyError:
# Else `name` must be an new container attribute to be set
self._setattr_with_args_check(name, value)
# TODO:
# - promote shortcuts to intellisence for IDE=code-completion
# - shortcuts currently only support overlead suffixes with only 1 word
# Only create shortcuts if:
# - `value` is a subclass of a container-ID class, meaning it's a framework or feature class
# - OR `value' is a user defined container-ID class
# - AND: `value` is not a instance of a integrable class, because in this case the
# shortcuts were already generated in `self.setattr_with_args_check()`
# NOTE:
# . The selective criteria for state container shortcuts generation are for 2 reasons:
# . so-called integrated (mainly PyTorch) classes create their shortcuts already during
# integration
# - `MetaValueContainer`s can set state objects directly from the state container and you
# do NOT want to set a state-CONTAINER-shortcut for a state-OBJECT (!), e.g.
# `state.configs.a_conf_var = 3` creates the state object
# `state.configs.user_defined_configs.a_conf_var`
user_defined_ctrID_clses = \
tuple(type(self.__class__)._user_defined_ctrID_classes[self.__class__.__name__])
integrated_clses = tuple(type(self.__class__)._integrated_state_component_attrs.keys())
if (isinstance(value, ContainerID) or isinstance(value, user_defined_ctrID_clses)) \
and not isinstance(value, integrated_clses):
# if isinstance(value, ContainerID) \
# or any([isinstance(value, integrated_cls) for integrated_cls in all_integrated_clses]):
# and all([base_cls not in type(self.__class__)._integrated_state_component_attrs[self.__class__.__name__].keys()
# for base_cls in value.__class__.mro()]) :
self._create_state_and_ctr_shortcuts(fso_or_ref_ctr_name=name, fso_or_ref_ctr=value)
@namespace_item
def __delattr__(self, name):
if name in type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__].keys():
delattr(type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__][name], name)
elif hasattr(self, name):
object.__delattr__(self, name)
# raise FrameworkRestrictionError(
# '`state` is read-only. The state container `{}` cannot be deleted.'.format(name))
else:
# Provoke regular `KeyError`
object.__delattr__(self, name)
# state containers that can be used in combintation with `with`-statement to add arguements,
# hyperparameters or helper-functions
# NOTE: Ugly, but inevitable (?)... frame state objects are not defined yet, so they are placeholdered
# by strings and exchanged in `__exit__`:
# . 'SCV': `StateConfigurationVariable`
# - 'SP': `StateParameter`
indentable_ctrs = {'Configurations': {'args_ctr_name': 'user_defined_configs',
'default_attr_type': 'SCV'},
'Parameters': {'args_ctr_name': 'user_defined_params',
'default_attr_type': 'SP'},
'Engines': {'args_ctr_name': 'hyperparams',
'default_attr_type': 'SP'},
'Maintenance': {'args_ctr_name': 'arguments',
'default_attr_type': 'SCV'},
'_placeholders_not_replaced_yet': True}
@namespace_item
def __enter__(self):
if not self.__class__.__name__ in indentable_ctrs.keys():
raise FrameworkRestrictionError(
'The `with`-statement is currently not implemented for `state.{}`. At the moment the `with`-'
'statement is only implemented for `state.configs`, `state.params` & `state.engines` to add '
'arguments, hyperparameters and helper functions. Additionally the `with`-statement is available '
'for adding state components (e.g. PyTorch `Module`, `DataLoader`, `Optimizer` etc.) in `state.'
'dataloaders`, `state.modules` & `state.optimizers`.'.format(self.get_state_ctr_name()))
# if isinstance(self, Engines):
# raise FrameworkRestrictionError(
# 'The `with`-statement is currently not implemented for `state.engines`. Any additional '
# '`state.engines` parameters (and helper functions) are regarded as hyperparameters and'
# 'can be set using the `with state.params as p: ...` statement. This will generate a parameter '
# 'container `state.engines.hpyerparams` with state parameters. Functions defined in the '
# '`with state.params as p: @p ...` statement will be stored in `state.engines.helpers` as '
# 'state function.')
self.__dict__['_entered_namespace'] = DecoratorAndAttributeToItemDictionary()
return self.__dict__['_entered_namespace']
# @namespace_item(only_under_condition=self.name == indentable_container_class_names)
@namespace_item
def __exit__(self, exc_type, exc_value, traceback):
if exc_type != None:
raise exc_type(exc_value, None, traceback)
# UGLY: Prelacing state base object paceholders of `indented_ctrs` because weren't defined yet when
# `indented_ctrs` was defined
if '_placeholders_not_replaced_yet' in indentable_ctrs.keys():
del indentable_ctrs['_placeholders_not_replaced_yet']
fso_placeholders = {'SCV': StateConfigurationVariable, 'SP': StateParameter}
for cls_name in indentable_ctrs.keys():
indentable_ctrs[cls_name]['default_attr_type'] = \
fso_placeholders[indentable_ctrs[cls_name]['default_attr_type']]
# `helpers` of `state.params` are stored in `state.engines.helpers` that's what the `if`s are for
indented_ctr = self
indented_cls_name = self.__class__.__name__
arg_or_hyperparams_ctr_name = indentable_ctrs[indented_cls_name]['args_ctr_name']
helpers_dict = self.__dict__['_entered_namespace'].pop('decorator_args')
if helpers_dict:
if hasattr(indented_ctr, 'helpers'):
if state.engines.state_status.state_run_started:
raise FrameworkRestrictionError(
'`state.{}.helpers` cannot be modified anymore after `state` started running. This '
'prevents possible corruptions of callbacks. Please move the `with`-statement code block '
'further up before `state.run()` was called.'.format(self.get_state_ctr_name()))
self.helpers._setattrs_from_additional_dict_as_fsos(attrs_dict=helpers_dict)
else:
# Create the state object container holding the entered functions as `StateFunction()`s
self._create_helpers_from_entered_namespace(helpers_dict)
# dict.__delitem__(self.__dict__['_entered_namespace'], 'decorator_args')
# Create the state object container holding the entered arguments and hyperparameters
if self.__dict__['_entered_namespace']:
if hasattr(indented_ctr, arg_or_hyperparams_ctr_name):
if state.engines.state_status.state_run_started:
raise FrameworkRestrictionError(
'`state.{}.{}` cannot be modified anymore after `state` started running. This '
'prevents possible corruptions of callbacks. Please move the `with`-statement code block '
'further up before `state.run()` was called.'.format(arg_or_hyperparams_ctr_name,
self.get_state_ctr_name()))
self.get(arg_or_hyperparams_ctr_name)._setattrs_from_additional_dict_as_fsos(
attrs_dict=self.__dict__['_entered_namespace'])
else:
if indented_ctr == state.engines:
self._create_hyperparameters_from_entered_namespace(
args_ctr_name=indentable_ctrs[indented_cls_name]['args_ctr_name'],
default_attr_type=indentable_ctrs[indented_cls_name]['default_attr_type'])
else:
self._create_arguments_from_entered_namespace(
args_ctr_name=indentable_ctrs[indented_cls_name]['args_ctr_name'],
default_attr_type=indentable_ctrs[indented_cls_name]['default_attr_type'])
dict.__delitem__(self.__dict__, '_entered_namespace')
@namespace_item
def register_user_defined_ctrID_class(self, id_class):
"""
Add/Register a new container-ID-class to a state cotnainer, e.g. a special tokenizer class to
`state.modules`.
As soon as the new ID-class is registered at the state container, instances of this ID-class
(or subclasses of the ID-class) can be set a s container attribute.
Args:
self: state container, e.g. `state.modules`, `state.optimizer`, etc.
id_class: a class from which a instance of this class the user wnats to attach to the state container
Returns:
"""
if not isclass(id_class):
raise FrameworkTypeError(
'ContainerID-classes must be class, but received `{}` instance.'.format(id_class.__class__))
type(self.__class__)._user_defined_ctrID_classes[self.__class__.__name__].add(id_class)
@namespace_item
def integrate_new_state_component_class(self, component_class, integrated_attrs_args_dict={}):
"""
Integrate new state component class, e.g. PyTorch torch.nn.NewModule, or append additional/change existing
integrated attributes to/of existing state component class.
Args:
self:
component_class: class of the new state component to be integrated
integrated_attrs_args_dict (dict):
dict(<attr_name>=(<bso_class>, <initial_value>, <component_attr_getter>,
<component_attr_setter>))
attr__get__: function that is forwarded to `instance__get__` of `BaseStateObject`
attr__set__: function that is forwarded to `instance__set__` of `BaseStateObject`
Returns:
"""
integrated_attrs_dict = type(self.__class__)._integrated_state_component_attrs
#
if not component_class in integrated_attrs_dict.keys():
integrated_attrs_dict[component_class] = {}
# Append/Add integrated attributes either additionally to existing or new state component class
for attr_name, integrated_attr_args in integrated_attrs_args_dict.items():
fso_cls, initial_value, instance__get__, instance__set__ = integrated_attr_args
integrated_attrs_dict[component_class][attr_name] = \
(fso_cls, initial_value,
instance__get__ if instance__get__ != '' else get_component_attr,
instance__set__ if instance__set__ != '' else set_component_attr)
# Register `component_class` as new ID-class of container `self`
self.register_user_defined_ctrID_class(id_class=component_class)
@namespace_item
def get_state_ctr_name(self):
return [state_ctr_name for state_ctr_name, state_ctr_obj in state.__class__.__dict__.items()
if state_ctr_obj == self][0]
@namespace_item
def get_bso_ctrs(self):
return [fso_ctr for fso_ctr in self.__dict__.values()
if any([fso_ctr_cls in containing_types for fso_ctr_cls in fso_ctr.__class__.mro()])]
@namespace_item
def get_bso_ctr_names(self):
return [fso_ctr.name for fso_ctr in self.get_bso_ctrs()]
@namespace_item
def _check_if_new_fso_ctr_already_exists(self, new_ctr_name, indented_ctr):
fso_ctr_names = \
[('helpers', 'function'), ('arguments', 'argument'), ('hyperparams', 'hyperparameter')]
for fso_ctr_name, fso_name in fso_ctr_names:
if new_ctr_name == fso_ctr_name:
ctr_attr_name = fso_name
return hasattr(indented_ctr, new_ctr_name)
# return True
# raise FrameworkRestrictionError(
# 'The {attr}s container `state.{ind_cont}.{new_cont}` has already been created. '
# 'All {attr}s of `state.{ind_cont}` must be added in one and the same `with`-statement. '
# 'Please add the {attr}(s) to the first statement `wtih state.{ind_cont}s as {i}: ...`.'.format(
# attr=ctr_attr_name, ind_cont=self.get_state_ctr_name(),
# new_cont=new_ctr_name, i=containing_name[0]))
@namespace_item
def _create_helpers_from_entered_namespace(self, helpers_dict):
# Create customized `Helpers` depending on container and entered attribute functions
cls_name = self.__class__.__name__ + 'Helpers'
Helpers = MetaEnteredFrameStateObjectContainer(
cls_name, (containing_types[0],), {},
entered_namespace=helpers_dict,
default_attr_fso_cls=StateFunction, ctr=self)
Helpers(name='helpers')
@namespace_item
def _create_arguments_from_entered_namespace(self, args_ctr_name, default_attr_type):
# Create customized `Argument` depending on container and entered attributes/frame state objects
cls_name = self.__class__.__name__ + 'Arguments'
Arguments = MetaEnteredFrameStateObjectContainer(
cls_name, (containing_types[0],), {}, entered_namespace=self.__dict__['_entered_namespace'],
default_attr_fso_cls=default_attr_type, ctr=self)
Arguments(name=args_ctr_name)
@namespace_item
def _create_hyperparameters_from_entered_namespace(self, args_ctr_name, default_attr_type):
# Note: As only 1 hyperparameter container exists (`state.engines.hyperparams`) the class name
# does not have to be individualized like required for non unique `helpers` or `arguments`
HyperParameter = MetaEnteredFrameStateObjectContainer(
'HyperParameter', (EnginesContainerID,), {},
entered_namespace=self.__dict__['_entered_namespace'], default_attr_fso_cls=default_attr_type,
ctr=self)
HyperParameter(name=args_ctr_name)
@namespace_item
def _create_state_and_ctr_shortcuts_for_all_fsos(self, fso_ctr):
for fso_name in [*type(fso_ctr.__class__)._variable_names[fso_ctr.__class__.__name__],
*type(fso_ctr.__class__)._parameter_names[fso_ctr.__class__.__name__],
*type(fso_ctr.__class__)._other_fso_names[fso_ctr.__class__.__name__]]:
self._create_state_and_ctr_shortcuts_for_fso(fso_name, fso_ctr)
@namespace_item
def _create_state_and_ctr_shortcuts_for_fso(self, fso_name, fso_ctr):
self._check_fso_name_is_not_equal_to_ctr_attr_name(fso_name, fso_ctr)
# SHORTCUT ADDING FOR STATE VARIABLES & PARAMETERS
self._create_state_and_ctr_shortcuts_of_fso_or_refcaller(
fso_or_refcaller_name=fso_name, value=fso_ctr)
@namespace_item
def _create_state_and_ctr_shortcuts_for_all_ref_callers(self, fso_or_ref_ctr):
for refcaller_name in fso_or_ref_ctr._ref_caller_names:
self._setattr_with_args_check(cls_name, fso_or_ref_ctr)
self._create_state_and_ctr_shortcuts_of_fso_or_refcaller(
fso_or_refcaller_name=refcaller_name, value=fso_or_ref_ctr)
@namespace_item
def _check_fso_name_is_not_equal_to_ctr_attr_name(self, fso_name, fso_ctr):
if fso_name == fso_ctr.name:
fso_ctr_cls_name = fso_ctr.__class__.__name__
raise FrameworkNameError(
'The frame state object name `{}` is identical to frame state object container `{}` and therefore '
'cannot be set. Please change the frame state object name `{}` and avoid the names: `{}`.'.format(
fso_name, fso_ctr.name, fso_ctr.__name__, fso_name,
[fso_ctr.name, *type(fso_ctr.__class__)._variable_names[fso_ctr_cls_name],
*type(fso_ctr.__class__)._parameter_names[fso_ctr_cls_name],
*type(fso_ctr.__class__)._other_fso_names[fso_ctr_cls_name]]))
@namespace_item
def _create_state_and_ctr_shortcuts(self, fso_or_ref_ctr_name, fso_or_ref_ctr):
# Note: overload suffix exclusion is treated in MetaTransition
# Create shortcuts
self._create_state_shortcut_for_ctr_attr(fso_or_ref_ctr_name, fso_or_ref_ctr)
if hasattr(type(fso_or_ref_ctr.__class__), '_other_fso_names'):
self._create_state_and_ctr_shortcuts_for_all_fsos(fso_ctr=fso_or_ref_ctr)
elif '_ref_caller_names' in fso_or_ref_ctr.__dict__.keys():
self._create_state_and_ctr_shortcuts_for_all_ref_callers(fso_or_ref_ctr)
@namespace_item
def _set_ctr_shortcut_of_fso(self, name, value):
setattr(type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__][name], name, value)
@namespace_item
def _setattr_with_args_check(self, name, value):
# Set container attribute name `name` wherease `name` must be unique and `value` allowed for the container
try:
self._check_ctrID_of_ctr_attr_value(name, value)
self._check_ctr_attr_name_ctr_specifically(name, value)
self._set_ctr_attr(name, value)
except FrameworkTypeError as error_msg:
# `MetaValueContainer`s can take any value type in contrast to all other state containers
self._check_ctr_attr_name_not_blocked(name, value)
self._handle_attr_value_error_ctr_specifically(name, value, error_msg)
@namespace_item
def _handle_attr_value_error_ctr_specifically(self, name, value, error_msg):
raise FrameworkTypeError(error_msg)
@namespace_item
def _set_ctr_attr(self, name, value):
# def _set_container_attr_container_specifically(self, name, value):
# Case: class or class(es) in MRO integrable `and` not yet integrated
if any([cls in type(self.__class__)._integrated_state_component_attrs.keys()
for cls in value.__class__.mro()]) and 'Ignite' not in value.__class__.__name__:
cls_name = 'Ignite' + value.__class__.__name__
# `bases` should normally be set to `(StateComponentContainerID,)` but it's not defined yet
IntegratedAttribute = MetaStateComponent(cls_name, (), {}, component_obj=value, ctr=self)
IntegratedAttribute(name)
# Case: `value` is not an integratable class instance and is set regualarly
else:
self.__dict__[name] = value
# @namespace_item
# def _set_container_attr_container_specifically(self, name, value):
# self.__dict__[name] = value
@namespace_item
def _check_ctr_attr_name_ctr_specifically(self, name, value):
self._check_ctr_attr_name_not_blocked(name, value)
@namespace_item
def _check_ctr_attr_name_not_blocked(self, name, value):
# Note: transition/component setting only occures during initialization => uncritical for performance
# NAMING RULES FOR TRANSITIONS: ONLY ALLOW UNIQUE TRANSITION/COMPONENT NAMES, meaning...
# ...DO NOT use state container, existing transition/component, state variable or parameter names
# for new transitions/components
if any([name in state._all_blocked_shortcuts_for_new_fso_ctr_names()]):
raise FrameworkNameError('The transition name `{}` may not be used, because e.g. it already names '
'another object in state. Rename the transition `{}` also avoiding the '
'following names: `{}`'.format(
name, value.__class__.__name__, state._all_blocked_shortcuts_for_new_fso_ctr_names()))
@namespace_item
def _check_ctrID_of_ctr_attr_value(self, name, value):
# Checking `isinstance(value, containing_type)` is required for `*ContainerID`-verification
if not (value in containing_types \
or any([isinstance(value, containing_type) for containing_type
in [*containing_types,
*type(self.__class__)._user_defined_ctrID_classes[self.__class__.__name__]]])):
raise FrameworkTypeError(
'`{name}` cannot be set. `state.{ctr}` attributes can only be set with values '
'that are either of type `{types}` or a instance of `{types}`, but given type `{typ}` '
'and MRO `{mro}`.'.format(name=name,
ctr=self.get_state_ctr_name(),
types=[containing_type.__name__ for containing_type in containing_types],
typ=type(value).__name__,
mro=value.__class__.mro()))
@namespace_item
def _create_state_shortcut_for_ctr_attr(self, fso_ctr_name, fso_ctr):
# STATE-SHORTCUT ADDING FOR TRANSITION
# Note: Trivially no shortcut for a transition is required in container (transitions are container attrs)
if fso_ctr_name not in MetaState._non_state_shortcuts:
MetaState._state_shortcuts['ctr_attrs'][fso_ctr_name] = fso_ctr
@namespace_item
def _create_state_and_ctr_shortcuts_of_fso_or_refcaller(
self, fso_or_refcaller_name, value):
self._create_state_shortcut_of_fso_or_refcaller(fso_or_refcaller_name, value)
self._create_ctr_shortcut_of_fso_or_refcaller(fso_or_refcaller_name, value)
@namespace_item
def _create_state_shortcut_of_fso_or_refcaller(self, fso_or_refcaller_name, value):
# Include overload suffixes
fso_suffixes = self._allowed_fso_suffixes(name=fso_or_refcaller_name,
value=value)
for suffix in fso_suffixes:
overloaded_variable = '_'.join([fso_or_refcaller_name, suffix])
if overloaded_variable[-1] == '_':
overloaded_variable = overloaded_variable[:-1]
# Container shortcut
# If already a shortcutted state variable with identical name exists in container, then delete
# shortcut and block this `object_name` for shortcuts as it`s not unique anymore
if overloaded_variable in type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__].keys():
del type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__]\
['_delete_item_' + overloaded_variable]
type(self.__class__)._blocked_ctr_shortcuts_for_fsos[self.__class__.__name__]\
.add(overloaded_variable)
if overloaded_variable not in type(self.__class__)._blocked_ctr_shortcuts_for_fsos[self.__class__.__name__]:
type(self.__class__)._ctr_shortcuts_for_fsos[self.__class__.__name__][overloaded_variable] = value
@namespace_item
def _create_ctr_shortcut_of_fso_or_refcaller(self, fso_or_refcaller_name, value):
# State shortcut
# Include overload suffixes
fso_suffixes = self._allowed_fso_suffixes(name=fso_or_refcaller_name,
value=value)
for suffix in fso_suffixes:
overloaded_variable = '_'.join([fso_or_refcaller_name, suffix])
if overloaded_variable[-1] == '_':
overloaded_variable = overloaded_variable[:-1]
if overloaded_variable in MetaState._state_shortcuts['fsos'].keys():
del MetaState._state_shortcuts['fsos']['_delete_item_' + overloaded_variable]
MetaState._blocked_state_shortcuts['redundant_variables'].add(fso_or_refcaller_name)
if overloaded_variable not in MetaState._blocked_state_shortcuts['redundant_variables']:
MetaState._state_shortcuts['fsos'][overloaded_variable] = value
@namespace_item
def _allowed_fso_suffixes(self, name, value):
if isinstance(value, FrameStateObjectsReference) or value._has_callbackoverloads(name):
return ['', *MetaState._variable_overload_suffixes]
return ['', 'ref']
return super().__new__(typ, cls_name, bases, namespace)
[docs]class MetaBaseTransitionsContainer(MetaStateContainer):
"""
The special containers (e.g. DataLoaders, Modules, etc.) are directly derived herefrom with a
defining a `BaseStateContainer` in between for 2 reasons:
1. no container specific `__init__()` method is required
2. pure framework subclasses, so no feature developer interface `BaseStateContainer` is required
"""
[docs]class MetaStateComponentsContainer(MetaStateContainer):
[docs] def __new__(typ, cls_name, bases, namespace, containing_name, containing_types):
# Decorator for adding methods to namespace ONLY IF they are not overriden by any baseclasses
namespace_item = namespace_decorator(cls_name, bases, namespace)
# DO NOT REMOVE shortcut dictionary attributes, are required as empty dummy for `for`-loops
# @namespace_item
# def __init__(self): pass
# # @namespace_item(only_under_condition=bases_have_attr(bases, attr_name='__setattr__'))
# @namespace_item(only_for_containers='dataloaders')
# def __setattr__(self, name, value):
# super().__setattr__(name, value)
# self.__setattr_(name, value)
@namespace_item
def __enter__(self):
# raise FrameworkNotImplementedError('`Containers of type `{}` are not implemented for the '
# '`with`-statement.'.format(type(self)))
# self.__dict__['_entered_namespace'] = AttributeToItemDictionary()
# return self.__dict__['_entered_namespace']
return self
@namespace_item
def __exit__(self, exc_type, exc_value, traceback):
# TODO:
# - Enhance assigned `Module`s with module wrapper that wrap relevant paramaeters/variables of the
# `Module`s into `StateVariable`s & `StateParameter`s etc.
if exc_type != None:
raise exc_type(exc_value, None, traceback)
@namespace_item
def get_bso_ctr_components(self):
return [fso_ctr.component for fso_ctr in self.get_bso_ctrs()]
# NOTE: METHOD `self.add` CURRENTLY NOT IN USE
#TODO: This is actually a nich feature, but currently not used. It does this:
# `state.dataloaders.add(trainer_dataloader)` and assignes the object with
# name detection as `state.dataloaders.trainer_dataloader`
@namespace_item
def add(self, *args, **kwargs):
# Create a signature to pull out positional and/or keyword arguments correctly
# TODO: args/kwargs are bound correctly, BUT signature in function # print(inspect.signature(<self>.add)`
# gives `(**kwargs)` instead of `(<containing_name>)` TO FIX!!!
# Signature for `add` method
# __signature__ = Signature(parameters=[Parameter(name='name', kind=Parameter.POSITIONAL_OR_KEYWORD),
# Parameter(name=containing_name, kind=Parameter.POSITIONAL_OR_KEYWORD)])
add.__signature__ = Signature(
parameters=[Parameter(name=containing_name, kind=Parameter.POSITIONAL_OR_KEYWORD)])
_bound = add.__signature__.bind(*args, **kwargs)
# self.set(_bound.arguments['name'], _bound.arguments[containing_name])
for keyword, value in _bound.arguments.items():
added_object_name = get_argument_name_from_frame_inspection(name=keyword,
obj=value,
default_name=containing_name)
# Assigne added object to attribute with the added object's global name outside the function
# print(keyword, added_object_name)
self.set(added_object_name, value)
return super().__new__(typ, cls_name, bases, namespace, containing_name, containing_types)
[docs]class MetaValuesContainer(MetaStateContainer):
[docs] def __new__(typ, cls_name, bases, namespace, containing_name, containing_types):
namespace_item = namespace_decorator(cls_name, bases, namespace)
@namespace_item
def _handle_attr_value_error_ctr_specifically(self, name, value, error_msg):
with self as value_ctr:
setattr(value_ctr, name, value)
return super().__new__(typ, cls_name, bases, namespace, containing_name, containing_types)
[docs]class MetaParametersContainer(MetaValuesContainer):
[docs] def __new__(typ, cls_name, bases, namespace, containing_name, containing_types):
namespace_item = namespace_decorator(cls_name, bases, namespace)
# Any fso-ctr assigned directly to `state.params` must be listed here to avoid:
# - raising a FrameworkNameError (because `name` is not equal to another existing fso-ctr name)
# - avoid that `state.params` tries to create a parameter referencing container of it
# NOTE: despite the beloew listed `_params_sepcific_fso_ctr_names`, `state.params` only holds
# `StateParametersReference()`s automatically created from all fso-ctr in `state` that contain
# state parameters.
typ._params_specific_fso_ctr_names = ['constants', 'user_defined_params']
@namespace_item
def _check_ctr_attr_name_ctr_specifically(self, name, value):
"""
Check if container attribute name `name` equals a container name as required.
Set container attribute name `name` where as `name` must be the container name which it is referencing
the parameters from or `helpers` or `argumnents`
"""
if name not in [*MetaState._blocked_state_shortcuts['ctrs'],
*MetaState._non_state_shortcuts,
*type(self.__class__)._params_specific_fso_ctr_names]:
raise FrameworkNameError('The `StateParametersReference()` must be named equal to the container of '
'which it is referencing the parameters, but given: `{}`'.format(name))
return super().__new__(typ, cls_name, bases, namespace, containing_name, containing_types)
[docs]class Configurations(metaclass=MetaValuesContainer, containing_name='config',
containing_types=[ConfigurationsContainerID, BaseTransitionsContainerID]):
"""
Any configuration applied to the code before `starting the engines`. This can be simple parameter switches
that change from their initialization value to their run value, e.g. turning on the event trigger mode of
all `StateVariables` from 'False' to 'True' just before starting the engines. But configurations can also
mean completely rewriting the complete code into highly efficient for-loops or into code compatible for
distributed parallel cloud-computing.
"""
pass
[docs]class Parameters(metaclass=MetaParametersContainer, containing_name='parameter',
containing_types=[ParametersContainerID]):
"""
Training parameter container.
"""
[docs]class Pipelines(metaclass=MetaBaseTransitionsContainer, containing_name='pipelines',
containing_types=[PipelinesContainerID, BaseTransitionsContainerID]):
"""
Container for all output handlers of e.g. metrics & loggers.
"""
[docs]class DataLoaders(metaclass=MetaStateComponentsContainer, containing_name='dataloader',
containing_types=[DataloadersContainerID, StateComponentID, DataLoader]):
"""
Container for all dataloaders.
"""
[docs]class Modules(metaclass=MetaStateComponentsContainer, containing_name='module',
containing_types=[ModulesContainerID, StateComponentID, Module]):
"""
Container for all `torch.nn.Module`\s included for training.
"""
[docs]class Optimizers(metaclass=MetaStateComponentsContainer, containing_name='optimizer',
containing_types=[OptimizersContainerID, StateComponentID,
_LRScheduler, ReduceLROnPlateau, Optimizer]):
"""
Container for all `torch.nn.Optimizer`\s.
"""
[docs]class Engines(metaclass=MetaBaseTransitionsContainer, containing_name='engine',
containing_types=[EnginesContainerID]):
"""
Container for all
"""
[docs]class OutputHandlers(metaclass=MetaBaseTransitionsContainer, containing_name='handler',
containing_types=[OutputhandlersContainerID, BaseTransitionsContainerID,
StateComponentID]):
"""
Container for all output handlers of e.g. metrics & loggers.
"""
class Metrics(metaclass=MetaBaseTransitionsContainer, containing_name='metric',
containing_types=[MetricsContainerID, BaseTransitionsContainerID]):
pass
[docs]class Loggers(metaclass=MetaBaseTransitionsContainer, containing_name='logger',
containing_types=[LoggersContainerID, BaseTransitionsContainerID]):
"""
Container for all loggers despite tensorboard.
"""
[docs]class TransitionStates(metaclass=MetaBaseTransitionsContainer, containing_name='transition',
containing_types=[TransitionsContainerID, BaseTransitionsContainerID]):
"""
`State container` containing all transitions which are no engines, metrics, maintainer, etc.. For example,
the ```Once/EveryNumthValue`` class that is created when calling e.g.
``state.trainer.n_epoch_iteration_started_every_10``.
"""
[docs]class Maintenance(metaclass=MetaBaseTransitionsContainer, containing_name='maintainer',
containing_types=[MaintenanceContainerID, BaseTransitionsContainerID]):
"""
Any maintainance/debugging/re-configuration processed after "starting the engines"
and may also be periodically repeated, e.g.:
- all loggers DESPITE tensorboard loggers
- after each epoch some kind of memory cleanups, etc..
- search tools fo return e.g. at which events a certain function/method is callbacked
- list all callbacks of all state variables
"""
pass
[docs]class TensorboardStates(metaclass=MetaBaseTransitionsContainer, containing_name='chart',
containing_types=[TensorboardContainerID, BaseTransitionsContainerID,
StateComponentID, SummaryWriter]):
"""
Live training state visualization in Tensorboard. All features therefore are contained in here.
"""
pass
[docs]class State(metaclass=MetaState):
"""
State is a singleton class containing the entire training state.
"""
# TODO: Optimize categorization by adding/removing/changing the containers
# - split off addition containers from `state.transitions`, e.g.:
# - `state.output_handlers` for better overview so user reuses calcs rather than recalculating
configs = Configurations()
params = Parameters()
# General `state.helpers` would be redundant as all helpers can be categorized into the state containers already
# helpers = HelperFunctions()
pipelines = Pipelines()
dataloaders = DataLoaders()
modules = Modules()
optimizers = Optimizers()
engines = Engines()
output_handlers = OutputHandlers()
metrics = Metrics()
transitions = TransitionStates()
tensorboard = TensorboardStates()
loggers = Loggers()
maintenance = Maintenance()
def _all_blocked_shortcuts_for_new_fso_ctr_names(self):
all_blocked_shortcuts = set([
'state',
*[blocked_shortcut for blocked_shortcuts in MetaState._blocked_state_shortcuts.values()
for blocked_shortcut in blocked_shortcuts],
*[shortcut for shortcuts in MetaState._state_shortcuts.values() for shortcut in shortcuts.keys()],
*[shortcut for ctr in MetaState._blocked_state_shortcuts['ctrs']
for shortcut in type(self.get(ctr).__class__)._ctr_shortcuts_for_fsos.keys()]])
# exculde attribute names which are allowed to be redundant in `state` ( not within a container),
# e.g. `arguments` in a state containers
return all_blocked_shortcuts - MetaState._non_state_shortcuts
# return set(all_blocked_shortcuts)
[docs] def run(self):
"""
User friendly method to "start the engines". This method breaks with the structure of `State`
as a static container for other static `MetaStateContainer`s (e.g. `state.engines`, `state.modules`)
that has no user methods.
Returns:
"""
# Start `state` initialization: config, maintenance, tensorboard triggered by callback
state.engines.state_status.state_init_started = True
# Complete initialization: config, maintenance, tensorboard triggered by callback
state.engines.state_status.state_init_completed = True
# Start the engines
# Note: Feature design rules normally forbid direct inter-fso-method calls, but as `state` is not fso-ctr
# calling `state.state_status.run()` inside `state.run()` is no break of the rule (even though
# it looks like one...)
state.engines.state_status.state_run_started = True
state = State()
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### S T A T E O B J E C T S R E F E R E N C E S ###
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
# class StateParametersReference(ParametersContainerID, FrameStateObjectsReference):
#
# def _create_reference_caller(self, owner_instance, bso_name, reference_caller_name):
# super()._create_reference_caller(owner_instance, bso_name, reference_caller_name)
# state.params._create_state_and_ctr_shortcuts_of_bso_or_refcaller(
# bso_or_refcaller_name=reference_caller_name,
# value=self)
[docs]class StateParametersReference(ParametersContainerID, FrameStateObjectsReference):
def _create_ref_caller(self, owner_instance, fso_name, ref_caller_name):
super()._create_ref_caller(owner_instance, fso_name, ref_caller_name)
state.params._create_state_and_ctr_shortcuts_of_fso_or_refcaller(
fso_or_refcaller_name=ref_caller_name,
value=self)
# ==================================================================================================================
### PARAMETER REFERENCE INITIALIZATION ###
# ==================================================================================================================
# Note: - The Referencing of the container in `state.params` is on purpose set directly in `__dict__` to avoid
# error raising beause the `StateVariablesReference`s' names are equal to the container names.
# - Shortcuts in `state.params` are anyway not set for `ParameterContainerID`s subclasses
# - On `state.params` ONLY shortcuts set for parameter-call-references (and their overload suffixes), not for the
# `StateVariablesReference`-instances themselves (excluded anyway, see above point).
for parameter_ref in [
'configs', 'dataloaders', 'modules', 'optimizers', 'engines', 'output_handlers', 'metrics', 'transitions',
'tensorboard', 'maintenance']:
# state.params.__dict__[parameter_reference]
setattr(state.params, parameter_ref, StateParametersReference(owner_instances=[], fso_names=[],
ref_caller_names=[]))
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### S T A T E O B J E C T C O N T A I N E R ###
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### META STATE OBJECT CONTAINER ###
# ==================================================================================================================
class MetaFrameStateObjectContainer(type):
_variable_names = {}
_parameter_names = {}
_other_fso_names = {}
# # Referencer tracking for data flow tracking
_fsos_refed_by_other_fso_ctrs = {}
def __new__(typ, cls_name, bases, namespace, ctr):
# Decorator for adding methods to namespace ONLY IF they are not overriden by any baseclasses
namespace_item = namespace_decorator(cls_name, bases, namespace)
# `BaseStateObject`-subclass (e.g. state objects/variables/parameters) instance creation steps:
# In `__new__()`:
# 1. Instantiate & assign to owner class (namesapce)
# 2. Append/Register `BaseStateObject`-instance name to/in `cls._variable/parameter_names/_other_bso_names`
# In `__init__()`:
# 3. Bind to owner instance: call `self._bind_bso_to_owner_instance()` in `__init__()`
# _other_bso_names = []
# typ._variable_names[cls_name] = []
# typ._parameter_names[cls_name] = []
# for key, value in namespace.items():
# if isinstance(value, BaseParameterStateObject):
# type._parameter_names[cls_name].append(key)
# # If instance of `BaseStateVariable` but NOT `StateParameter` (already fetched from first `if`)
# elif isinstance(value, BaseVariableStateObject):
# typ._variable_names[cls_name].append(key)
# # elif isinstance(value, StateObject) or isinstance(value, StateFunction):
# elif isinstance(value, BaseStateObject):
# _other_bso_names.append(key)
# _parameter_names = {}
# typ._other_fso_names = {}
typ._variable_names[cls_name] = [key for key, value in namespace.items() if isinstance(value, FrameVariableStateObject)]
typ._parameter_names[cls_name] = [key for key, value in namespace.items() if isinstance(value, FrameParameterStateObject)]
typ._other_fso_names[cls_name] = [key for key, value in namespace.items() if isinstance(value, FrameStateObject)
and key not in [*typ._variable_names[cls_name], *typ._parameter_names[cls_name]]]
# DO NOT include state variable overload suffixes in names
for fso_name in [*typ._other_fso_names[cls_name], *typ._variable_names[cls_name], *typ._parameter_names[cls_name]]:
if any([word in MetaState._variable_overload_suffixes for word in fso_name.split(sep='_')[-3:]]):
raise FrameworkNameError(
'The state variable name `{}` includes suffix(es) of the overload functionality in the last 3 '
'name parts. Please rename the `{}` instance and avoid the following name fragments `{}`.'.format(
fso_name, fso_name,
[*['_{}_'.format(suffix) for suffix in MetaState._variable_overload_suffixes],
*['_{}'.format(suffix) for suffix in MetaState._variable_overload_suffixes]]))
# EFFECTIVELY (BUT NOT REALLY) ENABLING SUPER-BINDING STATE VARIABLE DESCRIPTORS!!!!
# Any state variables (descriptor) assigned to any baseclass are reassigned to the new class (`namespace`)
for superclass in bases:
for key, value in superclass.__dict__.items():
# Only reassign if state variable name `key` is not overriden by new class
if key not in namespace.keys():
if isinstance(value, FrameParameterStateObject):
typ._parameter_names[cls_name].append(key)
namespace[key] = value
elif isinstance(value, FrameVariableStateObject):
typ._variable_names[cls_name].append(key)
namespace[key] = value
elif isinstance(value, FrameStateObject):
namespace[key] = value
@namespace_item
def __init__(self, name):
self._init_MetaBaseStateObjectContainer(name)
@namespace_item
def _init_MetaBaseStateObjectContainer(self, name):
# `StateVariable` (descriptor) initialization
for fso_name in [*type(self.__class__)._other_fso_names[self.__class__.__name__],
*type(self.__class__)._variable_names[self.__class__.__name__],
*type(self.__class__)._parameter_names[self.__class__.__name__]]:
self._bind_fso_to_owner_instance(fso_name)
# Set name of frame state object container
self.__dict__['name'] = name
# Initiate reference dictionary for data flow tracking of instance with instance name `name` as key
type(self.__class__)._fsos_refed_by_other_fso_ctrs[name] = {}
setattr(ctr, name, self)
# CREATE PARAMETER REFERENCES IN `state.params`
if len(type(self.__class__)._parameter_names[self.__class__.__name__]) != 0 \
and name not in MetaParametersContainer._params_specific_fso_ctr_names:
ctr_name = [key for key, value in state.__class__.__dict__.items()
if id(value) == id(ctr)][0]
getattr(state.params, ctr_name).add_bsos_or_ref_callers(
owner_instances=[self],
fso_names=type(self.__class__)._parameter_names[self.__class__.__name__],
ref_caller_names=['_'.join([name, parameter_name]) for parameter_name
in type(self.__class__)._parameter_names[self.__class__.__name__]])
# TODO:
# - get rid of `MetaEngine` by handing `centainer` directly in `Baseclass.__new__()`
# @namespace_item
# def __new__(typ, name, bases, namespace):
# return super().__new__(typ, name, bases, namespace, container=container)
@namespace_item
def __getattr__(self, name):
# OPTIONS FOR OVERLOADING `__getattr__()` based on examples `state.trainer.n_completed_started`:
# - AMetric(... , update_event=state.trainer.n_iteration_completed_ref,... ) =>
# suffix_len = 1
# - state.trainer.n_epoch_completed_callbacks =>
# suffix_len = 1
# - AMetric(... , update_event=state.trainer.n_iteration_completed_every/once_10_ref,... ) =>
# suffix_len = 3
# EXCLUDED:
# - state.*.n_epoch_started_every/once_10_callback # create once/every AND get callbacks could
# lead to cinfusion and accidentially created events
# print('`{}.__getattr__({})` was called.'.format(self.name, name))
return self._handle_getattr_overload(name)
@namespace_item
def __setattr__(self, name, value):
# '_ref' is never set using overload, only called for function argument assignment
# '_callbacks' list is never set with instance attribute overloading, but directly with
# `StateVariable.__set__()` overlaoding
# OPTIONS FOR OVERLOADING `__setattr__()` based on examples `state.trainer.n_completed_started`:
# - state.trainer.n_epoch_every/once = 10 / a_transition.run => suffix_len = 1,
# type-overload handled by '__set__()'
# - state.trainer.n_epoch_every/once_10 = a_transition.run => suffix_len = 2,
# type-overload handled by '__set__()'
# - state.trainer.n_epoch_every_10 = a_transition.run => suffix_len = 2,
# type-overload handled by '__set__()'
# - AMetric(... , update_event=state.trainer.n_iteration_completed_every/once_10_ref,... ) =>
# suffix_len = 3
# EXCLUDED:
# - After '_callbacks' no further overload suffix is excepted
# - After '_ref' no further overload suffix is excepted nor can it be set with `= value`.
# '_ref` is only used for function arguments setting, Excluded examples:
# - state.trainer.n_epoch_ref = 10 / a_transition.run
# - state.trainer.n_epoch_ref_every/once
# if isinstance(value, FrameStateObjectsReference):
# value._referencing_fso_ctrs[self.name] = (self, name)
if not any([word in MetaState._variable_overload_suffixes for word in name.split(sep='_')[-3:]]):
object.__setattr__(self, name, value)
else:
self._handle_setattr_overload(name, value)
@namespace_item
def get(self, *name_or_parts):
"""
Get method for parameterized attribute names, especially for state object containers, but not only.
Main reason is to simply avoid unhandy commands, e.g.
`getattr(state.transitions, '{}_{}'.format(name, suffix)` commands and simply write
`state.transitions.get(name, suffix)`.
"""
try:
#TODO: - Turn off optional `default` argument properly
return getattr(self, *name_or_parts, FrameworkAttributeError(
'`{c}` object has no attribute `a_value`. Note: You may find this error message unappropriately '
'assigned to a variable or raised within another error. This error message is a hacky workaround '
'to avoid the `default` value of the internal `gettattr(self, *name_ort_parts [, default])` call '
'in the method `{c}.get()`. It hinders `{c}.get()` from ignoring the last of two arguments due to '
'the optional `default` arguemnt of `getattr`. Sorry for the inconveniences, it may be fixed '
'soon.'.format(c=self.__class__.__name__)))
except TypeError:
return getattr(self, '_'.join([str(name_part) for name_part in name_or_parts]))
@namespace_item
def set(self, *name_or_parts_and_value):
try:
setattr(self, *name_or_parts_and_value[:-1],
name_or_parts_and_value[-1])
except TypeError:
setattr(self, '_'.join([str(name_part) for name_part in name_or_parts_and_value[:-1]]),
name_or_parts_and_value[-1])
@namespace_item
def get_bso_names(self):
return [fso.name for fso in self.__class__.__dict__.values() if isinstance(fso, FrameStateObject)]
# ==============================================================================================================
### SUFFIX OVERLOADING ### The following methods implement the state variable suffix overloading
# ==============================================================================================================
@namespace_item
def _bind_fso_to_owner_instance(self, fso_name):
"""
Defines how the value and callback list of the state variable is added to the owner instance.
Args:
self: frame state object container
bso_name: name of attribute
Returns:
"""
# self.__dict__[name] = self.__class__.__dict__[name]._initial_value
# self.__class__[name]._instance__set__(self, self.__class__, self.__class__.__dict__[name]._initial_value)
if fso_name in [*type(self.__class__)._variable_names[self.__class__.__name__],
*type(self.__class__)._parameter_names[self.__class__.__name__]]:
self.__dict__[fso_name + '_callbacks'] = CallbacksList()
# This 'do_not_set' value is required for:
# - `StateComponent` component object attributes that shouldn't be set after initialization
# - generally for any user-defined `_instance__get/set__` method to avoid possible `TypeError`s when
# setting the initial value (in case initial value type is wrong for user defined instance__get/set__)
if self.__class__.__dict__[fso_name]._initial_value != 'do_not_set':
self.__class__.__dict__[fso_name]._instance_attr__set__(
instance=self, value=self.__class__.__dict__[fso_name]._initial_value)
# self.set(name, self.__class__.__dict__[name]._initial_value)
#TODO: HACK ATTACK! ...TO BE REMOVED
# - This case avoids `KeyError: 'output'` in clase of `OuputHandler` that `output` is tried to be gotten
# before it was set. This error is also invoked when only checking `hasattr( <output_handler>, 'output')`
elif 'FrameOutputHandler' in [base_class.__name__ for base_class in self.__class__.mro()]:
# NOTE: Above type checking with type name instead of class circumvents importing `FrameOutputHandler`
# which would not work due to cross-import error
# The following string value is set to inform user/developer in case she/he gets before sets
self.__dict__[fso_name] = 'WARNING: This is a dummy initial value for `{s}.{f}` in case ' \
'`{s}.{f}` is gotten before set.'.format(s=self.name, f=fso_name)
@namespace_item
def _handle_getattr_overload(self, name):
"""
This handler deals with overloaded `__getattr__` for state variables. It identify and process correct
overload attributes. In case of incorrect overload calls, an error is raised
Args:
self:
name:
Returns:
"""
name_prefix, suffix_words, suffix_len, once_or_every_transition_prefix, \
once_or_every_transition_variable_name = self._filter_name_prefix_and_suffixes(name, get_or_set='get')
if suffix_len == -1:
return self._getattr_ctr_specifically(name)
# Only possibility for '__getattr__`: reference overloading 'ref'
if self._getting_ref_single_suffix(name_prefix, suffix_words, suffix_len):
return self._get_fso_reference(name=name_prefix)
elif self._getting_once_or_every_numth_double_suffix(name_prefix, suffix_words, suffix_len):
return self._get_or_set_once_or_every_conditional_call_and_return_name(
once_or_every_transition_prefix,
once_or_every_transition_variable_name,
name=name_prefix,
suffix=suffix_words[0],
num_calls=int(suffix_words[1]))
# Case: init value is returned of new variable state, e.g. `initial_value =
# state.trainer.n_epoch_started_every/once_10`
elif self._getting_once_or_every_numth_and_ref_or_callback_triple_suffix(name_prefix, suffix_words,
suffix_len):
return self._get_ref_or_callbacks_of_once_or_every_numth_transition(
name_prefix, suffix_words, once_or_every_transition_prefix, once_or_every_transition_variable_name)
@namespace_item
def _getattr_ctr_specifically(self, name):
"""
Required for `StateCompoennt`s where first the attribute is tried to get from teh component
`self.component` and only if that fails the attribute is gotten from `self`.
"""
# If no suffix matches an overload suffix then raise regular AttributeError
return object.__getattribute__(self, name)
@namespace_item
def _get_ref_or_callbacks_of_once_or_every_numth_transition(self, name_prefix, suffix_words,
once_or_every_transition_prefix,
once_or_every_transition_variable_name):
# Name identival to name inside `self._get_or_set_once_or_every_conditional_call_and_return_name()`
once_or_every_transition_name = '_'.join([once_or_every_transition_prefix, *suffix_words[:-1]])
# Create every/once `StateVariable`
if not hasattr(state.transitions, once_or_every_transition_name):
_ = self.get(name_prefix, *suffix_words[:2])
# Then return reference or callbacks list depending on last suffix word
return state.transitions.get(once_or_every_transition_name)\
.get(once_or_every_transition_variable_name, suffix_words[2])
@namespace_item
def _handle_setattr_overload(self, name, value):
"""
This handler deals with overloaded `__setattr__` for state variables. It identify and process correct
overload attributes. In case of incorrect overload calls, an error is raised
Args:
self:
name:
value:
Returns:
"""
name_prefix, suffix_words, suffix_len, once_or_every_transition_prefix, \
once_or_every_transition_variable_name = self._filter_name_prefix_and_suffixes(name, get_or_set='set')
if suffix_len == -1:
if isinstance(value, FrameStateObjectsReference):
for ref_caller_name in value._ref_caller_names:
owner_instance = value._owner_instances[ref_caller_name]
fso_name = value._fso_names[ref_caller_name]
owner_instance_metacls = type(owner_instance.__class__)
if fso_name not in owner_instance_metacls._fsos_refed_by_other_fso_ctrs[owner_instance.name].keys():
owner_instance_metacls._fsos_refed_by_other_fso_ctrs[owner_instance.name][fso_name] = {}
#TODO: - Discuss: 'fso_ctr' is not necessary thanks to unique fso-ctr names and
# could be omitted
owner_instance_metacls._fsos_refed_by_other_fso_ctrs[owner_instance.name][fso_name][self.name] = \
{'attr_name': name,
'ref_caller_name': ref_caller_name}
self._setattr_ctr_specifically(name, value)
return
# `suffix_len` for `__setattr__` brings a benefit only up to `2` at max (for `__getattr__` 3 is also
# useful):
# Examples for "useless-3-setter" that can be replaced by "useful-2-setter":
# - `<trans>.n_epoch_started_every_9_ref = func` <=> `<trans>.n_epoch_started_every_9 = func`
# - `<trans>.n_epoch_starte_every_9_callbacks = [func1, funx2]` => `<trans>.n_epoch_started_every_9 = [
# func1, func2]`
if self._setting_once_or_every_single_suffix(suffix_words, suffix_len):
# Only `every` & `once` possible, whereas `callbacks` and `ref` are not allowed to be set as an
# attribute
# DO NOT OVERRIDE/SET callback list or reference which are both stored in owner instance's `__dict__`
self._get_or_set_once_or_every_conditional_call_and_return_name(
once_or_every_transition_prefix,
once_or_every_transition_variable_name,
name=name_prefix,
suffix=suffix_words[0],
num_calls=value)
return
elif self._setting_once_or_every_numth_double_suffix(suffix_words, suffix_len):
# Only 'once` & `every` followed by integer string as suffix possible
# The set-`value` refers to the value passed to `__set__()` of the new variable state that is created
# below
self._get_or_set_once_or_every_conditional_call_and_return_name(
once_or_every_transition_prefix,
once_or_every_transition_variable_name,
name=name_prefix,
suffix=suffix_words[0],
num_calls=int(suffix_words[1]),
value=value)
return
# Not allowed for `__setattr__` and raise by `self.filter_name_predix_and_suffixes`: suffix_len == 3
else:
# Raise `RuntimeError` in case of any bad suffix combinations
self._raise_setattr_runtime_error(name, get_or_set='set')
@namespace_item
def _setattr_ctr_specifically(self, name, value):
object.__setattr__(self, name, value)
@namespace_item
def _filter_name_prefix_and_suffixes(self, name, get_or_set):
"""
Prefilters all suffix comninations allowed for `__getattr__` AND `__setattr__` overload.
If conditions ar not met, then getter/setter specific error is raised.
The everloads functinality of '_every' & '_once' are realized by implementing a new `Every/OnceNumthCall`
transition with the standardized name `<self.name>_every/once_<num_calls>` that is stored in
`state.transitions`. The overloaded `__getattr__` & `__setattr__` "just" handles the instantiation
triggering and the referencing/forwarding values, calls and other overloads.
Args:
self:
name:
get_or_set:
Returns:
"""
words = name.split(sep='_')
# Starting with `suffix_len=1` as full `name` is not an attribute of `self` which is tested before this
# function call.
for suffix_len in range(1, min(len(words), 3) + 1):
name_prefix_is_a_variable = False
name_prefix = '_'.join(words[:-suffix_len])
# Hardcoded standardized `Every/OnceNumthCounter` name, see doc above
every_or_once_transition_prefix = '_'.join([self.name, name_prefix])
every_or_once_transition_variable_name = 'n_conditional_counts'
suffix_words = words[-suffix_len:]
name_parts = name_prefix, suffix_words, suffix_len, \
every_or_once_transition_prefix, every_or_once_transition_variable_name
if self._is_fso(name_prefix):
name_prefix_is_a_variable = True
if suffix_words[0] in MetaState._variable_overload_suffixes:
if suffix_len == 1:
return name_parts
if suffix_len == 2 and int(suffix_words[1]) > 0:
return name_parts
if suffix_len == 3 and suffix_words[2] in MetaState._variable_overload_suffixes \
and get_or_set == 'get':
return name_parts
if get_or_set == 'get':
# container specific non-fso attribute getting `_getattr_container_specifically`
# and `suffix_len = -1` is the flag therefore
# Note: This may provoke an `AttributeError` later on if `self._getattr_container_specifically` in
# the next step finds no attribute `name`
return name, [], -1, '', ''
elif get_or_set == 'set' and name_prefix_is_a_variable:
# Raise `RuntimeError` in case trying to set a 3-word-suffix (note: getting with 3 words is possible)
self._raise_setattr_runtime_error(name, get_or_set)
elif get_or_set not in ['get', 'set']:
raise ValueError('The argument `get_or_set` must have value `get` or `set`, but got: {}'.format(
get_or_set))
else:
# regular attribute is to be set by `__setattr__` and `suffix_len = -1` is the flag therefore
return name, [], -1, '', ''
@namespace_item
def _getting_ref_single_suffix(self, name_prefix, suffix_words, suffix_len):
return suffix_len == 1 and self._is_fso(name=name_prefix) and suffix_words[0] == 'ref'
@namespace_item
def _get_fso_reference(self, name):
"""
Singleton like reference call which creates reference once, stored it in self.__dict__ and will be
returned by self.__getattribute__()` in any following ref-get-calls.
Args:
self:
name:
Returns:
"""
return StateObjectsReference(owner_instances=[self], fso_names=[name])
@namespace_item
def _getting_once_or_every_numth_double_suffix(self, name_prefix, suffix_words, suffix_len):
return suffix_len == 2 and self._has_callbackoverloads(name=name_prefix) \
and suffix_words[0] in ['once', 'every'] \
and int(suffix_words[1]) > 0
@namespace_item
def _get_or_set_once_or_every_conditional_call_and_return_name(self, once_or_every_transition_prefix,
once_or_every_transition_variable_name,
name, suffix, num_calls, value=''):
"""
Args:
once_or_every_transition_prefix:
once_or_every_transition_variable_name:
name:
suffix:
num_calls:
value: setting value AND implicit `get_or_set`-flag, whereas `''` is equivalent to `get_or_set='get'`
resulting in a getter-like-call, and any other value invokes a setter-like-call wiht `value` as
setter-value
Returns:
"""
once_or_every_transition_name = '_'.join([once_or_every_transition_prefix, suffix, str(num_calls)])
if not hasattr(state.transitions, once_or_every_transition_name):
if suffix == 'once':
OnceOrEveryCounter = OnceNumthValue
elif suffix == 'every':
OnceOrEveryCounter = EveryNumthValue
OnceOrEveryCounter(name=once_or_every_transition_name,
conditional_value_ref=self.get(name + '_ref'),
threshold_or_num_calls=num_calls)
if value == '':
return getattr(getattr(state.transitions, once_or_every_transition_name),
once_or_every_transition_variable_name)
else:
setattr(getattr(state.transitions, once_or_every_transition_name),
once_or_every_transition_variable_name, value)
return
@namespace_item
def _getting_once_or_every_numth_and_ref_or_callback_triple_suffix(self, name_prefix, suffix_words, suffix_len):
return suffix_len == 3 and self._has_callbackoverloads(name=name_prefix) \
and suffix_words[0] in ['once', 'every'] \
and int(suffix_words[1]) > 0 \
and suffix_words[2] in ['ref', 'callbacks']
@namespace_item
def _setting_once_or_every_single_suffix(self, suffix_words, suffix_len):
return suffix_len == 1 and suffix_words[0] in ['once', 'every']
@namespace_item
def _setting_once_or_every_numth_double_suffix(self, suffix_words, suffix_len):
return suffix_len == 2 and suffix_words
@namespace_item
def _is_fso(self, name):
if name in self.__class__.__dict__.keys():
value = self.__class__.__dict__[name]
if isinstance(value, FrameStateObject):
return True
return False
@namespace_item
def _has_callbackoverloads(self, name):
if name in self.__class__.__dict__.keys(): # Recursion issue with `hasattr(self, name)`
value = self.__class__.__dict__[name]
if isinstance(value, FrameCallbackoverloadsStateObject):
return True
return False
@namespace_item
def _raise_setattr_runtime_error(self, name, get_or_set):
if get_or_set == 'get':
method_name = '__get__()'
elif get_or_set == 'set':
method_name = '__set__()'
elif get_or_set not in ['get', 'set']:
raise ValueError('The argument `get_or_set` must have value `get` or `set`, but got: {}'.format(
get_or_set))
raise FrameworkNameError('`{}` is an unspecified overload expression for `StateVariable.{}`. '
'State variable suffixes that are no specified overload suffixes should '
'not be set by users. Which overload is specified also depends on being '
'called by `__get__()` or `__set__()`. Allowed are generally overloaded '
'`StateVariable` names with suffixes i.e. `_ref`, `_callbacks`-getter, '
'`_once/_every`-setter, `_once/_every_<int>`-getter/setter, '
'`_every/once_<int>_ref`-getter, but got for `{}`: {}`-setter'.format(
name, method_name, name, method_name))
# ===========================================
### END STATE VARIABLE SUFFPX OVERLOADING ###
namespace['name'] = cls_name
# namespace['_variable_names'] = _variable_names[cls_name]
# namespace['_parameter_names'] = _parameter_names
# namespace['_other_fso_names'] = _other_fso_names
# namespace['__new__'] = __new__
# Only define `-__init__` for base class creation (i.e. when called by base class). When
# `MetaTransition.__new__()` is called for feature creation by a subclass of
# `BaseSomeTransition` that already defines a `__init__()` DO NOT override it
# if not (any([hasattr(base, '__init__') for base in bases]) or '__init__' in namespace.keys()):
# namespace['__init__'] = __init__
# if not '__new__' in namespace.keys():
# namespace['__new__'] = __new__
return type.__new__(typ, cls_name, bases, namespace)
[docs]class MetaEnteredFrameStateObjectContainer(MetaFrameStateObjectContainer):
"""
Metaclass to create new classes with user defined frame state objects at runtime.
"""
_blocked_argument_and_hyperparam_names = ['device', 'run_mode', 'init_mode']
[docs] def __new__(typ, cls_name, bases, namespace, entered_namespace, default_attr_fso_cls, ctr):
namespace_item = namespace_decorator(cls_name, bases, namespace)
for attr_name, attr_value in entered_namespace.items():
if attr_name in typ._blocked_argument_and_hyperparam_names:
raise FrameworkAttributeError(
'The argument or hyperparameter name `{}` cannot be set, as it is used already as a framework '
'configuration parameter. Please rename the argument/hyperparameter and avoid the names '
'`{}`.'.format(attr_name, typ._blocked_argument_and_hyperparam_names))
if not isinstance(attr_value, FrameStateObject):
attr_value = default_attr_fso_cls(initial_value=attr_value)
namespace[attr_name] = attr_value
@namespace_item
def __enter__(self):
self.__dict__['_entered_namespace'] = DecoratorAndAttributeToItemDictionary()
return self.__dict__['_entered_namespace']
@namespace_item
def __exit__(self, exc_type, exc_value, traceback):
if exc_type != None:
raise exc_type(exc_value, None, traceback)
func_dict = self.__dict__['_entered_namespace'].pop('decorator_args')
for func in func_dict.values():
self.add_new_state_function(func=func)
for fso_name, fso_value in self.__dict__['_entered_namespace'].items():
self.add_new_bso(name=fso_name, value=fso_value)
dict.__delitem__(self.__dict__, '_entered_namespace')
@namespace_item
def add_new_state_function(self, func):
if not isinstance(func, StateFunction):
if type(func) in CALLABLE_TYPES:
name = func.__name__
value = StateFunction(initial_value=func)
self.add_new_bso(name, value)
else:
raise FrameworkTypeError('The argument `func` in `add_new_state_function` must be one of types `{}`,'
'but got: `{}`'.fomrat(CALLABLE_TYPES, type(func).__name__))
@namespace_item
def add_new_bso(self, name, value):
self._setattr_as_new_fso(name, value)
@namespace_item
def _setattrs_from_additional_dict_as_fsos(self, attrs_dict):
# Set class attribute
for attr_name, attr_value in attrs_dict.items():
self._setattr_as_new_fso(name=attr_name, value=attr_value)
@namespace_item
def _setattr_as_new_fso(self, name, value):
# NOTE: This method holds redundant code to `MetaBaseStateObjectContainer.__new__()`
# which couldn't be avoided so far.
# Check `attr_name`
# No overload-suffixes allowed in `attr_name`
if any([word in MetaState._variable_overload_suffixes for word in name.split(sep='_')]):
raise FrameworkNameError(
'The state variable name `{}` includes suffix(es) of the overload functionality. '
'Please rename the `{}` instance and avoid the following name fragments `{}`.'.format(
name, name,
[*['_{}_'.format(suffix) for suffix in MetaState._variable_overload_suffixes],
*['_{}'.format(suffix) for suffix in MetaState._variable_overload_suffixes]]))
# Check if `attr_name` is not already a frame state object or class/instance attr
if hasattr(self, name):
raise FrameworkNameError('Attribute with name `{}` already exists in `{}` and cannot '
'be overwritten. Please change the name.'.format(name, self.name))
# If `default_attr_bso_type` is not meant for functions but value is a function...
if isinstance(value, CALLABLE_TYPES) and FrameFunctionStateObject not in default_attr_fso_cls.mro():
# ...then convert the function to `StateFunction`
# Note: Setting the value of a NON-`BaseFunctionStateObject` with a function messes things up.
value = StateFunction(initial_value=value)
# Convert to `default_attr_type` in case `attr_value` is not yet a `BaseStateObject`
if not isinstance(value, FrameStateObject):
value = default_attr_fso_cls(initial_value=value)
# Assign frame state object to the class of the container instance `self`
setattr(self.__class__, name, value)
# Set frame state object `name`
setattr(self.__class__.__dict__[name], 'name', name)
# register/append according to its type
if FrameParameterStateObject in value.__class__.mro():
type(self.__class__)._parameter_names[self.__class__.__name__].append(name)
# If instance of `BaseStateVariable` but NOT `StateParameter` (already fetched from first `if`)
elif FrameVariableStateObject in value.__class__.mro():
type(self.__class__)._variable_names[self.__class__.__name__].append(name)
# elif isinstance(attr_value, StateObject) or isinstance(attr_value, StateFunction):
elif FrameStateObject in value.__class__.mro():
type(self.__class__)._other_fso_names[self.__class__.__name__].append(name)
# Create shortcuts in state and state container
ctr._create_state_and_ctr_shortcuts_for_fso(fso_name=name, fso_ctr=self)
self._bind_fso_to_owner_instance(fso_name=name)
return super().__new__(typ, cls_name, bases, namespace, ctr)
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### S T A T E C O M P O N E N T ### Mainly PyTorch classes, but any external class can be integrated
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### PYTORCH CLASS INTEGRATION ###
# ==================================================================================================================
# Pytorch class instance attrs that are integrated in state as frame state object, ordered from sub- so superclasses
MetaStateContainer._integrated_state_component_attrs = INTEGRATED_STATE_COMPONENT_ATTRS
# ==================================================================================================================
### META STATE COMPONENT ###
# ==================================================================================================================
[docs]class MetaStateComponent(MetaFrameStateObjectContainer):
# TODO:
# - implement this metaclass as its base & subclasses
"""
Metaclass of wrapper class integrating PyTorch object (e.g. `DataLoader()`\s, `Module()`\s, `Optimizer()`\s, etc.)
into `state`\.
When assigning a PyTorch object to the `state` this object is wrapped into a subclass of `BaseStateComponent`
which integrates all its relevant parameters and variables into `state` to expose (make them available) to the
user. For example, when the trainer's dataloader `trainer_dataloader = DataLoader(...)`
is assigned to `state.dataloaders.trainer_dataloader = trainer_dataloader`, the batch size of the dataloader
should be integrated as a `StateParameter` which will make it available as hyperparameter in
`state.params.dataloaders.trainer_dataloader_batch_size` (shortcut: `state.trainer_dataloader_batch_size`)
with all the functionalities like suffix overloading & callbakcks.
"""
[docs] def __new__(typ, cls_name, bases, namespace, component_obj, ctr):
bases = tuple(list(bases) + [StateComponentID])
namespace_item = namespace_decorator(cls_name, bases, namespace)
# Collect component attributes which are integrated as a frame state object subclass
to_be_integrated_component_attrs_dicts = \
[(name, *attrs) for sub_cls in component_obj.__class__.mro()
if sub_cls in type(ctr)._integrated_state_component_attrs.keys()
for name, attrs in type(ctr.__class__)._integrated_state_component_attrs[sub_cls].items()]
for fso_name, fso_type, initial_value, instance_attr__get__, instance_attr__set__ \
in to_be_integrated_component_attrs_dicts:
# Exclude redundant attrs of (to be integrated Pytorch) mro-superclasses:
if fso_name not in namespace.keys():
# if hasattr(component_obj, bso_name):
# initial_value = getattr(component_obj, bso_name)
# Case: `initial_value == ''` triggers the frame state object to set it default `initial_value`
# Case: `initial_value == 'do_not_set'` skips setting a intial value
# In case `instance_attr__get/set__ == None` frame state object sets its default `__get/set__`
namespace[fso_name] = \
fso_type(initial_value, instance_attr__get__, instance_attr__set__)
@namespace_item
def __init__(self, name):
# Copy+paste component object `__dict__` to `__dict__` of state component object
self.__dict__['component'] = component_obj
# Initialize superclass, equivalent to `super().__init__()` for regular class inheritance
self._init_MetaBaseStateObjectContainer(name)
if hasattr(component_obj, '__call__'):
sig = signature(component_obj)
@namespace_item
def __call__(self, *args, **kwargs):
__call__.__signature__ = sig
return component_obj(*args, **kwargs)
@namespace_item
def _getattr_ctr_specifically(self, name):
try:
# If `self` is instance of `StateComponent` then name may be attribute of `self.component`
return getattr(self.__dict__['component'], name)
except AttributeError:
# If no suffix matches an overload suffix then raise regular AttributeError
return object.__getattribute__(self, name)
@namespace_item
def _setattr_ctr_specifically(self, name, value):
# In `StateComponent`s non-`BaseStateObject` attributes are only set in `self.component`
setattr(self.__dict__['component'], name, value)
return super().__new__(typ, cls_name, bases, namespace, ctr)
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### T R A N S I T I O N S ###
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
class MetaTransition(MetaFrameStateObjectContainer):
def __new__(typ, cls_name, bases, namespace, ctr):
namespace_item = namespace_decorator(cls_name, bases, namespace)
# @namespace_item
# def run(self):
# raise_run_not_implemented_and_appendable_to_callback_error(transition=self)
return super().__new__(typ, cls_name, bases, namespace, ctr)
# class MetaTensorboardComponent(MetaBaseStateObjectContainer):
# def __new__(typ, name, bases, namespace):
# return super().__new__(typ, name, bases, namespace, ctr=state.tensorboard)
[docs]class MetaEngine(MetaTransition):
"""Define the St"""
[docs] def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.engines)
class MetaMetric(MetaTransition):
def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.metrics)
class MetaBaseTransition(MetaTransition):
def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.transitions)
class MetaMaintainer(MetaTransition):
def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.maintenance)
class MetaTensorboard(MetaTransition):
def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.tensorboard)
class MetaLogger(MetaTransition):
def __new__(typ, name, bases, namespace):
return super().__new__(typ, name, bases, namespace, ctr=state.loggers)
class MetaStateContainerStatus(MetaTransition):
def __new__(typ, cls_name, bases, namespace, ctr, variable_name_prefix):
namespace_item = namespace_decorator(cls_name, bases, namespace)
# Boolean variables describing the state of the container
# Note:
# - `*_started` and `*_completed` must be separated as different callables may be triggered a those events
# - when e.g. container initialization is completed (of course `*_init_completed` is set `True` but)
# `*_init_started` cannot be re-set `False` (which may seem appropriate) because this would retrigger the
# initial callbacks
# - 4 different container states are offered for many possibilities for individual use cases
started_init_var_name = variable_name_prefix + '_init_started'
completed_init_var_name = variable_name_prefix + '_init_completed'
started_run_var_name = variable_name_prefix + '_run_started'
completed_run_var_name = variable_name_prefix + '_run_completed'
for variable_name in [started_init_var_name, completed_init_var_name,
started_run_var_name, completed_run_var_name]:
namespace[variable_name] = StateBooleanVariable()
@namespace_item
def __init__(self, name):
# Initialize superclass, equivalent to `super().__init__()` for regular class inheritance
self._init_MetaBaseStateObjectContainer(name)
# Append the container transitions' `run`/`terminate` to
# `state.engines.state_status.state_run_started/completed`
if self.__class__.__name__ != 'StateStatus':
# Synchronize `*_init_started/completed` of all state container transitions involved in infracture
# processes (i.e. `state.configs_status`, `state.maintenance.maintenance_status`,
# `state.tensorboard.tensorboard_status`)
self.set(started_init_var_name, state.engines.state_status.state_init_started_ref)
self.set(completed_init_var_name, state.engines.state_status.state_init_completed_ref)
self.set(started_run_var_name, state.engines.state_status.state_run_started_ref)
self.set(completed_run_var_name, state.engines.state_status.state_run_completed_ref)
# state.engines.state_status.state_run_started = self # appends `self.run`
# state.engines.state_status.state_run_completed = (self.name + '.terminate', self.terminate)
# @namespace_item
# def run(self):
# self.set(started_run_var_name, True)
#
# @namespace_item
# def terminate(self):
# self.set(completed_run_var_name, True)
return super().__new__(typ, cls_name, bases, namespace, ctr)
# ==================================================================================================================
### FRAME TRANSITIONS ###
# ==================================================================================================================
[docs]class FrameEngine(EnginesContainerID, metaclass=MetaEngine):
"""
Make `BaseEngine.__init__()` that will instantiate the descriptors. Any subclasses with its own
`__init__()` must call `super().__init__(name)`.
"""
# class BaseLogger(LoggersContainerID, metaclass=MetaLogger):
# """
#
# """
[docs]class FrameBaseTransition(BaseTransitionsContainerID, metaclass=MetaBaseTransition):
"""
Make `FrameBaseTransition.__init__()` that will instantiate the descriptors. Any subclasses with its own
`__init__()` must call `super().__init__(name)`.
"""
[docs]class FrameMaintainer(MaintenanceContainerID, metaclass=MetaMaintainer):
"""
Make `BaseTransition.__init__()` that will instantiate the descriptors. Any subclasses with its own
`__init__()` must call `super().__init__(name)`.
"""
[docs]class FrameMaintainerMetric(MaintenanceContainerID, MetricID, metaclass=MetaBaseTransition):
"""
Metric of infrastructure resources.
"""
# class FrameTensorboardComponent(TensorboardContainerID, metaclass=MetaTensorboardComponent):
# """
#
# """
class MetaChart(MetaTransition):
def __new__(typ, cls_name, bases, namespace):
# #TODO:
# # - Discuss if adding `transform_input/output` may be of interest? => would become output handler
# namespace = add_output_handler_methods_and_attrs_to_namespace(name, bases, namespace,
# output_name='summary_writer_arguments')
return super().__new__(typ, cls_name, bases, namespace, ctr=state.tensorboard)
# ==================================================================================================================
### FRAME PLUGINS ###
# ==================================================================================================================
#TODO:
# - build `Frame/Base` interface that work for `StateFunctions`
# class FrameMaintainerPlugin(MaintenanceContainerID, metaclass=MetaEnteredFrameStateObjectContainer,
# entered_namespace={},default_attr_fso_cls=StateFunction, ctr=state.maintenance): pass
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### O U T P U T H A N D L E R S ###
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
[docs]def add_output_handler_methods_and_attrs_to_namespace(name, bases, namespace, output_name):
"""
All methods target to set the `output` state variable arguments `instance_attr__get/set__` in the
instance of the class of the output handler metaclass (in `self.__init__`) while `output` has
already been initialized in the metaclass. So the arguments normally would have to be available
when the class is generated by the metaclass (`typ.__new__()`), but are first available during
instance initialization (`self.__init__()`).
Args:
name:
bases:
namespace:
Returns:
"""
namespace_item = namespace_decorator(name, bases, namespace)
# @namespace_item
# def __new__(cls, *args, **kwargs):
#
# # Workaround to define `staticmethod`s because `@staticmethod` and `staticmethod(func)` does not work
# cls._transform_input = _transform_input
# cls._transform_output = _transform_output
# return cls
def _instance_attr__get__(self, instance, owner):
# _instance_attr__get__.__name__ = '_instance_attr__get__'
value = FrameStateObject._default_instance_attr__get__(self, instance, owner)
return instance._transform_output(value)
def _instance_attr__set__(self, instance, value):
# _instance_attr__set__.__name__ = '_instance_attr__set__'
FrameStateObject._default_instance_attr__set__(self, instance,
value=instance._transform_input(value))
@namespace_item
@staticmethod
def _transform_input(value):
"""
Abstract static method to transfrom the `__set__` `value` before assigning it.
If `_transform_input` is not overriden by a subclass
Args:
value:
Returns:
"""
return value
@namespace_item
@staticmethod
def _transform_output(value):
return value
namespace[output_name] = StateVariable(instance_attr__get__=_instance_attr__get__,
instance_attr__set__=_instance_attr__set__)
return namespace
class MetaOutputHandler(MetaTransition):
def __new__(typ, cls_name, bases, namespace):
namespace = add_output_handler_methods_and_attrs_to_namespace(cls_name, bases, namespace, output_name='output')
return super().__new__(typ, cls_name, bases, namespace, ctr=state.output_handlers)
# ==================================================================================================================
### FRAME OUTPUTHANDLER ### CALLBACK STATE VARIABLES
# ==================================================================================================================
[docs]class FrameOutputHandler(OutputhandlersContainerID, metaclass=MetaOutputHandler):
"""
Base class of any output handler.
Output handlers are used when certain output transformation are required by multiple
transitions or metrics. This way repeated calculations of the same output transformation
which perhaps may me very costly can be avoided.
"""
[docs]class FrameChart(TensorboardContainerID, metaclass=MetaChart):
"""
Framework class of any tensorboard chart.
"""
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
### C O M P O S I T I O N S ###
# ==================================================================================================================
# ==================================================================================================================
# ==================================================================================================================
class MetaFrameStateObjectContainerComposition(MetaFrameStateObjectContainer):
pass
class MetaChartComposition(MetaFrameStateObjectContainerComposition):
def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.tensorboard)
class MetaPipeline(MetaFrameStateObjectContainerComposition):
def __new__(typ, cls_name, bases, namespace):
return super().__new__(typ, cls_name, bases, namespace, ctr=state.pipelines)
# ==================================================================================================================
### FRAME STATE OBJECT CONTAINER COMPOSITIONS ###
# ==================================================================================================================
# ==================================================================================================================
### STATE CONTAINER STATUS' ### CALLBACK STATE VARIABLES
# ==================================================================================================================
[docs]class ConfigurationsStatus(BaseTransitionsContainerID,
metaclass=MetaStateContainerStatus,
ctr=state.configs,
variable_name_prefix='config'):
"""Trigger for all configuration work."""
[docs]class StateStatus(EnginesContainerID,
metaclass=MetaStateContainerStatus,
ctr=state.engines,
variable_name_prefix='state'):
"""
Transition starting and endpoint of the `state.engines` state container.
The endpoints are defined by the state variable counters `state.engines.state_status.state_run_started` &
`state.engines.state_status.state_run_completed`.
"""
# # DEBUGGING
# object_a = StateObject(initial_value=12345)
# object_b = StateObject()
# object_c = StateObject()
# param_a = StateParameter()
# param_b = StateParameter()
# param_c = StateParameter(initial_value=100000)
def run(self):
state.engines.state_status.state_run_started = True
state.engines.state_status.state_run_completed = True
metaclass=MetaStateContainerStatus,
ctr=state.maintenance,
variable_name_prefix='maintenance'): pass
metaclass=MetaStateContainerStatus,
ctr=state.tensorboard,
variable_name_prefix='tensorboard'): pass
metaclass=MetaStateContainerStatus,
ctr=state.loggers,
variable_name_prefix='logger'): pass
# # NICE EXAMPLE FOR `inspect.Signature` use case!!!!! DO NOT DELETE BEFORE COPYING TO STOOPEDIA
# class TensorboardLogger(BaseTransitionContainerID,
# metaclass=MetaStateContainerTransition,
# ctr=state.tensorboard,
# variable_name_prefix='tensorboard'):
#
# def __init__(self, *args, **kwargs):
# # Signature handling
# init_parameters = signature(SummaryWriter).parameters.copy()
# # Generate `OrderedDict` of `TensorboardLogger` parameters
# init_parameters = OrderedDict(
# name=Parameter(name='name', kind=Parameter.POSITIONAL_OR_KEYWORD)).update(init_parameters)
# # Create `TensorboardLogger` signature
# self.__init__.__signature__ = Signature(*init_parameters.values())
# # Distribute arguments to super- & subclass
# _bound = OrderedDict(self.__init__.__signature__.bind(*args, **kwargs).arguments)
# # Name initialization
# self.name = _bound.pop('name')
# # `SummaryWriter` initialization
# self.writer = SummaryWriter(**_bound)
# IMPORTANT: Order of initialization defines which `run` is added first to the callbacks of
# `state.engines.state_statuss.state_run_started`
StateStatus(name='state_status')
ConfigurationsStatus(name='configs_status')
MaintenanceStatus(name='maintenance_status')
TensoboardStatus(name='tensorboard_status')
LoggersStatus(name='loggers_status')
# Setting configuration variable relations and transitions during state run
# Entangling `state.init_mode` & `state.run_mode`
state.engines.state_status.state_init_started
# ==================================================================================================================
### TRANSITIONS ###
# ==================================================================================================================
[docs]class FrameConditionalCounter(FrameBaseTransition):
n_conditional_counts = StateVariable(initial_value=0)
[docs] def __init__(self, name, conditional_value_ref, threshold_or_num_calls, caller_refs='', min_or_max='max'):
super().__init__(name)
self.threshold_value_or_num_calls = threshold_or_num_calls
self._internal_counter_value = 0
# The `to_be_set_value` is the value which `self.n_conditional_counts` will be set. This value can be
# modified by `self.condition`
# Initial value is set to the variable reference's initial value so first setting/callbacks-trigger only happens
# after `self.to_be_set_value` was modified by `self.condition` for the first time.
# self.conditional_set_value = caller_variable_ref.caller_name
# Attach event
self._conditional_value_ref = conditional_value_ref
caller_refs = [conditional_value_ref] if caller_refs == '' else caller_refs \
if isinstance(caller_refs, list) else [caller_refs]
for caller_ref in caller_refs:
caller_ref.caller_name = self
self.min_or_max = min_or_max
# def condition(self):
# # `StateVariable.__set__()` only executes callbacks if the new value differs from the current value
# self.conditional_set_value = self._internal_call_counts // self.num_calls
[docs] def condition(self):
""""
Modify `set.conditional_set_value` based on user conditions. For triggering the callbacks list of
the state variable 'self.n_conditional_counts` the `self.conditional_set_value` has to be increased by '+1`.
"""
raise_or_warn_abstract_method_not_implemented_error(instance=self, method=condition)
def true_transform(self):
raise_or_warn_abstract_method_not_implemented_error(instance=self, method=true_transform)
def false_transform(self):
raise_or_warn_abstract_method_not_implemented_error(instance=self, method=false_transform)
def run(self):
if self.condition():
self.true_transform()
else:
self.false_transform()
[docs]class OnceNumthValue(FrameConditionalCounter):
# TODO:
# - Would it make sense to remove the callback `OnceNumthCounter.run` from the state variable
# callback list after it was triggered once?
"""
"""
def __init__(self, name, conditional_value_ref, threshold_or_num_calls, caller_refs='', min_or_max='max'):
super().__init__(
name=name, conditional_value_ref=conditional_value_ref, threshold_or_num_calls=threshold_or_num_calls,
caller_refs=caller_refs, min_or_max=min_or_max)
self._internal_counter_value = self._get_new_internal_counter_value()
self._initial_value = self.__class__.__dict__['n_conditional_counts']._initial_value
def _get_new_internal_counter_value(self):
return self._conditional_value_ref.caller_name - self.threshold_value_or_num_calls
[docs] def condition(self):
# If state variable hast not yet been change before, proceed till it was changed ONCE
if self.n_conditional_counts == self._initial_value:
# Each time a multiple of a threshold is surpassed by `self.caller_variable_ref.caller_name` de[ending on
# `min_or_max`
# new_internal_counter_value = self._get_new_internal_counter_value()
# if new_internal_counter_value == 0:
# return True
return self._conditional_value_ref.caller_name >= self.threshold_value_or_num_calls if self.min_or_max == \
'max' \
else self._conditional_value_ref.caller_name <= self.threshold_value_or_num_calls
return False
[docs]class EveryNumthValue(FrameConditionalCounter):
[docs] def condition(self):
# Each time a multiple of a threshold is surpassed by `self.caller_variable_ref.caller_name` de[ending on
# `min_or_max`
new_internal_counter_value = self._conditional_value_ref.caller_name // self.threshold_value_or_num_calls
condition_state = new_internal_counter_value > self._internal_counter_value if self.min_or_max == 'max' \
else new_internal_counter_value < self._internal_counter_value
if condition_state:
self._internal_counter_value = new_internal_counter_value
return condition_state
