Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
""" Contains named expression classes for Research """
import os
from .results import Results
from ..named_expr import NamedExpression
class ResearchNamedExpression(NamedExpression):
""" NamedExpression base class for Research objects """
def _get(self, **kwargs):
name = self._get_name(**kwargs)
return name, kwargs
class ResearchExecutableUnit(ResearchNamedExpression):
""" NamedExpression for ExecutableUnit """
def _get(self, **kwargs):
_, kwargs = super()._get(**kwargs)
experiment = kwargs['experiment']
return experiment
def get(self, **kwargs):
experiment = self._get(**kwargs)
if isinstance(experiment, (list, tuple)):
_experiment = experiment
def update_bar(bar, bar_desc, **kwargs):
""" Update bar with description and one step."""
current_iter = bar.n
if bar_desc:
if callable(bar_desc) and not isinstance(bar_desc, NamedExpression):
desc = bar_desc()
if current_iter == 0:
# During the first iteration we can't get items from empty containers (lists, dicts, etc)
try:
desc = eval_expr(bar_desc, **kwargs)
desc = str(desc)
except LookupError:
desc = None
else:
desc = eval_expr(bar_desc, **kwargs)
desc = str(desc)
bar.set_description(desc)
bar.update(1)
return total
if 'ratio'.startswith(name):
if total is None:
raise ValueError('Total number of iterations is not defined!')
ratio = pipeline._iter_params['_n_iters'] / total # pylint:disable=protected-access
return ratio
raise ValueError('Unknown key for named expresssion I: %s' % name)
def assign(self, *args, **kwargs):
""" Assign a value by calling a callable """
_ = args, kwargs
raise NotImplementedError("Assigning a value to an iteration number is not supported")
class F(NamedExpression):
""" A function, method or any other callable that takes a batch or a pipeline and possibly other arguments
Examples
--------
::
F(MyBatch.rotate)(angle=30)
F(make_data)
F(prepare_data)(115, item=10)
Notes
-----
Take into account that the actual calls will look like `current_batch.rotate(angle=30)`,
`make_data(current_batch)` and `prepare_data(current_batch, 115, item=10)`.
"""
def __init__(self, name, mode='w', _pass=True):
value = dataset
elif hasattr(dataset, name):
value = getattr(dataset, name)
else:
raise KeyError("Attribute does not exist in the dataset", name)
return value
def assign(self, value, **kwargs):
""" Assign a value to a dataset attribute """
name, dataset, _ = self._get(**kwargs)
if name is None:
raise ValueError('Assigning a value to D() is not possible.')
setattr(dataset, name, value)
class R(NamedExpression):
""" A random value
Notes
-----
If `size` is needed, it should be specified as a named, not a positional argument.
Examples
--------
::
R('normal', 0, 1)
R('poisson', lam=5.5, seed=42, size=3)
R(['metro', 'taxi', 'bike'], p=[.6, .1, .3], size=10)
"""
def __init__(self, name, *args, state=None, seed=None, size=None, **kwargs):
super().__init__(name)
def assign(self, *args, **kwargs):
""" Assign a value """
_ = args, kwargs
raise NotImplementedError("Assigning a value to a random variable is not supported")
def __repr__(self):
repr_str = 'R(' + str(self.name)
if self.args:
repr_str += ', ' + ', '.join(str(a) for a in self.args)
if self.kwargs:
repr_str += ', ' + str(self.kwargs)
return repr_str + (', size=' + str(self.size) + ')' if self.size else ')')
class W(NamedExpression):
""" A wrapper which returns the wrapped named expression without evaluating it
Examples
--------
::
W(V('variable'))
W(B(copy=True))
W(R('normal', 0, 1, size=B('size')))
"""
def get(self, **kwargs):
""" Return a wrapped named expression """
if not isinstance(self.name, NamedExpression):
raise ValueError("Named expressions is expected, but given %s" % self.name)
self.name.set_params(**kwargs)
return self.name
super().__init__(AN_EXPR, mode='w')
self.op = op
self.a = a
self.b = b
def get(self, **kwargs):
""" Return a value of an algebraic expression """
kwargs = self.get_params(**kwargs)
a = eval_expr(self.a, **kwargs)
b = eval_expr(self.b, **kwargs)
if self.op in UNARY_OPS:
return OPERATIONS[self.op](a)
return OPERATIONS[self.op](a, b)
class B(NamedExpression):
""" Batch component or attribute name
Notes
-----
``B()`` return the batch itself.
To avoid unexpected data changes the copy of the batch may be returned, if ``copy=True``.
Examples
--------
::
B('size')
B('images_shape')
B(copy=True)
"""
def default_name(self):
""": str - the model class name (serve as a default for a model name) """
if isinstance(self.model_class, NamedExpression):
raise ValueError("Model name should be explicitly set if a model class is a named expression",
self.model_class)
return self.model_class.__name__
name, batch, _ = self._get(**kwargs)
if isinstance(batch, _DummyBatch):
raise ValueError("Batch expressions are not allowed in static models: B('%s')" % name)
if name is None:
return batch.copy() if self.copy else batch
return getattr(batch, name)
def assign(self, value, **kwargs):
""" Assign a value to a batch component """
name, batch, _ = self._get(**kwargs)
if name is not None:
setattr(batch, name, value)
class PipelineNamedExpression(NamedExpression):
""" Base class for pipeline expressions """
def _get(self, **kwargs):
name, kwargs = super()._get(**kwargs)
pipeline = kwargs['batch'].pipeline
return name, pipeline, kwargs
class C(PipelineNamedExpression):
""" A pipeline config option
Notes
-----
``C()`` return config itself.
Examples
--------
::
def update_bar(bar, bar_desc, **kwargs):
""" Update bar with description and one step."""
current_iter = bar.n
if bar_desc:
if callable(bar_desc) and not isinstance(bar_desc, NamedExpression):
desc = bar_desc()
if current_iter == 0:
# During the first iteration we can't get items from empty containers (lists, dicts, etc)
try:
desc = NamedExpression.eval_expr(bar_desc, **kwargs)
desc = str(desc)
except LookupError:
desc = None
else:
desc = NamedExpression.eval_expr(bar_desc, **kwargs)
desc = str(desc)
bar.set_description(desc)
bar.update(1)
name = partial(name, *args)
return name() if self._call else name
def assign(self, *args, **kwargs):
""" Assign a value by calling a callable """
_ = args, kwargs
raise NotImplementedError("Assigning a value with a callable is not supported")
class L(F):
""" A function, method or any other callable """
def __init__(self, name, mode='w'):
super().__init__(name, mode=mode, _pass=False)
class D(NamedExpression):
""" Dataset attribute or dataset itself
Examples
--------
::
D()
D('classes')
D('organization')
"""
def _get(self, **kwargs):
name, kwargs = super()._get(**kwargs)
batch = kwargs['batch']
dataset = batch.dataset or kwargs['batch'].pipeline.dataset
if dataset is None:
raise ValueError("Dataset is not set", self)