Skip to content
Snippets Groups Projects
Commit 9a16e9ca authored by Philip ABBET's avatar Philip ABBET
Browse files

Refactoring: Move the 'Library' and 'Algorithm' classes into beat.backend.python

parent 9b9d4f3e
No related branches found
No related tags found
1 merge request!14Add support to serve databases from a docker container
Hide whitespace changes
Inline Side-by-side
beat/core/algorithm.py
+ 5
− 419
View file @ 9a16e9ca
......@@ -39,155 +39,15 @@ from . import dataformat
from . import library
from . import schema
from . import prototypes
from . import loader
from . import utils
class Storage(utils.CodeStorage):
"""Resolves paths for algorithms
from beat.backend.python.algorithm import Storage
from beat.backend.python.algorithm import Runner
from beat.backend.python.algorithm import Algorithm as BackendAlgorithm
Parameters:
prefix (str): Establishes the prefix of your installation.
name (str): The name of the algorithm object in the format
``<user>/<name>/<version>``.
"""
def __init__(self, prefix, name, language=None):
if name.count('/') != 2:
raise RuntimeError("invalid algorithm name: `%s'" % name)
self.username, self.name, self.version = name.split('/')
self.prefix = prefix
self.fullname = name
path = utils.hashed_or_simple(self.prefix, 'algorithms', name)
super(Storage, self).__init__(path, language)
class Runner(object):
'''A special loader class for algorithms, with specialized methods
Parameters:
module (module): The preloaded module containing the algorithm as
returned by :py:func:`beat.core.loader.load_module`.
obj_name (str): The name of the object within the module you're interested
on
exc (class): The class to use as base exception when translating the
exception from the user code. Read the documention of :py:func:`run`
for more details.
algorithm (object): The algorithm instance that is used for parameter
checking.
*args: Constructor parameters for the algorithm (normally none)
**kwargs: Constructor parameters for the algorithm (normally none)
'''
def __init__(self, module, obj_name, algorithm, exc=None, *args,
**kwargs):
try:
class_ = getattr(module, obj_name)
except Exception as e:
if exc is not None:
type, value, traceback = sys.exc_info()
six.reraise(exc, exc(value), traceback)
else:
raise #just re-raise the user exception
self.obj = loader.run(class_, '__new__', exc, *args, **kwargs)
self.name = module.__name__
self.algorithm = algorithm
self.exc = exc
# if the algorithm does not have a 'setup' method, it is ready by default
self.ready = not hasattr(self.obj, 'setup')
def _check_parameters(self, parameters):
"""Checks input parameters from the user and fill defaults"""
user_keys = set(parameters.keys())
algo_parameters = self.algorithm.parameters or {}
valid_keys = set(algo_parameters.keys())
# checks the user is not trying to set an undeclared parameter
if not user_keys.issubset(valid_keys):
err_keys = user_keys - valid_keys
message = "parameters `%s' are not declared for algorithm `%s' - " \
"valid parameters are `%s'" % (
','.join(err_keys),
self.name,
','.join(valid_keys),
)
exc = self.exc or KeyError
raise exc(message)
# checks all values set by the user are in range (if a range is set)
retval = dict() #dictionary with checked user parameters and defaults
for key, definition in algo_parameters.items():
if key in parameters:
try:
value = self.algorithm.clean_parameter(key, parameters[key])
except Exception as e:
message = "parameter `%s' cannot be safely cast to the declared " \
"type on algorithm `%s': %s" % (key, self.name, e)
exc = self.exc or ValueError
raise exc(message)
else: #user has not set a value, use the default
value = algo_parameters[key]['default']
# in the end, set the value on the dictionary to be returned
retval[key] = value
return retval
def setup(self, parameters, *args, **kwargs):
'''Sets up the algorithm, only effective on the first call'''
if self.ready: return self.ready
completed_parameters = self._check_parameters(parameters) #may raise
kwargs['parameters'] = completed_parameters
if hasattr(self.obj, 'setup'):
self.ready = loader.run(self.obj, 'setup', self.exc, *args, **kwargs)
return self.ready
else:
return True
def process(self, *args, **kwargs):
'''Runs through data'''
if not self.ready:
message = "algorithm `%s' is not yet setup" % (self.name,)
exc = self.exc or RuntimeError
raise self.exc(message)
return loader.run(self.obj, 'process', self.exc, *args, **kwargs)
def __getattr__(self, key):
'''Returns an attribute of the algorithm - only called at last resort'''
return getattr(self.obj, key)
class Algorithm(object):
class Algorithm(BackendAlgorithm):
"""Algorithms represent runnable components within the platform.
This class can only parse the meta-parameters of the algorithm (i.e., input
......@@ -276,19 +136,8 @@ class Algorithm(object):
"""
def __init__(self, prefix, data, dataformat_cache=None, library_cache=None):
super(Algorithm, self).__init__(prefix, data, dataformat_cache, library_cache)
self._name = None
self.storage = None
self.prefix = prefix
self.dataformats = {}
self.libraries = {}
self.groups = []
dataformat_cache = dataformat_cache if dataformat_cache is not None else {}
library_cache = library_cache if library_cache is not None else {}
self._load(data, dataformat_cache, library_cache)
def _load(self, data, dataformat_cache, library_cache):
"""Loads the algorithm"""
......@@ -556,269 +405,6 @@ class Algorithm(object):
(library, self.libraries[library].language, self.language))
@property
def name(self):
"""Returns the name of this object
"""
return self._name or '__unnamed_algorithm__'
@name.setter
def name(self, value):
if self.data['language'] == 'unknown':
raise RuntimeError("algorithm has no programming language set")
self._name = value
self.storage = Storage(self.prefix, value, self.data['language'])
@property
def schema_version(self):
"""Returns the schema version"""
return self.data.get('schema_version', 1)
@property
def language(self):
"""Returns the current language set for the executable code"""
return self.data['language']
@language.setter
def language(self, value):
"""Sets the current executable code programming language"""
if self.storage: self.storage.language = value
self.data['language'] = value
def clean_parameter(self, parameter, value):
"""Checks if a given value against a declared parameter
This method checks if the provided user value can be safe-cast to the
parameter type as defined on its specification and that it conforms to any
parameter-imposed restrictions.
Parameters:
parameter (str): The name of the parameter to check the value against
value (object): An object that will be safe cast into the defined
parameter type.
Returns:
The converted value, with an appropriate numpy type.
Raises:
KeyError: If the parameter cannot be found on this algorithm's
declaration.
ValueError: If the parameter cannot be safe cast into the algorithm's
type. Alternatively, a ``ValueError`` may also be raised if a range or
choice was specified and the value does not obbey those settings
estipulated for the parameter
"""
if (self.parameters is None) or (parameter not in self.parameters):
raise KeyError(parameter)
retval = self.parameters[parameter]['type'].type(value)
if 'choice' in self.parameters[parameter] and \
retval not in self.parameters[parameter]['choice']:
raise ValueError("value for `%s' (%r) must be one of `[%s]'" % \
(parameter, value, ', '.join(['%r' % k for k in \
self.parameters[parameter]['choice']])))
if 'range' in self.parameters[parameter] and \
(retval < self.parameters[parameter]['range'][0] or \
retval > self.parameters[parameter]['range'][1]):
raise ValueError("value for `%s' (%r) must be picked within " \
"interval `[%r, %r]'" % (parameter, value,
self.parameters[parameter]['range'][0],
self.parameters[parameter]['range'][1]))
return retval
@property
def valid(self):
"""A boolean that indicates if this algorithm is valid or not"""
return not bool(self.errors)
@property
def uses(self):
return self.data.get('uses')
@uses.setter
def uses(self, value):
self.data['uses'] = value
return value
@property
def results(self):
return self.data.get('results')
@results.setter
def results(self, value):
self.data['results'] = value
return value
@property
def parameters(self):
return self.data.get('parameters')
@parameters.setter
def parameters(self, value):
self.data['parameters'] = value
return value
@property
def splittable(self):
return self.data.get('splittable', False)
@splittable.setter
def splittable(self, value):
self.data['splittable'] = value
return value
def uses_dict(self):
"""Returns the usage dictionary for all dependent modules"""
if self.data['language'] == 'unknown':
raise RuntimeError("algorithm has no programming language set")
if not self._name:
raise RuntimeError("algorithm has no name")
retval = {}
if self.uses is not None:
for name, value in self.uses.items():
retval[name] = dict(
path=self.libraries[value].storage.code.path,
uses=self.libraries[value].uses_dict(),
)
return retval
def runner(self, klass='Algorithm', exc=None):
"""Returns a runnable algorithm object.
Parameters:
klass (str): The name of the class to load the runnable algorithm from
exc (class): If passed, must be a valid exception class that will be
used to report errors in the read-out of this algorithm's code.
Returns:
:py:class:`beat.core.algorithm.Runner`: An instance of the algorithm,
which will be constructed, but not setup. You **must** set it up
before using the ``process`` method.
"""
if not self._name:
exc = exc or RuntimeError
raise exc("algorithm has no name")
if self.data['language'] == 'unknown':
exc = exc or RuntimeError
raise exc("algorithm has no programming language set")
if not self.valid:
message = "cannot load code for invalid algorithm (%s)" % (self.name,)
exc = exc or RuntimeError
raise exc(message)
# loads the module only once through the lifetime of the algorithm object
try:
self.__module = getattr(self, 'module',
loader.load_module(self.name.replace(os.sep, '_'),
self.storage.code.path, self.uses_dict()))
except Exception as e:
if exc is not None:
type, value, traceback = sys.exc_info()
six.reraise(exc, exc(value), traceback)
else:
raise #just re-raise the user exception
return Runner(self.__module, klass, self, exc)
@property
def description(self):
"""The short description for this object"""
return self.data.get('description', None)
@description.setter
def description(self, value):
"""Sets the short description for this object"""
self.data['description'] = value
@property
def documentation(self):
"""The full-length description for this object"""
if not self._name:
raise RuntimeError("algorithm has no name")
if self.storage.doc.exists():
return self.storage.doc.load()
return None
@documentation.setter
def documentation(self, value):
"""Sets the full-length description for this object"""
if not self._name:
raise RuntimeError("algorithm has no name")
if hasattr(value, 'read'):
self.storage.doc.save(value.read())
else:
self.storage.doc.save(value)
def hash(self):
"""Returns the hexadecimal hash for the current algorithm"""
if not self._name:
raise RuntimeError("algorithm has no name")
return self.storage.hash()
def result_dataformat(self):
"""Generates, on-the-fly, the dataformat for the result readout"""
if not self.results:
raise TypeError("algorithm `%s' is a block algorithm, not an analyzer" \
% (self.name))
format = dataformat.DataFormat(self.prefix,
dict([(k, v['type']) for k,v in self.results.items()]))
format.name = 'analysis:' + self.name
return format
def json_dumps(self, indent=4):
"""Dumps the JSON declaration of this object in a string
......
beat/core/library.py
+ 4
− 172
View file @ 9a16e9ca
......@@ -39,32 +39,12 @@ from . import prototypes
from . import loader
from . import utils
class Storage(utils.CodeStorage):
"""Resolves paths for libraries
from beat.backend.python.library import Storage
from beat.backend.python.library import Library as BackendLibrary
Parameters:
prefix (str): Establishes the prefix of your installation.
name (str): The name of the library object in the format
``<user>/<name>/<version>``.
"""
def __init__(self, prefix, name, language=None):
if name.count('/') != 2:
raise RuntimeError("invalid library name: `%s'" % name)
self.username, self.name, self.version = name.split('/')
self.prefix = prefix
self.fullname = name
path = utils.hashed_or_simple(self.prefix, 'libraries', name)
super(Storage, self).__init__(path, language)
class Library(object):
class Library(BackendLibrary):
"""Librarys represent independent algorithm components within the platform.
This class can only parse the meta-parameters of the library. The actual
......@@ -121,20 +101,8 @@ class Library(object):
"""
def __init__(self, prefix, data, library_cache=None):
super(Library, self).__init__(prefix, data, library_cache)
self._name = None
self.storage = None
self.prefix = prefix
self.libraries = {}
library_cache = library_cache if library_cache is not None else {}
try:
self._load(data, library_cache)
finally:
if self._name is not None: #registers it into the cache, even if failed
library_cache[self._name] = self
def _load(self, data, library_cache):
"""Loads the library"""
......@@ -251,142 +219,6 @@ class Library(object):
(library, self.libraries[library].language, self.language))
def uses_dict(self):
"""Returns the usage dictionary for all dependent modules"""
if self.data['language'] == 'unknown':
raise RuntimeError("library has no programming language set")
if not self._name:
raise RuntimeError("library has no name")
retval = {}
if self.uses is not None:
for name, value in self.uses.items():
retval[name] = dict(
path=self.libraries[value].storage.code.path,
uses=self.libraries[value].uses_dict(),
)
return retval
def load(self):
"""Loads the Python module for this library resolving all references
Returns the loaded Python module.
"""
if self.data['language'] == 'unknown':
raise RuntimeError("library has no programming language set")
if not self._name:
raise RuntimeError("library has no name")
return loader.load_module(self.name.replace(os.sep, '_'),
self.storage.code.path, self.uses_dict())
@property
def name(self):
"""Returns the name of this object
"""
return self._name or '__unnamed_library__'
@property
def schema_version(self):
"""Returns the schema version"""
return self.data.get('schema_version', 1)
@name.setter
def name(self, value):
if self.data['language'] == 'unknown':
raise RuntimeError("library has no programming language set")
self._name = value
self.storage = Storage(self.prefix, value, self.data['language'])
@property
def language(self):
"""Returns the current language set for the library code"""
return self.data['language']
@language.setter
def language(self, value):
"""Sets the current executable code programming language"""
if self.storage: self.storage.language = value
self.data['language'] = value
self._check_language_consistence()
@property
def valid(self):
"""A boolean that indicates if this library is valid or not"""
return not bool(self.errors)
@property
def uses(self):
return self.data.get('uses')
@uses.setter
def uses(self, value):
self.data['uses'] = value
return value
@property
def description(self):
"""The short description for this object"""
return self.data.get('description', None)
@description.setter
def description(self, value):
"""Sets the short description for this object"""
self.data['description'] = value
@property
def documentation(self):
"""The full-length description for this object"""
if not self._name:
raise RuntimeError("library has no name")
if self.storage.doc.exists():
return self.storage.doc.load()
return None
@documentation.setter
def documentation(self, value):
"""Sets the full-length description for this object"""
if not self._name:
raise RuntimeError("library has no name")
if hasattr(value, 'read'):
self.storage.doc.save(value.read())
else:
self.storage.doc.save(value)
def hash(self):
"""Returns the hexadecimal hash for the current library"""
if not self._name:
raise RuntimeError("library has no name")
return self.storage.hash()
def json_dumps(self, indent=4):
"""Dumps the JSON declaration of this object in a string
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment