linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add features

Browse Source 
+ split out metaclass and params

Change-Id: I4d9092827cd74da6757ef1f30d2460471e5e5ef3
Reviewed-on: https://forge.frm2.tum.de/review/18190
Tested-by: JenkinsCodeReview <bjoern_pedersen@frm2.tum.de>
Reviewed-by: Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
This commit is contained in:
Enrico Faulhaber
2018-06-18 18:03:54 +02:00
parent 33e6ded5b8
commit 48382852b8
4 changed files with 511 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
secop/features.py Normal file
View File

@ -0,0 +1,164 @@
# -*- coding: utf-8 -*-
# *****************************************************************************
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# Module authors:
# Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
#
# *****************************************************************************
"""Define Mixin Features for real Modules implemented in the server"""
from __future__ import print_function
from secop.datatypes import EnumType, TupleOf, StringType, FloatRange, StructOf, ArrayOf, BoolType
from secop.modules import Parameter, Command
from secop.metaclass import ModuleMeta, add_metaclass
@add_metaclass(ModuleMeta)
class Feature(object):
"""all things belonging to a small, predefined functionality influencing the working of a module"""
pass
class HAS_PID(Feature):
# note: implementors should either use p,i,d or pid, but ECS must be handle both cases
# note: if both p,i,d and pid are implemented, it MUST NOT matter which one gets a change, the final result should be the same
# note: if there are additional custom accessibles with the same name as an element of the struct, the above applies
# note: (i would still but them in the same group, though)
# note: if extra elements are implemented in the pid struct they MUST BE
# properly described in the description of the pid Parameter
accessibles = {
'use_pid' : Parameter('use the pid mode', datatype=EnumType(openloop=0, pid_control=1), ),
'p' : Parameter('proportional part of the regulation', datatype=FloatRange(0), ),
'i' : Parameter('(optional) integral part', datatype=FloatRange(0), optional=True),
'd' : Parameter('(optional) derivative part', datatype=FloatRange(0), optional=True),
'base_output' : Parameter('(optional) minimum output value', datatype=FloatRange(0), optional=True),
'pid': Parameter('(optional) Struct of p,i,d, minimum output value',
datatype=StructOf(p=FloatRange(0),
i=FloatRange(0),
d=FloatRange(0),
base_output=FloatRange(0),
), optional=True,
), # note: struct may be extended with custom elements (names should be prefixed with '_')
'output' : Parameter('(optional) output of pid-control', datatype=FloatRange(0), optional=True, readonly=False),
}
class Has_PIDTable(HAS_PID):
accessibles = {
'use_pidtable' : Parameter('use the zoning mode', datatype=EnumType(fixed_pid=0, zone_mode=1)),
'pidtable' : Parameter('Table of pid-values vs. target temperature', datatype=ArrayOf(TupleOf(FloatRange(0),
StructOf(p=FloatRange(0),
i=FloatRange(0),
d=FloatRange(0),
_heater_range=FloatRange(0),
_base_output=FloatRange(0),),),), optional=True), # struct may include 'heaterrange'
}
class HAS_Persistent(Feature):
#extra_Status {
# 'decoupled' : Status.OK+1, # to be discussed.
# 'coupling' : Status.BUSY+1, # to be discussed.
# 'coupled' : Status.BUSY+2, # to be discussed.
# 'decoupling' : Status.BUSY+3, # to be discussed.
#}
accessibles = {
'persistent_mode': Parameter('Use persistent mode',
datatype=EnumType(off=0,on=1),
default=0, readonly=False),
'is_persistent': Parameter('current state of persistence',
datatype=BoolType(), optional=True),
'stored_value': Parameter('current persistence value, often used as the modules value',
datatype='main', unit='$', optional=True),
'driven_value': Parameter('driven value (outside value, syncs with stored_value if non-persistent)',
datatype='main', unit='$' ),
}
class HAS_Tolerance(Feature):
# detects IDLE status by checking if the value lies in a given window:
# tolerance is the maximum allowed deviation from target, value must lie in this interval
# for at least ´timewindow´ seconds.
accessibles = {
'tolerance': Parameter('Half height of the Window',
datatype=FloatRange(0), default=1, unit='$'),
'timewindow': Parameter('Length of the timewindow to check',
datatype=FloatRange(0), default=30, unit='s',
optional=True),
}
class HAS_Timeout(Feature):
accessibles = {
'timeout': Parameter('timeout for movement',
datatype=FloatRange(0), default=0, unit='s'),
}
class HAS_Pause(Feature):
# just a proposal, can't agree on it....
accessibles = {
'pause': Command('pauses movement', arguments=[], result=None),
'go': Command('continues movement or start a new one if target was change since the last pause',
arguments=[], result=None),
}
class HAS_Ramp(Feature):
accessibles = {
'ramp': Parameter('speed of movement', unit='$/min',
datatype=FloatRange(0)),
'use_ramp': Parameter('use the ramping of the setpoint, or jump',
datatype=EnumType(disable_ramp=0, use_ramp=1),
optional=True),
'setpoint': Parameter('currently active setpoint',
datatype=FloatRange(0), unit='$',
readonly=True, ),
}
class HAS_Speed(Feature):
accessibles = {
'speed' : Parameter('(maximum) speed of movement (of the main value)',
unit='$/s', datatype=FloatRange(0)),
}
class HAS_Accel(HAS_Speed):
accessibles = {
'accel' : Parameter('acceleration of movement', unit='$/s^2',
datatype=FloatRange(0)),
'decel' : Parameter('deceleration of movement', unit='$/s^2',
datatype=FloatRange(0), optional=True),
}
class HAS_MotorCurrents(Feature):
accessibles = {
'movecurrent' : Parameter('Current while moving',
datatype=FloatRange(0)),
'idlecurrent' : Parameter('Current while idle',
datatype=FloatRange(0), optional=True),
}
class HAS_Curve(Feature):
# proposed, not yet agreed upon!
accessibles = {
'curve' : Parameter('Calibration curve', datatype=StringType(80), default='<unset>'),
# XXX: tbd. (how to upload/download/select a curve?)
}

176
secop/metaclass.py Normal file
View File

@ -0,0 +1,176 @@
# -*- coding: utf-8 -*-
# *****************************************************************************
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# Module authors:
# Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
#
# *****************************************************************************
"""Define Metaclass for Modules/Features"""
from __future__ import print_function
try:
# pylint: disable=unused-import
from six import add_metaclass # for py2/3 compat
except ImportError:
# copied from six v1.10.0
def add_metaclass(metaclass):
"""Class decorator for creating a class with a metaclass."""
def wrapper(cls):
orig_vars = cls.__dict__.copy()
slots = orig_vars.get('__slots__')
if slots is not None:
if isinstance(slots, str):
slots = [slots]
for slots_var in slots:
orig_vars.pop(slots_var)
orig_vars.pop('__dict__', None)
orig_vars.pop('__weakref__', None)
return metaclass(cls.__name__, cls.__bases__, orig_vars)
return wrapper
import time
from secop.errors import ProgrammingError
from secop.datatypes import EnumType
from secop.params import Parameter
EVENT_ONLY_ON_CHANGED_VALUES = True
# warning: MAGIC!
class ModuleMeta(type):
"""Metaclass
joining the class's properties, parameters and commands dicts with
those of base classes.
also creates getters/setter for parameter access
and wraps read_*/write_* methods
(so the dispatcher will get notfied of changed values)
"""
def __new__(mcs, name, bases, attrs):
newtype = type.__new__(mcs, name, bases, attrs)
if '__constructed__' in attrs:
return newtype
# merge properties, Parameter and commands from all sub-classes
for entry in ['properties', 'parameters', 'commands']:
newentry = {}
for base in reversed(bases):
if hasattr(base, entry):
newentry.update(getattr(base, entry))
newentry.update(attrs.get(entry, {}))
setattr(newtype, entry, newentry)
# apply Overrides from all sub-classes
newparams = getattr(newtype, 'parameters')
for base in reversed(bases):
overrides = getattr(base, 'overrides', {})
for n, o in overrides.items():
newparams[n] = o.apply(newparams[n].copy())
for n, o in attrs.get('overrides', {}).items():
newparams[n] = o.apply(newparams[n].copy())
# Correct naming of EnumTypes
for k, v in newparams.items():
if isinstance(v.datatype, EnumType) and not v.datatype._enum.name:
v.datatype._enum.name = k
# check validity of Parameter entries
for pname, pobj in newtype.parameters.items():
# XXX: allow dicts for overriding certain aspects only.
if not isinstance(pobj, Parameter):
raise ProgrammingError('%r: Parameters entry %r should be a '
'Parameter object!' % (name, pname))
# XXX: create getters for the units of params ??
# wrap of reading/writing funcs
rfunc = attrs.get('read_' + pname, None)
for base in bases:
if rfunc is not None:
break
rfunc = getattr(base, 'read_' + pname, None)
def wrapped_rfunc(self, maxage=0, pname=pname, rfunc=rfunc):
if rfunc:
self.log.debug("rfunc(%s): call %r" % (pname, rfunc))
value = rfunc(self, maxage)
else:
# return cached value
self.log.debug("rfunc(%s): return cached value" % pname)
value = self.parameters[pname].value
setattr(self, pname, value) # important! trigger the setter
return value
if rfunc:
wrapped_rfunc.__doc__ = rfunc.__doc__
if getattr(rfunc, '__wrapped__', False) is False:
setattr(newtype, 'read_' + pname, wrapped_rfunc)
wrapped_rfunc.__wrapped__ = True
if not pobj.readonly:
wfunc = attrs.get('write_' + pname, None)
for base in bases:
if wfunc is not None:
break
wfunc = getattr(base, 'write_' + pname, None)
def wrapped_wfunc(self, value, pname=pname, wfunc=wfunc):
self.log.debug("wfunc(%s): set %r" % (pname, value))
pobj = self.parameters[pname]
value = pobj.datatype.validate(value)
if wfunc:
self.log.debug('calling %r(%r)' % (wfunc, value))
value = wfunc(self, value) or value
# XXX: use setattr or direct manipulation
# of self.parameters[pname]?
setattr(self, pname, value)
return value
if wfunc:
wrapped_wfunc.__doc__ = wfunc.__doc__
if getattr(wfunc, '__wrapped__', False) is False:
setattr(newtype, 'write_' + pname, wrapped_wfunc)
wrapped_wfunc.__wrapped__ = True
def getter(self, pname=pname):
return self.parameters[pname].value
def setter(self, value, pname=pname):
pobj = self.parameters[pname]
value = pobj.datatype.validate(value)
pobj.timestamp = time.time()
if (not EVENT_ONLY_ON_CHANGED_VALUES) or (value != pobj.value):
pobj.value = value
# also send notification
if self.parameters[pname].export:
self.log.debug('%s is now %r' % (pname, value))
self.DISPATCHER.announce_update(self, pname, pobj)
setattr(newtype, pname, property(getter, setter))
# also collect/update information about Command's
setattr(newtype, 'commands', getattr(newtype, 'commands', {}))
for attrname in attrs:
if attrname.startswith('do_'):
if attrname[3:] not in newtype.commands:
raise ProgrammingError('%r: command %r has to be specified '
'explicitly!' % (name, attrname[3:]))
attrs['__constructed__'] = True
return newtype

293
secop/modules.py
View File

@ -30,297 +30,12 @@ from __future__ import print_function
import time
try:
from six import add_metaclass # for py2/3 compat
except ImportError:
# copied from six v1.10.0
def add_metaclass(metaclass):
"""Class decorator for creating a class with a metaclass."""
def wrapper(cls):
orig_vars = cls.__dict__.copy()
slots = orig_vars.get('__slots__')
if slots is not None:
if isinstance(slots, str):
slots = [slots]
for slots_var in slots:
orig_vars.pop(slots_var)
orig_vars.pop('__dict__', None)
orig_vars.pop('__weakref__', None)
return metaclass(cls.__name__, cls.__bases__, orig_vars)
return wrapper
from secop.lib import formatExtendedStack, mkthread, unset_value
from secop.lib.enum import Enum
from secop.errors import ConfigError, ProgrammingError
from secop.datatypes import DataType, EnumType, TupleOf, StringType, FloatRange, get_datatype
EVENT_ONLY_ON_CHANGED_VALUES = False
class CountedObj(object):
ctr = [0]
def __init__(self):
cl = self.__class__.ctr
cl[0] += 1
self.ctr = cl[0]
class Parameter(CountedObj):
"""storage for Parameter settings + value + qualifiers
if readonly is False, the value can be changed (by code, or remote)
if no default is given, the parameter MUST be specified in the configfile
during startup, value is initialized with the default value or
from the config file if specified there
poll can be:
- False (never poll this parameter)
- True (poll this ever pollinterval)
- positive int (poll every N(th) pollinterval)
- negative int (normally poll every N(th) pollinterval, if module is busy, poll every pollinterval)
note: Drivable (and derived classes) poll with 10 fold frequency if module is busy....
"""
def __init__(self,
description,
datatype=None,
default=unset_value,
unit='',
readonly=True,
export=True,
group='',
poll=False,
value=unset_value,
timestamp=0,
optional=False,
ctr=None):
super(Parameter, self).__init__()
if not isinstance(datatype, DataType):
if issubclass(datatype, DataType):
# goodie: make an instance from a class (forgotten ()???)
datatype = datatype()
else:
raise ValueError(
'datatype MUST be derived from class DataType!')
self.description = description
self.datatype = datatype
self.default = default
self.unit = unit
self.readonly = readonly
self.export = export
self.group = group
self.optional = optional
# note: auto-converts True/False to 1/0 which yield the expected
# behaviour...
self.poll = int(poll)
# internal caching: value and timestamp of last change...
self.value = default
self.timestamp = 0
def __repr__(self):
return '%s_%d(%s)' % (self.__class__.__name__, self.ctr, ', '.join(
['%s=%r' % (k, v) for k, v in sorted(self.__dict__.items())]))
def copy(self):
# return a copy of ourselfs
return Parameter(**self.__dict__)
def for_export(self):
# used for serialisation only
res = dict(
description=self.description,
readonly=self.readonly,
datatype=self.datatype.export_datatype(),
)
if self.unit:
res['unit'] = self.unit
if self.group:
res['group'] = self.group
return res
def export_value(self):
return self.datatype.export_value(self.value)
class Override(CountedObj):
"""Stores the overrides to ba applied to a Parameter
note: overrides are applied by the metaclass during class creating
"""
def __init__(self, **kwds):
super(Override, self).__init__()
self.kwds = kwds
self.kwds['ctr'] = self.ctr
def apply(self, paramobj):
if isinstance(paramobj, Parameter):
for k, v in self.kwds.items():
if hasattr(paramobj, k):
setattr(paramobj, k, v)
return paramobj
else:
raise ProgrammingError(
"Can not apply Override(%s=%r) to %r: non-existing property!" %
(k, v, paramobj))
else:
raise ProgrammingError(
"Overrides can only be applied to Parameter's, %r is none!" %
paramobj)
class Command(CountedObj):
"""storage for Commands settings (description + call signature...)
"""
def __init__(self, description, arguments=None, result=None, optional=False):
super(Command, self).__init__()
# descriptive text for humans
self.description = description
# list of datatypes for arguments
self.arguments = arguments or []
# datatype for result
self.resulttype = result
# whether implementation is optional
self.optional = optional
def __repr__(self):
return '%s_%d(%s)' % (self.__class__.__name__, self.ctr, ', '.join(
['%s=%r' % (k, v) for k, v in sorted(self.__dict__.items())]))
def for_export(self):
# used for serialisation only
return dict(
description=self.description,
arguments=[arg.export_datatype() for arg in self.arguments],
resulttype=self.resulttype.export_datatype() if self.resulttype else None,
)
# Meta class
# warning: MAGIC!
class ModuleMeta(type):
"""Metaclass
joining the class's properties, parameters and commands dicts with
those of base classes.
also creates getters/setter for parameter access
and wraps read_*/write_* methods
(so the dispatcher will get notfied of changed values)
"""
def __new__(mcs, name, bases, attrs):
newtype = type.__new__(mcs, name, bases, attrs)
if '__constructed__' in attrs:
return newtype
# merge properties, Parameter and commands from all sub-classes
for entry in ['properties', 'parameters', 'commands']:
newentry = {}
for base in reversed(bases):
if hasattr(base, entry):
newentry.update(getattr(base, entry))
newentry.update(attrs.get(entry, {}))
setattr(newtype, entry, newentry)
# apply Overrides from all sub-classes
newparams = getattr(newtype, 'parameters')
for base in reversed(bases):
overrides = getattr(base, 'overrides', {})
for n, o in overrides.items():
newparams[n] = o.apply(newparams[n].copy())
for n, o in attrs.get('overrides', {}).items():
newparams[n] = o.apply(newparams[n].copy())
# Check naming of EnumType
for k, v in newparams.items():
if isinstance(v.datatype, EnumType) and not v.datatype._enum.name:
v.datatype._enum.name = k
# check validity of Parameter entries
for pname, pobj in newtype.parameters.items():
# XXX: allow dicts for overriding certain aspects only.
if not isinstance(pobj, Parameter):
raise ProgrammingError('%r: Parameters entry %r should be a '
'Parameter object!' % (name, pname))
# XXX: create getters for the units of params ??
# wrap of reading/writing funcs
rfunc = attrs.get('read_' + pname, None)
for base in bases:
if rfunc is not None:
break
rfunc = getattr(base, 'read_' + pname, None)
def wrapped_rfunc(self, maxage=0, pname=pname, rfunc=rfunc):
if rfunc:
self.log.debug("rfunc(%s): call %r" % (pname, rfunc))
value = rfunc(self, maxage)
else:
# return cached value
self.log.debug("rfunc(%s): return cached value" % pname)
value = self.parameters[pname].value
setattr(self, pname, value) # important! trigger the setter
return value
if rfunc:
wrapped_rfunc.__doc__ = rfunc.__doc__
if getattr(rfunc, '__wrapped__', False) is False:
setattr(newtype, 'read_' + pname, wrapped_rfunc)
wrapped_rfunc.__wrapped__ = True
if not pobj.readonly:
wfunc = attrs.get('write_' + pname, None)
for base in bases:
if wfunc is not None:
break
wfunc = getattr(base, 'write_' + pname, None)
def wrapped_wfunc(self, value, pname=pname, wfunc=wfunc):
self.log.debug("wfunc(%s): set %r" % (pname, value))
pobj = self.parameters[pname]
value = pobj.datatype.validate(value)
if wfunc:
self.log.debug('calling %r(%r)' % (wfunc, value))
value = wfunc(self, value) or value
# XXX: use setattr or direct manipulation
# of self.parameters[pname]?
setattr(self, pname, value)
return value
if wfunc:
wrapped_wfunc.__doc__ = wfunc.__doc__
if getattr(wfunc, '__wrapped__', False) is False:
setattr(newtype, 'write_' + pname, wrapped_wfunc)
wrapped_wfunc.__wrapped__ = True
def getter(self, pname=pname):
return self.parameters[pname].value
def setter(self, value, pname=pname):
pobj = self.parameters[pname]
value = pobj.datatype.validate(value)
pobj.timestamp = time.time()
if (not EVENT_ONLY_ON_CHANGED_VALUES) or (value != pobj.value):
pobj.value = value
# also send notification
if self.parameters[pname].export:
self.log.debug('%s is now %r' % (pname, value))
self.DISPATCHER.announce_update(self, pname, pobj)
setattr(newtype, pname, property(getter, setter))
# also collect/update information about Command's
setattr(newtype, 'commands', getattr(newtype, 'commands', {}))
for attrname in attrs:
if attrname.startswith('do_'):
if attrname[3:] not in newtype.commands:
raise ProgrammingError('%r: command %r has to be specified '
'explicitly!' % (name, attrname[3:]))
attrs['__constructed__'] = True
return newtype
from secop.errors import ConfigError
from secop.datatypes import EnumType, TupleOf, StringType, FloatRange, get_datatype
from secop.metaclass import add_metaclass, ModuleMeta
from secop.params import Command, Parameter, Override
@add_metaclass(ModuleMeta)

167
secop/params.py Normal file
View File

@ -0,0 +1,167 @@
# -*- coding: utf-8 -*-
# *****************************************************************************
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# Module authors:
# Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
#
# *****************************************************************************
"""Define classes for Parameters/Commands and Overriding them"""
from secop.lib import unset_value
from secop.errors import ProgrammingError
from secop.datatypes import DataType
EVENT_ONLY_ON_CHANGED_VALUES = False
class CountedObj(object):
ctr = [0]
def __init__(self):
cl = self.__class__.ctr
cl[0] += 1
self.ctr = cl[0]
class Parameter(CountedObj):
"""storage for Parameter settings + value + qualifiers
if readonly is False, the value can be changed (by code, or remote)
if no default is given, the parameter MUST be specified in the configfile
during startup, value is initialized with the default value or
from the config file if specified there
poll can be:
- False (never poll this parameter)
- True (poll this ever pollinterval)
- positive int (poll every N(th) pollinterval)
- negative int (normally poll every N(th) pollinterval, if module is busy, poll every pollinterval)
note: Drivable (and derived classes) poll with 10 fold frequency if module is busy....
"""
def __init__(self,
description,
datatype=None,
default=unset_value,
unit='',
readonly=True,
export=True,
group='',
poll=False,
value=unset_value,
timestamp=0,
optional=False,
ctr=None):
super(Parameter, self).__init__()
if not isinstance(datatype, DataType):
if issubclass(datatype, DataType):
# goodie: make an instance from a class (forgotten ()???)
datatype = datatype()
else:
raise ValueError(
'datatype MUST be derived from class DataType!')
self.description = description
self.datatype = datatype
self.default = default
self.unit = unit
self.readonly = readonly
self.export = export
self.group = group
self.optional = optional
# note: auto-converts True/False to 1/0 which yield the expected
# behaviour...
self.poll = int(poll)
# internal caching: value and timestamp of last change...
self.value = default
self.timestamp = 0
def __repr__(self):
return '%s_%d(%s)' % (self.__class__.__name__, self.ctr, ', '.join(
['%s=%r' % (k, v) for k, v in sorted(self.__dict__.items())]))
def copy(self):
# return a copy of ourselfs
return Parameter(**self.__dict__)
def for_export(self):
# used for serialisation only
res = dict(
description=self.description,
readonly=self.readonly,
datatype=self.datatype.export_datatype(),
)
if self.unit:
res['unit'] = self.unit
if self.group:
res['group'] = self.group
return res
def export_value(self):
return self.datatype.export_value(self.value)
class Override(CountedObj):
"""Stores the overrides to ba applied to a Parameter
note: overrides are applied by the metaclass during class creating
"""
def __init__(self, **kwds):
super(Override, self).__init__()
self.kwds = kwds
self.kwds['ctr'] = self.ctr
def apply(self, paramobj):
if isinstance(paramobj, Parameter):
for k, v in self.kwds.items():
if hasattr(paramobj, k):
setattr(paramobj, k, v)
return paramobj
else:
raise ProgrammingError(
"Can not apply Override(%s=%r) to %r: non-existing property!" %
(k, v, paramobj))
else:
raise ProgrammingError(
"Overrides can only be applied to Parameter's, %r is none!" %
paramobj)
class Command(CountedObj):
"""storage for Commands settings (description + call signature...)
"""
def __init__(self, description, arguments=None, result=None, optional=False):
super(Command, self).__init__()
# descriptive text for humans
self.description = description
# list of datatypes for arguments
self.arguments = arguments or []
# datatype for result
self.resulttype = result
# whether implementation is optional
self.optional = optional
def __repr__(self):
return '%s_%d(%s)' % (self.__class__.__name__, self.ctr, ', '.join(
['%s=%r' % (k, v) for k, v in sorted(self.__dict__.items())]))
def for_export(self):
# used for serialisation only
return dict(
description=self.description,
arguments=[arg.export_datatype() for arg in self.arguments],
resulttype=self.resulttype.export_datatype() if self.resulttype else None,
)