frappy_psi: add triple current source (leiden)

frappy_psi/tcs.py

@@ -0,0 +1,74 @@
+# *****************************************************************************
+# 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
+# this program; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+#
+# Module authors:
+#   Markus Zolliker <markus.zolliker@psi.ch>
+# *****************************************************************************
+
+from frappy.core import StringIO, HasIO, Writable, Parameter, Property, FloatRange, IntRange, BoolType, \
+    ERROR
+from frappy.errors import CommunicationFailedError
+
+
+class IO(StringIO):
+    end_of_line = '\n'
+    identification = [('ID?', '0\t.*')]  # senden: ID?<lf>, Antwort: 0<ht> str version information<lf>
+    #default_settings = {}
+
+
+class Heater(HasIO, Writable):
+
+    ioClass = IO    # define IO class for automatic creation of the IO module
+
+    channel = Property('channel (source number)', IntRange(1, 3))
+    value = Parameter('current reading', FloatRange(0, 0.1, unit='A'))
+    target = Parameter('current target value', FloatRange(0, 0.1, unit='A'), readonly=False)
+    on = Parameter('turn current on/off', BoolType(), readonly=False)
+
+    def query_status(self):
+        reply, txtvalue = self.communicate('STATUS?').split('\t')
+        if reply != '0':
+            raise CommunicationFailedError(f'Bad reply: {reply}')
+        return txtvalue.split(',')
+
+    def write_on(self, value):
+        for _ in range(2):
+            txtvalue = self.query_status()
+            on_idx = (self.channel - 1) * 4 + 2 + 1
+            if txtvalue[on_idx] == str(int(value)):
+                break    
+            txtvalue[on_idx] = str(int(value))
+            answer = ','.join(txtvalue)
+            reply = self.communicate(f'SETUP {answer}')
+            if reply != '0/t':
+                raise CommunicationFailedError(f'Bad reply: {reply}')
+        else:
+            return ERROR, 'unable to turn device on/off'
+
+    def write_target(self, target):
+        reply = self.communicate(f'SETDAC {self.channel} 0 {int(target * 1e6)}')  # 0: autorange
+        if reply != '0':  # not as in manual
+            raise CommunicationFailedError(f'Bad reply: {reply!r}')
+
+    def read_target(self):
+        txtvalue = self.query_status()
+        current_range = txtvalue[(self.channel - 1) * 4 + 1]
+        current = txtvalue[(self.channel - 1) * 4 + 1 + 1]   # percent of range
+        multipliers = {'1': 99e-6, '2': 990e-6, '3': 9900e-6, '4': 99e-3}
+        value = float(current) / 100 * float(multipliers[current_range])
+        return value
+    
+    # no measured value available
+    read_value = read_target