<jupyter_start><jupyter_text>Auto Sampler script Elveflow Instruments* OB1 pressure controller* MUX Wire*   Import Libraries<jupyter_code>%reset -f

# Import Elveflow library
import sys
import os
from email.header import UTF8
# SDK_HOME = 'C:/Users/Admin/ELVEFLOW/SDK_V3_04_04'
SDK_HOME = 'C:/Users/Admin/ELVEFLOW/ESI_V3_04_01/SDK V3_04_01'
sys.path.append(SDK_HOME+'/DLL64/DLL64')#add the path of the library here
sys.path.append(SDK_HOME+'/Python_64')#add the path of the LoadElveflow.py

from ctypes import *

from array import array

from Elveflow64 import *<jupyter_output><empty_output><jupyter_text>Instrument Class definitions<jupyter_code>Z_regulator_type = {
    'none': 0,
    '0_200_mbar': 1,
    '0_2000_mbar': 2,    
    '0_8000_mbar': 3,
    'm1000_1000_mbar': 4,
    'm1000_6000_mbar': 5
}
Z_sensor_type = {
    'none': 0,
    'Flow_1_5_uL_min': 1,
    'Flow_7_uL_min': 2,
    'Flow_50_uL_min': 3,
    'Flow_80_uL_min': 4,
    'Flow_1000_uL_min': 5,
    'Flow_5000_uL_min': 6,
    'Press_340_mbar': 7,
    'Press_1_bar': 8,
    'Press_2_bar': 9,
    'Press_7_bar': 10,
    'Press_16_bar': 11,
    'Level': 12
}

def _check_error(task, error):
    err_dict = {
        -8000: 'No Digital Sensor found',
        -8001: 'No pressure sensor compatible with OB1 MK3',
        -8002: 'No Digital pressure sensor compatible with OB1 MK3+',
        -8003: 'No Digital Flow sensor compatible with OB1 MK3',
        -8004: 'No IPA config for this sensor',
        -8005: 'Sensor not compatible with AF1',
        -8006: 'No Instrument with selected ID'
    }
    if error != 0:
        if error in err_dict:
            raise RuntimeError('{} failed with error code {}\n{}'.format(task, error, err_dict[error]))
        else:
            raise RuntimeError('{} failed with error code {}\n{}'.format(task, error, 'Unknown error, see http://www.ni.com/pdf/manuals/321551a.pdf'))
            
class ob1():
    def __init__(self, address='01EF64C3', calibrate=False):
        self.Instr_ID = c_int32()
        print('Instrument name and regulator types hardcoded in the python script'.encode('utf-8'))
        # see User guide to determine regulator type NI MAX to determine the instrument name
        # channel 1: -1000~1000 mBar, channel 2: none, channel 3: none, channel 4:none
        error = OB1_Initialization(address.encode('ascii'), 
                                   Z_regulator_type['m1000_1000_mbar'], 
                                   Z_regulator_type['none'], 
                                   Z_regulator_type['none'], 
                                   Z_regulator_type['none'], 
                                   byref(self.Instr_ID))
        # all functions will return error code to help you to debug your code, for further information see user guide
        _check_error('OB1 Initialization', error)

        # add one digital flow sensor to OB1 channel 1, 1mL/min, digital, water calibration, 16bit resolution, 
        error=OB1_Add_Sens(self.Instr_ID, 1, Z_sensor_type['Flow_1000_uL_min'], 1, 0, 7, 0)
        _check_error('Adding digital flow sensor', error)

        self.calib_path = os.path.abspath('ob1_calibration.txt')
        print(self.calib_path)
        self.Calib = (c_double * 1000)()
        if calibrate:
            print ('Starting calibration')
            OB1_Calib(self.Instr_ID.value, self.Calib, 1000)
            error = Elveflow_Calibration_Save(self.calib_path.encode('ascii'), byref(self.Calib), 1000)
            print ('Calibration finished')
            print ('Calibration saved in file %s' % self.calib_path.encode('ascii'))
        else:
            if not os.path.isfile(self.calib_path):
                raise IOError('Calibration file "{}" does not exist'.format(self.calib_path))               
            error = Elveflow_Calibration_Load(self.calib_path.encode('ascii'), byref(self.Calib), 1000)
            _check_error('Loading calibration file', error)            
            
        self.set_pressure(0)

    def meas_flowrate(self, channel=1):
        """
        Measures the instantaneous flow rate, on designated port.
        Parameters:
            channel: ob1 channel to set (1-4), defaults to 1
        return: (flowrate, error code)
            flowrate in uL/min
            0 : successful
            -1 : parameter error
            otherwise: NI error
        """
        if channel <1 or channel > 4:
            print('ob1.meas_flowrate: channel must be within 1~4')
            return -1
        
        set_channel = c_int32(channel)  # convert to c_int32
        get_flowrate = c_double(0.0)
        error = OB1_Get_Sens_Data(self.Instr_ID.value, set_channel, 1, byref(get_flowrate))  # Acquire_data =1 -> Read all the analog value
        _check_error('Getting data from flow sensor', error)
        
        return get_flowrate.value, error

    def set_pressure(self, pressure, channel=1):
        """
        Sets the pressure, on designated port.
        Parameters:
            pressure: target pressure in mBar (-1000 to 8000)
            channel: ob1 channel to set (1-4), defaults to 1
        return: error code
            0 : successful
            -1 : parameter error
            otherwise: NI error
        """      
        
        # check parameters
        if pressure >8000.0 or pressure < -1000.0:
            print('ob1.set_pressure: Error pressure must be within -1000~8000')
            return -1
        if channel <1 or channel > 4:
            print('set_pressure: channel must be within 1~4')
            return -1
        
        set_channel=c_int32(channel)#convert to c_int32        
        set_pressure=c_double(pressure) #convert to c_double        
        error=OB1_Set_Press(self.Instr_ID.value, set_channel, set_pressure, byref(self.Calib),1000) 
        _check_error('Setting pressure', error)
        
        return error
    
    def meas_pressure(self, channel=1):
        """
        Measures the instantaneous pressure, on designated channel.
        Parameters:
            channel: ob1 channel to set (1-4), defaults to 1
        return: (pressure, error code)
            flowrate in mBar
            0 : successful
            -1 : parameter error
            otherwise: NI error
        """
        if channel <1 or channel > 4:
            print('ob1.meas_pressure: channel must be within 1~4')
            return -1
        
        set_channel = c_int32(channel)  # convert to c_int32
        get_pressure = c_double(0.0)
        error = OB1_Get_Press(self.Instr_ID.value, set_channel, Acquire_Data1True0False=1, Calib_array_in=self.Calib, Pressure=byref(get_pressure), Calib_Array_len=1000)  # Acquire_data =1 -> Read all the analog value
        _check_error('Getting pressure from OB1', error)
        
        return get_pressure.value, error
    
    def __del__(self):
        self.set_pressure(0)
        error=OB1_Destructor(self.Instr_ID.value)
        
class mux_wire():
    def __init__(self, address='Dev2'):
        self.Instr_ID = c_int32()
        # see User guide to determine regulator type NI MAX to determine the instrument name
        error = MUX_Initialization(address.encode('ascii'), byref(self.Instr_ID))
        # all functions will return error code to help you to debug your code, for further information see user guide
        _check_error('MUX Initialization', error)
        print(error)
        
        self.state = [0]*8
        
    def set_all_valves(self, state):
        """
        Sets state of all valves
        Parameters:
            state: 8 integer (0: close, 1: open)
        return: error code
            0 : successful
            -1 : parameter error
            otherwise: NI error
        """
        if not (isinstance(state, list) and len(state) == 8):
            print('Error mux_wire.set_all_valves: state has to be a list of 8 integers')
            return -1
        
        self.state = state
        valve_state=(c_int32*16)(0)
        for i in range(8):
            valve_state[i]=c_int32(state[i]) 
            
        error=MUX_Wire_Set_all_valves(self.Instr_ID.value, valve_state, 16)
        
        return error     
    
    def set_valve(self, channel, state):
        """
        Sets state of a certain valve
        Parameters:
            channel: integer (1-8)
            state: integer (0: close to N.O., 1: open to N.C.)
        return: error code
            0 : successful
            -1 : parameter error
            otherwise: NI error
        """
        if not isinstance(state, int):
            print('Error mux_wire.set_valve: state has to be an integer')
            return -1
        if not (isinstance(channel, int) and channel>0 and channel<9) :
            print('Error mux_wire.set_valve: channel has to be integer (1-8)')
            return -1
                
        self.state[channel-1] = state
        valve_state=(c_int32*16)(0)
        for i in range(8):
            valve_state[i]=c_int32(self.state[i]) 
            
        error=MUX_Wire_Set_all_valves(self.Instr_ID.value, valve_state, 16)
        
    def get_state(self):
        """
        Returns state of current valves: list of 8 integers
        """
        return self.state
            
    def __del__(self):
        error=MUX_Destructor(self.Instr_ID.value)
        
class msr():
    def __init__(self, address='01EE3A2D'):
        self.Instr_ID = c_int32()
        # see User guide to determine regulator type NI MAX to determine theerror=M_S_R_D_Initialization('01DAA568'.encode('ascii'),5,0,0,0,0,0,byref(Instr_ID)) 
        # channel 1: Z_sensor _type_Flow_5000_muL_min, channel 2: none, channel 3: none, channel 4:none
        error = M_S_R_D_Initialization(address.encode('ascii'),6,0,0,0,0,0,byref(Instr_ID))
        # all functions will return error code to help you to debug your code, for further information see user guide
        _check_error('MUX Initialization', error)
        print(error)       
            
    def __del__(self):
        error=M_S_R_D_Destructor(self.Instr_ID.value)
        
class sampler():
    def __init__(self):
        """
        Initialization
        """
        self.ob1 = ob1(calibrate=False)
        self.mux_wire = mux_wire()
        
    def set_valves(self, reservoir, target):
        """
        Sets valve configuration of sampling system
        """
        state = self.mux_wire.get_state()
        if reservoir=='air':
            print('sampler.set_valves: Set air reservoir')
            
        if target=='waste':
            state[0] = 0
        elif target == 'tflask':
            state[0] = 1
            
        error = self.mux_wire.set_all_valves(state)
        
        return error
    
    def pushpull(self, mode, target, pressure = -10):
        """
        Push or pulls a certain volume or for some time using OB1 and connected flow sensor        
        Parameters:
            mode: integer (0: time mode, 1: volume mode)
            target: float (in seconds and uL)
            pressure: float (pressure in mbar)
        return: timestamp, volume
        """
        from time import time, sleep
        timeout = 30 # seconds
        time_step = 0.5
        
        if mode >1 or mode <0:
            print('Error sampler.pull: mode has to be integer 0 or 1')
            return -1
        if target < 0 :
            print('Error sampler.pull: target has to be a positive float')
            return -1
        vol_data = []
        time_data = []
        cur_vol = 0.0
        
        
        self.ob1.set_pressure(pressure)
        start_time = time()
        cur_time = time()
        if mode == 0: # timer mode            
            while cur_time-start_time < target:                
                flowrate,error = self.ob1.meas_flowrate()
                
                next_time = time()
                cur_vol = cur_vol + flowrate/60*(next_time-cur_time)
                cur_time = time()
                
                vol_data.append(cur_vol)
                time_data.append(cur_time)
                print('{} uL/min, cumulative {} uL at {} sec'.format(flowrate, cur_vol, cur_time-start_time))
                
                sleep(time_step)
                
        elif mode == 1:   # volume mode
            if pressure <0:
                pol = -1.0
            else:
                pol = 1.0
            while cur_time-start_time < timeout and pol*cur_vol < target:
                flowrate,error = self.ob1.meas_flowrate()
                
                next_time = time()
                cur_vol = cur_vol + flowrate/60*(next_time-cur_time)
                cur_time = time()
                
                vol_data.append(cur_vol)
                time_data.append(cur_time)
                print('{} uL/min, cumulative {} uL at {} sec'.format(flowrate, cur_vol, cur_time-start_time))
                
                sleep(time_step)
                
        self.ob1.set_pressure(0)
        
        return time_data, vol_data
    
    def __del__(self):
        
        self.set_valves('air', 'waste')
        self.ob1.set_pressure(0)
        
poe = sampler()

poe.set_valves('air', 'waste')
poe.ob1.set_pressure(5)

poe.set_valves('air', 'waste')
poe.set_valves('air', 'tflask')
poe.ob1.set_pressure(100)
poe.ob1.set_pressure(5)
poe.ob1.meas_flowrate()
import matplotlib.pyplot as plt
poe.set_valves('air', 'tflask')
time,vol = poe.pushpull(mode=1, target=30, pressure=-20)
poe.set_valves('air', 'waste')
time_rel = [t-time[0] for t in time]
plt.plot(time_rel, vol)
plt.xlabel('Time [sec]')
plt.ylabel('Volume [uL]')
plt.show()
del poe
# kenobi=ob1(calibrate=True)
kenobi=ob1(calibrate=False)
p,e = kenobi.meas_pressure()

print(p)
kenobi.set_pressure(-50)
kenobi.meas_flowrate()
mux = mux_wire()
state = [1, 0, 0, 0, 0, 0, 0, 0]
mux.set_all_valves(state)
isinstance(state, list)
print(os.path.expanduser(r'./ob1_calibration.txt'))
print(os.getcwd())
print(os.path.relpath('ob1_calibration.txt'))
print(os.path.abspath('ob1_calibration.txt'))
del kenobi
del mux
print([0]*8)<jupyter_output>[0, 0, 0, 0, 0, 0, 0, 0]
{'fraction_non_alphanumeric': 0.10128022759601707, 'fraction_numerical': 0.033357041251778095, 'mean_word_length': 2.0435064935064937, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '525903', 'license': 'MIT', 'n_tokens_mistral': 4712, 'n_tokens_neox': 4314, 'n_words': 1120, 'path': 'control_scripts/sampler.ipynb', 'repo_name': 'jhwnkim/fluid-auto-sampler', 'size': '14060'}