-
Notifications
You must be signed in to change notification settings - Fork 179
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* modified get serial number method for pass test device id * adjust humidity and temperature threthold * modified test method for pressure, adjust pressure threthold * test 5ul in test droplets * check jaws aligned while doing tip motors assembly
- Loading branch information
1 parent
ff46b4d
commit b2a8545
Showing
6 changed files
with
436 additions
and
47 deletions.
There are no files selected for viewing
139 changes: 139 additions & 0 deletions
139
hardware-testing/hardware_testing/drivers/sealed_pressure_fixture.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,139 @@ | ||
| # encoding:utf-8 | ||
|
|
||
| import time | ||
| from typing import Union | ||
| import serial | ||
| import serial.tools.list_ports | ||
|
|
||
| ReceiveBuffer = 100 | ||
|
|
||
|
|
||
| class SerialDriver: | ||
|
|
||
| @classmethod | ||
| def get_com_list(cls): | ||
| port_list = serial.tools.list_ports.comports() | ||
| return port_list | ||
|
|
||
| def __init__(self): | ||
| self.device = None | ||
| self.com = None | ||
|
|
||
| def get_device(self): | ||
| """ | ||
| select device | ||
| :return: | ||
| """ | ||
| port_list = SerialDriver.get_com_list() | ||
| print("=" * 5 + "PORT LIST" + "=" * 5) | ||
| for index, p in enumerate(port_list): | ||
| print(f"{index + 1} >>{p.device}") | ||
| select = input("Select Port Number(输入串口号对应的数字):") | ||
| self.device = port_list[int(select.strip()) - 1].device | ||
|
|
||
| def init_serial(self, baud): | ||
|
|
||
| """ | ||
| init connection | ||
| :param baud: | ||
| :return: | ||
| """ | ||
| self.com = serial.Serial(self.device, baud, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, | ||
| bytesize=serial.EIGHTBITS, timeout=1) | ||
| if self.com.isOpen(): | ||
| print(f"{self.device} Opened! \n") | ||
| # settings | ||
| self.com.bytesize = serial.EIGHTBITS # 数据位 8 | ||
| self.com.parity = serial.PARITY_NONE # 无校验 | ||
| self.com.stopbits = serial.STOPBITS_ONE # 停止位 1 | ||
|
|
||
| def close(self): | ||
| """ | ||
| close com | ||
| :return: | ||
| """ | ||
| self.com.close() | ||
| print(f"{self.device} Closed! \n") | ||
|
|
||
| def init(self, baud): | ||
| """ | ||
| main | ||
| :return: | ||
| """ | ||
| self.get_device() | ||
| try: | ||
| self.init_serial(baud) | ||
| except: | ||
| print("Can't find device") | ||
|
|
||
| def write_and_get_buffer(self, send: Union[str, int, bytes], only_write=False, delay=None, times=30): | ||
| """ | ||
| send cmd | ||
| :return: | ||
| """ | ||
| if self.com is None: | ||
| return | ||
| if type(send) is not bytes: | ||
| send = (send + "\r\n").encode('utf-8') | ||
| self.com.flushInput() | ||
| self.com.flushOutput() | ||
| self.com.write(send) | ||
| time.sleep(0.1) | ||
| if delay is None: | ||
| pass | ||
| else: | ||
| time.sleep(delay) | ||
| if only_write is True: | ||
| return | ||
| for i in range(times): | ||
| data = self.com.read(ReceiveBuffer) | ||
| if type(data) is not bytes: | ||
| if "OK" not in data.decode('utf-8') or "busy" in data.decode('utf-8'): | ||
| time.sleep(1) | ||
| continue | ||
| else: | ||
| return data | ||
| return data.decode('utf-8') | ||
|
|
||
| def read_buffer(self): | ||
| """ | ||
| 读取缓存 | ||
| :return: | ||
| """ | ||
| try: | ||
| self.com.flushInput() | ||
| self.com.flushOutput() | ||
| except: | ||
| pass | ||
| time.sleep(3) | ||
| length = ReceiveBuffer | ||
| data = self.com.read(length) | ||
| self.com.flushInput() | ||
| self.com.flushOutput() | ||
| return data.decode('utf-8') | ||
|
|
||
| def get_pressure(self): | ||
| """ | ||
| analyze pressure value | ||
| """ | ||
| for _i in range(5): | ||
| try: | ||
| respond = self.read_buffer() | ||
| respond_list = respond.split('|') | ||
| respond_value = respond_list[1] | ||
|
|
||
| average_value = respond_value.split('\r\n')[0].split('\t')[1].strip() | ||
| average_value = float(average_value) | ||
| return average_value | ||
| except: | ||
| print(f"get pressure fail at {_i} times") | ||
| pass | ||
|
|
||
|
|
||
|
|
||
| if __name__ == '__main__': | ||
| s = SerialDriver() | ||
| s.init(9600) | ||
| for i in range(100): | ||
| result = s.get_pressure() | ||
| print(result) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.