Master_Project/XP3000.py at master · Kaffejunge/Master_Project · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import serial
import time


class XP3000:

    term_char = chr(13).encode()

    def __init__(self, _port):
        # pump settings
        self.plunger_pos = 0

        # serial settings
        self._port = _port
        self._baudrate = 9600
        # !try ser = serial.rs485.RS485
        self.ser = serial.Serial(
            port=self._port, baudrate=self._baudrate, timeout=5)

        # open the connection
        if(self.ser.isOpen()):
            self.ser.close()
        self.ser.open()
        self.ser.reset_input_buffer()

        # initializing the pump
        self.ser.write(b'/1ZR'+XP3000.term_char)
        self.wait_for_ready()
        print(f'Opened successful connection to XP3000 at port: {self._port}')

    def is_open(self):
        return self.ser.isOpen()

    def flush_buffers(self):
        self.ser.reset_input_buffer()
        self.ser.reset_output_buffer()

    def report(self):
        self.ser.write(b'/1?R' + XP3000.term_char)
        self.plunger_pos = self.ser.readline()
        print(f'Absolute position of plunger: {self.plunger_pos}')

        self.ser.write(b'/1?1R' + XP3000.term_char)
        start_vel = self.ser.readline()
        print(f'Start velocity of plunger: {start_vel}')

        self.ser.write(b'/1?2R' + XP3000.term_char)
        top_vel = self.ser.readline()
        print(f'Top velocity of plunger {top_vel}')

        self.ser.write(b'/1?3R' + XP3000.term_char)
        cutoff_vel = self.ser.readline()
        print(f'Cutoff velocity of plunger: {cutoff_vel}')

        self.ser.write(b'/1?4R' + XP3000.term_char)
        actual_pos = self.ser.readline()
        print(f'Actual position of plunger: {actual_pos}')

        self.ser.write(b'/1?12R' + XP3000.term_char)
        backlash_steps = self.ser.readline()
        print(f'Backlash steps: {backlash_steps}')

        self.ser.write(b'/1?22R' + XP3000.term_char)
        fluid_val = self.ser.readline()
        print(f'Fluid Sensor value (0 = wet): {fluid_val}')

        self.ser.write(b'/1&R' + XP3000.term_char)
        firmware = self.ser.readline()
        print(f'Firmware: {firmware}')

    def wait_for_ready(self):
        # ser.in_waiting() / ser.out_waiting() checks how many bytes at port
        # !check if ser.read_until("'", size=?) works.
        self.flush_buffers()
        string_to_send = b'/1QR' + XP3000.term_char
        ready = False
        while(not ready):
            self.ser.write(string_to_send)
            time.sleep(1)
            response = self.ser.readline()
            ready = "`" in str(response)
            # print(response, ready)

    def get_syr_vol(self, mL, pos, speed=16):
        # set plunger so syringe to contain set numer of mL
        muL = int(mL * 1000)
        self.plunger_pos = str(int(muL * (3000/5000))).encode()
        solvents = ['acetone', 'h2o', 'rct_slv', 'wash_slv']
        if(pos in solvents):
            string_to_send = b'/1S' + \
                str(speed).encode() + b'IA' + \
                self.plunger_pos + b'R' + XP3000.term_char
        else:
            string_to_send = b'/1S' + \
                str(speed).encode() + b'OA' + \
                self.plunger_pos + b'R' + XP3000.term_char
        # resulting string /1<speed><valve_position><set_plunger_pos>R
        self.ser.write(string_to_send)
        time.sleep(1)

        self.wait_for_ready()

    def empty_syr(self, speed=14):
        string_to_send = b'/1OS' + \
            str(speed).encode() + b'A0R' + XP3000.term_char
        self.ser.write(string_to_send)
        self.wait_for_ready()

    def terminate_movement(self):
        self.ser.write(b'/1T' + XP3000.term_char)
        self.ser.write(b'/1?R' + XP3000.term_char)
        self.wait_for_ready()

    def send_string(self, string_to_send):
        string_to_send = string_to_send.encode() + XP3000.term_char
        self.ser.write(string_to_send)
        time.sleep(0.5)
        # !notice the lack of wait_for_ready()

    def read_line(self):
        print(self.ser.readline())

    def set_speed(self, speed):
        # enter number between 1-40 (1 = fastes)
        string_to_send = b'/1S' + str(speed).encode() + b'R' + XP3000.term_char
        self.ser.write(string_to_send)
        self.wait_for_ready()
        print(f'Speed was set to {speed}.')

    def close(self):
        self.ser.close()
        print(f'Closed the connection to XP3000 at port: {self._port}')