state as of 01.07.2026

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',
+# )

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):

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
 
 

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 <markus.zolliker@psi.ch>
+#   Anik Stark <anik.stark@psi.ch>
+# *****************************************************************************
+"""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]

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}