Flowmeter

basil/HL/bronkhorst_elflow.py

@@ -7,6 +7,7 @@
 
 import logging
 import struct
+import time
 
 from basil.HL.RegisterHardwareLayer import HardwareLayer
 
@@ -16,160 +17,100 @@
 
 class Bronkhorst_ELFLOW(HardwareLayer):
     ''' Bronkhorst ELFLOW
+    Manual can be found here:
+    https://www.bronkhorst.com/getmedia/77a1438f-e547-4a79-95ad-53e81fd38a97/917027-Manual-RS232-interface.pdf
     '''
 
+    CMDS = {
+        'get_measure_flow': ':06800401210120',
+        'get_capacity': ':068004014D014D',
+        'get_control_mode': ':06800401040104',
+        'set_control_mode': ':0580010104',
+        'set_setpoint': ':0680010121',
+        'get_setpoint': ':06800401210121',
+        'get_valve': ':06800472417241',
+    }
+
     def __init__(self, intf, conf):
         self.debug = 0
         self.node = "80"
         super(Bronkhorst_ELFLOW, self).__init__(intf, conf)
+        self.pre_time = time.time()
 
     def init(self):
         super(Bronkhorst_ELFLOW, self).init()
 
     def write(self, cmd):
-        cmd_s = ""
-        for c in cmd:
-            cmd_s = cmd_s + "%02X" % c
-        cmd_s = ":%02X%s%s" % (len(cmd) + 1, self.node, cmd_s)
-        if self.debug != 0:
-            logger.debug("ELFLOW.write() %s" % str(cmd_s))
-        self._intf.write(cmd_s)
+        if time.time() - self.pre_time < 1.0:
+            time.sleep(1.0)
+        self._intf.write(str(cmd))
+        self.pre_time = time.time()
 
     def read(self):
-        ret_s = self._intf.read()
-        if self.debug != 0:
-            logger.debug("ELFLOW.read() %s" % str(ret_s))
-        if len(ret_s) < 5 or ret_s[0] != ":" or ret_s[3:5] != self.node:
-            logger.debug("ELFLOW.read() format error ret=%s" % str(ret_s))
-            return []
-        ret_len = int(ret_s[1:3])
-        if ret_len * 2 != len(ret_s[3:-2]):
-            logger.debug("ELFLOW.read() data lenth error ret=%s" % str(ret_s))
-            return []
-        ret = []
-        for i in range(ret_len - 1):
-            ret.append(int(ret_s[5 + 2 * i:5 + 2 * (i + 1)], 16))
-        return ret
+        ret = self._intf.read()
+        if len(ret) < 2 or ret[-2:] != "\r\n":
+            logger.warning("read() termination error")
+        return ret.strip()
+
+    def get_valve_output(self):
+        self._intf.write(self.CMDS['get_valve'])
+        ret = int(self.read()[11:], 16)
+        return ret * 61.7 / 10345949  # converts int in percentage
 
     def set_setpoint(self, value):
-        cmd = [1, 1, 0x21, (value >> 8) & 0xFF, value & 0xFF]
-        self.write(cmd)
+        """value range from 0 - 32000
+        """
+
+        if not isinstance(value, int):
+            raise ValueError(f"Given value has to be of type integer, is {type(value)}!")
+
+        hex_val = hex(value)[2:]        # [2:] to remove the 0x from the beginning of the hex number
+        command = f"{self.CMDS['set_setpoint']}" + f"{hex_val.zfill(4)}"  # hex should have at least 4 digits
+        self._intf.write(command)
         ret = self.read()
-        if len(ret) != 3:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 0 and ret[1] == 0 and ret[2] == 5:
-            return 0
-        else:
-            logger.debug("ELFLOW.set_setpoint() ret error ret=%s" % str(ret))
-            return -1
+        return ret
 
     def get_setpoint(self):
-        cmd = [4, 1, 0x21, 1, 0x21]
-        self.write(cmd)
+        self._intf.write(self.CMDS['get_setpoint'])
         ret = self.read()
-        if len(ret) != 5:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 2 and ret[1] == cmd[1] and ret[2] == cmd[2]:
-            return ((ret[3] << 8) & 0xFF00) | (ret[4] & 0xFF)
-        else:
-            logger.debug("ELFLOW.set_setpoint() ret error ret=%s" % str(ret))
-            return -1
-
-    def set_control_mode(self, value):
+        answer_in_hex = ret[11:]  # read from the 11th digits to translate what point is set
+        answer = int(answer_in_hex, 16)
+        return answer
+
+    def set_mode(self, value):
         """ 0 setpoint source RS232
             3 valve close
-            4 freeze valuve out
+            4 freeze valve out
             8 valve fully open
-            20 valve steering (valve=setpoint)"""
-        cmd = [1, 1, 4, value & 0xFF]
-        self.write(cmd)
-        ret = self.read()
-        if len(ret) != 3:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 0 and ret[1] == 0 and ret[2] == 4:
-            return 0
-        else:
-            logger.debug("ELFLOW.set_setpoint() ret error ret=%s" % str(ret))
-            return -1
-
-    def get_control_mode(self):
-        cmd = [4, 1, 1, 1, 4]
-        self.write(cmd)
+            20 valve steering """
+        hex_val = hex(value)[2:]        # [2:] to remove the 0x from the beginning of the hex number
+        command = f"{self.CMDS['set_control_mode']}" + f"{hex_val.zfill(2)}"  # hex should have at least two digits
+        self._intf.write(command)
         ret = self.read()
-        if len(ret) != 4:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 2 and ret[1] == cmd[1] and ret[2] == cmd[2]:
-            return ret[3]
-        else:
-            logger.debug("ELFLOW.set_setpoint() ret error ret=%s" % str(ret))
-            return -1
-
-    def set_valve_output(self, value):
-        cmd = [1, 114, 0x41, (value >> 24) & 0xFF, (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF]
-        self.write(cmd)
-        ret = self.read()
-        if len(ret) != 3:
-            logger.debug("ELFLOW.set_valve_output() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 0 and ret[1] == 0 and ret[2] == 7:
-            return 0
-        else:
-            logger.debug("ELFLOW.set_valve_output() ret error ret=%s" % str(ret))
-            return -1
+        return ret
 
-    def get_valve_output(self):
-        cmd = [4, 114, 0x41, 114, 0x41]
-        self.write(cmd)
-        ret = self.read()
-        if len(ret) != 7:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 2 and ret[1] == cmd[1] and ret[2] == cmd[2]:
-            return ((ret[3] << 24) & 0xFF000000) | ((ret[4] << 16) & 0xFF0000) | ((ret[5] << 8) & 0xFF00) | (ret[6] & 0xFF)
-        else:
-            logger.debug("ELFLOW.get_valve_output() ret error ret=%s" % str(ret))
-            return -1
-
-    def set_controller_speed(self, value):
-        value = struct.unpack('<I', struct.pack('<f', value))[0]
-        cmd = [1, 114, 0x40 + 30, (value >> 24) & 0xFF, (value >> 16) & 0xFF, (value >> 8) & 0xFF, value & 0xFF]
-        self.write(cmd)
-        ret = self.read()
-        if len(ret) != 3:
-            logger.debug("ELFLOW.set_controller_speed() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 0 and ret[1] == 0 and ret[2] == 7:
-            return 0
-        else:
-            logger.debug("ELFLOW.set_controller_speed() ret error ret=%s" % str(ret))
-            return -1
-
-    def get_controller_speed(self):
-        cmd = [4, 114, 0x41, 114, 0x40 + 30]
-        self.write(cmd)
+    def get_mode(self):
+        self._intf.write(self.CMDS['get_control_mode'])
         ret = self.read()
-        if len(ret) != 7:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 2 and ret[1] == cmd[1] and ret[2] == cmd[2]:
-            return struct.unpack('!f', chr(ret[3]) + chr(ret[4]) + chr(ret[5]) + chr(ret[6]))[0]
-        else:
-            logger.debug("ELFLOW.get_valve_output() ret error ret=%s" % str(ret))
-            return -1
-
-    def get_measure(self):
-        cmd = [4, 1, 0x21, 1, 0x20]
-        self.write(cmd)
+        answer_in_hex = ret[11:]        # read from the 11th digits to translate what mode is on
+        answer = int(answer_in_hex, 16)
+        return answer
+
+    def get_flow(self):
+        """This should give the flow in l/min
+        """
+
+        # first get the max capacity in %
+        self._intf.write(self.CMDS['get_capacity'])
         ret = self.read()
-        if len(ret) != 5:
-            logger.debug("ELFLOW.set_setpoint() data lenth error ret=%s" % str(ret))
-            return -1
-        elif ret[0] == 2 and ret[1] == cmd[1] and ret[2] == cmd[2]:
-            return ((ret[3] << 8) & 0xFF00) | (ret[4] & 0xFF)
-        else:
-            logger.debug("ELFLOW.get_valve_output() ret error ret=%s" % str(ret))
-            return -1
+        answer_in_hex = ret[11:]  # read from the 11th digits to translate what the capacity is
+        cap_100 = struct.unpack('!f', bytes.fromhex(answer_in_hex))[0]
+
+        # now measure the flow
+        self._intf.write(self.CMDS['get_measure_flow'])
+        ret1 = self.read()
+        answer_in_hex = ret1[11:]
+        answer = int(answer_in_hex, 16)
+
+        val = answer / 32000 * cap_100
+        return val

examples/lab_devices/bronkhorsteflow.py

@@ -0,0 +1,30 @@
+#
+# ------------------------------------------------------------
+# Copyright (c) All rights reserved
+# SiLab, Institute of Physics, University of Bonn
+# ------------------------------------------------------------
+#
+
+''' This script shows how to use EFLOW
+'''
+
+from basil.dut import Dut
+
+dut = Dut('bronkhorstELFLOW_pyserial.yaml')
+dut.init()
+
+# setting set point
+dut["hot_n2"].set_mode(0)
+dut["hot_n2"].set_setpoint(10000)
+dut["hot_n2"].set_mode(0)
+print("setpoint", dut["hot_n2"].get_setpoint())
+
+# controlling valve
+dut["hot_n2"].set_mode(20)
+
+# measuring flow rate
+# print("Flow",dut["hot_n2"].get_flow())
+
+# Measuring of valve opening in %
+valve = dut["hot_n2"].get_valve_output()
+print("Valve opened in %", valve)