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,102 @@
 
 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
+    and here: https://www.bronkhorst.com/getmedia/257147fc-7f5d-4628-9a47-0533cf68ac08/917099-Manual-EL-FLOW-Select.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)}!")
+
+        if value > 32000:
+            raise ValueError(f"The valid range is 0 to 32000, the set value is {value}!. Setting setpoint to 32000")
+            value = 32000
+
+        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 = int(ret[11:], 16)  # read from the 11th digits to translate what point is set
+        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)
+            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) != 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)
-        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)
+    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 ((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)
+        answer = int(ret[11:], 16)        # read from the 11th digits to translate what mode is on
+        return answer
+
+    def get_flow(self):
+        """This should give the flow in l/min
+        """
+        # first get the max capacity in capacity unit
+        # the max capacity depends on e.g the fluid, which can be changed
+        # Capacity unit is a command, which returns the used unit in hex
+        self._intf.write(self.CMDS['get_capacity'])
         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)
-        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)
-        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
+        cap_100 = struct.unpack('!f', bytes.fromhex(ret[11:]))[0]  # read from the 11th digits to translate what the capacity is
+
+        # now measure the flow
+        self._intf.write(self.CMDS['get_measure_flow'])
+        ret1 = self.read()
+        answer = int(ret1[11:], 16)  # convert reply from hex to integer
+
+        val = answer / 32000 * cap_100
+        return val

basil/utils/USBBinds.py

@@ -37,7 +37,7 @@ def query_identification(rm, resource, baud_rate, read_termination=None, write_t
 def find_usb_binds(rm, log,
                    instruments,
                    binds_to_skip=[],
-                   memorized_binds=[],
+                   memorized_binds={},
                    timeout=1000 * 4
                    ):
     """
@@ -54,8 +54,8 @@ def find_usb_binds(rm, log,
             - 'identification': The identification string for the instrument.
         binds_to_skip (list, optional): List of binds to skip during the search.
             Defaults to an empty list.
-        memorized_binds (list, optional): List of memorized binds.
-            Defaults to an empty list.
+        memorized_binds (dict, optional): Dictionary of memorized binds.
+            Defaults to an empty dictionary.
         timeout (int, optional): Timeout value in milliseconds.
             Defaults to 4000.
 
@@ -97,29 +97,34 @@ def find_usb_binds(rm, log,
             try:
                 log.debug(f"Trying {res} with baud rate {instrument['baud_rate']}")
 
-                if any(res in bind for bind in memorized_binds):
+                if memorized_binds.get(res):
                     log.debug(f"Found memorized bind {res}")
                     result = memorized_binds[res]
                 else:
                     result = query_identification(rm, res, instrument['baud_rate'], instrument['read_termination'], instrument['write_termination'], timeout=timeout)
 
-                    memorized_binds.append({res, result})
-
+                    memorized_binds[res] = result
                     log.debug(f"Found {result.strip()}")
 
-                if result.lower().strip() in [inst["identification"].lower().strip() for inst in instruments]:
-                    substring = res.split("/")[2].split("::")[0]
+                for inst in instruments:
+                    if result.lower().strip() in inst["identification"].lower().strip():
+                        substring = res.split("/")[2].split("::")[0]
 
-                    log.info(f"Matched instrument {instrument['identification']} to /dev/{str(substring)}")
-                    skip_binds.append(f"/dev/{str(substring)}")
+                        log.info(f"Matched instrument {inst['identification']} to /dev/{str(substring)}")
+                        skip_binds.append(f"/dev/{str(substring)}")
 
-                    results[result.lower().strip()] = f"/dev/{str(substring)}"
+                        results[result.lower().strip()] = f"/dev/{str(substring)}"
 
-                    if len(results) == len(instruments):
-                        return results
+                        if len(results) == len(instruments):
+                            return results
 
-                    log.debug(f"Found {len(results)} out of {len(instruments)}")
+                        log.debug(f"Found {len(results)} out of {len(instruments)}")
+
+                        break
+                else:
+                    continue
 
+                if inst["identification"].lower().strip() in instrument["identification"].lower().strip():
                     break
 
             except pyvisa.VisaIOError:

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)