fix new statemachine with ips magfield

+ add on_error, on_restart etc. to states.py
+ add n_retry argument to mercury change/multichange
This commit is contained in:
2022-12-19 15:12:00 +01:00
parent 397ca350f8
commit b7a1f17e5e
4 changed files with 119 additions and 67 deletions

View File

@@ -123,13 +123,14 @@ class MercuryChannel(HasIO):
else:
raise HardwareError(msg) from None
def multichange(self, adr, values, convert=as_float, tolerance=0):
def multichange(self, adr, values, convert=as_float, tolerance=0, n_retry=3):
"""set parameter(s) in mercury syntax
:param adr: as in see multiquery method
:param values: [(name1, value1), (name2, value2) ...]
:param convert: a converter function (converts given value to string and replied string to value)
:param tolerance: tolerance for readback check
:param n_retry: number of retries or 0 for no readback check
:return: the values as tuple
Example:
@@ -143,7 +144,7 @@ class MercuryChannel(HasIO):
adr = self._complete_adr(adr)
params = ['%s:%s' % (k, convert(v)) for k, v in values]
cmd = 'SET:%s:%s' % (adr, ':'.join(params))
for _ in range(3): # try 3 times or until readback result matches
for _ in range(max(1, n_retry)): # try n_retry times or until readback result matches
t = time.time()
reply = self.communicate(cmd)
head = 'STAT:SET:%s:' % adr
@@ -158,6 +159,8 @@ class MercuryChannel(HasIO):
except (AssertionError, AttributeError, ValueError) as e:
time.sleep(0.1) # in case of missed replies this might help to skip garbage
raise HardwareError('invalid reply %r to cmd %r' % (reply, cmd)) from e
if n_retry == 0:
return [v[1] for v in values] # no readback check
keys = [v[0] for v in values]
debug = []
readback = self.multiquery(adr, keys, convert, debug)
@@ -182,9 +185,9 @@ class MercuryChannel(HasIO):
adr, _, name = adr.rpartition(':')
return self.multiquery(adr, [name], convert)[0]
def change(self, adr, value, convert=as_float, tolerance=0):
def change(self, adr, value, convert=as_float, tolerance=0, n_retry=3):
adr, _, name = adr.rpartition(':')
return self.multichange(adr, [(name, value)], convert, tolerance)[0]
return self.multichange(adr, [(name, value)], convert, tolerance, n_retry)[0]
class TemperatureSensor(MercuryChannel, Readable):