843 lines
30 KiB
Python
843 lines
30 KiB
Python
import os
|
|
import sys
|
|
import datetime
|
|
from datetime import datetime, timedelta
|
|
import prottcp as prot
|
|
from PyQt5.QtCore import *
|
|
from PyQt5.QtGui import *
|
|
from PyQt5.QtWidgets import *
|
|
import threading
|
|
import time
|
|
import serial
|
|
import serial.tools.list_ports
|
|
import coder_params as cpars
|
|
import coder_check as chk
|
|
|
|
|
|
|
|
|
|
# 批检仪赋码上位机
|
|
|
|
|
|
STR_RED="\033[1;31m"
|
|
STR_BLUE="\033[1;34m"
|
|
STR_END="\033[0m"
|
|
|
|
|
|
|
|
# 获取北京时间
|
|
def get_date():
|
|
now_time = datetime.utcnow()
|
|
utc_time = now_time + timedelta(hours=8) # UTC只是比北京时间提前了8个小时
|
|
utc_time = utc_time.strftime("%Y%m%d")
|
|
return utc_time
|
|
|
|
# 获取北京时间
|
|
def get_time():
|
|
now_time = datetime.utcnow()
|
|
utc_time = now_time + timedelta(hours=8) # UTC只是比北京时间提前了8个小时
|
|
utc_time = utc_time.strftime("%Y-%m-%d %H:%M:%S")
|
|
return utc_time
|
|
|
|
_QSS="""
|
|
QListView {
|
|
border: 1px solid white;
|
|
border-radius: 3px;
|
|
show-decoration-selected: 1;
|
|
}
|
|
|
|
QWidget {
|
|
font-size: 16px;
|
|
}
|
|
|
|
"""
|
|
|
|
|
|
|
|
class coder(QObject):
|
|
|
|
def __init__(self) -> None:
|
|
QObject.__init__(self)
|
|
self.app = QApplication(sys.argv)
|
|
self.code_id=0
|
|
self.run_times=0
|
|
self.slave_num=10
|
|
self.code_list_lock=[]
|
|
self.save_file="code_list.csv"
|
|
self.ser_is_open = False
|
|
self.recv_handler=None
|
|
self.autoinc_id=False
|
|
self.openignore_flag=True
|
|
self.autotest_is_open=0
|
|
self.uid_length=16
|
|
self.code_list_backcolor=False
|
|
self.ser=prot.protu()
|
|
self.widget = QWidget()
|
|
self.widget.setStyleSheet (_QSS)
|
|
self.widget.resize(1500, 900)
|
|
self.widget.setWindowTitle("批检仪赋码工具")
|
|
self.com_but_init()
|
|
self.combsp_init()
|
|
self.com_init()
|
|
self.code_list_init()
|
|
self.dot_fat_init()
|
|
self.channel_init()
|
|
self.sig_list_init()
|
|
self.moterup_init()
|
|
self.infotext_init()
|
|
self.autoinc_init()
|
|
self.check_but_init()
|
|
self.posend_but_init()
|
|
self.autotest_but_init()
|
|
self.stop_but_init()
|
|
self.code_but_init()
|
|
self.moterdown_init()
|
|
self.openignore_init()
|
|
self.recv_handler_table_init()
|
|
self.device_type_init()
|
|
self.writebuff_init()
|
|
self.widget.destroyed.connect(self.quit)
|
|
|
|
def quit(self):
|
|
self.close_serial()
|
|
print("app exit.")
|
|
# 程序退出
|
|
sys.exit(1)
|
|
|
|
# 初始化
|
|
def init(self,detonator_fat:bytearray,chip_fat:bytearray,signature_code:bytearray,slave_num:int):
|
|
self.set_params(detonator_fat,chip_fat,signature_code,slave_num)
|
|
date=get_date().encode('utf-8')
|
|
self.date=date[-5:]
|
|
self.year=date[:4]
|
|
# self.date=b'91114'
|
|
# self.year=b'2023'
|
|
self.save_file=date.decode("utf-8")+".csv"
|
|
if(os.path.exists(self.save_file)):
|
|
with open(self.save_file) as f:
|
|
lines=f.readlines()
|
|
length=len(lines)
|
|
self.code_id=length
|
|
for i in lines:
|
|
self.code_list_add(i)
|
|
print("code id start as:",self.code_id)
|
|
self.code_list_to_bottom()
|
|
|
|
# 设置雷管厂，芯片厂，特征码
|
|
def set_params(self,detonator_fat:bytearray,chip_fat:bytearray,signature_code:bytearray,slave_num:int):
|
|
self.detonator_fat=detonator_fat
|
|
self.chip_fat=chip_fat
|
|
self.slave_num=slave_num
|
|
self.signature_code=signature_code
|
|
info="已更换编码参数，现在的编码参数为："+"雷管厂:"+detonator_fat.decode("utf-8")+",芯片厂:"+chip_fat.decode("utf-8")
|
|
info+=",特征码:"+signature_code.decode("utf-8")+",通道数:"+str(slave_num)
|
|
self.set_infotext(info)
|
|
|
|
# 选择波特率
|
|
def combsp_init(self):
|
|
self.combsp = QComboBox(self.widget)
|
|
self.combsp.setObjectName(u"combsp")
|
|
self.combsp.setGeometry(QRect(465, 10, 85, 25))
|
|
self.combsp.setEditable(True)
|
|
self.combsp.currentIndexChanged.connect(self.com_changed)
|
|
for i in cpars.uartbsp_list():
|
|
self.combsp.addItem(i)
|
|
self.combsp_label = QLabel(self.widget)
|
|
self.combsp_label.setObjectName(u"label")
|
|
self.combsp_label.setGeometry(QRect(405, 16, 72, 15))
|
|
self.combsp_label.setText("波特率:")
|
|
|
|
# 选择雷管厂家
|
|
def dot_fat_init(self):
|
|
self.dot_fat = QComboBox(self.widget)
|
|
self.dot_fat.setObjectName(u"dot_fat")
|
|
self.dot_fat.setGeometry(QRect(645, 10, 400, 25))
|
|
self.dot_fat.setEditable(False)
|
|
self.dot_fat.currentIndexChanged.connect(self.params_changed)
|
|
for i in cpars.dot_fat_list():
|
|
self.dot_fat.addItem(i)
|
|
self.dot_fat_label = QLabel(self.widget)
|
|
self.dot_fat_label.setObjectName(u"label")
|
|
self.dot_fat_label.setGeometry(QRect(570, 16, 72, 15))
|
|
self.dot_fat_label.setText("雷管厂家:")
|
|
|
|
# com口
|
|
def com_init(self):
|
|
self.com = QComboBox(self.widget)
|
|
self.com.setObjectName(u"com")
|
|
self.com.setGeometry(QRect(95, 10, 300, 25))
|
|
self.com.setEditable(True)
|
|
self.com.currentIndexChanged.connect(self.com_changed)
|
|
# self.com.addItem("utcp:7777")
|
|
ports_list = list(serial.tools.list_ports.comports())
|
|
for comport in ports_list:
|
|
# print(comport.name,comport.description)
|
|
self.com.addItem(comport.name+":"+comport.description)
|
|
self.com_label = QLabel(self.widget)
|
|
self.com_label.setObjectName(u"label")
|
|
self.com_label.setGeometry(QRect(30, 16, 72, 15))
|
|
self.com_label.setText("COM口:")
|
|
|
|
# 特征码
|
|
def sig_list_init(self):
|
|
self.sig_list = QComboBox(self.widget)
|
|
self.sig_list.setObjectName(u"sig_list")
|
|
self.sig_list.setGeometry(QRect(1285, 10, 85, 25))
|
|
self.sig_list.setEditable(False)
|
|
self.sig_list.currentIndexChanged.connect(self.params_changed)
|
|
for i in cpars.code_signature_code_list():
|
|
self.sig_list.addItem(i)
|
|
self.sig_list_label = QLabel(self.widget)
|
|
self.sig_list_label.setObjectName(u"label")
|
|
self.sig_list_label.setGeometry(QRect(1220, 16, 72, 15))
|
|
self.sig_list_label.setText("特征码:")
|
|
|
|
# 通道数
|
|
def channel_init(self):
|
|
self.channel = QComboBox(self.widget)
|
|
self.channel.setObjectName(u"channel")
|
|
self.channel.setGeometry(QRect(1125, 10, 85, 25))
|
|
self.channel.setEditable(True)
|
|
self.channel.currentIndexChanged.connect(self.params_changed)
|
|
for i in cpars.code_channel_list():
|
|
self.channel.addItem(i)
|
|
self.channel_label = QLabel(self.widget)
|
|
self.channel_label.setObjectName(u"label")
|
|
self.channel_label.setGeometry(QRect(1055, 16, 72, 15))
|
|
self.channel_label.setText("通道数:")
|
|
|
|
# 提示信息
|
|
def infotext_init(self):
|
|
self.infotext=QLabel(self.widget)
|
|
self.infotext.setGeometry(QRect(30, 865, 1200, 15))
|
|
self.infotext.setText("请打开串口以开始赋码")
|
|
def set_infotext(self,text:str):
|
|
try:
|
|
self.infotext.setText(text)
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 初始化打开端口按钮
|
|
def com_but_init(self):
|
|
self.com_but = QPushButton(self.widget)
|
|
self.com_but.setObjectName(u"com_but")
|
|
self.com_but.setGeometry(QRect(1370, 90, 100, 28))
|
|
self.com_but.setText("打开端口")
|
|
self.com_but.clicked.connect(self.com_but_clicked)
|
|
|
|
# 电机上升
|
|
def moterup_init(self):
|
|
self.moterup=QPushButton(self.widget)
|
|
self.moterup.setObjectName(u"moteerup")
|
|
self.moterup.setGeometry(QRect(1250, 90, 100, 28))
|
|
self.moterup.setText("电机上升")
|
|
self.moterup.clicked.connect(self.cmd_moter_up)
|
|
|
|
# 自动增加id号
|
|
def autoinc_init(self):
|
|
self.autoinc=QPushButton(self.widget)
|
|
self.autoinc.setObjectName(u"autoinc")
|
|
self.autoinc.setGeometry(QRect(1250, 130, 100, 28))
|
|
self.autoinc.clicked.connect(self.autoinc_clicked)
|
|
self.autoinc_clicked()
|
|
|
|
# 开始检测
|
|
def check_but_init(self):
|
|
self.check_but=QPushButton(self.widget)
|
|
self.check_but.setObjectName(u"check_but")
|
|
self.check_but.setGeometry(QRect(1250, 170, 100, 28))
|
|
self.check_but.clicked.connect(self.check_but_clicked)
|
|
self.check_but.setText("开始检测")
|
|
|
|
# 到位感应
|
|
def posend_but_init(self):
|
|
self.posend_but=QPushButton(self.widget)
|
|
self.posend_but.setObjectName(u"posend_but")
|
|
self.posend_but.setGeometry(QRect(1250, 210, 100, 28))
|
|
self.posend_but.clicked.connect(self.posend_but_clicked)
|
|
self.posend_but.setText("到位感应")
|
|
|
|
# 自动测试
|
|
def autotest_but_init(self):
|
|
self.autotest_but=QPushButton(self.widget)
|
|
self.autotest_but.setObjectName(u"autotest_but")
|
|
self.autotest_but.setGeometry(QRect(1250, 250, 100, 28))
|
|
self.autotest_but.clicked.connect(self.autotest_but_clicked)
|
|
self.autotest_but.setText("开自动测试")
|
|
|
|
# 急停
|
|
def stop_but_init(self):
|
|
self.stop_but=QPushButton(self.widget)
|
|
self.stop_but.setObjectName(u"stop_but")
|
|
self.stop_but.setGeometry(QRect(1250, 290, 100, 28))
|
|
self.stop_but.clicked.connect(self.stop_but_clicked)
|
|
self.stop_state=True
|
|
self.stop_but_clicked()
|
|
|
|
# 发送赋码指令
|
|
def code_but_init(self):
|
|
self.code_but=QPushButton(self.widget)
|
|
self.code_but.setObjectName(u"code_but")
|
|
self.code_but.setGeometry(QRect(1250, 330, 100, 28))
|
|
self.code_but.clicked.connect(self.code_but_clicked)
|
|
self.code_but.setText("赋码")
|
|
|
|
# 电机下降
|
|
def moterdown_init(self):
|
|
self.moterdown=QPushButton(self.widget)
|
|
self.moterdown.setObjectName(u"moteerup")
|
|
self.moterdown.setGeometry(QRect(1250, 370, 100, 28))
|
|
self.moterdown.setText("电机下降")
|
|
self.moterdown.clicked.connect(self.cmd_moter_down)
|
|
|
|
# 忽略接触异常
|
|
def openignore_init(self):
|
|
self.openignore=QPushButton(self.widget)
|
|
self.openignore.setObjectName(u"moteerup")
|
|
self.openignore.setGeometry(QRect(1250, 410, 100, 28))
|
|
self.openignore.clicked.connect(self.openignore_clicked)
|
|
self.openignore_clicked()
|
|
|
|
# 写缓存
|
|
def writebuff_init(self):
|
|
self.writebuff=QPushButton(self.widget)
|
|
self.writebuff.setObjectName(u"moteerup")
|
|
self.writebuff.setText("写缓存数据")
|
|
self.writebuff.setGeometry(QRect(1250, 450, 100, 28))
|
|
self.writebuff.clicked.connect(self.writebuff_clicked)
|
|
self.writebuff_edit = QLineEdit(self.widget)
|
|
self.writebuff_edit.setObjectName(u"writebuff_edit")
|
|
self.writebuff_edit.setGeometry(QRect(1250, 490, 100, 28))
|
|
self.writebuff_edit.setText("0,0,0,0")
|
|
|
|
# 初始化设备类型选择框
|
|
def device_type_init(self):
|
|
self.device_type_list = QComboBox(self.widget)
|
|
self.device_type_list.setObjectName(u"device_type_list")
|
|
self.device_type_list.setGeometry(QRect(1250, 50, 93+120, 28))
|
|
self.device_type_list.setEditable(False)
|
|
self.device_type_list.currentIndexChanged.connect(self.device_type_changed)
|
|
for i in self.recv_handler_table.items():
|
|
self.device_type_list.addItem(i[0])
|
|
# self.device_type_list_label = QLabel(self.widget)
|
|
# self.device_type_list_label.setObjectName(u"label")
|
|
# self.device_type_list_label.setGeometry(QRect(1220, 16, 72, 15))
|
|
# self.device_type_list_label.setText("特征码:")
|
|
self.device_type_changed(0)
|
|
|
|
# 注码结果列表
|
|
def code_list_init(self):
|
|
self.code_list=QTableWidget(self.widget)
|
|
self.code_list.setObjectName(u"code_list")
|
|
self.code_list.setGeometry(QRect(30, 50, 1200, 800))
|
|
self.code_list.setColumnCount(6)
|
|
self.code_list.setHorizontalHeaderLabels(list(["创建时间","通道号","注码结果","管壳码","UID","密码"]))
|
|
self.code_list.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
|
|
|
|
# 添加数据
|
|
def code_list_add(self,code_line:str):
|
|
stat=self.code_list_backcolor
|
|
if(stat==True):
|
|
bkcolor=QColor()
|
|
bkcolor.setRgb(150,150,150,255)
|
|
else:
|
|
bkcolor=QColor()
|
|
bkcolor.setRgb(255,255,255,255)
|
|
code_strs=code_line.split(",")
|
|
errcode=int(code_strs[2])
|
|
code_strs[2]=cpars.code_errinfo(errcode)
|
|
RowCont=self.code_list.rowCount()
|
|
self.code_list.insertRow(RowCont)
|
|
for i in range(6):
|
|
self.code_list.setItem(RowCont,i,QTableWidgetItem(code_strs[i]))
|
|
self.code_list.item(RowCont,i).setBackground(bkcolor)
|
|
if(errcode!=0):
|
|
self.code_list.item(RowCont,2).setBackground(Qt.GlobalColor.red)
|
|
# 滚动到底部
|
|
def code_list_to_bottom(self):
|
|
self.code_list.scrollToBottom()
|
|
|
|
def com_but_clicked(self):
|
|
print("com but clicked")
|
|
if(self.ser_is_open==False):
|
|
item=self.com.itemText(self.com.currentIndex())
|
|
bsp=self.combsp.itemText(self.combsp.currentIndex())
|
|
com=item.split(":")[0]
|
|
if(com!="utcp"):
|
|
item=com+":"+bsp
|
|
self.open_serial(item)
|
|
else:
|
|
self.close_serial()
|
|
def set_port_state(self,state:bool):
|
|
self.ser_is_open=state
|
|
if(state==True):
|
|
self.com_but.setText("关闭端口")
|
|
else:
|
|
self.com_but.setText("打开端口")
|
|
|
|
# 修改id自增状态
|
|
def autoinc_clicked(self):
|
|
if(self.autoinc_id==True):
|
|
self.autoinc_id=False
|
|
self.autoinc.setText("打开ID自增")
|
|
self.set_infotext("已关闭ID自增")
|
|
else:
|
|
self.autoinc_id=True
|
|
self.autoinc.setText("关闭ID自增")
|
|
self.set_infotext("已打开ID自增")
|
|
|
|
def openignore_clicked(self):
|
|
if(self.openignore_flag==False):
|
|
self.openignore_flag=True
|
|
self.openignore.setText("拦截接触异常")
|
|
self.set_infotext("当前为忽略接触异常，接触异常时也会跑码。")
|
|
else:
|
|
self.openignore_flag=False
|
|
self.openignore.setText("忽略接触异常")
|
|
self.set_infotext("当前为拦截接触异常，接触异常时不会跑码。")
|
|
|
|
def writebuff_clicked(self):
|
|
print("writebuff cmd.")
|
|
buff=self.writebuff_edit.text()
|
|
data=buff.split(',')
|
|
byts=bytearray()
|
|
for i in data:
|
|
byts.append(int(i)&0xff)
|
|
def bytes_to_ints(byts):
|
|
ints=""
|
|
for i in byts:
|
|
ints+=str(int(i))+','
|
|
return ints
|
|
print("writebuff:",bytes_to_ints(byts))
|
|
self.cmd_user(0x45,byts)
|
|
|
|
def check_but_clicked(self):
|
|
print("send start check cmd.")
|
|
self.cmd_user(0x90,bytearray([2]))
|
|
|
|
def posend_but_clicked(self):
|
|
print("send in place cmd.")
|
|
self.cmd_user(0x90,bytearray([3]))
|
|
|
|
def stop_but_clicked(self):
|
|
print("send stop cmd.")
|
|
if(self.stop_state==False):
|
|
# 发送急停命令
|
|
self.cmd_user(0x90,bytearray([0]))
|
|
self.stop_but.setText("运行")
|
|
self.stop_state=True
|
|
else:
|
|
# 发送启动命令
|
|
self.cmd_user(0x90,bytearray([1]))
|
|
self.stop_but.setText("急停")
|
|
self.stop_state=False
|
|
|
|
def autotest_but_clicked(self):
|
|
if(self.autotest_is_open==0):
|
|
self.autotest_but.setText("启动中...")
|
|
self.autotest_is_open=2
|
|
self.autotest_but.setEnabled(False)
|
|
self.start_autotest()
|
|
elif(self.autotest_is_open==1):
|
|
self.autotest_but.setText("关闭中...")
|
|
self.autotest_is_open=3
|
|
self.autotest_but.setEnabled(False)
|
|
|
|
def code_but_clicked(self):
|
|
self.set_infotext("发送赋码指令")
|
|
self.cmd_code(self.slave_num)
|
|
|
|
# 无效化编码参数设置项
|
|
def params_active(self,power:bool):
|
|
self.dot_fat.setEnabled(power)
|
|
self.sig_list.setEnabled(power)
|
|
self.channel.setEnabled(power)
|
|
self.device_type_list.setEnabled(power)
|
|
|
|
# 更换管壳码生成参数
|
|
def params_changed(self,index:int):
|
|
try:
|
|
dot_fat=self.dot_fat.itemText(self.dot_fat.currentIndex()).split(":")[0].encode("utf-8")
|
|
chip_fat=b"AD"
|
|
sig_code=self.sig_list.itemText(self.sig_list.currentIndex()).encode("utf-8")
|
|
slave_num=self.channel.itemText(self.channel.currentIndex())
|
|
self.set_params(dot_fat,chip_fat,sig_code,int(slave_num))
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 设备类型发生了改变
|
|
def device_type_changed(self,index:int):
|
|
device_type=self.device_type_list.itemText(index)
|
|
self.recv_handler=self.recv_handler_table[device_type][0]
|
|
self.uid_length=self.recv_handler_table[device_type][1]
|
|
try:
|
|
for i in self.recv_handler_table:
|
|
if(i!=device_type):
|
|
self.recv_handler_table[i][3]()
|
|
self.recv_handler_table[device_type][2]()
|
|
except Exception as err:
|
|
print(str(err))
|
|
self.set_infotext("设备已切换为："+device_type)
|
|
|
|
# 更换端口时关闭之前的
|
|
def com_changed(self,index:int):
|
|
print("com changed")
|
|
self.set_infotext("串口参数有变动，将自动关闭打开的串口")
|
|
self.close_serial()
|
|
# 打开通信接口
|
|
def open_serial(self,com:str):
|
|
if(self.ser.init(com)==True):
|
|
self.ser.recv_signal.connect(self.recv_slot)
|
|
self.ser.start_recv()
|
|
self.set_port_state(True)
|
|
print("open serial.")
|
|
self.set_infotext("打开串口成功，开始监听请求赋码指令")
|
|
else:
|
|
self.set_infotext("打开串口失败，是否有其他程序占用了串口？")
|
|
print("open serial failed.")
|
|
def close_serial(self):
|
|
try:
|
|
self.ser.recv_signal.disconnect(self.recv_slot)
|
|
except Exception as a:
|
|
pass
|
|
self.ser.close()
|
|
self.set_infotext("串口已关闭")
|
|
self.set_port_state(False)
|
|
|
|
def recv_slot(self,cmd:int,data:bytearray,err:str):
|
|
print("recv:","cmd:",hex(cmd),"data:",data.hex(' '),"errstr:",err)
|
|
if(self.run_times<=0):
|
|
print("run times end.")
|
|
self.set_infotext("运行次数已结束，不再响应检测赋码指令")
|
|
return
|
|
self.run_times-=1
|
|
if (self.recv_handler!=None):
|
|
self.recv_handler(cmd,data,err)
|
|
|
|
# 2通道赋码设备的处理函数
|
|
def recv_deal_coder2ch(self,cmd:int,data:bytearray,err:str):
|
|
ack=False
|
|
if(cmd==0x8b):
|
|
self.params_active(False)
|
|
self.set_infotext("已接收请求赋码命令")
|
|
self.cmd_code(self.slave_num)
|
|
elif(cmd==0x82):
|
|
self.set_infotext("已接收赋码结果")
|
|
ack=self.decode_code(data)
|
|
if(ack!=True):
|
|
print("code failed.")
|
|
self.set_infotext("赋码出现异常，本次赋码结果被丢弃，请排查异常后重试")
|
|
self.params_active(True)
|
|
elif(cmd==0x8c):
|
|
s="设备信息：{d1},{d2}".format(d1=int(data[0]),d2=data[1:].decode("utf-8"))
|
|
print(s)
|
|
self.set_infotext(s)
|
|
elif(cmd==0x8a):
|
|
# print("recv keepliver .data:",data.hex(' '))
|
|
if(data[0]==0):
|
|
# 只在正常的心跳数据做回复
|
|
self.cmd_user(0x8a,bytearray([0]))
|
|
elif(cmd==0x91):
|
|
# 输入通道状态
|
|
print("input:channel({d1})={d2}".format(d1=int(data[0]),d2=int(data[1])))
|
|
elif(cmd==0x92):
|
|
# 输出通道状态
|
|
print("output:channel({d1})={d2}".format(d1=int(data[0]),d2=int(data[1])))
|
|
|
|
|
|
# 批检仪的处理函数
|
|
def recv_code_after_check_checker(self,cmd:int,data:bytearray,err:str):
|
|
ack=False
|
|
if(cmd==0x81):
|
|
self.set_infotext("已接收检测结果")
|
|
ack=self.decode_check(data)
|
|
ack=True
|
|
if(ack==True):
|
|
self.cmd_code(self.slave_num)
|
|
else:
|
|
print("check failed.")
|
|
self.set_infotext("检测出现异常,未进行赋码，请排查异常后重试")
|
|
self.cmd_moter_up()
|
|
self.params_active(True)
|
|
elif(cmd==0x82):
|
|
self.set_infotext("已接收赋码结果")
|
|
ack=self.decode_code(data)
|
|
if(ack!=True):
|
|
print("code failed.")
|
|
self.set_infotext("赋码出现异常，本次赋码结果被丢弃，请排查异常后重试")
|
|
self.cmd_moter_up()
|
|
self.params_active(True)
|
|
elif(cmd==0x37):
|
|
print("key press.")
|
|
self.params_active(False)
|
|
self.set_infotext("已接收请求检测命令")
|
|
self.cmd_moter_down()
|
|
time.sleep(2)
|
|
self.cmd_check(self.slave_num)
|
|
elif(cmd==0x45):
|
|
print("write buff.")
|
|
if(data[0]==0):
|
|
self.set_infotext("写缓存数据成功")
|
|
else:
|
|
self.set_infotext("写缓存数据失败")
|
|
# 批检仪检测时赋码处理函数
|
|
def recv_code_with_check_checker(self,cmd:int,data:bytearray,err:str):
|
|
ack=False
|
|
if(cmd==0x82):
|
|
self.set_infotext("已接收赋码结果")
|
|
ack=self.decode_code(data)
|
|
if(ack!=True):
|
|
print("code failed.")
|
|
self.set_infotext("赋码出现异常，本次赋码结果被丢弃，请排查异常后重试")
|
|
self.cmd_moter_up()
|
|
self.params_active(True)
|
|
elif(cmd==0x37):
|
|
print("key press.")
|
|
self.params_active(False)
|
|
self.set_infotext("已接收请求赋码命令")
|
|
self.cmd_moter_down()
|
|
time.sleep(2)
|
|
self.cmd_code(self.slave_num)
|
|
elif(cmd==0x45):
|
|
print("write buff.")
|
|
if(data[0]==0):
|
|
self.set_infotext("写缓存数据成功")
|
|
else:
|
|
self.set_infotext("写缓存数据失败")
|
|
|
|
# 初始化处理函数列表
|
|
def recv_handler_table_init(self):
|
|
self.recv_handler_table={"批检仪检测后赋码(XT)":(self.recv_code_after_check_checker,16,self.change_enter_checker,self.change_exit_checker),
|
|
"批检仪检测时赋码(JQ)":(self.recv_code_with_check_checker,15,self.change_enter_checker,self.change_exit_checker),
|
|
"赋码控制器":(self.recv_deal_coder2ch,15,self.change_enter_coder2ch,self.change_exit_coder2ch)}
|
|
|
|
# 切换设备类型时进行的额外处理
|
|
def change_enter_coder2ch(self):
|
|
self.check_but.setEnabled(True)
|
|
self.posend_but.setEnabled(True)
|
|
self.autotest_but.setEnabled(True)
|
|
self.stop_but.setEnabled(True)
|
|
def change_exit_coder2ch(self):
|
|
self.check_but.setEnabled(False)
|
|
self.posend_but.setEnabled(False)
|
|
self.autotest_but.setEnabled(False)
|
|
self.stop_but.setEnabled(False)
|
|
def change_enter_checker(self):
|
|
self.moterup.setEnabled(True)
|
|
def change_exit_checker(self):
|
|
self.moterup.setEnabled(False)
|
|
|
|
|
|
def run(self,times:int):
|
|
self.run_times=times*4
|
|
self.widget.show()
|
|
ret=self.app.exec()
|
|
self.close_serial()
|
|
sys.exit(ret)
|
|
|
|
# 生成管壳码
|
|
def calc_shell_code(self,id:int):
|
|
shell=bytearray()
|
|
shell+=self.detonator_fat
|
|
shell+=self.date
|
|
shell+=self.signature_code
|
|
shell+="{d:05d}".format(d=id).encode("utf-8")
|
|
return shell
|
|
|
|
# 生成n通道的注码数据
|
|
def creat_code_data(self,id:int,num:int):
|
|
d=bytearray()
|
|
d.append(num&0xff)
|
|
d+=self.year
|
|
for i in range(num):
|
|
d.append(i&0xff)
|
|
d+=self.calc_shell_code(id+i)
|
|
return d
|
|
# 发送检测命令
|
|
def cmd_check(self,num:int):
|
|
d=bytearray()
|
|
d.append(num&0xff)
|
|
d.append(0)
|
|
self.ser.send(1,d)
|
|
print("check start.")
|
|
# 电机上升
|
|
def cmd_moter_up(self):
|
|
d=bytearray()
|
|
d.append(0x02)
|
|
try:
|
|
self.ser.send(0x40,d)
|
|
except Exception as e:
|
|
self.set_infotext("发送命令失败，是否没有打开串口？")
|
|
print("moter up.")
|
|
# 电机下降
|
|
def cmd_moter_down(self):
|
|
d=bytearray()
|
|
d.append(0x03)
|
|
try:
|
|
self.ser.send(0x40,d)
|
|
except Exception as e:
|
|
self.set_infotext("发送命令失败，是否没有打开串口？")
|
|
print("moter down.")
|
|
|
|
# 发送赋码指令
|
|
def cmd_code(self,num:int):
|
|
try:
|
|
self.ser.send(2,self.creat_code_data(self.code_id,num))
|
|
except Exception as e:
|
|
self.set_infotext("发送命令失败，是否没有打开串口？")
|
|
print("code start.")
|
|
|
|
# 发送任意命令
|
|
def cmd_user(self,cmd:int,data:bytearray):
|
|
try:
|
|
self.ser.send(cmd,data)
|
|
except Exception as e:
|
|
self.set_infotext("发送命令失败，是否没有打开串口？")
|
|
|
|
# 解析检测结果
|
|
def decode_check(self,data:bytearray):
|
|
num=data[0]
|
|
ack=True
|
|
acks_list=[]
|
|
for i in range(num):
|
|
ack_i=data[6*i+3]
|
|
str_start=""
|
|
if(ack_i!=0):
|
|
# 接触异常则忽略
|
|
if(ack_i!=3):
|
|
ack=False
|
|
elif(self.openignore_flag==False):
|
|
ack=False
|
|
str_start=STR_RED
|
|
print(str_start+"addr:",int(data[6*i+2]),"ack:",ack_i,STR_END)
|
|
acks_list.append(str(ack_i))
|
|
try:
|
|
self.ch_dlog.close()
|
|
self.co_dlog.close()
|
|
except Exception as e:
|
|
pass
|
|
if(ack!=True):
|
|
self.ch_dlog=chk.check_dlog(self.widget,self.slave_num,"检测结果查看")
|
|
self.ch_dlog.ack_list_init(acks_list)
|
|
self.ch_dlog.show()
|
|
return ack
|
|
|
|
# 解析注码结果
|
|
def decode_code(self,data:bytearray):
|
|
num=data[0]
|
|
ack=True
|
|
code_list=[]
|
|
# 生成注码结果数据
|
|
for i in range(num):
|
|
item_len=13+self.uid_length+8+2
|
|
ack_i=data[item_len*i+2]
|
|
d=data[item_len*i+3:item_len*i+3+item_len-1]
|
|
try:
|
|
shell_code=d[:13].decode("utf-8")
|
|
uid_code=d[13:13+self.uid_length].decode("utf-8")
|
|
psw_code=d[13+self.uid_length:13+self.uid_length+8].decode("utf-8")
|
|
except Exception as e:
|
|
shell_code="unknown"
|
|
uid_code="unknown"
|
|
psw_code="unknown"
|
|
# print("psw:",d.decode("utf-8"))
|
|
code_list.append(((i+1),ack_i,shell_code,uid_code,psw_code))
|
|
|
|
# 统计错误信息
|
|
length=len(code_list)
|
|
if(len(self.code_list_lock)==0):
|
|
# 第一次失败，保存失败数据
|
|
self.code_list_lock=code_list
|
|
else:
|
|
# 不是第一次失败，比对失败数据，把成功过的统计出来
|
|
if (len(self.code_list_lock)>length):
|
|
self.code_list_lock=self.code_list_lock[:length]
|
|
for i in range(length):
|
|
# 长度不足的地方填充为本次的结果
|
|
if(i>=len(self.code_list_lock)):
|
|
self.code_list_lock.append(code_list[i])
|
|
if(self.code_list_lock[i][0]==code_list[i][0]):
|
|
if(code_list[i][1]==0):
|
|
self.code_list_lock[i]=code_list[i]
|
|
else:
|
|
if(self.code_list_lock[i][1]!=0):
|
|
self.code_list_lock[i]=code_list[i]
|
|
# 校验注码结果
|
|
acks_list=[]
|
|
for i in self.code_list_lock:
|
|
str_start=""
|
|
ack_i=i[1]
|
|
if(ack_i!=0):
|
|
# 接触异常则忽略
|
|
if((ack_i!=3) and (ack_i!=200)):
|
|
ack=False
|
|
elif(self.openignore_flag==False):
|
|
ack=False
|
|
str_start=STR_RED
|
|
print(str_start+"addr:",i[0],"ack:",ack_i,STR_END)
|
|
print("\t","shell_code:",i[2])
|
|
print("\t","uid_code:",i[3])
|
|
print("\t","psw_code:",i[4])
|
|
acks_list.append(str(ack_i))
|
|
if(ack==True):
|
|
self.code_list_backcolor=not self.code_list_backcolor
|
|
s=""
|
|
time=get_time()
|
|
for i in self.code_list_lock:
|
|
s=time+","+str(i[0])+','+str(i[1])+','+i[2]+','+i[3]+','+i[4]
|
|
self.code_list_add(s)
|
|
with open(self.save_file,"+a") as f:
|
|
f.write(s+'\n')
|
|
self.code_list_to_bottom()
|
|
print("save to file:",self.save_file)
|
|
# 全部成功，清除错误记录
|
|
self.code_list_lock.clear()
|
|
if(self.autoinc_id==True):
|
|
self.code_id+=self.slave_num
|
|
try:
|
|
self.co_dlog.close()
|
|
except Exception as e:
|
|
pass
|
|
if(ack!=True):
|
|
self.co_dlog=chk.check_dlog(self.widget,self.slave_num,"注码结果查看")
|
|
self.co_dlog.ack_list_init(acks_list)
|
|
self.co_dlog.show()
|
|
return ack
|
|
|
|
def start_autotest(self):
|
|
self.thread_ = threading.Thread(target=self.autotest, args=())
|
|
self.thread_.start()
|
|
|
|
def autotest(self):
|
|
self.autotest_but.setText("关自动测试")
|
|
self.autotest_is_open=1
|
|
self.autotest_but.setEnabled(True)
|
|
while(self.autotest_is_open==1):
|
|
if(self.ser_is_open!=True):
|
|
print("ser was closed,stop autotest.")
|
|
break
|
|
self.check_but_clicked()
|
|
time.sleep(0.3)
|
|
self.posend_but_clicked()
|
|
time.sleep(11.3)
|
|
self.autotest_but.setText("开自动测试")
|
|
self.autotest_is_open=0
|
|
self.autotest_but.setEnabled(True)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
QCoreApplication.setAttribute(Qt.ApplicationAttribute.AA_EnableHighDpiScaling)
|
|
c=coder()
|
|
c.init(b'00',b'AD',b'A',int(cpars.code_channel_list()[0]))
|
|
c.run(10000)
|
|
|