Status and autorange implemented for AC resistane bridge

Change-Id: I8c94660c3b76cc78886e9e074b4ce8114fbb7f9e
This commit is contained in:
Oksana Shliakhtun 2024-06-04 16:46:52 +02:00 committed by Markus Zolliker
parent 7c96b83aff
commit d38b672a9a
2 changed files with 226 additions and 56 deletions

View File

@ -6,28 +6,62 @@ Node('bridge.psi.ch',
Mod('io',
'frappy_psi.bridge.BridgeIO',
'communication to sim900',
uri='serial:///dev/cu.usbserial-14440',
uri='serial:///dev/cu.usbserial-14240',
)
Mod('Resistance_1',
'frappy_psi.bridge.SIM921',
Mod('res1',
'frappy_psi.bridge.Resistance',
'module communication',
io='io',
port=1,
# channel='A'
)
Mod('Resistance_2',
'frappy_psi.bridge.SIM921',
Mod('res2',
'frappy_psi.bridge.Resistance',
'module communication',
io='io',
port=3,
)
Mod('Resistance_3',
'frappy_psi.bridge.SIM921',
Mod('res3',
'frappy_psi.bridge.Resistance',
'module communication',
io='io',
port=5,
)
Mod('phase1',
'frappy_psi.bridge.Phase',
'module communication',
resistance='res1',
)
Mod('phase2',
'frappy_psi.bridge.Phase',
'module communication',
resistance='res2',
)
Mod('phase3',
'frappy_psi.bridge.Phase',
'module communication',
resistance='res3',
)
Mod('dev1',
'frappy_psi.bridge.Deviation',
'module communication',
resistance='res1',
)
Mod('dev2',
'frappy_psi.bridge.Deviation',
'module communication',
resistance='res1',
)
Mod('dev3',
'frappy_psi.bridge.Deviation',
'module communication',
resistance='res3',
)

View File

@ -16,91 +16,227 @@
# Module authors: Oksana Shliakhtun <oksana.shliakhtun@psi.ch>
# *****************************************************************************
import time
import re
from frappy.core import StringIO, HasIO, Readable, Drivable, \
from frappy.core import StringIO, HasIO, Readable, \
Parameter, FloatRange, IntRange, EnumType, \
Enum, Property, StringType
# from SR830 import string_to_value
Property, Attached, IDLE, ERROR, WARN
def string_to_value(value):
value_with_unit = re.compile(r'(\d+)([pnumkMG]?)')
value, pfx = value_with_unit.match(value).groups()
pfx_dict = {'p': 1e-12, 'n': 1e-9, 'u': 1e-6, 'm': 1e-3, 'k': 1e3, 'M': 1e6, 'G': 1e9}
if pfx in pfx_dict:
value = round(float(value) * pfx_dict[pfx], 12)
return float(value)
def find_idx(list_of_values, target):
target = float(target)
cl_idx = None
cl_value = float('inf')
for idx, value in enumerate(list_of_values):
if value >= target:
diff = value - target
if diff < cl_value:
cl_value = value
cl_idx = idx
return cl_idx, cl_value
class BridgeIO(StringIO):
end_of_line = ('\r\n', '') # read terminator / rmpty write terminator
identification = [('*IDN?\r\n', r'Stanford_Research_Systems,.*'),
('RPER 510\r\nRPER?\r\n', '510')]
end_of_line = '\n'
identification = [('_\n*IDN?', r'Stanford_Research_Systems,.*')]
class Base(HasIO):
port = Property('modules port', IntRange(0, 15))
def command(self, command, replylines=1):
with self._lock:
reply = super().communicate(f'sndt {self.port:x}, "{command}"\r\n')
for _ in range(replylines-1):
super().communicate('')
head, tail = reply.split('#', 1)
assert head[:5] == f'MSG {self.port:x},'
return tail[tail[1:int(tail[0])+1]:]
def communicate(self, command):
return self.io.communicate(f'_\nconn {self.port:x},"_\n"\n{command}')
def query(self, command):
return float(self.communicate(command))
class SIM921(Base, Drivable):
# channel = Property('sim921 module', StringType())
setpoint = Parameter('temperature deviation', datatype=FloatRange, unit='K')
class Resistance(Base, Readable):
value = Parameter('resistance', datatype=FloatRange, unit='ohm')
dev = Parameter('resistance deviation', FloatRange())
output_offset = Parameter('temperature deviation', datatype=FloatRange, unit='Ohm')
phase_hold = Parameter('phase hold', EnumType('phase hold', off=0, on=1))
RES_RANGE = ['20mOhm', '200mOhm', '2Ohm', '20Ohm', '200Ohm', '2kOhm', '20kOhm', '200kOhm',
'2MOhm', '20MOhm']
irange = Parameter('resistance range index', EnumType('resistance range index',
{name: idx for idx, name in enumerate(RES_RANGE)}), readonly=False)
range = Parameter('resistance range value', FloatRange(2e-2, 2e7), unit='Om', readonly=False)
TIME_CONST = ['0.3s', '1s', '3s', '10s', '30s', '100s', '300s']
EXCT_RANGE = ['0', '3uV', '10uV', '30uV', '100uV', '300uV', '1mV', '3mV', '10mV', '30mV']
irange = Parameter('range index', EnumType('resistance range index',
{name: idx for idx, name in enumerate(RES_RANGE)}), readonly=False)
phase = Parameter('phase', FloatRange, unit='ang')
tc = Parameter('time constant value', FloatRange(1e-1, 3e2), unit='s', readonly=False)
itc = Parameter('time constant index',
EnumType('time const. index range',
{name: value for value, name in enumerate(TIME_CONST)}), readonly=False)
tc = Parameter('time constant value', FloatRange(1e-1, 3e2), unit='s', readonly=False)
EXCT_RANGE = ['0', '3uV', '10uV', '30uV', '100uV', '300uV', '1mV', '3mV', '10mV', '30mV']
iexct = Parameter('excitation index', EnumType('excitation index range',
{name: idx for idx, name in enumerate(EXCT_RANGE, start=-1)}), readonly=False)
exct = Parameter('excitation value', FloatRange(0, 3e-2), unit='s', default=300, readonly=False)
autorange = Parameter('autorange_on', EnumType('autorange', off=0, on=1),
readonly=False, default=0)
iexct = Parameter('excitation index', IntRange)
RES_RANGE_values = [string_to_value(value) for value in RES_RANGE]
TIME_CONST_values = [string_to_value(value) for value in TIME_CONST]
EXCT_RANGE_values = [string_to_value(value) for value in EXCT_RANGE]
ioClass = BridgeIO
def doPoll(self):
super().doPoll()
max_res = abs(self.value)
if self.autorange == 1:
if max_res >= 0.9 * self.range and self.irange < 9:
self.write_irange(self.irange + 1)
elif max_res <= 0.3 * self.range and self.irange > 0:
self.write_irange(self.irange - 1)
def read_status(self):
esr = int(self.communicate('*esr?')) # standart event status byte
ovsr = int(self.communicate('ovsr?')) # overload status
cesr = int(self.communicate('cesr?')) # communication error status
if esr & (1 << 1):
return ERROR, 'input error, cleared'
if esr & (1 << 2):
return ERROR, 'query error'
if esr & (1 << 4):
return ERROR, 'execution error'
if esr & (1 << 5):
return ERROR, 'command error'
if cesr & (1 << 0):
return ERROR, 'parity error'
if cesr & (1 << 2):
return ERROR, 'noise error'
if cesr & (1 << 4):
return ERROR, 'input overflow, cleared'
if cesr & (1 << 3):
return ERROR, 'hardware overflow'
if ovsr & (1 << 0):
return ERROR, 'output overload'
if cesr & (1 << 7):
return WARN, 'device clear'
if ovsr & (1 << 2):
return WARN, 'current saturation'
if ovsr & (1 << 3):
return WARN, 'under servo'
if ovsr & (1 << 4):
return WARN, 'over servo'
return IDLE, ''
def read_value(self):
reply = self.command('rval?', 2)
return reply
return self.query('rval?')
def read_phase(self):
return self.command('phase?', 2)
def read_irange(self):
return self.query('rang?')
def read_setpoint(self):
return self.command('rset?')
def write_irange(self, idx):
value = int(idx)
self.query(f'rang {value}; rang?')
self.read_range()
return value
# def write_setpoint(self, setpoint):
# return self.command('rset', setpoint)
def read_range(self):
idx = self.read_irange()
name = self.RES_RANGE[idx]
return string_to_value(name)
def write_range(self, target):
cl_idx, cl_value = find_idx(self.RES_RANGE_values, target)
self.query(f'rang {cl_idx}; rang?')
return cl_value
def read_output_offset(self):
return self.query('rset?')
def write_output_offset(self, output_offset):
self.query(f'rset {output_offset};rset?')
def read_itc(self):
return self.query('tcon?')
def write_itc(self, itc):
self.read_itc()
value = int(itc)
return self.query(f'tcon {value}; tcon?')
def read_tc(self):
return self.command('tcon?')
idx = self.read_itc()
name = self.TIME_CONST[idx]
return string_to_value(name)
# def write_tc(self, tc):
# return self.command('tcon', tc)
def read_dev(self):
return self.command('rdev?')
def write_tc(self, target):
cl_idx, cl_value = find_idx(self.TIME_CONST_values, target)
self.query(f'tcon {cl_idx};tcon?')
return cl_value
def read_autorange(self):
return self.command('agai?')
return self.autorange
# def write_autorange(self, autorange):
# return self.command('agai', autorange)
def write_autorange(self, value):
self.query(f'agai {value:d};agai?')
return value
def read_iexct(self):
return self.command('exci?')
return int(self.query('exci?'))
# def write_iexct(self, iexct):
# return self.command('exci', iexct)
def write_iexct(self, iexct):
value = int(iexct)
return self.query(f'exci {value};exci?')
def write_exct(self, target):
target = float(target)
cl_idx = None
cl_value = float('inf')
min_diff = float('inf')
for idx, value in enumerate(self.EXCT_RANGE_values):
diff = abs(value - target)
if diff < min_diff:
min_diff = diff
cl_value = value
cl_idx = idx
self.write_iexct(cl_idx)
return cl_value
def read_exct(self):
idx = int(self.read_iexct())
name = self.EXCT_RANGE[idx + 1]
return string_to_value(name)
def read_phase_hold(self):
return int(self.communicate('phld?'))
def write_phase_hold(self, phase_hold):
self.communicate(f'phld {phase_hold}')
return self.read_phase_hold()
class Phase(Readable):
resistance = Attached()
value = Parameter('phase', FloatRange, default=0, unit='deg')
def read_value(self):
return self.resistance.query('phas?')
class Deviation(Readable):
resistance = Attached()
value = Parameter('resistance deviation', FloatRange(), unit='Ohm')
def read_value(self):
return self.resistance.query('rdev?')