create qnw driver

Change-Id: I7b466caf91b7a2a177fa94ac84cdfc315475f959

cfg/QnwTC1_cfg.py

@@ -0,0 +1,24 @@
+Node('QnwTC1test.psi.ch',
+     'QnwTC1 test',
+     'tcp://5000',
+)
+
+Mod('io',
+    'frappy_psi.qnw.QnwIO',
+    'connection for Quantum northwest',
+    uri= 'tcp://ldmcc01-ts:3004',
+)
+
+Mod('T',
+    'frappy_psi.qnw.TemperatureLoopTC1',
+    'holder temperature',
+    channel='CT',
+    io='io',
+)
+
+Mod('Th',
+    'frappy_psi.qnw.SensorTC1',
+    'heat exch. temperature',
+    channel='HT',
+    io='io',
+)

frappy_psi/qnw.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+#  -*- coding: utf-8 -*-
+# *****************************************************************************
+# 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:
+#   Oksana Shliakhtun <oksana.shliakhtun@psi.ch>
+# *****************************************************************************
+
+
+from frappy.core import Readable, Parameter, FloatRange, IDLE, ERROR, BoolType,\
+    StringIO, HasIO, Property, Writable, Drivable, BUSY, StringType
+from frappy.errors import InternalError
+
+
+class QnwIO(StringIO):
+    """communication with TC1"""
+    end_of_line = ']'  # no line feed!
+    identification = [('[F1 VN ?', r'.F1 VN .*')]  # Controller Firmware Version, holder number
+
+
+class SensorTC1(HasIO, Readable):
+    ioClass = QnwIO
+    value = Parameter(unit='degC')
+    channel = Property('channel name', StringType())
+
+    ERROR_MAP = {
+        -1: (IDLE, ''),
+        5: (ERROR, 'Cell T out of range (Loose cable? Sensor failure?)'),
+        6: (ERROR, 'Cell and heat exchanger T out of range (Loose cable?)'),
+        7: (ERROR, 'Heat exchanger T out of range (Loose cable? Sensor failure?)'),
+        8: (ERROR, 'Inadequate coolant (check flow). Temperature control has shut down'),
+        9: (ERROR, 'Syntax error')
+    }
+
+    def set_param(self, adr, value=None):
+        short = adr.split()[0]
+        # try 3 times in case we got an asynchronous message
+        for _ in range(3):
+            if value is None:
+                reply = self.communicate(f'[F1 {adr} ?').split()
+            else:
+                reply = self.communicate(f'[F1 {adr} {value}][F1 {short} ?').split()
+            if reply[1] == short:
+                break
+        else:
+            raise InternalError(f'bad reply {reply}')
+        try:
+            return float(reply[2])
+        except ValueError:
+            return reply[2]
+
+    def get_param(self, adr):
+        return self.set_param(adr)
+
+    def read_value(self):
+        return self.get_param(self.channel)
+
+    def read_status(self):
+        reply = self.get_param('IS') # instrument status
+        if reply[0] == 1:
+            return self.ERROR_MAP[int(self.get_param('ER'))]
+        return IDLE, ''
+
+
+class TemperatureLoopTC1(SensorTC1, Drivable):
+    value = Parameter('temperature', unit='degC')
+    target = Parameter('setpoint', unit='degC')
+    control = Parameter('temperature control flag', BoolType(), readonly=False)
+    ramp = Parameter('ramping value', FloatRange, unit='degC/min', readonly=False)
+    target_min = Parameter('lowest target temperature', FloatRange, unit='degC')
+    target_max = Parameter('maximum target temperature', FloatRange, unit='degC')
+
+    def read_target_min(self):
+        return self.get_param('LT')
+
+    def read_target_max(self):
+        return self.get_param('MT')
+
+    def read_status(self):
+        reply = self.get_param('IS') # instrument status
+        if reply[0] == 1:
+            return self.ERROR_MAP[int(self.get_param('ER'))]
+        self.control = reply[2] == '+'
+        if reply[3] == 'C':
+            return BUSY, 'changing'
+        return IDLE, ''
+
+    def write_target(self, target):
+        self.write_control(True)
+        return self.set_param('TT S', target)
+
+    def read_target(self):
+        return self.get_param('TT')
+
+    def write_control(self, control):
+        sign = '-+'[control]
+        return self.set_param('TC', sign)
+
+    def read_ramp(self):
+        reply = self.get_param('RR')
+        if reply == 'W':
+            return 'waiting'
+        try:
+            return float(reply)
+        except ValueError:
+            return reply
+
+    def write_ramp(self, rate):
+        return self.set_param('RR S', rate)
+
+
+
+    # def write_target(self, target):
+    #     target_ = self.communicate(f'[F1 TT S {target}')
+    #     T_high_lim = float(self.communicate(f'[F1 MT ?'))
+    #     T_low_lim = float(self.communicate(f'[F1 LT ?'))
+    #     if T_low_lim < target_ < T_high_lim:
+    #         return self.communicate(f'[F1 TC +')
+    #     return 'Error'
+