diff --git a/frappy/extparams.py b/frappy/extparams.py index 6f7b9dd4..f74e15de 100644 --- a/frappy/extparams.py +++ b/frappy/extparams.py @@ -25,12 +25,150 @@ special parameter classes with some automatic functionality import re from frappy.core import Parameter, Property -from frappy.datatypes import DataType, ValueType, EnumType, \ - StringType, FloatRange, DataTypeType +from frappy.datatypes import BoolType, DataType, DataTypeType, EnumType, \ + FloatRange, StringType, StructOf, ValueType from frappy.errors import ProgrammingError -# TODO: insert StructParam here +class StructParam(Parameter): + """convenience class to create a struct Parameter together with individual params + + Usage: + + class Controller(Drivable): + + ... + + ctrlpars = StructParam('ctrlpars struct', [ + ('pid_p', 'p', Parameter('control parameter p', FloatRange())), + ('pid_i', 'i', Parameter('control parameter i', FloatRange())), + ('pid_d', 'd', Parameter('control parameter d', FloatRange())), + ], readonly=False) + + ... + + then implement either read_ctrlpars and write_ctrlpars or + read_pid_p, read_pid_i, read_pid_d, write_pid_p, write_pid_i and write_pid_d + + the methods not implemented will be created automatically + """ + + # use properties, as simple attributes are not considered on copy() + paramdict = Property('dict of Parameter(...)', ValueType()) + hasStructRW = Property('has a read_ or write_ method', + BoolType(), default=False) + + insideRW = 0 # counter for avoiding multiple superfluous updates + + def __init__(self, description=None, paramdict=None, prefix_or_map='', *, datatype=None, readonly=False, **kwds): + """create a struct parameter together with individual parameters + + in addition to normal Parameter arguments: + + :param paramdict: dict of Parameter(...) + :param prefix_or_map: either a prefix for the parameter name to add to the member name + or a dict or + """ + if isinstance(paramdict, DataType): + raise ProgrammingError('second argument must be a dict of Param') + if datatype is None and paramdict is not None: # omit the following on Parameter.copy() + if isinstance(prefix_or_map, str): + prefix_or_map = {m: prefix_or_map + m for m in paramdict} + for membername, param in paramdict.items(): + param.name = prefix_or_map[membername] + datatype = StructOf(**{m: p.datatype for m, p in paramdict.items()}) + kwds['influences'] = [p.name for p in paramdict.values()] + self.updateEnable = {} + super().__init__(description, datatype, paramdict=paramdict, readonly=readonly, **kwds) + + def __set_name__(self, owner, name): + # names of access methods of structed param (e.g. ctrlpars) + struct_read_name = f'read_{name}' # e.g. 'read_ctrlpars' + struct_write_name = f'write_{name}' # e.h. 'write_ctrlpars' + self.hasStructRW = hasattr(owner, struct_read_name) or hasattr(owner, struct_write_name) + + for membername, param in self.paramdict.items(): + pname = param.name + changes = { + 'readonly': self.readonly, + 'influences': set(param.influences) | {name}, + } + param.ownProperties.update(changes) + param.init(changes) + setattr(owner, pname, param) + param.__set_name__(owner, param.name) + + if self.hasStructRW: + rname = f'read_{pname}' + + if not hasattr(owner, rname): + def rfunc(self, membername=membername, struct_read_name=struct_read_name): + return getattr(self, struct_read_name)()[membername] + + rfunc.poll = False # read_ is polled only + setattr(owner, rname, rfunc) + + if not self.readonly: + wname = f'write_{pname}' + if not hasattr(owner, wname): + def wfunc(self, value, membername=membername, + name=name, rname=rname, struct_write_name=struct_write_name): + valuedict = dict(getattr(self, name)) + valuedict[membername] = value + getattr(self, struct_write_name)(valuedict) + return getattr(self, rname)() + + setattr(owner, wname, wfunc) + + if not self.hasStructRW: + if not hasattr(owner, struct_read_name): + def struct_read_func(self, name=name, flist=tuple( + (m, f'read_{p.name}') for m, p in self.paramdict.items())): + pobj = self.parameters[name] + # disable updates generated from the callbacks of individual params + pobj.insideRW += 1 # guarded by self.accessLock + try: + return {m: getattr(self, f)() for m, f in flist} + finally: + pobj.insideRW -= 1 + + setattr(owner, struct_read_name, struct_read_func) + + if not (self.readonly or hasattr(owner, struct_write_name)): + + def struct_write_func(self, value, name=name, funclist=tuple( + (m, f'write_{p.name}') for m, p in self.paramdict.items())): + pobj = self.parameters[name] + pobj.insideRW += 1 # guarded by self.accessLock + try: + return {m: getattr(self, f)(value[m]) for m, f in funclist} + finally: + pobj.insideRW -= 1 + + setattr(owner, struct_write_name, struct_write_func) + + super().__set_name__(owner, name) + + def finish(self, modobj=None): + """register callbacks for consistency""" + super().finish(modobj) + if modobj: + + if self.hasStructRW: + def cb(value, modobj=modobj, structparam=self): + for membername, param in structparam.paramdict.items(): + setattr(modobj, param.name, value[membername]) + + modobj.valueCallbacks[self.name].append(cb) + else: + for membername, param in self.paramdict.items(): + def cb(value, modobj=modobj, structparam=self, membername=membername): + if not structparam.insideRW: + prev = dict(getattr(modobj, structparam.name)) + prev[membername] = value + setattr(modobj, structparam.name, prev) + + modobj.valueCallbacks[param.name].append(cb) class FloatEnumParam(Parameter): diff --git a/frappy/structparam.py b/frappy/structparam.py deleted file mode 100644 index 2534dda9..00000000 --- a/frappy/structparam.py +++ /dev/null @@ -1,164 +0,0 @@ -# ***************************************************************************** -# -# 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: -# Markus Zolliker -# -# ***************************************************************************** -"""convenience class to create a struct Parameter together with indivdual params - -Usage: - - class Controller(Drivable): - - ... - - ctrlpars = StructParam('ctrlpars struct', [ - ('pid_p', 'p', Parameter('control parameter p', FloatRange())), - ('pid_i', 'i', Parameter('control parameter i', FloatRange())), - ('pid_d', 'd', Parameter('control parameter d', FloatRange())), - ], readonly=False) - - ... - - then implement either read_ctrlpars and write_ctrlpars or - read_pid_p, read_pid_i, read_pid_d, write_pid_p, write_pid_i and write_pid_d - - the methods not implemented will be created automatically -""" - -from frappy.core import Parameter, Property -from frappy.datatypes import BoolType, DataType, StructOf, ValueType -from frappy.errors import ProgrammingError - - -class StructParam(Parameter): - """create a struct parameter together with individual parameters - - in addition to normal Parameter arguments: - - :param paramdict: dict of Parameter(...) - :param prefix_or_map: either a prefix for the parameter name to add to the member name - or a dict or - """ - # use properties, as simple attributes are not considered on copy() - paramdict = Property('dict of Parameter(...)', ValueType()) - hasStructRW = Property('has a read_ or write_ method', - BoolType(), default=False) - - insideRW = 0 # counter for avoiding multiple superfluous updates - - def __init__(self, description=None, paramdict=None, prefix_or_map='', *, datatype=None, readonly=False, **kwds): - if isinstance(paramdict, DataType): - raise ProgrammingError('second argument must be a dict of Param') - if datatype is None and paramdict is not None: # omit the following on Parameter.copy() - if isinstance(prefix_or_map, str): - prefix_or_map = {m: prefix_or_map + m for m in paramdict} - for membername, param in paramdict.items(): - param.name = prefix_or_map[membername] - datatype = StructOf(**{m: p.datatype for m, p in paramdict.items()}) - kwds['influences'] = [p.name for p in paramdict.values()] - self.updateEnable = {} - super().__init__(description, datatype, paramdict=paramdict, readonly=readonly, **kwds) - - def __set_name__(self, owner, name): - # names of access methods of structed param (e.g. ctrlpars) - struct_read_name = f'read_{name}' # e.g. 'read_ctrlpars' - struct_write_name = f'write_{name}' # e.h. 'write_ctrlpars' - self.hasStructRW = hasattr(owner, struct_read_name) or hasattr(owner, struct_write_name) - - for membername, param in self.paramdict.items(): - pname = param.name - changes = { - 'readonly': self.readonly, - 'influences': set(param.influences) | {name}, - } - param.ownProperties.update(changes) - param.init(changes) - setattr(owner, pname, param) - param.__set_name__(owner, param.name) - - if self.hasStructRW: - rname = f'read_{pname}' - - if not hasattr(owner, rname): - def rfunc(self, membername=membername, struct_read_name=struct_read_name): - return getattr(self, struct_read_name)()[membername] - - rfunc.poll = False # read_ is polled only - setattr(owner, rname, rfunc) - - if not self.readonly: - wname = f'write_{pname}' - if not hasattr(owner, wname): - def wfunc(self, value, membername=membername, - name=name, rname=rname, struct_write_name=struct_write_name): - valuedict = dict(getattr(self, name)) - valuedict[membername] = value - getattr(self, struct_write_name)(valuedict) - return getattr(self, rname)() - - setattr(owner, wname, wfunc) - - if not self.hasStructRW: - if not hasattr(owner, struct_read_name): - def struct_read_func(self, name=name, flist=tuple( - (m, f'read_{p.name}') for m, p in self.paramdict.items())): - pobj = self.parameters[name] - # disable updates generated from the callbacks of individual params - pobj.insideRW += 1 # guarded by self.accessLock - try: - return {m: getattr(self, f)() for m, f in flist} - finally: - pobj.insideRW -= 1 - - setattr(owner, struct_read_name, struct_read_func) - - if not (self.readonly or hasattr(owner, struct_write_name)): - - def struct_write_func(self, value, name=name, funclist=tuple( - (m, f'write_{p.name}') for m, p in self.paramdict.items())): - pobj = self.parameters[name] - pobj.insideRW += 1 # guarded by self.accessLock - try: - return {m: getattr(self, f)(value[m]) for m, f in funclist} - finally: - pobj.insideRW -= 1 - - setattr(owner, struct_write_name, struct_write_func) - - super().__set_name__(owner, name) - - def finish(self, modobj=None): - """register callbacks for consistency""" - super().finish(modobj) - if modobj: - - if self.hasStructRW: - def cb(value, modobj=modobj, structparam=self): - for membername, param in structparam.paramdict.items(): - setattr(modobj, param.name, value[membername]) - - modobj.valueCallbacks[self.name].append(cb) - else: - for membername, param in self.paramdict.items(): - def cb(value, modobj=modobj, structparam=self, membername=membername): - if not structparam.insideRW: - prev = dict(getattr(modobj, structparam.name)) - prev[membername] = value - setattr(modobj, structparam.name, prev) - - modobj.valueCallbacks[param.name].append(cb) diff --git a/test/test_extparams.py b/test/test_extparams.py index 0ca4648b..ee16627a 100644 --- a/test/test_extparams.py +++ b/test/test_extparams.py @@ -24,8 +24,7 @@ from test.test_modules import LoggerStub, ServerStub import pytest from frappy.core import FloatRange, Module, Parameter -from frappy.structparam import StructParam -from frappy.extparams import FloatEnumParam +from frappy.extparams import StructParam, FloatEnumParam from frappy.errors import ProgrammingError