remove IOHandler stuff
as all code using IO handlers has been changed to use secop.rwhandler, IO handlers can be removed Change-Id: Id57fbc4ce2744dbe73bb8792fd45449373f76bb5 Reviewed-on: https://forge.frm2.tum.de/review/c/sine2020/secop/playground/+/28526 Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de> Reviewed-by: Enrico Faulhaber <enrico.faulhaber@frm2.tum.de> Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
parent
6a0261c728
commit
1357ead435
@ -28,7 +28,6 @@
|
||||
# pylint: disable=unused-import
|
||||
from secop.datatypes import ArrayOf, BLOBType, BoolType, EnumType, \
|
||||
FloatRange, IntRange, ScaledInteger, StringType, StructOf, TupleOf
|
||||
from secop.iohandler import IOHandler, IOHandlerBase
|
||||
from secop.lib.enum import Enum
|
||||
from secop.modules import Attached, Communicator, \
|
||||
Done, Drivable, Feature, Module, Readable, Writable, HasAccessibles
|
||||
|
||||
@ -1,332 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- 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:
|
||||
# Markus Zolliker <markus.zolliker@psi.ch>
|
||||
# *****************************************************************************
|
||||
"""IO handler
|
||||
|
||||
Utility class for cases, where multiple parameters are treated with a common command,
|
||||
or in cases, where IO can be parametrized.
|
||||
The support for LakeShore and similar protocols is already included.
|
||||
|
||||
For read, instead of the methods read_<parameter> we write one method analyze_<group>
|
||||
for all parameters with the same handler. Before analyze_<group> is called, the
|
||||
reply is parsed and converted to values, which are then given as arguments.
|
||||
|
||||
def analyze_<group>(self, value1, value2, ...):
|
||||
# here we have to calculate parameters from the values (value1, value2 ...)
|
||||
# and return a dict with parameter names as keys and new values.
|
||||
|
||||
It is an error to have a read_<parameter> method implemented on a parameter with a
|
||||
handler.
|
||||
|
||||
For write, instead of the methods write_<parameter>" we write one method change_<group>
|
||||
for all parameters with the same handler.
|
||||
|
||||
def change_<group>(self, change):
|
||||
# Change contains the to be changed parameters as attributes, and also the unchanged
|
||||
# parameters taking part to the handler group. If the method needs the current values
|
||||
# from the hardware, it can read them with change.getValues(). This call does also
|
||||
# update the values of the attributes of change, which are not subject to change.
|
||||
# In addtion, the method may call change.toBeChanged(<parameter name>) to determine,
|
||||
# whether a specific parameter is subject to change.
|
||||
# The return value must be either a sequence of values to be written to the hardware,
|
||||
# which will be formatted by the handler, or None. The latter is used only in some
|
||||
# special cases, when nothing has to be written.
|
||||
|
||||
A write_<parameter> method may be implemented in addition. In that case, the handlers write
|
||||
method has to be called explicitly int the write_<parameter> method, if needed.
|
||||
"""
|
||||
import re
|
||||
|
||||
from secop.errors import ProgrammingError
|
||||
from secop.modules import Done
|
||||
|
||||
|
||||
class CmdParser:
|
||||
"""helper for parsing replies
|
||||
|
||||
using a subset of old style python formatting.
|
||||
The same format can be used for formatting command arguments
|
||||
"""
|
||||
|
||||
# make a map of cast functions
|
||||
CAST_MAP = {letter: cast
|
||||
for letters, cast in (
|
||||
('d', int),
|
||||
('s', str), # 'c' is treated separately
|
||||
('o', lambda x: int(x, 8)),
|
||||
('xX', lambda x: int(x, 16)),
|
||||
('eEfFgG', float),
|
||||
) for letter in letters}
|
||||
# pattern for characters to be escaped
|
||||
ESC_PAT = re.compile(r'([\|\^\$\-\.\+\*\?\(\)\[\]\{\}\<\>])')
|
||||
# format pattern
|
||||
FMT_PAT = re.compile('(%%|%[^diouxXfFgGeEcrsa]*(?:.|$))')
|
||||
|
||||
def __init__(self, argformat):
|
||||
self.fmt = argformat
|
||||
spl = self.FMT_PAT.split(argformat)
|
||||
spl_iter = iter(spl)
|
||||
|
||||
def escaped(text):
|
||||
return self.ESC_PAT.sub(lambda x: '\\' + x.group(1), text)
|
||||
|
||||
casts = []
|
||||
# the first item in spl is just plain text
|
||||
pat = [escaped(next(spl_iter))]
|
||||
todofmt = None # format set aside to be treated in next loop
|
||||
|
||||
# loop over found formats and separators
|
||||
for fmt, sep in zip(spl_iter, spl_iter):
|
||||
if fmt == '%%':
|
||||
if todofmt is None:
|
||||
pat.append('%' + escaped(sep)) # plain text
|
||||
continue
|
||||
fmt = todofmt
|
||||
todofmt = None
|
||||
sep = '%' + sep
|
||||
elif todofmt:
|
||||
raise ValueError("a separator must follow '%s'" % todofmt)
|
||||
cast = self.CAST_MAP.get(fmt[-1], None)
|
||||
if cast is None: # special or unknown case
|
||||
if fmt != '%c':
|
||||
raise ValueError("unsupported format: '%s'" % fmt)
|
||||
# we do not need a separator after %c
|
||||
pat.append('(.)')
|
||||
casts.append(str)
|
||||
pat.append(escaped(sep))
|
||||
continue
|
||||
if sep == '': # missing separator. postpone handling for '%%' case or end of pattern
|
||||
todofmt = fmt
|
||||
continue
|
||||
casts.append(cast)
|
||||
# accepting everything up to a separator
|
||||
pat.append('([^%s]*)' % escaped(sep[0]) + escaped(sep))
|
||||
if todofmt:
|
||||
casts.append(cast)
|
||||
pat.append('(.*)')
|
||||
self.casts = casts
|
||||
self.pat = re.compile(''.join(pat))
|
||||
try:
|
||||
argformat % ((0,) * len(casts)) # validate argformat
|
||||
except ValueError as e:
|
||||
raise ValueError("%s in %r" % (e, argformat)) from None
|
||||
|
||||
def format(self, *values):
|
||||
return self.fmt % values
|
||||
|
||||
def parse(self, reply):
|
||||
match = self.pat.match(reply)
|
||||
if not match:
|
||||
raise ValueError('reply "%s" does not match pattern "%s"' % (reply, self.fmt))
|
||||
return [c(v) for c, v in zip(self.casts, match.groups())]
|
||||
|
||||
|
||||
class Change:
|
||||
"""contains new values for the call to change_<group>
|
||||
|
||||
A Change instance is used as an argument for the change_<group> method.
|
||||
Getting the value of change.<parameter> returns either the new, changed value or the
|
||||
current one from the module, if there is no new value.
|
||||
"""
|
||||
def __init__(self, handler, module, valuedict):
|
||||
self._handler = handler
|
||||
self._module = module
|
||||
self._valuedict = valuedict
|
||||
self._to_be_changed = set(self._valuedict)
|
||||
self._reply = None
|
||||
|
||||
def __getattr__(self, key):
|
||||
"""return attribute from module key is not in self._valuedict"""
|
||||
if key in self._valuedict:
|
||||
return self._valuedict[key]
|
||||
return getattr(self._module, key)
|
||||
|
||||
def doesInclude(self, *args):
|
||||
"""check whether one of the specified parameters is to be changed"""
|
||||
return bool(set(args) & self._to_be_changed)
|
||||
|
||||
def readValues(self):
|
||||
"""read values from the hardware
|
||||
|
||||
and update our parameter attributes accordingly (i.e. do not touch the new values)
|
||||
"""
|
||||
if self._reply is None:
|
||||
self._reply = self._handler.send_command(self._module)
|
||||
result = self._handler.analyze(self._module, *self._reply)
|
||||
result.update(self._valuedict)
|
||||
self._valuedict.update(result)
|
||||
return self._reply
|
||||
|
||||
|
||||
class IOHandlerBase:
|
||||
"""abstract IO handler
|
||||
|
||||
IO handlers for parametrized access should inherit from this
|
||||
"""
|
||||
|
||||
def get_read_func(self, modclass, pname):
|
||||
"""get the read function for parameter pname"""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_write_func(self, pname):
|
||||
"""get the write function for parameter pname"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class IOHandler(IOHandlerBase):
|
||||
"""IO handler for cases, where multiple parameters are treated with a common command
|
||||
|
||||
This IO handler works for a syntax, where the reply of a query command has
|
||||
the same format as the arguments for the change command.
|
||||
Examples: devices from LakeShore, PPMS
|
||||
|
||||
:param group: the handler group (used for analyze_<group> and change_<group>)
|
||||
:param querycmd: the command for a query, may contain named formats for cmdargs
|
||||
:param replyfmt: the format for reading the reply with some scanf like behaviour
|
||||
:param changecmd: the first part of the change command (without values), may be
|
||||
omitted if no write happens
|
||||
"""
|
||||
CMDARGS = [] #: list of properties or parameters to be used for building some of the the query and change commands
|
||||
CMDSEPARATOR = None #: if not None, it is possible to join a command and a query with the given separator
|
||||
|
||||
def __init__(self, group, querycmd, replyfmt, changecmd=None):
|
||||
"""initialize the IO handler
|
||||
|
||||
group: the handler group (used for analyze_<group> and change_<group>)
|
||||
querycmd: the command for a query, may contain named formats for cmdargs
|
||||
replyfmt: the format for reading the reply with some scanf like behaviour
|
||||
changecmd: the first part of the change command (without values), may be
|
||||
omitted if no write happens
|
||||
"""
|
||||
self.group = group
|
||||
self.parameters = set()
|
||||
self._module_class = None
|
||||
self.querycmd = querycmd
|
||||
self.replyfmt = CmdParser(replyfmt)
|
||||
self.changecmd = changecmd
|
||||
|
||||
def parse_reply(self, reply):
|
||||
"""return values from a raw reply"""
|
||||
return self.replyfmt.parse(reply)
|
||||
|
||||
def make_query(self, module):
|
||||
"""make a query"""
|
||||
return self.querycmd % {k: getattr(module, k, None) for k in self.CMDARGS}
|
||||
|
||||
def make_change(self, module, *values):
|
||||
"""make a change command"""
|
||||
changecmd = self.changecmd % {k: getattr(module, k, None) for k in self.CMDARGS}
|
||||
return changecmd + self.replyfmt.format(*values)
|
||||
|
||||
def send_command(self, module, changecmd=''):
|
||||
"""send a command (query or change+query) and parse the reply into a list
|
||||
|
||||
If changecmd is given, it is prepended before the query. changecmd must
|
||||
contain the command separator at the end.
|
||||
"""
|
||||
querycmd = self.make_query(module)
|
||||
reply = module.communicate(changecmd + querycmd)
|
||||
return self.parse_reply(reply)
|
||||
|
||||
def send_change(self, module, *values):
|
||||
"""compose and send a command from values
|
||||
|
||||
and send a query afterwards, or combine with a query command.
|
||||
Override this method, if the change command already includes a reply.
|
||||
"""
|
||||
changecmd = self.make_change(module, *values)
|
||||
if self.CMDSEPARATOR is None:
|
||||
module.communicate(changecmd) # ignore result
|
||||
return self.send_command(module)
|
||||
return self.send_command(module, changecmd + self.CMDSEPARATOR)
|
||||
|
||||
def get_read_func(self, modclass, pname):
|
||||
"""returns the read function passed to the metaclass
|
||||
|
||||
and registers the parameter in this handler
|
||||
"""
|
||||
self._module_class = self._module_class or modclass
|
||||
if self._module_class != modclass:
|
||||
raise ProgrammingError("the handler '%s' for '%s.%s' is already used in module '%s'"
|
||||
% (self.group, modclass.__name__, pname, self._module_class.__name__))
|
||||
# self.change might be needed even when get_write_func was not called
|
||||
self.change = getattr(self._module_class, 'change_' + self.group, None)
|
||||
self.parameters.add(pname)
|
||||
self.analyze = getattr(modclass, 'analyze_' + self.group)
|
||||
return self.read
|
||||
|
||||
def read(self, module):
|
||||
# read values from module
|
||||
assert module.__class__ == self._module_class
|
||||
try:
|
||||
# do a read of the current hw values
|
||||
reply = self.send_command(module)
|
||||
# convert them to parameters
|
||||
result = self.analyze(module, *reply)
|
||||
for pname, value in result.items():
|
||||
setattr(module, pname, value)
|
||||
for pname in self.parameters:
|
||||
if module.parameters[pname].readerror:
|
||||
# clear errors on parameters, which were not updated.
|
||||
# this will also inform all activated clients
|
||||
setattr(module, pname, getattr(module, pname))
|
||||
except Exception as e:
|
||||
# set all parameters of this handler to error
|
||||
for pname in self.parameters:
|
||||
module.announceUpdate(pname, None, e)
|
||||
raise
|
||||
return Done
|
||||
|
||||
def get_write_func(self, pname):
|
||||
"""returns the write function passed to the metaclass
|
||||
|
||||
:param pname: the parameter name
|
||||
|
||||
May be overriden to return None, if not used
|
||||
"""
|
||||
|
||||
def wfunc(module, value, hdl=self, pname=pname):
|
||||
hdl.write(module, pname, value)
|
||||
return Done
|
||||
|
||||
return wfunc
|
||||
|
||||
def write(self, module, pname, value):
|
||||
# write value to parameter pname of the module
|
||||
assert module.__class__ == self._module_class
|
||||
force_read = False
|
||||
valuedict = {pname: value}
|
||||
if module.writeDict: # collect other parameters to be written
|
||||
for p in self.parameters:
|
||||
if p in module.writeDict:
|
||||
valuedict[p] = module.writeDict.pop(p)
|
||||
elif p not in valuedict:
|
||||
force_read = True
|
||||
change = Change(self, module, valuedict)
|
||||
if force_read:
|
||||
change.readValues()
|
||||
values = self.change(module, change)
|
||||
if values is None: # this indicates that nothing has to be written
|
||||
return
|
||||
# send the change command and a query command
|
||||
reply = self.send_change(module, *values)
|
||||
result = self.analyze(module, *reply)
|
||||
for k, v in result.items():
|
||||
setattr(module, k, v)
|
||||
@ -117,21 +117,9 @@ class HasAccessibles(HasProperties):
|
||||
continue
|
||||
|
||||
rfunc = getattr(cls, 'read_' + pname, None)
|
||||
# TODO: remove handler stuff here
|
||||
rfunc_handler = pobj.handler.get_read_func(cls, pname) if pobj.handler else None
|
||||
wrapped = getattr(rfunc, 'wrapped', False) # meaning: wrapped or auto generated
|
||||
if rfunc_handler:
|
||||
if 'read_' + pname in cls.__dict__:
|
||||
if pname in cls.__dict__:
|
||||
raise ProgrammingError("parameter '%s' can not have a handler "
|
||||
"and read_%s" % (pname, pname))
|
||||
# read_<pname> overwrites inherited handler
|
||||
else:
|
||||
rfunc = rfunc_handler
|
||||
wrapped = False
|
||||
|
||||
# create wrapper except when read function is already wrapped
|
||||
if not wrapped:
|
||||
# create wrapper except when read function is already wrapped or auto generatoed
|
||||
if not getattr(rfunc, 'wrapped', False):
|
||||
|
||||
if rfunc:
|
||||
|
||||
@ -164,15 +152,8 @@ class HasAccessibles(HasProperties):
|
||||
|
||||
wfunc = getattr(cls, 'write_' + pname, None)
|
||||
if not pobj.readonly or wfunc: # allow write_ method even when pobj is not readonly
|
||||
wrapped = getattr(wfunc, 'wrapped', False) # meaning: wrapped or auto generated
|
||||
if (wfunc is None or wrapped) and pobj.handler:
|
||||
# ignore the handler, if a write function is present
|
||||
# TODO: remove handler stuff here
|
||||
wfunc = pobj.handler.get_write_func(pname)
|
||||
wrapped = False
|
||||
|
||||
# create wrapper except when write function is already wrapped
|
||||
if not wrapped:
|
||||
# create wrapper except when write function is already wrapped or auto generated
|
||||
if not getattr(wfunc, 'wrapped', False):
|
||||
|
||||
if wfunc:
|
||||
|
||||
|
||||
@ -141,9 +141,6 @@ class Parameter(Accessible):
|
||||
optional = Property(
|
||||
'[internal] is this parameter optional?', BoolType(),
|
||||
export=False, settable=False, default=False)
|
||||
handler = Property(
|
||||
'[internal] overload the standard read and write functions', ValueType(),
|
||||
export=False, default=None, settable=False)
|
||||
initwrite = Property(
|
||||
'''[internal] write this parameter on initialization
|
||||
|
||||
|
||||
@ -184,7 +184,7 @@ def proxy_class(remote_class, name=None):
|
||||
for aname, aobj in rcls.accessibles.items():
|
||||
if isinstance(aobj, Parameter):
|
||||
pobj = aobj.copy()
|
||||
pobj.merge(dict(handler=None, needscfg=False))
|
||||
pobj.merge(dict(needscfg=False))
|
||||
attrs[aname] = pobj
|
||||
|
||||
def rfunc(self, pname=aname):
|
||||
|
||||
@ -1,202 +0,0 @@
|
||||
# -*- 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:
|
||||
# Markus Zolliker <markus.zolliker@psi.ch>
|
||||
#
|
||||
# *****************************************************************************
|
||||
"""test commandhandler."""
|
||||
|
||||
import pytest
|
||||
|
||||
from secop.datatypes import FloatRange, IntRange, Property, StringType
|
||||
from secop.errors import ProgrammingError
|
||||
from secop.iohandler import CmdParser, IOHandler
|
||||
from secop.modules import Module, Parameter
|
||||
|
||||
|
||||
@pytest.mark.parametrize('fmt, text, values, text2', [
|
||||
('%d,%d', '2,3', [2,3], None),
|
||||
('%c%s', '222', ['2', '22'], None), # %c does not need a spearator
|
||||
('%s%%%d_%%', 'abc%12_%', ['abc', 12], None), # % as separator and within separator
|
||||
('%s.+', 'string,without+period.+', ['string,without+period',], None), # special characters
|
||||
('%d,%.3f,%x,%o', '1,1.2346,fF,17', [1, 1.2346, 255, 15], '1,1.235,ff,17'), # special formats
|
||||
])
|
||||
def test_CmdParser(fmt, text, values, text2):
|
||||
parser = CmdParser(fmt)
|
||||
print(parser.format(*values))
|
||||
assert parser.parse(text) == values
|
||||
if text2 is None:
|
||||
text2 = text
|
||||
assert parser.format(*values) == text2
|
||||
|
||||
def test_CmdParser_ex():
|
||||
with pytest.raises(ValueError):
|
||||
CmdParser('%d%s') # no separator
|
||||
|
||||
|
||||
class Data:
|
||||
"""a cyclic list where we put data to be checked or used during test"""
|
||||
def __init__(self):
|
||||
self.data = []
|
||||
|
||||
def push(self, tag, *args):
|
||||
if len(args) == 1:
|
||||
args = args[0]
|
||||
self.data.append((tag, args))
|
||||
|
||||
def pop(self, expected):
|
||||
tag, data = self.data.pop(0)
|
||||
print('pop(%s) %r' % (tag, data))
|
||||
if tag != expected:
|
||||
raise ValueError('expected tag %s, not %s' % (expected, tag))
|
||||
return data
|
||||
|
||||
def empty(self):
|
||||
return not self.data
|
||||
|
||||
|
||||
class DispatcherStub:
|
||||
# the first update from the poller comes a very short time after the
|
||||
# initial value from the timestamp. However, in the test below
|
||||
# the second update happens after the updates dict is cleared
|
||||
# -> we have to inhibit the 'omit unchanged update' feature
|
||||
omit_unchanged_within = 0
|
||||
|
||||
def __init__(self, updates):
|
||||
self.updates = updates
|
||||
|
||||
def announce_update(self, module, pname, pobj):
|
||||
if pobj.readerror:
|
||||
self.updates['error', pname] = str(pobj.readerror)
|
||||
else:
|
||||
self.updates[pname] = pobj.value
|
||||
|
||||
|
||||
class LoggerStub:
|
||||
def debug(self, *args):
|
||||
pass
|
||||
info = warning = exception = debug
|
||||
|
||||
|
||||
class ServerStub:
|
||||
def __init__(self, updates):
|
||||
self.dispatcher = DispatcherStub(updates)
|
||||
|
||||
|
||||
def test_IOHandler():
|
||||
class Hdl(IOHandler):
|
||||
CMDARGS = ['channel', 'loop']
|
||||
CMDSEPARATOR ='|'
|
||||
|
||||
def __init__(self, name, querycmd, replyfmt):
|
||||
changecmd = querycmd.replace('?', ' ')
|
||||
if not querycmd.endswith('?'):
|
||||
changecmd += ','
|
||||
super().__init__(name, querycmd, replyfmt, changecmd)
|
||||
|
||||
group1 = Hdl('group1', 'SIMPLE?', '%g')
|
||||
group2 = Hdl('group2', 'CMD?%(channel)d', '%g,%s,%d')
|
||||
|
||||
|
||||
class Module1(Module):
|
||||
channel = Property('the channel', IntRange(), default=3)
|
||||
loop = Property('the loop', IntRange(), default=2)
|
||||
simple = Parameter('a readonly', FloatRange(), default=0.77, handler=group1)
|
||||
real = Parameter('a float value', FloatRange(), default=12.3, handler=group2, readonly=False)
|
||||
text = Parameter('a string value', StringType(), default='x', handler=group2, readonly=False)
|
||||
|
||||
def communicate(self, command):
|
||||
assert data.pop('command') == command
|
||||
return data.pop('reply')
|
||||
|
||||
def analyze_group1(self, val):
|
||||
assert data.pop('val') == val
|
||||
return dict(simple=data.pop('simple'))
|
||||
|
||||
def analyze_group2(self, gval, sval, dval):
|
||||
assert data.pop('gsv') == (gval, sval, dval)
|
||||
real, text = data.pop('rt')
|
||||
return dict(real=real, text=text)
|
||||
|
||||
def change_group2(self, change):
|
||||
gval, sval, dval = change.readValues()
|
||||
assert data.pop('old') == (gval, sval, dval)
|
||||
assert data.pop('self') == (self.real, self.text)
|
||||
assert data.pop('new') == (change.real, change.text)
|
||||
return data.pop('changed')
|
||||
|
||||
data = Data()
|
||||
updates = {}
|
||||
module = Module1('mymodule', LoggerStub(), {'.description': ''}, ServerStub(updates))
|
||||
print(updates)
|
||||
updates.clear() # get rid of updates from initialisation
|
||||
|
||||
# for communicate
|
||||
data.push('command', 'SIMPLE?')
|
||||
data.push('reply', '4.51')
|
||||
# for analyze_group1
|
||||
data.push('val', 4.51)
|
||||
data.push('simple', 45.1)
|
||||
value = module.read_simple()
|
||||
assert value == 45.1
|
||||
assert module.simple == value
|
||||
assert data.empty()
|
||||
assert updates.pop('simple') == 45.1
|
||||
assert not updates
|
||||
|
||||
# for communicate
|
||||
data.push('command', 'CMD?3')
|
||||
data.push('reply', '1.23,text,5')
|
||||
# for analyze_group2
|
||||
data.push('gsv', 1.23, 'text', 5)
|
||||
data.push('rt', 12.3, 'string')
|
||||
value = module.read_real()
|
||||
assert module.real == value
|
||||
assert module.real == updates.pop('real') == 12.3
|
||||
assert module.text == updates.pop('text') == 'string'
|
||||
assert data.empty()
|
||||
assert not updates
|
||||
|
||||
# for communicate
|
||||
data.push('command', 'CMD?3')
|
||||
data.push('reply', '1.23,text,5')
|
||||
# for analyze_group2
|
||||
data.push('gsv', 1.23, 'text', 5)
|
||||
data.push('rt', 12.3, 'string')
|
||||
# for change_group2
|
||||
data.push('old', 1.23, 'text', 5)
|
||||
data.push('self', 12.3, 'string')
|
||||
data.push('new', 12.3, 'FOO')
|
||||
data.push('changed', 1.23, 'foo', 9)
|
||||
# for communicate
|
||||
data.push('command', 'CMD 3,1.23,foo,9|CMD?3')
|
||||
data.push('reply', '1.23,foo,9')
|
||||
# for analyze_group2
|
||||
data.push('gsv', 1.23, 'foo', 9)
|
||||
data.push('rt', 12.3, 'FOO')
|
||||
value = module.write_text('FOO')
|
||||
assert module.text == value
|
||||
assert module.text == updates.pop('text') == 'FOO'
|
||||
assert module.real == updates.pop('real') == 12.3
|
||||
assert data.empty()
|
||||
assert not updates
|
||||
|
||||
with pytest.raises(ProgrammingError): # can not use a handler for different modules
|
||||
# pylint: disable=unused-variable
|
||||
class Module2(Module):
|
||||
simple = Parameter('a readonly', FloatRange(), default=0.77, handler=group1)
|
||||
Loading…
x
Reference in New Issue
Block a user