From a4ede7cfef93216cb25db75f9c7da045d497d7dc Mon Sep 17 00:00:00 2001 From: Anik Stark Date: Wed, 1 Jul 2026 16:43:48 +0200 Subject: [PATCH] state as of 01.07.2026 --- cfg/lakeshore370_test_cfg.py | 164 +++++++++++ frappy_psi/lakeshore.py | 207 ++++++++------ frappy_psi/lakeshore340.py | 8 +- frappy_psi/lakeshore370.py | 537 +++++++++++++++++++++++++++++++++++ frappy_psi/ls372.py | 22 +- 5 files changed, 840 insertions(+), 98 deletions(-) create mode 100644 cfg/lakeshore370_test_cfg.py create mode 100644 frappy_psi/lakeshore370.py diff --git a/cfg/lakeshore370_test_cfg.py b/cfg/lakeshore370_test_cfg.py new file mode 100644 index 00000000..4e4ca1cd --- /dev/null +++ b/cfg/lakeshore370_test_cfg.py @@ -0,0 +1,164 @@ +Node('ls370test.psi.ch', + 'Lsc370 Test', + interface='tcp://5000', +) + +### temperature monitoring lakeshore (tmon) ### + +IO('io_tmon', 'dil3-ts:3003') + +Mod('tmon', + 'frappy_psi.lakeshore370.Device', + 'monitoring lakeshore 370', + curve_handling = True, + io = 'io_tmon', + ) + +Mod('switcher', + 'frappy_psi.lakeshore370.Switcher', + '', + io = 'io_tmon', + ) + +Mod('stillt', + 'frappy_psi.lakeshore370.Sensor', + 'tmon stillt', + io = 'io_tmon', + device = 'tmon', + channel = 10, + switcher = 'switcher', + calcurve = 'ruoxm0', + enabled = True, +) + +Mod('sorb', + 'frappy_psi.lakeshore370.Sensor', + 'tmon sorb', + io = 'io_tmon', + device = 'tmon', + channel = 12, + switcher = 'switcher', + calcurve = 'c270', + enabled = True, +) + +Mod('sample_mon', + 'frappy_psi.lakeshore370.Sensor', + 'tmon sample', + io = 'io_tmon', + device = 'tmon', + channel = 3, + switcher = 'switcher', + calcurve = 'rx078', + enabled = False, +) + +Mod('mix', + 'frappy_psi.lakeshore370.Sensor', + 'tmon mix', + io = 'io_tmon', + device = 'tmon', + channel = 4, + switcher = 'switcher', + calcurve = 'ruoxm0', + enabled = True, +) + +Mod('samplehtr', + 'frappy_psi.lakeshore370.Sensor', + 'tmon samplehtr', + io = 'io_tmon', + device = 'tmon', + channel = 1, + switcher = 'switcher', + calcurve = 'rx078', + enabled = False, # disable when used as heater +) + +Mod('onek', + 'frappy_psi.lakeshore370.RawSensor', + 'tmon onek', + io = 'io_tmon', + device = 'tmon', + channel = 11, + switcher = 'switcher', + # calcurve = 'ruoxm0', + enabled = True, +) + +### temperature regulating lakeshore (treg) ### + +IO('io_treg', 'dil3-ts:3001') + +Mod('treg', + 'frappy_psi.lakeshore370.Device', + 'regulating lakeshore 370', + io = 'io_treg', + curve_handling = True, + ) + +Mod('switcher_treg', + 'frappy_psi.lakeshore370.Switcher', + '', + io = 'io_treg', + ) + +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('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', +# enabled = True, +# ) + + +### lakeshore 370 in lab (ldmse3-ts:3014') ### + +# IO('io', 'ldmse3-ts:3014') + +# Mod('T1', +# 'frappy_psi.lakeshore370.RawSensor', +# '', +# io = 'io', +# channel = 1, +# switcher = 'switcher', +# ) + +# Mod('T5', +# 'frappy_psi.lakeshore370.RawSensor', +# '', +# io = 'io', +# channel = 5, +# switcher = 'switcher', +# ) \ No newline at end of file diff --git a/frappy_psi/lakeshore.py b/frappy_psi/lakeshore.py index 385626a8..bf78431e 100644 --- a/frappy_psi/lakeshore.py +++ b/frappy_psi/lakeshore.py @@ -32,6 +32,7 @@ from frappy.errors import CommunicationFailedError, ConfigError, \ HardwareError, DisabledError, ImpossibleError, secop_error, SECoPError from frappy.lib.units import NumberWithUnit, format_with_unit from frappy.lib import formatStatusBits +from frappy.lib.enum import EnumMember from frappy_psi.calcurve import CalCurve from frappy_psi.convergence import HasConvergence from frappy.mixins import HasOutputModule, HasControlledBy @@ -104,7 +105,7 @@ class HasLscIO(HasIO): self.communicate(f'{msghead};*OPC?') return None # when an argument is given as integer, it might be that this argument might be a float - converters = [string_to_num if isinstance(a, int) else type(a) for a in args] + converters = [string_to_num if isinstance(a, (int, EnumMember)) else type(a) for a in args] values = [a if isinstance(a, str) else f'{a:g}' for c, a in zip(converters, args)] if ' ' in msghead: @@ -162,7 +163,8 @@ class Device(HasLscIO, Module): if np.array_equal(prev_request.points, req.points): # a request is already running and the curve content has not changed self.log.info('already installing %s', sensor.calcurve) - req.add_sensor(sensor) + prev_request.add_sensor(sensor) + # self.log.info('sensors for req %r', prev_request.sensors) else: self.log.info('file has changed, restart treating %s', sensor.calcurve) self._requests[sensor.calcurve] = req @@ -367,7 +369,7 @@ class Device(HasLscIO, Module): # write a temporary header # (in case loading will not finish, the curve should be marked as empty) header = list(request.crvhdr) - self.put_header(request.curve_no, 'loading...', '', *header[2:]) + self.put_header(request.curve_no, 'loading...', '_', *header[2:]) request.pointer = 0 return self.load_points @@ -423,7 +425,7 @@ class Device(HasLscIO, Module): def put_header(self, curve_no, name, sn, fmt, limit, coef): """write header""" - self.communicate(f'CRVHDR {curve_no},"{name}","{sn}",{fmt},{limit},{coef};*OPC?') + self.command(f'CRVHDR {curve_no}', name, sn or '_', fmt, limit, coef) def is_equal(self, left, right, fixeps=(1.1e-5, 1.1e-5), significant=6): """check whether a returned calibration point is equal within curve point precision""" @@ -595,19 +597,81 @@ class Base(HasLscIO): super().initModule() -class Sensor(Base, Readable): +class SensorBase(Base, Readable): """base class for sensors""" value = Parameter(unit='K') status = Parameter(datatype=StatusType(Readable, 'DISABLED', 'BUSY')) - raw = Parameter('raw sensor value', datatype=FloatRange(unit='Ohm'), default=0) channel = Property('used channel', StringType()) # input - calcurve = Parameter('calibration curve name', StringType(), readonly=False) enabled = Parameter('enable flag', BoolType(), readonly=False) curve_no = None STATUS_BIT_LABELS = 'invalid_reading old_reading b2 b3 t_under t_over units_zero units_overrange'.split() - _read_error = None + _value_error = None _raw_error = None - _do_read = True + # TODO: implement alarms + + def doPoll(self): + self.read_status() # polls also value and raw + + def get_data(self): + """get reading status, raw and Kelvin value with minimal delay""" + status = IDLE, '' + raw_error = None + value_error = None + if self.enabled: + rdgst, raw, value = self.get_internal_data() + rdgst &= 0xfd # suppress old reading + if rdgst: + statuslist = formatStatusBits(rdgst, self.STATUS_BIT_LABELS) + status = ERROR, statuslist[-1] # show only the most fatal error + value_error = HardwareError(statuslist[-1]) + while statuslist and statuslist[-1].startswith('t_'): + statuslist.pop() + if statuslist: + raw_error = HardwareError(statuslist[-1]) + elif self._curve_handling: + value_error = self.device.get_calib_state(self) + if value_error: + status = ERROR, str(value_error) + else: + value_error = raw_error = DisabledError('disabled') + value = raw = 0 + status = DISABLED, 'disabled' + self.enable = False + self._value_error = value_error + self._raw_error = raw_error + return status, value, raw + + +class RawSensor(SensorBase): + """class for sensors using raw value (w/o calibration curve)""" + value = Parameter('raw value', unit='Ohm') + _curve_handling = False + + def get_internal_data(self): + ch = self.channel + rdgst, raw = self.query(f'RDGST?{ch};SRDG?{ch}', int, float) + 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) + return status + + @nopoll + def read_value(self): + self.status, value, _ = self.get_data() + if self._raw_error: + raise self._raw_error + return value + + +class Sensor(SensorBase): + """class for sensors using value (incl. calibration curve)""" + value = Parameter('calibrated value', unit='K') + raw = Parameter('raw', datatype=FloatRange(unit='Ohm')) + calcurve = Parameter('calibration curve name', StringType(), readonly=False) + curve_no = None _curve_handling = False # True when a device is present and device.curve_handling is True # TODO: implement alarms @@ -615,62 +679,40 @@ class Sensor(Base, Readable): super().initModule() if self.device: self._curve_handling = self.device.curve_handling + self.log.info('curve handling %r', self._curve_handling) if not self._curve_handling: self.parameters['calcurve'].setProperty('export', False) - def doPoll(self): - self.read_status() # polls also value and raw + def get_internal_data(self): + ch = self.channel + rdgst, raw, value = self.query(f'RDGST?{ch};SRDG?{ch};KRDG?{ch}', int, float, float) + return rdgst, raw, value - @nopoll - def read_value(self): - self.status, value, raw = self.get_data() - if isinstance(value, SECoPError): - raise value - if not isinstance(raw, SECoPError): - self.raw = raw - return value + def read_status(self): + status, self.value, self.raw = self.get_data() + if self._raw_error: + self.announceUpdate('raw', err=self._raw_error) + if self._value_error: + self.announceUpdate('value', err=self._value_error) + return status @nopoll def read_raw(self): - self.status, value, raw = self.get_data() - if isinstance(raw, SECoPError): - raise raw - self.value = value + self.status, self.value, raw = self.get_data() + if self._value_error: + self.announceUpdate('value', err=self._value_error) + if self._raw_error: + raise self._raw_error return raw - def read_status(self): - try: - status, self.value, self.raw = self.get_data() - except Exception as e: - self.raw = self.value = secop_error(e) - raise - return status - - def get_data(self): - """get reading status, raw and Kelvin value with minimal delay""" - status = IDLE, '' - if self.enabled: - ch = self.channel - rdgst, raw, value = self.query(f'RDGST?{ch};SRDG?{ch};KRDG?{ch}', int, float, float) - rdgst &= 0xfd # suppress old reading - if rdgst: - statuslist = formatStatusBits(rdgst, self.STATUS_BIT_LABELS) - status = ERROR, statuslist[-1] # show only the most fatal error - value = HardwareError(statuslist[-1]) - while statuslist and statuslist[-1].startswith('t_'): - statuslist.pop() - if statuslist: - raw = HardwareError(statuslist[-1]) - elif self._curve_handling: - value_error = self.device.get_calib_state(self) - if value_error: - value = value_error - status = ERROR, str(value_error) - else: - raw = value = DisabledError('disabled') - status = DISABLED, 'disabled' - self.enable = False - return status, value, raw + @nopoll + def read_value(self): + self.status, value, self.raw = self.get_data() + if self._raw_error: + self.announceUpdate('raw', err=self._raw_error) + if self._value_error: + raise self._value_error + return value def write_calcurve(self, calibname): if not self._curve_handling: @@ -728,7 +770,7 @@ class Output(Base, HasControlledBy, Writable): ''', FloatRange(0, 100, unit='W'), readonly=False) # for the case when several outputs are possible: output_no = Parameter('lakeshore output or loop number', IntRange(1, 4), default=1) - resistance = Parameter('heater resistance', FloatRange(10, 100), readonly=False, default=25) + resistance = Parameter('heater resistance', FloatRange(10, 100, unit='Ohm'), readonly=False, default=25) Extension = None AMP_VOLT_WATT = NumberWithUnit('A', 'V', 'W') imax = None @@ -751,27 +793,6 @@ class Output(Base, HasControlledBy, Writable): """ raise NotImplementedError - def set_closed_loop(self, loop): - """set to control mode - - :param loop: the temperature loop module - """ - if self._control_loop != loop: - self.log.info(f'set output to be controlled on channel {loop.channel}') - self._control_loop = loop - self.configure() - else: - self.fix_heater_range() - - def set_open_loop(self): - """set to open loop""" - if self._control_loop is not None: - self.log.info('put output into manual mode') - self._control_loop = None - self.configure() - else: - self.fix_heater_range() - def fix_heater_range(self): """switch on heater range, if off""" @@ -890,10 +911,10 @@ class MainOutput(Output): self.put_manual_power(self._control_loop.power_offset) def read_max_power(self): - curidx, self._user_current = self.query(f'HTRSET? {self.output_no}', int, int, float)[1:3] + curidx, self._user_current = self.query(f'HTRSET? {self.output_no}', int, int, float)[1:3] # ! cmd not found in ls370 manual (HTRRNG?) if not self._user_current: self._user_current = 2.0 ** (curidx * 0.5 - 1) - self._htr_range = self.query(f'RANGE?{self.output_no}', int) or self._htr_range + self._htr_range = self.query(f'RANGE?{self.output_no}', int) or self._htr_range # ! cmd not found in ls370 manual (HTRRNG?) self._power_scale = self.imax ** 2 * self.heater_ranges[self._htr_range] / 1e4 return self.calc_power(100) @@ -964,13 +985,16 @@ class Loop(HasConvergence, HasOutputModule, Sensor): }, prefix='pid_', readonly=False) power_offset = Parameter('power offset\n\n(using LakeShores Manual Output)', FloatRange(0, 100, unit='W'), readonly=False, default=0) + + def loop(self): + return f' {self.output_module.output_no}' def write_ctrlpars(self, pid): - pid = self.command(f'PID {self.output_module.output_no}', *[pid[k] for k in 'pid']) + pid = self.command(f'PID{self.loop()}', *[pid[k] for k in 'pid']) return dict(zip('pid', pid)) def read_ctrlpars(self): - pid = self.query(f'PID?{self.output_module.output_no}', float, float, float) + pid = self.query(f'PID?{self.loop()}', float, float, float) return dict(zip('pid', pid)) def write_target(self, value): @@ -978,7 +1002,6 @@ class Loop(HasConvergence, HasOutputModule, Sensor): if sensor_status[0] >= BUSY: raise ImpossibleError(f'can not control while status is {sensor_status}') self.activate_control() - self.output_module.set_closed_loop(self) self.set_target(value) def write_power_offset(self, value): @@ -989,10 +1012,24 @@ class Loop(HasConvergence, HasOutputModule, Sensor): return self.get_target() def set_target(self, value): - return self.command(f'SETP {self.output_module.output_no}', float(value)) + return self.command(f'SETP{self.loop()}', float(value)) def get_target(self): - return self.query(f'SETP?{self.output_module.output_no}', float) + return self.query(f'SETP?{self.loop()}', float) + + def set_control_active(self, active): + if active: + controlled_by = self.name + if self.output_module.controlled_by != controlled_by: + self.log.info(f'set output to be controlled on channel {self.channel}') + self.output_module.controlled_by = controlled_by + self.output_module.configure() + else: + self.output_module.fix_heater_range() + else: + self.output_module.write_target(0) + self.log.info('put output into manual loop') + # TODO: or should we keep output value? class Device2(Device): diff --git a/frappy_psi/lakeshore340.py b/frappy_psi/lakeshore340.py index b6b3c7dd..4b103e12 100644 --- a/frappy_psi/lakeshore340.py +++ b/frappy_psi/lakeshore340.py @@ -61,7 +61,11 @@ class Device(ls.Device): return super().is_equal(left, right, fixeps, significant) -class Sensor340(ls.Sensor): +class RawSensor(ls.RawSensor): + model = 340 + + +class Sensor(ls.Sensor): model = 340 intype = None # arguments for the intype command @@ -100,7 +104,7 @@ class Sensor340(ls.Sensor): self.command(f'INSET {self.channel}', 1, 1) -class Loop340(ls.Loop, Sensor340): +class Loop340(ls.Loop, Sensor): pass diff --git a/frappy_psi/lakeshore370.py b/frappy_psi/lakeshore370.py new file mode 100644 index 00000000..73ec340b --- /dev/null +++ b/frappy_psi/lakeshore370.py @@ -0,0 +1,537 @@ +#!/usr/bin/env python +# ***************************************************************************** +# 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 +# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA +# +# Module authors: +# Markus Zolliker +# Anik Stark +# ***************************************************************************** +"""LakeShore 370""" + +import time +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 +from frappy_psi.channelswitcher import Channel, ChannelSwitcher +import frappy_psi.lakeshore as ls + + +Status = Drivable.Status + + +STATUS_BIT_LABELS = 'CS_OVL VCM_OVL VMIX_OVL VDIF_OVL R_OVER R_UNDER T_OVER T_UNDER'.split() + + +class Ls370: + model = 370 + channel = Property('used channel', IntRange(1, 17)) + + +class IO(ls.IO): + identification = [('*IDN?', 'LSCI,MODEL370,.*')] + wait_before = 0.05 + + +class Device(ls.Device): + # ioClass = IO + model = 370 + channels = list(range(1, 17)) + user_curves = (1, 21) # the last curve is 20 + log_formats = True, True # log Ohm / log K is supported + cmds_per_line = 1 + _crvsav_deadline = None + + def disable_channel(self, channel): + self.command(f'INSET {channel}', 0, 1, 3, 0, 1) + + def put_header(self, curve_no, name, sn, fmt, limit, coef): + self.command(f'CRVHDR {curve_no}', f'{name:15s}', f'{sn:10s}', fmt, limit, coef) + + 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) + + +class Switcher(ls.HasLscIO, ChannelSwitcher): + 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', + BoolType(), readonly=False, default=False) + common_pause = Parameter('pause with common delays', FloatRange(3, 200, unit='s'), readonly=False, default=3) + ioClass = IO + fast_poll = 1 + _measure_delay = None + _switch_delay = None + + def startModule(self, start_events): + super().startModule(start_events) + # disable unused channels + for ch in range(1, 16): + if ch not in self.channels: + self.command(f'INSET {ch}', 0, 1, 3, 0, 1) + channelno, autoscan = self.query('SCAN?', int, int) + if channelno in self.channels and self.channels[channelno].enabled: + if not autoscan: + return # nothing to do + else: + channelno = self.next_channel(channelno) + if channelno is None: + self.status = 'ERROR', 'no enabled channel' + return + self.command(f'SCAN {channelno}', 0) + + def doPoll(self): + """poll buttons + + 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: + # 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.command('SCAN', self.value, 0) + if channelno != self.value: + # channel changed by keyboard + if channelno in self.channels: + self.write_target(channelno) + else: + channelno = self.value + chan = self.channels.get(channelno) + if chan is None: + channelno = self.next_channel(channelno) + if channelno is None: + 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): + self._switch_delay = value + return super().write_switch_delay(value) + + def write_measure_delay(self, value): + self._measure_delay = value + return super().write_measure_delay(value) + + def write_use_common_delays(self, value): + if value: + # use values from a previous change, instead of + # the values from the current channel + if self._measure_delay is not None: + self.measure_delay = self._measure_delay + if self._switch_delay is not None: + self.switch_delay = self._switch_delay + return value + + def set_delays(self, chan): + if self.use_common_delays: + 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_) + else: + # switch_delay and measure_delay is changing with 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) + + +class SensorBase(Ls370, ls.SensorBase, Channel): + """temperature channel on Lakeshore 370""" + + RES_RANGE = {key: i+1 for i, key in list( + enumerate(mag % val for mag in ['%gmOhm', '%gOhm', '%gkOhm', '%gMOhm'] + for val in [2, 6.32, 20, 63.2, 200, 632]))[:-2]} + CUR_RANGE = {key: i + 1 for i, key in list( + enumerate(mag % val for mag in ['%gpA', '%gnA', '%guA', '%gmA'] + for val in [1, 3.16, 10, 31.6, 100, 316]))[:-2]} + VOLT_RANGE = {key: i + 1 for i, key in list( + enumerate(mag % val for mag in ['%guV', '%gmV'] + for val in [2, 6.32, 20, 63.2, 200, 632]))} + RES_SCALE = [2 * 10 ** (0.5 * i) for i in range(-7, 16)] # RES_SCALE[0] is not used + MAX_RNG = len(RES_SCALE) - 1 + + status = Parameter(datatype=StatusType(Drivable, 'DISABLED')) + pollinterval = Parameter(visibility=3, default=1) + range = Parameter('reading range', readonly=False, + datatype=EnumType(**RES_RANGE)) + minrange = Parameter('minimum range for software autorange', readonly=False, default=1, + datatype=EnumType(**RES_RANGE)) + autorange = Parameter('autorange', datatype=BoolType(), + readonly=False, default=1) + iexc = Parameter('current excitation', datatype=EnumType(off=0, **CUR_RANGE), readonly=False) + vexc = Parameter('voltage excitation', datatype=EnumType(off=0, **VOLT_RANGE), readonly=False) + 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) + + _toggle_autorange = 'init' # flag to toggle autorange + _prev_rdgrng = (1, 1) # last read values for icur and exc + _last_range_change = 0 + rdgrng_params = 'range', 'iexc', 'vexc' + inset_params = 'enabled', 'pause', 'dwell' + _curve = None + tempco = None + + def initModule(self): + # take io from switcher + # pylint: disable=unsupported-assignment-operation + self.attachedModules['io'] = self.switcher.io # pylint believes this is None + super().initModule() + + def is_switching(self, now, last_switch, switch_delay): + last_switch = max(last_switch, self._last_range_change) + if now + 0.5 > last_switch + self.pause: + self.get_data() # adjust range only + return super().is_switching(now, last_switch, switch_delay) + + @CommonReadHandler(rdgrng_params) + 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 + 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): + 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 + + @CommonWriteHandler(rdgrng_params) + def write_rdgrng(self, change): + self.read_range() # make sure autorange is handled + if 'vexc' in change: # in case vext is changed, do not consider iexc + change['iexc'] = 0 + if change['iexc']: + iscur = 1 + exc = change['iexc'] + excoff = 0 + elif change['vexc']: + iscur = 0 + exc = change['vexc'] + excoff = 0 + else: + iscur, exc = self._prev_rdgrng # set to last read values + excoff = 1 + rng = change['range'] + if self.autorange: + rng = max(rng, self.minrange) + self.command(f'RDGRNG {self.channel}', iscur, exc, rng, 0, excoff) + 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): + # 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): + _, _, _, 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) + elif self.switcher.target == self.channel: + self.switcher.set_delays(self) + + def read_filter(self): + on, settle = self.query(f'FILTER?{self.channel}', int, int) + return settle if on else 0 + + def write_filter(self, value): + on = 1 if value else 0 + value = max(1, value) + on, settle, _ = self.command(f'FILTER {self.channel}', on, value, self.channel) + if not on: + settle = 0 + return settle + + +class RawSensor(SensorBase, ls.RawSensor): + + def get_internal_data(self): + ch = self.channel + rdgst = self.query(f'RDGST?{ch}', int) + raw = self.query(f'RDGR?{ch}', float) + return rdgst, raw, raw + + def get_data(self): + if not self.enabled: + self.status = DISABLED, 'disabled' + if not self.channel == self.switcher.value == self.switcher.target: + return self.status, self.value, self.value + return super().get_data() + + +class Sensor(SensorBase, ls.Sensor): + + def get_internal_data(self): + ch = self.channel + rdgst = self.query(f'RDGST?{ch}', int) + raw = self.query(f'RDGR?{ch}', float) + value = self.query(f'RDGK?{ch}', float) + return rdgst, raw, value + + def get_data(self): + if not self.enabled: + self.status = DISABLED, 'disabled' + if not self.channel == self.switcher.value == self.switcher.target: + return self.status, self.value, self.raw + return super().get_data() + + def install_sensor(self): + if self.query(f'INSET?{self.channel}', int, int, int, int, int)[4] != self.tempco: + self.write_enabled(True) + + 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) + if reply[3] != self.curve_no: + self.enabled, self.dwell, self.pause, _, _ = self.command( + f'INSET {self.channel}', self.enabled, self.dwell, self.pause, + self.curve_no, self.tempco) + + def get_curve_type(self, calcurve): + unit = calcurve.options.get('unit', 'Ohm') + logformat = True, False + range_limit = None + if unit == 'Ohm': + if calcurve.ptc: + self.tempco = 2 # PTC + else: + self.tempco = 1 # NTC + return logformat, range_limit + + +class SorbHeater(ls.HasLscIO, Writable): + value = Parameter('heater output', FloatRange(0, unit='W')) + target = Parameter('heater output', FloatRange(0, unit='$'), readonly=False, default=0) + output = Property('output channel', IntRange(0, 2), default=1) + HTRRNG = {n: i for i, n in enumerate(['off', '31.6uA', '100uA', '316uA', '1mA', '3.16mA', '10mA', '31.6mA', '100mA'])} + heater_range = Parameter('heater range', EnumType(HTRRNG), readonly=False) + target_max = Parameter('max power', FloatRange(0, unit='$'), readonly=False, default=5) + fullpower = Parameter('power at 100 %', FloatRange(0, unit='$'), readonly=False, default=5) + + def write_target(self, value): + percent = 100 * sqrt(value / self.fullpower) + on = bool(value) + if on != bool(self.heater_range): + self.write_heater_range(on) + self.command(f'MOUT {self.output},{percent}') + # percent = parse1(self.communicate(f'MOUT {ch},{percent};MOUT?{ch}')) + # percent = self.query(f'MOUT?{ch}', float) + # return (percent * 0.01) ** 2 * self.fullpower + return self.read_value() + + def read_value(self): + percent = self.query(f'MOUT?{self.output}', float) + # percent = parse1(self.communicate(f'MOUT?{self.output}')) + return (percent * 0.01) ** 2 * self.fullpower + + def write_heater_range(self, value): + # ch = self.output + # return parse1(self.communicate(f'RANGE {ch},{value};RANGE?{ch}')) + self.command(f'HTRRNG ', value) + return self.read_heater_range() + + def read_heater_range(self): + # return parse1(self.communicate(f'RANGE?{self.output}')) + return self.query(f'HTRRNG?', int) + + +class TemperatureLoop(ls.Loop, Sensor, Drivable): + loop = Property('lakshore loop', IntRange(0, 1), default=0) # TODO: implemented specific issues of loop 1 + target = Parameter('setpoint', FloatRange(0, unit='$')) + control_active = Parameter('we are controlling', BoolType(), default=0) + + _control_active = False + + def loop(self): + return '' + + def doPoll(self): + super().doPoll() + self.read_control_active() + + def read_control_active(self): + if self._control_active: + self.output_module.fix_heater_range() + return self._control_active + + # def read_target(self): + # if self._control_active: + # return self.query('SETP?', float) + # return 0 + + def set_target(self, target): + outmode = 1 + prev = self.query(f'CMODE?', int) + if outmode != prev: + self.command(f'CMODE', outmode) + for chan in self.switcher.channels.values(): + chan._control_active = False + self._control_active = True + self.read_control_active() + self.convergence_start() + # do not return the readback value, as it might not yet be correct + self.command('SETP', target) + return target + + +HEATER_RANGES = {8 - i: 10 ** -i for i in range(8)} + +class MainOutput(ls.MainOutput): + ioClass = IO + model = 370 + HTRRNG = {n: i for i, n in enumerate(['off', '31.6uA', '100uA', '316uA', '1mA', '3.16mA', '10mA', '31.6mA', '100mA'])} + # htrrng = Parameter('', EnumType(HTRRNG), readonly=False) + htr = Parameter('heater percentage', FloatRange(unit='%')) + minheater = Parameter('minimal heater current', FloatRange(0, 0.01, unit='A'), readonly=False, default=0) + resistance = Parameter(datatype=FloatRange(1,100000, unit='Ohm')) + heater_ranges = HEATER_RANGES + HTRST_MAP = { + 0: (IDLE, ''), + 1: (ERROR, 'Open heater load'), + } + _manual_output = 0.0 # TODO: check how to set this + vmax = 50 + imax = 2 + max_currents = {4 - i: 2 ** (1 - i) for i in range(4)} + SETPOINTLIMS = 1500.0 + sorted_factors = sorted([(fhtr * fcur ** 2, (i, h)) + for i, fcur in max_currents.items() + for h, fhtr in HEATER_RANGES.items() + ]) + + def get_status(self): + st = self.query(f'HTRST?', int) + return self.HTRST_MAP[st] + + def configure(self): + if self._desired_max_power is None: + self.log.info(f'max_heater {self.writeDict} {self.max_heater}') + self.write_max_heater(self.max_heater) + icurrent, htr_range = self.get_best_power_idx(self._desired_max_power, 1.1) + self._power_scale = self.max_currents[icurrent] ** 2 * self.heater_ranges[htr_range] / 1e4 + self._htr_range = htr_range + if self._control_loop is None: + mode = self.query(f'CMODE?', int) + if mode != 3: # open loop + self.command(f'CSET', '0', 1, 1, 1, 1, self._htr_range, self.resistance) # control off + self.command(f'CMODE', 3) # open loop + self.command('HTRRNG', self._htr_range) + self.put_manual_power(self._manual_output) + else: + self.command(f'CSET', self._control_loop.channel, 1, 1, 1, 1, self._htr_range, self.resistance) # control on + self.command(f'CMODE', 1) # pid + self.command('HTRRNG', self._htr_range) + self.put_manual_power(self._control_loop.power_offset) + + def fix_heater_range(self): + # switch heater range on, if needed + irng = self.query('HTRRNG?', int) + if irng != self._htr_range: + if irng: + self.log.info('output range was changed manually') + self._htr_range = irng + else: + self.log.info('output was off - switch on again') + self.command('HTRRNG', self._htr_range) + + def read_max_power(self): + htr_range = self.query(f'CSET?', int, int, int, int, int, int, float)[5] + htr_range = self.query('HTRRNG?', int) or htr_range + self._htr_range = htr_range + self._power_scale = 0.1 ** 2 * self.heater_ranges[htr_range] / 1e4 + return self.calc_power(100) + + def read_htr(self): + """read heater output in percent or power depending on the heater output selection""" + return self.query('HTR?', float) + + @Command + def control_off(self): + """switch control off""" + self._control_active = False + self.command(f'HTRRNG', 0) + self.read_control_active() + + def set_htrrng(self): + if self._control_active: + newhtr = int(self.htrrng) if self._control_active else 0 + htrrng = self.query('HTRRNG?', int) + if htrrng != newhtr: + if newhtr: + self.log.info('switched heater on %d', newhtr) + self.command('HTRRNG', newhtr) + + def put_manual_power(self, value): + self.command(f'MOUT ', value) + + # def write_htrrng(self, value): + # if self._control_active: + # self.command('HTRRNG', int(value)) + # return value + + +class AnalogOutput(ls.HasLscIO, Writable): + + target = Parameter('target', datatype=FloatRange(0, 100, unit='%')) + value = Parameter('value', datatype=FloatRange(0, 100, unit='%')) + output_no = Property('output number', datatype=IntRange(1, 2)) + + def write_target(self, target): + if target > 0: + self.command(f'ANALOG {self.output_no}', 1, 2, 1, 1, 0.0, 0.0, target) # manual (2) + return + self.command(f'ANALOG {self.output_no}', 1, 0, 1, 1, 0.0, 0.0, target) # off (0) + + def read_value(self): + return self.query(f'ANALOG? {self.output_no}', int, int, int, int, float, float, float)[6] diff --git a/frappy_psi/ls372.py b/frappy_psi/ls372.py index bbec51f7..f057dc27 100644 --- a/frappy_psi/ls372.py +++ b/frappy_psi/ls372.py @@ -414,11 +414,11 @@ class TemperatureLoop(HasConvergence, TemperatureChannel, Drivable): minheater = Parameter('minimal heater current', FloatRange(0, 0.01, unit='A'), readonly=False, default=0) HTRRNG = {n: i for i, n in enumerate(['off', '30uA', '100uA', '300uA', '1mA', '3mA', '10mA', '30mA', '100mA'])} htrrng = Parameter('', EnumType(HTRRNG), readonly=False) - ctrlpars = StructParam('control parameters struct', { - 'p': Parameter('proportional heat parameter', FloatRange()), - 'i': Parameter('integral heat parameter', FloatRange()), - 'd': Parameter('derivative heat parameter', FloatRange()), - }, readonly=False, export=False) + # ctrlpars = StructParam('control parameters struct', { + # 'p': Parameter('proportional heat parameter', FloatRange()), + # 'i': Parameter('integral heat parameter', FloatRange()), + # 'd': Parameter('derivative heat parameter', FloatRange()), + # }, readonly=False, export=False) htr = Parameter('heater percentage', FloatRange(unit='%')) _control_active = False @@ -481,10 +481,10 @@ class TemperatureLoop(HasConvergence, TemperatureChannel, Drivable): self.communicate(f'SETP {self.loop},{target};*OPC?') return target - def write_ctrlpars(self, ctrlpars): - p, i, d = self.change(f'PID {self.loop}', ctrlpars['p'], ctrlpars['i'], ctrlpars['d']) - return {'p': p, 'i': i, 'd': d} + # def write_ctrlpars(self, ctrlpars): + # p, i, d = self.change(f'PID {self.loop}', ctrlpars['p'], ctrlpars['i'], ctrlpars['d']) + # return {'p': p, 'i': i, 'd': d} - def read_ctrlpars(self): - p, i, d = self.query(f'PID?{self.loop}') - return {'p': p, 'i': i, 'd': d} + # def read_ctrlpars(self): + # p, i, d = self.query(f'PID?{self.loop}') + # return {'p': p, 'i': i, 'd': d}