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

allow super calls on read_/write_ methods

Browse Source 
instead of wrapping the access methods on the class directly,
create a wrapper class with the wrapped methods.

Change-Id: I93f3985bd06d6956b42a6690c087fb125e460ef9
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/30448
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
Reviewed-by: Alexander Zaft <a.zaft@fz-juelich.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker 2023-02-17 07:49:36 +01:00
parent ca4297e3a6
commit 4c577cf83d
4 changed files with 147 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
frappy/modules.py
View File

@ -26,7 +26,6 @@
import time
import threading
from collections import OrderedDict
from functools import wraps
from frappy.datatypes import ArrayOf, BoolType, EnumType, FloatRange, \
IntRange, StatusType, StringType, TextType, TupleOf, DiscouragedConversion
@ -43,6 +42,8 @@ Done = UniqueObject('Done')
indicating that the setter is triggered already"""
wrapperClasses = {}
class HasAccessibles(HasProperties):
"""base class of Module
@ -52,9 +53,14 @@ class HasAccessibles(HasProperties):
wrap read_*/write_* methods
(so the dispatcher will get notified of changed values)
"""
isWrapped = False
checkedMethods = set()
@classmethod
def __init_subclass__(cls): # pylint: disable=too-many-branches
super().__init_subclass__()
if cls.isWrapped:
return
# merge accessibles from all sub-classes, treat overrides
# for now, allow to use also the old syntax (parameters/commands dict)
accessibles = OrderedDict() # dict of accessibles
@ -102,26 +108,20 @@ class HasAccessibles(HasProperties):
# declarations within the same class
cls.accessibles = accessibles
# Correct naming of EnumTypes
# moved to Parameter.__set_name__
# check validity of Parameter entries
cls.wrappedAttributes = {'isWrapped': True}
# create wrappers for access methods
wrapped_name = '_' + cls.__name__
for pname, pobj in accessibles.items():
# XXX: create getters for the units of params ??
# wrap of reading/writing funcs
if not isinstance(pobj, Parameter):
# nothing to do for Commands
continue
rfunc = getattr(cls, 'read_' + pname, None)
# create wrapper except when read function is already wrapped or auto generatoed
if not getattr(rfunc, 'wrapped', False):
rname = 'read_' + pname
rfunc = getattr(cls, rname, None)
# create wrapper
if rfunc:
@wraps(rfunc) # handles __wrapped__ and __doc__
def new_rfunc(self, pname=pname, rfunc=rfunc):
with self.accessLock:
try:
@ -143,19 +143,17 @@ class HasAccessibles(HasProperties):
return getattr(self, pname)
new_rfunc.poll = False
new_rfunc.__doc__ = 'auto generated read method for ' + pname
new_rfunc.wrapped = True # indicate to subclasses that no more wrapping is needed
setattr(cls, 'read_' + pname, new_rfunc)
wfunc = getattr(cls, 'write_' + pname, None)
if not pobj.readonly or wfunc: # allow write_ method even when pobj is not readonly
# create wrapper except when write function is already wrapped or auto generated
if not getattr(wfunc, 'wrapped', False):
new_rfunc.__name__ = rname
new_rfunc.__qualname__ = wrapped_name + '.' + rname
new_rfunc.__module__ = cls.__module__
cls.wrappedAttributes[rname] = new_rfunc
wname = 'write_' + pname
wfunc = getattr(cls, wname, None)
if wfunc:
# allow write method even when parameter is readonly, but internally writable
@wraps(wfunc) # handles __wrapped__ and __doc__
def new_wfunc(self, value, pname=pname, wfunc=wfunc):
with self.accessLock:
pobj = self.accessibles[pname]
@ -170,16 +168,22 @@ class HasAccessibles(HasProperties):
return getattr(self, pname)
setattr(self, pname, new_value) # important! trigger the setter
return new_value
elif pobj.readonly:
new_wfunc = None
else:
def new_wfunc(self, value, pname=pname):
setattr(self, pname, value)
return value
new_wfunc.__doc__ = 'auto generated write method for ' + pname
if new_wfunc:
new_wfunc.__name__ = wname
new_wfunc.__qualname__ = wrapped_name + '.' + wname
new_wfunc.__module__ = cls.__module__
cls.wrappedAttributes[wname] = new_wfunc
new_wfunc.wrapped = True # indicate to subclasses that no more wrapping is needed
setattr(cls, 'write_' + pname, new_wfunc)
cls.checkedMethods.update(cls.wrappedAttributes)
# check for programming errors
for attrname in dir(cls):
@ -189,7 +193,7 @@ class HasAccessibles(HasProperties):
if prefix == 'do':
raise ProgrammingError('%r: old style command %r not supported anymore'
% (cls.__name__, attrname))
if prefix in ('read', 'write') and not getattr(getattr(cls, attrname), 'wrapped', False):
if prefix in ('read', 'write') and attrname not in cls.checkedMethods:
raise ProgrammingError('%s.%s defined, but %r is no parameter'
% (cls.__name__, attrname, pname))
@ -206,6 +210,13 @@ class HasAccessibles(HasProperties):
res[param][pn] = pv
cls.configurables = res
def __new__(cls, *args, **kwds):
wrapper_class = wrapperClasses.get(cls)
if wrapper_class is None:
wrapper_class = type('_' + cls.__name__, (cls,), cls.wrappedAttributes)
wrapperClasses[cls] = wrapper_class
return super().__new__(wrapper_class)
class Feature(HasAccessibles):
"""all things belonging to a small, predefined functionality influencing the working of a module

19
frappy/rwhandler.py
View File

@ -72,9 +72,8 @@ def wraps(func):
class Handler:
func = None
method_names = set() # this is shared among all instances of handlers!
wrapped = True # allow to use read_* or write_* as name of the decorated method
prefix = None # 'read_' or 'write_'
method_names = set() # global set registering used method names
poll = None
def __init__(self, keys):
@ -86,12 +85,13 @@ class Handler:
def __call__(self, func):
"""decorator call"""
self.func = func
if func.__qualname__ in self.method_names:
if (func.__module__, func.__qualname__) in self.method_names:
# make sure method name is not used twice
# (else __set_name__ will not be called)
raise ProgrammingError('duplicate method %r' % func.__qualname__)
self.func = func
func.wrapped = False
# __qualname__ used here (avoid conflicts between different modules)
self.method_names.add(func.__qualname__)
self.method_names.add((func.__module__, func.__qualname__))
return self
def __get__(self, obj, owner=None):
@ -102,8 +102,9 @@ class Handler:
def __set_name__(self, owner, name):
"""create the wrapped read_* or write_* methods"""
self.method_names.discard(self.func.__qualname__)
# at this point, this 'method_names' entry is no longer used -> delete
self.method_names.discard((self.func.__module__, self.func.__qualname__))
owner.checkedMethods.add(name)
for key in self.keys:
wrapped = self.wrap(key)
method_name = self.prefix + key
@ -112,7 +113,7 @@ class Handler:
# wrapped.poll is False when the nopoll decorator is applied either to self.func or to self
wrapped.poll = getattr(wrapped, 'poll', self.poll)
func = getattr(owner, method_name, None)
if func and not func.wrapped:
if func and method_name in owner.__dict__:
raise ProgrammingError('superfluous method %s.%s (overwritten by %s)'
% (owner.__name__, method_name, self.__class__.__name__))
setattr(owner, method_name, wrapped)

39
test/test_handler.py
View File

@ -84,10 +84,9 @@ def test_handler():
data.append(pname)
return value
assert Mod.read_a.poll is True
assert Mod.read_b.poll is True
m = Mod()
assert m.read_a.poll is True
assert m.read_b.poll is True
data.append(1.2)
assert m.read_a() == 1.2
@ -130,9 +129,10 @@ def test_common_handler():
self.b = values['b']
data.append('write_hdl')
assert set([Mod.read_a.poll, Mod.read_b.poll]) == {True, False}
m = Mod(a=1, b=2)
assert set([m.read_a.poll, m.read_b.poll]) == {True, False}
assert m.writeDict == {'a': 1, 'b': 2}
m.write_a(3)
assert m.a == 3
@ -172,8 +172,10 @@ def test_nopoll():
def read_hdl(self):
pass
assert Mod1.read_a.poll is True
assert Mod1.read_b.poll is True
m = Mod1()
print(m, m.read_a)
assert m.read_a.poll is True
assert m.read_b.poll is True
class Mod2(ModuleTest):
a = Parameter('', FloatRange(), readonly=False)
@ -183,8 +185,9 @@ def test_nopoll():
def read_hdl(self):
pass
assert Mod2.read_a.poll is True
assert Mod2.read_b.poll is False
m = Mod2()
assert m.read_a.poll is True
assert m.read_b.poll is False
class Mod3(ModuleTest):
a = Parameter('', FloatRange(), readonly=False)
@ -195,8 +198,9 @@ def test_nopoll():
def read_hdl(self):
pass
assert Mod3.read_a.poll is False
assert Mod3.read_b.poll is False
m = Mod3()
assert m.read_a.poll is False
assert m.read_b.poll is False
class Mod4(ModuleTest):
a = Parameter('', FloatRange(), readonly=False)
@ -207,8 +211,9 @@ def test_nopoll():
def read_hdl(self):
pass
assert Mod4.read_a.poll is False
assert Mod4.read_b.poll is False
m = Mod4()
assert m.read_a.poll is False
assert m.read_b.poll is False
class Mod5(ModuleTest):
a = Parameter('', FloatRange(), readonly=False)
@ -219,8 +224,9 @@ def test_nopoll():
def read_hdl(self):
pass
assert Mod5.read_a.poll is False
assert Mod5.read_b.poll is False
m = Mod5()
assert m.read_a.poll is False
assert m.read_b.poll is False
class Mod6(ModuleTest):
a = Parameter('', FloatRange(), readonly=False)
@ -231,5 +237,6 @@ def test_nopoll():
def read_hdl(self):
pass
assert Mod6.read_a.poll is False
assert Mod6.read_b.poll is False
m = Mod6()
assert m.read_a.poll is False
assert m.read_b.poll is False

40
test/test_modules.py
View File

@ -687,3 +687,43 @@ def test_deferred_main_unit(config, dynamicunit, finalunit, someunit):
assert m.parameters['ramp'].datatype.unit == finalunit + '/min'
# when someparam.unit is configured, this differs from finalunit
assert m.parameters['someparam'].datatype.unit == someunit
def test_super_call():
class Base(Readable):
def read_status(self):
return Readable.Status.IDLE, 'base'
class Mod(Base):
def read_status(self):
code, text = super().read_status()
return code, text + ' (extended)'
class DispatcherStub1:
def __init__(self, updates):
self.updates = updates
def announce_update(self, modulename, pname, pobj):
if pobj.readerror:
raise pobj.readerror
self.updates.append((modulename, pname, pobj.value))
class ServerStub1:
def __init__(self, updates):
self.dispatcher = DispatcherStub1(updates)
updates = []
srv = ServerStub1(updates)
b = Base('b', logger, {'description': ''}, srv)
b.read_status()
assert updates == [('b', 'status', ('IDLE', 'base'))]
updates.clear()
m = Mod('m', logger, {'description': ''}, srv)
m.read_status()
# in the version before change 'allow super calls on read_/write_ methods'
# updates would contain two items
assert updates == [('m', 'status', ('IDLE', 'base (extended)'))]
assert type(m).__name__ == '_Mod'
assert type(m).__mro__[1:5] == (Mod, Base, Readable, Module)