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

state as of 05.08.26

Browse Source
This commit is contained in:
stark_a
2026-08-05 17:23:48 +02:00
parent 7be29bc4d1
commit e91b61dc0b
6 changed files with 156 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cfg/lakeshore370_test_cfg.py
+51 -54
View File
@@ -3,12 +3,14 @@ Node('ls370test.psi.ch',
interface='tcp://5000',
)
IO('io_treg', 'dil2-ts:3001')
IO('io_tmon', 'dil2-ts:3003')
# IO('io_treg', 'dil2-ts:3001')
# IO('io_tmon', 'dil2-ts:3003')
### temperature monitoring lakeshore (tmon) ###
IO('io_tmon', 'linse-6837-ts:3001')
Mod('tmon',
'frappy_psi.lakeshore370.Device',
'monitoring lakeshore 370',
@@ -20,122 +22,117 @@ Mod('switcher',
'frappy_psi.lakeshore370.Switcher',
'',
io = 'io_tmon',
device = 'tmon',
)
Mod('stillt',
Mod('ch10',
'frappy_psi.lakeshore370.Sensor',
'tmon stillt',
io = 'io_tmon',
device = 'tmon',
channel = 10,
switcher = 'switcher',
calcurve = 'ruoxm0',
enabled = True,
)
Mod('sorb',
Mod('ch12',
'frappy_psi.lakeshore370.Sensor',
'tmon sorb',
io = 'io_tmon',
device = 'tmon',
channel = 12,
switcher = 'switcher',
calcurve = 'c270',
enabled = True,
)
Mod('sample_mon',
Mod('ch3',
'frappy_psi.lakeshore370.Sensor',
'tmon sample',
io = 'io_tmon',
device = 'tmon',
channel = 3,
switcher = 'switcher',
calcurve = 'rx078',
enabled = False,
enabled = True,
)
Mod('mix',
Mod('ch4',
'frappy_psi.lakeshore370.Sensor',
'tmon mix',
io = 'io_tmon',
device = 'tmon',
channel = 4,
switcher = 'switcher',
calcurve = 'ruoxm0',
enabled = True,
)
Mod('samplehtr',
Mod('ch1',
'frappy_psi.lakeshore370.Sensor',
'tmon samplehtr',
io = 'io_tmon',
device = 'tmon',
channel = 1,
switcher = 'switcher',
calcurve = 'rx078',
enabled = False, # disable when used as heater
enabled = True, # disable when used as heater
)
Mod('onek',
Mod('ch11',
'frappy_psi.lakeshore370.RawSensor',
'tmon onek',
io = 'io_tmon',
device = 'tmon',
channel = 11,
switcher = 'switcher',
# calcurve = 'ruoxm0',
enabled = True,
)
### temperature regulating lakeshore (treg) ###
# ### temperature regulating lakeshore (treg) ###
Mod('treg',
'frappy_psi.lakeshore370.Device',
'regulating lakeshore 370',
io = 'io_treg',
curve_handling = True,
)
# IO('io_treg', 'dil3-ts:3001')
Mod('switcher_treg',
'frappy_psi.lakeshore370.Switcher',
'',
io = 'io_treg',
)
# Mod('treg',
# 'frappy_psi.lakeshore370.Device',
# 'regulating lakeshore 370',
# io = 'io_treg',
# curve_handling = True,
# )
Mod('sample',
'frappy_psi.lakeshore370.TemperatureLoop',
'treg sample',
io = 'io_treg',
device = 'treg',
channel = 6,
switcher = 'switcher_treg',
output_module = 'htr',
enabled = True,
calcurve = 'cx078',
)
# Mod('switcher_treg',
# 'frappy_psi.lakeshore370.Switcher',
# '',
# io = 'io_treg',
# )
Mod('htr',
'frappy_psi.lakeshore370.MainOutput',
'treg htr',
io = 'io_treg',
max_heater = '100mA',
max_power = 320e-6,
resistance = 316,
)
# Mod('sample',
# 'frappy_psi.lakeshore370.TemperatureLoop',
# 'treg sample',
# io = 'io_treg',
# channel = 6,
# switcher = 'switcher_treg',
# output_module = 'htr',
# enabled = True,
# calcurve = 'cx078',
# )
Mod('relais',
'frappy_psi.lakeshore370.AnalogOutput',
'relais to switch between mix heater (0%) and sample holder heater (100%)',
io = 'io_treg',
output_no = 2,
)
# Mod('htr',
# 'frappy_psi.lakeshore370.MainOutput',
# 'treg htr',
# io = 'io_treg',
# max_heater = '100mA',
# max_power = 320e-6,
# resistance = 316,
# )
# Mod('relais',
# 'frappy_psi.lakeshore370.AnalogOutput',
# 'relais to switch between mix heater (0%) and sample holder heater (100%)',
# io = 'io_treg',
# output_no = 2,
# )
# Mod('splehtr',
# 'frappy_psi.lakeshore370.TemperatureLoop',
# 'treg heater',
# io = 'io',
# device = 'treg',
# channel = 8,
# switcher = 'switcher',
# calcurve = 'cx078',
frappy/modulebase.py
+1
View File
@@ -824,6 +824,7 @@ class Module(HasAccessibles):
# in the mean time, a poller or handler might already have done it
if value is not Done:
wfunc = getattr(self, 'write_' + pname, None)
# self.log.info('writeInit %r, %r', self, wfunc)
if wfunc is None:
setattr(self, pname, value)
else:
frappy/rwhandler.py
+1
View File
@@ -113,6 +113,7 @@ class Handler:
if func and method_name in owner.__dict__:
raise ProgrammingError(f'superfluous method {owner.__name__}.' \
f'{method_name} (overwritten by {self.__class__.__name__})')
# print('set_name %r, %r, %r'%(owner, method_name, wrapped))
setattr(owner, method_name, wrapped)
def wrap(self, key):
frappy_psi/channelswitcher.py
+2 -1
View File
@@ -129,8 +129,9 @@ class ChannelSwitcher(Drivable):
self.target = self.value
next_measure = self._last_measure + chan.pollinterval
if now + self._time_tol > next_measure:
chan.read_value()
chan.read_status()
if chan.status[0] < ERROR:
chan.read_value()
self._last_measure = next_measure
if not self.autoscan or now + self._time_tol < self._start_measure + self.measure_delay:
return self.status
frappy_psi/lakeshore.py
+3 -3
View File
@@ -637,6 +637,7 @@ class SensorBase(Base, Readable):
value_error = None
if self.enabled:
rdgst, raw, value = self.get_internal_data()
self.log.info('rdgst %r, raw %r', rdgst, raw)
rdgst &= 0xfd # suppress old reading
if rdgst:
statuslist = formatStatusBits(rdgst, self.STATUS_BIT_LABELS)
@@ -671,9 +672,8 @@ class RawSensor(SensorBase):
return rdgst, raw, raw
def read_status(self):
status, self.value, _ = self.get_data()
if self._raw_error:
self.announceUpdate('value', err=self._raw_error)
status, _, raw = self.get_data()
self.announceUpdate('value', raw, self._raw_error)
return status
@nopoll
frappy_psi/lakeshore370.py
+98 -27
View File
@@ -24,7 +24,7 @@ from math import sqrt
from frappy.datatypes import BoolType, EnumType, FloatRange, IntRange, StatusType
from frappy.lib import formatStatusBits
from frappy.core import Done, Drivable, Parameter, Property, CommonReadHandler, CommonWriteHandler, Writable, \
IDLE, ERROR, DISABLED, Command
IDLE, ERROR, DISABLED, Command, Attached
from frappy_psi.channelswitcher import Channel, ChannelSwitcher
import frappy_psi.lakeshore as ls
@@ -47,6 +47,8 @@ class IO(ls.IO):
class Device(ls.Device):
# ioClass = IO
remote_mode = Parameter('mode', datatype=EnumType(local=0, remote=1),
value='remote', readonly=False)
model = 370
channels = list(range(1, 17))
user_curves = (1, 21) # the last curve is 20
@@ -63,9 +65,37 @@ class Device(ls.Device):
def is_equal(self, left, right, fixeps=(1.1e-5, 1.1e-4), significant=6):
# for whatever reason, the number of digits after decimal point for the T column is only 4
return super().is_equal(left, right, fixeps, significant)
def load_points(self, request):
"""load the next point(s)
:param request: the curve request
:return: next action
"""
try:
given = request.points[request.pointer:request.pointer+5]
first = request.pointer + 1
cmds = [f'CRVPT {request.curve_no},{first + n},{x:g},{y:g};'
for n, (x, y) in enumerate(given)]
cmds.append('*OPC?')
self.communicate(';'.join(cmds))
request.pointer += 5
# TODO: or should we return self.check_points instead of None?
return None if request.pointer >= len(request.points) else self.load_points
except Exception as e:
self.log.exception('error in load_points %s', e)
raise
def read_remote_mode(self):
return self.query('MODE?', int) # number of last key pressed
def write_remote_mode(self, mode):
self.command('MODE', mode)
return self.read_remote_mode()
class Switcher(ls.HasLscIO, ChannelSwitcher):
device = Attached(Device)
value = Parameter('channel', datatype=IntRange(1, 16))
target = Parameter('channel', datatype=IntRange(1, 16))
use_common_delays = Parameter('use switch_delay and measure_delay instead of the channels pause and dwell',
@@ -75,6 +105,7 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
fast_poll = 1
_measure_delay = None
_switch_delay = None
_toggle_autorange = False # flag to toggle autorange
def startModule(self, start_events):
super().startModule(start_events)
@@ -101,10 +132,9 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
super().doPoll()
# self.channels[self.target].get_raw_value() # check range or read # ! no get_raw_value()
channelno, autoscan = self.query('SCAN?', int, int)
if autoscan:
if autoscan and self.device.remote_mode == 'remote':
# pressed autoscan button: switch off HW autoscan and toggle soft autoscan
self.autoscan = not self.autoscan
# self.communicate(f'SCAN {self.value},0;SCAN?')
self.log.info('switch off HW autoscan')
self.command('SCAN', self.value, 0)
if channelno != self.value:
# channel changed by keyboard
@@ -119,8 +149,6 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
raise ValueError('no channels enabled')
self.write_target(channelno)
chan = self.channels.get(self.value)
chan.read_autorange()
chan.fix_autorange() # check for toggled autorange button
return Done
def write_switch_delay(self, value):
@@ -158,8 +186,13 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
def set_active_channel(self, chan):
self.command(f'SCAN {chan.channel}', 0)
self.value = chan.channel
chan._last_range_change = time.monotonic()
# chan._last_range_change = time.monotonic()
self.set_delays(chan)
def next_channel(self, channelno):
if self.device.remote_mode == 'local':
return channelno
return super().next_channel(channelno)
class SensorBase(Ls370, ls.SensorBase, Channel):
@@ -192,7 +225,7 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
dwell = Parameter('dwell time with autoscan', datatype=FloatRange(1, 200, unit='s'), readonly=False)
filter = Parameter('filter time', datatype=FloatRange(1, 200, unit='s'), readonly=False)
_toggle_autorange = 'init' # flag to toggle autorange
_new_autorange = True
_prev_rdgrng = (1, 1) # last read values for icur and exc
_last_range_change = 0
rdgrng_params = 'range', 'iexc', 'vexc'
@@ -201,10 +234,44 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
tempco = None
STATUS_BIT_LABELS = 'cs_ovl vcm_ovl vmix_ovl vdif_ovl r_over r_under t_over t_under'.split()
def doPoll(self):
now = time.monotonic()
self.log.info('last r ch %r', self._last_range_change)
if self._last_range_change == 0 and self.channel == self.switcher.value:
self.log.warn('first read')
self._last_range_change = now
if now + 0.5 < max(self._last_range_change, self.switcher._start_switch) + self.pause + 3:
return None
self.status, value, raw = self.get_data()
self.log.info('status %r, raw %r', self.status, raw)
if self.autorange and self.device.remote_mode == 'remote':
# if now + 0.5 > self._last_range_change + self.pause:
rng = int(max(self.minrange, self.range)) # convert from enum to int
self.log.info('rng %r, self.range %r, time %r', rng, self.range, now-self._last_range_change)
if self._raw_error is None:
if abs(raw) > self.RES_SCALE[rng]:
if rng < 22:
rng += 1
else:
lim = 0.2
while rng > self.minrange and abs(raw) < lim * self.RES_SCALE[rng]:
self.log.info('%r < %r, %r', raw, lim * self.RES_SCALE[rng], rng)
rng -= 1
lim -= 0.05 # not more than 4 steps at once
# effectively: <0.16 %: 4 steps, <1%: 3 steps, <5%: 2 steps, <20%: 1 step
elif rng < self.MAX_RNG:
rng = min(self.MAX_RNG, rng + 4)
self.log.warn('raw %r', self._raw_error)
if rng != self.range:
self.write_range(rng)
self._last_range_change = now
self.announce_data(raw, value)
def initModule(self):
# take io from switcher
# pylint: disable=unsupported-assignment-operation
self.attachedModules['io'] = self.switcher.io # pylint believes this is None
self.attachedModules['device'] = self.switcher.device
super().initModule()
def is_switching(self, now, last_switch, switch_delay):
@@ -217,14 +284,12 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
def read_rdgrng(self):
iscur, exc, rng, autorange, excoff = self.query(f'RDGRNG?{self.channel}', int, int, int, int, int)
self._prev_rdgrng = iscur, exc
if autorange:
# pressed autorange button: toggle software autorange
# we always disable hardware autorange
if not self._toggle_autorange:
self._toggle_autorange = True
if autorange and self.device.remote_mode == 'remote':
# disable hardware autorange
self.command(f'RDGRNG {self.channel}', iscur, exc, rng, 0, excoff)
iexc = 0 if excoff or not iscur else exc
vexc = 0 if excoff or iscur else exc
if (rng, iexc, vexc) != (self.range, self.iexc, self.vexc):
if (rng, iexc, vexc) != (self.range, self.iexc, self.vexc) and self._last_range_change:
self._last_range_change = time.monotonic()
try:
self.range, self.iexc, self.vexc = rng, iexc, vexc
@@ -253,31 +318,30 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
if self.autorange:
rng = max(rng, self.minrange)
self.command(f'RDGRNG {self.channel}', iscur, exc, rng, 0, excoff)
self.log.info('write_rdgrng %g', rng)
self.read_range()
def fix_autorange(self):
if self._toggle_autorange:
if self._toggle_autorange == 'init':
self.write_autorange(True)
else:
self.write_autorange(not self.autorange)
self._toggle_autorange = False
@CommonReadHandler(inset_params)
def read_inset(self):
# self.log.info('read_inset %r, %x', self.write_enabled, id(type(self).write_enabled))
# ignore curve no and temperature coefficient
self.enabled, self.dwell, self.pause, self.curve_no, self.tempco \
= self.query(f'INSET?{self.channel}', int, int, int, int, int)
@CommonWriteHandler(inset_params)
def write_inset(self, change):
# self.log.info('write_inset %r', change)
_, _, _, curve_no, tempco = self.query(f'INSET?{self.channel}', int, int, int, int, int)
self.enabled, self.dwell, self.pause, _, _ = self.command(
f'INSET {self.channel}', change['enabled'], change['dwell'], change['pause'],
curve_no, tempco)
if 'enabled' in change and change['enabled']:
# switch to enabled channel
self.switcher.write_target(self.channel)
if 'enabled' in change:
# self.log.info('channel %r, enable %r', self.channel, change['enabled'])
if change['enabled']:
# switch to enabled channel
self.switcher.write_target(self.channel)
else:
self.device.disable_channel(self.channel)
elif self.switcher.target == self.channel:
self.switcher.set_delays(self)
@@ -309,6 +373,9 @@ class RawSensor(SensorBase, ls.RawSensor):
return self.status, self.value, self.value
return super().get_data()
def announce_data(self, raw, value):
self.announceUpdate('value', raw, self._raw_error)
class Sensor(SensorBase, ls.Sensor):
@@ -328,12 +395,12 @@ class Sensor(SensorBase, ls.Sensor):
def install_sensor(self):
if self.query(f'INSET?{self.channel}', int, int, int, int, int)[4] != self.tempco:
self.write_enabled(True)
self.write_enabled(self.enabled)
def install_curve(self):
"""already done in install_sensor()"""
reply = self.query(f'INSET?{self.channel}', int, int, int, int, int)
self.log.info('install curve %r %r', reply, self.curve_no)
self.log.warn('install curve %r %r %r', reply, self.curve_no, self.enabled)
if reply[3] != self.curve_no:
self.enabled, self.dwell, self.pause, _, _ = self.command(
f'INSET {self.channel}', self.enabled, self.dwell, self.pause,
@@ -349,6 +416,10 @@ class Sensor(SensorBase, ls.Sensor):
else:
self.tempco = 1 # NTC
return logformat, range_limit
def announce_data(self, raw, value):
self.announceUpdate('value', value, self._value_error)
self.announceUpdate('raw', raw, self._raw_error)
class SorbHeater(ls.HasLscIO, Writable):