# -*- coding: utf-8 -*-
|
__author__ = 'wangzhibo 2022.01.20'
|
|
import serial
|
import serial.tools.list_ports
|
|
|
class DustCleaner():
|
# 初始化
|
def __init__(self, com, bps, timeout):
|
self.port = com
|
self.bps = bps
|
self.timeout = timeout
|
self.com_port = None
|
self.Open()
|
|
# 打印设备基本信息
|
def Print_Name(self):
|
print(self.com_port.name) # 设备名字
|
print(self.com_port.port) # 读或者写端口
|
print(self.com_port.baudrate) # 波特率
|
print(self.com_port.bytesize) # 字节大小
|
print(self.com_port.parity) # 校验位
|
print(self.com_port.stopbits) # 停止位
|
print(self.com_port.timeout) # 读超时设置
|
print(self.com_port.writeTimeout) # 写超时
|
print(self.com_port.xonxoff) # 软件流控
|
print(self.com_port.rtscts) # 软件流控
|
print(self.com_port.dsrdtr) # 硬件流控
|
print(self.com_port.interCharTimeout) # 字符间隔超时
|
|
# 打开串口
|
def Open(self) -> bool:
|
try:
|
# 打开串口，并得到串口对象
|
self.com_port = serial.Serial(self.port, self.bps, timeout=self.timeout)
|
# 判断是否打开成功
|
if (self.com_port.is_open):
|
return True
|
else:
|
return False
|
except Exception as _:
|
return False
|
|
# 关闭串口
|
def Close(self) -> None:
|
self.com_port.close()
|
print(self.com_port.is_open) # 检验串口是否打开
|
|
# 打印可用串口列表
|
@classmethod
|
def Print_Used_Com():
|
port_list = list(serial.tools.list_ports.comports())
|
print(port_list)
|
|
# 发数据
|
def Send(self, data):
|
if (not self.com_port is None) and self.com_port.is_open():
|
self.com_port.write(data)
|
self.Recive()
|
|
def SendNoChecksum(self, data):
|
checksum = 0
|
array = bytearray(data)
|
for i in array[:-1]:
|
checksum += i
|
checksum = checksum % 256
|
array[-1] = checksum
|
self.Send(bytes(array))
|
|
def Send_onetime_clean_self(self): # 发送函数
|
self.Send(bytes([0x5A, 0x04, 0x70, 0xCE])) # 用write函数向串口发送数据
|
|
def Send_onetime_clean_camera(self): # 发送函数
|
self.Send(bytes([0x5A, 0x04, 0x71, 0xCF])) # 用write函数向串口发送数据
|
|
def Send_close_period_clean_self(self):
|
self.Send(bytes([0x5A, 0x06, 0x74, 0x00, 0x00, 0xD4])) # close self period clean
|
|
def Send_close_period_clean_camera(self):
|
self.Send(bytes([0x5A, 0x06, 0x75, 0x00, 0x00, 0xD5])) # close self period clean
|
|
def Send_period_clean_self(self, period: int):
|
minute_H = period // 256
|
minute_L = period % 256
|
self.SendNoChecksum(bytes([0x5A, 0x06, 0x74, minute_H, minute_L, 0x00]))
|
|
def Send_period_clean_camera(self, period: int):
|
minute_H = period // 256
|
minute_L = period % 256
|
self.SendNoChecksum(bytes([0x5A, 0x06, 0x75, minute_H, minute_L, 0x00]))
|
|
def Send_self_clean_params(self, start_angle: int, end_angle: int, num: int):
|
self.SendNoChecksum(bytes([0x5A, 0x07, 0x72, start_angle, end_angle, num, 0x00]))
|
|
def Send_camera_clean_params(self, start_angle: int, end_angle: int, num: int):
|
self.SendNoChecksum(bytes([0x5A, 0x07, 0x73, start_angle, end_angle, num, 0x00]))
|
|
def Recive(self):
|
return self.main_engine.read_all()
|
|
|
if __name__ == '__main__':
|
# DustCleaner.Print_Used_Com()
|
cleaner = DustCleaner("com11", 115200, 1)
|
if cleaner.Open():
|
cleaner.Send_close_period_clean_camera()
|
cleaner.Recive()
|
