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

state as of 12.08.26

Browse Source
This commit is contained in:
stark_a
2026-08-12 16:24:08 +02:00
parent a9047ca6f4
commit ea0260e586
4 changed files with 79 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cfg/lakeshore370_test_cfg.py
+21 -20
View File
@@ -24,24 +24,14 @@ Mod('switcher',
device = 'tmon',
)
Mod('ch10',
Mod('ch1',
'frappy_psi.lakeshore370.Sensor',
'tmon stillt',
'tmon samplehtr',
io = 'io_tmon',
channel = 10,
channel = 1,
switcher = 'switcher',
calcurve = 'ruoxm0',
enabled = True,
)
Mod('ch12',
'frappy_psi.lakeshore370.Sensor',
'tmon sorb',
io = 'io_tmon',
channel = 12,
switcher = 'switcher',
calcurve = 'c270',
enabled = True,
calcurve = 'rx078',
enabled = False, # disable when used as heater
)
Mod('ch3',
@@ -64,14 +54,14 @@ Mod('ch4',
enabled = True,
)
Mod('ch1',
Mod('ch10',
'frappy_psi.lakeshore370.Sensor',
'tmon samplehtr',
'tmon stillt',
io = 'io_tmon',
channel = 1,
channel = 10,
switcher = 'switcher',
calcurve = 'rx078',
enabled = True, # disable when used as heater
calcurve = 'ruoxm0',
enabled = True,
)
Mod('ch11',
@@ -84,6 +74,17 @@ Mod('ch11',
enabled = True,
)
Mod('ch12',
'frappy_psi.lakeshore370.Sensor',
'tmon sorb',
io = 'io_tmon',
channel = 12,
switcher = 'switcher',
calcurve = 'c270',
enabled = True,
)
# ### temperature regulating lakeshore (treg) ###
# IO('io_treg', 'dil3-ts:3001')
frappy_psi/channelswitcher.py
+7 -1
View File
@@ -97,7 +97,7 @@ class ChannelSwitcher(Drivable):
if mod.enabled:
if first_channel is None:
first_channel = ch
if next_channel == ch:
if next_channel == ch: # find the current channel
next_channel = None
elif next_channel is None:
next_channel = ch
@@ -160,6 +160,12 @@ class ChannelSwitcher(Drivable):
self.setFastPoll(True, self.fast_poll)
return channel
def write_measure_delay(self, delay):
return delay
def write_switch_delay(self, delay):
return delay
class Channel(Readable):
"""base class for channels
frappy_psi/lakeshore.py
+11 -1
View File
@@ -139,6 +139,7 @@ class Device(HasLscIO, Module):
self._requests = {} # dict <calcurve> of CurveRequest
self._sensors = {} # dict <channel> of sensors
self._disable_channels = set(self.channels)
self._raw_channels = set()
self._request_lock = threading.Lock()
self.io.setProperty('model', self.model)
super().initModule()
@@ -209,6 +210,7 @@ class Device(HasLscIO, Module):
return
self.finish_curve(self._requests.pop(key))
# no more requests pending
self._disable_channels -= self._raw_channels
while self._disable_channels:
ch = self._disable_channels.pop()
sensor = self._sensors.get(ch)
@@ -484,6 +486,9 @@ class Device(HasLscIO, Module):
def disable_channel(self, channel):
self.command(f'INTYPE {channel},0')
def register_raw(self, sensor):
self._raw_channels.add(sensor.channel)
class CurveRequest:
invalidate = None
@@ -637,7 +642,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)
# self.log.info('rdgst %r, raw %r', rdgst, raw)
rdgst &= 0xfd # suppress old reading
if rdgst:
statuslist = formatStatusBits(rdgst, self.STATUS_BIT_LABELS)
@@ -666,6 +671,11 @@ class RawSensor(SensorBase):
value = Parameter('raw value', unit='Ohm')
_curve_handling = False
def initModule(self):
if self.device:
self.device.register_raw(self)
super().initModule()
def get_internal_data(self):
ch = self.channel
rdgst, raw = self.query(f'RDGST?{ch};SRDG?{ch}', int, float)
frappy_psi/lakeshore370.py
+40 -27
View File
@@ -100,6 +100,8 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
use_common_delays = Parameter('use switch_delay and measure_delay instead of the channels pause and dwell',
BoolType(), readonly=False, default=False)
common_pause = Parameter('pause with common delays', FloatRange(3, 200, unit='s'), readonly=False, default=3)
switch_delay = Parameter(default=3, export=False)
common_filter = Parameter('common filter time', datatype=FloatRange(0, 200, unit='s'), readonly=False, default=1)
ioClass = IO
fast_poll = 1
_measure_delay = None
@@ -129,11 +131,9 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
and check autorange during filter time
"""
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 and self.device.remote_mode == 'remote':
# pressed autoscan button: switch off HW autoscan and toggle soft autoscan
self.log.info('switch off HW autoscan')
# pressed autoscan button: switch off hardware autoscan and toggle software autoscan
self.command('SCAN', self.value, 0)
if channelno != self.value:
# channel changed by keyboard
@@ -170,22 +170,22 @@ class Switcher(ls.HasLscIO, ChannelSwitcher):
def set_delays(self, chan):
if self.use_common_delays:
# write common delays to delay parameters of active channel
if chan.dwell != self.measure_delay:
chan.write_dwell(self.measure_delay)
if chan.pause != self.common_pause:
chan.write_pause(self.common_pause)
filter_ = max(0, self.switch_delay - self.common_pause)
if chan.filter != filter_:
chan.write_filter(filter_)
self.switch_delay = self.common_pause + self.common_filter
if chan.filter != self.common_filter:
chan.write_filter(self.common_filter)
else:
# switch_delay and measure_delay is changing with channel
# change switch_delay and measure_delay for active channel
self.switch_delay = chan.pause + chan.filter
self.measure_delay = chan.dwell
def set_active_channel(self, chan):
self.command(f'SCAN {chan.channel}', 0)
self.value = chan.channel
# chan._last_range_change = time.monotonic()
self.set_delays(chan)
def next_channel(self, channelno):
@@ -222,7 +222,7 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
enabled = Parameter('is this channel enabled?', datatype=BoolType(), readonly=False)
pause = Parameter('pause after channel change', datatype=FloatRange(3, 60, unit='s'), readonly=False)
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)
filter = Parameter('filter time', datatype=FloatRange(0, 200, unit='s'), readonly=False)
_new_autorange = True
_prev_rdgrng = (1, 1) # last read values for icur and exc
@@ -230,40 +230,49 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
rdgrng_params = 'range', 'iexc', 'vexc'
inset_params = 'enabled', 'pause', 'dwell'
_curve = None
_direction = 0 # change in range (lower or higher)
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')
if self.channel != self.switcher.value:
return
if self._last_range_change == 0:
self._last_range_change = now
if now + 0.5 < max(self._last_range_change, self.switcher._start_switch) + self.pause + 3:
return None
if now + 0.5 < max(self._last_range_change, self.switcher._start_switch) + self.pause:
return
self.status, value, raw = self.get_data()
self.log.info('status %r, raw %r', self.status, raw)
direction = 0
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
direction = 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)
# self.log.info('%r < %r, %r', raw, lim * self.RES_SCALE[rng], rng)
rng -= 1
direction = -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)
rng = min(self.MAX_RNG, rng + 1)
direction = 1
self.log.warn('raw %r', self._raw_error)
if rng != self.range:
self.write_range(rng)
self._last_range_change = now
if direction * self._direction < 0:
# avoid staying too long on a noisy channel
self.switcher.next_channel(self.channel)
self._direction = direction
else:
self._direction = 0
self.announce_data(raw, value)
def initModule(self):
@@ -281,21 +290,26 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
@CommonReadHandler(rdgrng_params)
def read_rdgrng(self):
# rdgrng_params = 'range', 'iexc', 'vexc'
iscur, exc, rng, autorange, excoff = self.query(f'RDGRNG?{self.channel}', int, int, int, int, int)
self.log.info('ch %r, rng %r', self.channel, rng)
self._prev_rdgrng = iscur, exc
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) and self._last_range_change:
self._last_range_change = time.monotonic()
try:
self.range, self.iexc, self.vexc = rng, iexc, vexc
except Exception:
# avoid raising errors on disabled channel
if self.enabled:
raise
if (rng, iexc, vexc) != (self.range, self.iexc, self.vexc):
if self._last_range_change:
self._last_range_change = time.monotonic()
try:
self.range, self.iexc, self.vexc = rng, iexc, vexc
self.log.warn('%r', self.range)
except Exception:
# avoid raising errors on disabled channel
self.log.exception('error')
if self.enabled:
raise
@CommonWriteHandler(rdgrng_params)
def write_rdgrng(self, change):
@@ -329,7 +343,6 @@ class SensorBase(Ls370, ls.SensorBase, Channel):
@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'],