Source code for orchestrator.utils.module_factory
from inspect import isclass
from abc import ABC, abstractmethod
from .recorder import Recorder
from ..utils.exceptions import ModuleAlreadyInFactoryError
[docs]
class ModuleFactory:
"""
Factory for spawning concrete modules given a token.
This class is not instantiated directly, but through a concrete
:class:`ModuleBuilder`. A factory object is used for one module type,
specified at instantiation, to avoid token confusion/conflicts. Specific
module factories, used by module builders, are defined in each module.
:param base_class: The ABC base class defining the module that this
factory will spawn.
:type base_class: ``ABC``
"""
[docs]
def __init__(self, base_class):
self.base_class = base_class
self._creators = {}
[docs]
def add_new_module(self, module_type, module):
"""
Add a concrete class to the factory.
:param module_type: Token to specify the module. The convention is to
use all caps.
:type module_type: str
:param module: Implementation of the base_class to be added to the
factory.
:type module: type(self.base_class)
:raises ValueError: If the module_type is already registered or the
module is not a subclass of the base_class.
"""
if module_type in self._creators:
raise ModuleAlreadyInFactoryError(
f'Trying to add {module_type} to ModuleFactory, but it is '
'already listed!')
else:
if isclass(module) and self.base_class in module.__mro__:
self._creators[module_type] = module
else:
raise ValueError(f'Supplied module is not a '
f'{self.base_class.__name__} class!')
[docs]
def select_module(self, module_type):
"""
Select the type of module to be trained.
The specified module must have been added to the factory. Available
modules can be listed with the :meth:`list_modules` function.
:param module_type: Token to specify the desired module.
:type module_type: str
:returns: Requested concrete implementation of a module.
:rtype: Module
:raises ValueError: If the module_type has not been added to the
factory.
"""
if module_type in self._creators:
return self._creators[module_type]
raise ValueError(f'Module type {module_type} has not been added '
'to the factory yet! If you expected this module '
'to be present, it may be an optional dependency '
'which has not been installed in the environment.')
[docs]
def list_modules(self):
"""
Print the currently available modules in the factory.
This method is primarily designed as a utility for interactive use
of the orchestrator.
"""
print('The factory has the following Modules available (Key-Class):')
for module_type, module_class in self._creators.items():
print(f'{module_type} - {module_class}')
[docs]
class ModuleBuilder(Recorder, ABC):
"""
Abstract base class for module constructors.
This class sets the factory to be used for the builder. The default is to
use the specific module_factory generated by specific modules. A user-
defined :class:`ModuleFactory` can optionally be supplied instead.
:param factory: A module factory, typically defined in specific
implementations.
:type factory: ModuleFactory
"""
[docs]
def __init__(self, factory: ModuleFactory):
"""
Constructor for the ModuleBuilder, sets the factory to build from.
:param factory: A module factory, typically defined in specific
implementations.
:type factory: ModuleFactory
"""
super().__init__()
self.factory = factory
[docs]
def list_modules(self):
"""
Print the currently available modules in the builder.
This method is primarily designed as a utility for interactive use
of the orchestrator.
"""
self.factory.list_modules()
[docs]
@abstractmethod
def build(self):
"""
Return an instance of the specified module.
The build method signature is not specified by this class, as different
modules require different parameters to instantiate.
"""
pass