from concurrent.futures import process from PyQt5 import QtCore, QtGui, QtWidgets import os import sys import re import ast from urllib import parse import json import time import datetime import requests import pymongo import pymysql from xlrd import open_workbook from xlsxwriter.workbook import Workbook save_filename = "" pwd_str = "yf6021" toji_format = { 'font_name' : '宋体', 'font_size': 14, 'font_color': 'black', 'text_wrap': True, 'bold': False, 'fg_color': '92D050', 'align': 'center', 'valign': 'vcenter', 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } title_format = { 'font_name' : '宋体', 'font_size': 12, 'bold': True, 'align': 'center', 'valign': 'vcenter', 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } merge_title_format = { 'font_name' : '宋体', 'font_size': 26, 'bold': True, 'align': 'center', 'valign': 'vcenter', "fg_color": "8DB4E2", 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } explain_formal = { 'font_name' : '宋体', 'font_size': 11, 'font_color': 'black', 'text_wrap': True, 'align': 'justify', 'valign': 'vcenter', 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } formal_format = { 'font_name' : '宋体', 'font_size': 11, 'font_color': 'black', 'fg_color': '77E88C', 'text_wrap': True, 'align': 'center', 'valign': 'vcenter', 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } common_format = { 'font_name' : '宋体', 'font_size': 11, 'font_color': 'black', "fg_color": 'E7EC73', 'text_wrap': True, 'align': 'center', 'valign': 'vcenter', 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } error_format = { 'font_name' : '宋体', 'font_size': 11, 'font_color': 'black', "fg_color": 'F4746A', 'text_wrap': True, 'align': 'center', 'valign': 'vcenter', 'border': 1, 'top': 1, 'left': 1, 'right': 1, 'bottom': 1 } class Ui_MainWindow(object): """GUI界面""" def setupUi(self, MainWindow): MainWindow.setObjectName("MainWindow") MainWindow.resize(701, 644) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(12) MainWindow.setFont(font) icon = QtGui.QIcon("logo.ico") MainWindow.setWindowIcon(icon) self.centralwidget = QtWidgets.QWidget(MainWindow) self.centralwidget.setObjectName("centralwidget") self.pageTitleLabel = QtWidgets.QLabel(self.centralwidget) self.pageTitleLabel.setGeometry(QtCore.QRect(139, 50, 429, 42)) font = QtGui.QFont() font.setFamily("楷体") font.setPointSize(28) self.pageTitleLabel.setFont(font) self.pageTitleLabel.setTextFormat(QtCore.Qt.AutoText) self.pageTitleLabel.setObjectName("pageTitleLabel") self.inputFileLabel = QtWidgets.QLabel(self.centralwidget) self.inputFileLabel.setGeometry(QtCore.QRect(180, 130, 121, 21)) self.inputFileLabel.setObjectName("inputFileLabel") self.inputFileEdit = QtWidgets.QLineEdit(self.centralwidget) self.inputFileEdit.setGeometry(QtCore.QRect(310, 130, 151, 21)) self.inputFileEdit.setFocusPolicy(QtCore.Qt.NoFocus) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(8) self.inputFileEdit.setFont(font) self.inputFileEdit.setObjectName("inputFileEdit") self.inputFileTool = QtWidgets.QToolButton(self.centralwidget) self.inputFileTool.setGeometry(QtCore.QRect(470, 130, 71, 21)) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(10) self.inputFileTool.setFont(font) self.inputFileTool.setObjectName("inputFileTool") self.savePathLabel = QtWidgets.QLabel(self.centralwidget) self.savePathLabel.setGeometry(QtCore.QRect(180, 170, 121, 21)) self.savePathLabel.setObjectName("savePathLabel") self.savePathEdit = QtWidgets.QLineEdit(self.centralwidget) self.savePathEdit.setGeometry(QtCore.QRect(310, 170, 151, 21)) self.savePathEdit.setFocusPolicy(QtCore.Qt.NoFocus) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(8) self.savePathEdit.setFont(font) self.savePathEdit.setObjectName("savePathEdit") self.savePathTool = QtWidgets.QToolButton(self.centralwidget) self.savePathTool.setGeometry(QtCore.QRect(470, 170, 71, 21)) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(10) self.savePathTool.setFont(font) self.savePathTool.setObjectName("savePathTool") self.startTimeLabel = QtWidgets.QLabel(self.centralwidget) self.startTimeLabel.setGeometry(QtCore.QRect(180, 210, 121, 21)) self.startTimeLabel.setObjectName("startTimeLabel") self.startTimeEdit = QtWidgets.QDateTimeEdit(self.centralwidget) self.startTimeEdit.setGeometry(QtCore.QRect(310, 210, 151, 21)) self.startTimeEdit.setMaximumDateTime(QtCore.QDateTime(QtCore.QDate(2025, 12, 31), QtCore.QTime(23, 59, 59))) self.startTimeEdit.setMinimumDateTime(QtCore.QDateTime(QtCore.QDate(2023, 6, 9), QtCore.QTime(12, 0, 0))) self.startTimeEdit.setObjectName("startTimeEdit") self.endTimeLabel = QtWidgets.QLabel(self.centralwidget) self.endTimeLabel.setGeometry(QtCore.QRect(180, 250, 121, 21)) self.endTimeLabel.setObjectName("endTimeLabel") self.endTimeEdit = QtWidgets.QDateTimeEdit(self.centralwidget) self.endTimeEdit.setGeometry(QtCore.QRect(310, 250, 151, 21)) self.endTimeEdit.setMaximumDateTime(QtCore.QDateTime(QtCore.QDate(2025, 12, 31), QtCore.QTime(23, 59, 59))) self.endTimeEdit.setMinimumDateTime(QtCore.QDateTime(QtCore.QDate(2023, 6, 12), QtCore.QTime(6, 0, 0))) self.endTimeEdit.setObjectName("endTimeEdit") self.platLabel = QtWidgets.QLabel(self.centralwidget) self.platLabel.setGeometry(QtCore.QRect(180, 290, 121, 21)) self.platLabel.setObjectName("dianjiLabel") self.platBox = QtWidgets.QComboBox(self.centralwidget) self.platBox.setGeometry(QtCore.QRect(310, 290, 151, 21)) self.platBox.addItems(['大数据平台','四情平台']) # 产品名称 self.productLabel = QtWidgets.QLabel(self.centralwidget) self.productLabel.setGeometry(QtCore.QRect(180, 440, 121, 21)) self.productLabel.setObjectName("productLabel") self.productBox = QtWidgets.QComboBox(self.centralwidget) self.productBox.setGeometry(QtCore.QRect(310, 440, 151, 21)) self.productBox.addItems(['1.0测报灯','4.0测报灯']) # 供电选择 self.powerSupplyLabel = QtWidgets.QLabel(self.centralwidget) self.powerSupplyLabel.setGeometry(QtCore.QRect(180, 470, 121, 21)) self.powerSupplyLabel.setObjectName("powerSupplyLabel") self.powerSupplyBox = QtWidgets.QComboBox(self.centralwidget) self.powerSupplyBox.setGeometry(QtCore.QRect(310, 470, 151, 21)) self.powerSupplyBox.addItems(['AC','DC']) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(11) self.platBox.setFont(font) self.platBox.setObjectName("platBox") self.stm8vsLabel = QtWidgets.QLabel(self.centralwidget) self.stm8vsLabel.setGeometry(QtCore.QRect(180, 330, 121, 21)) self.stm8vsLabel.setObjectName("stm8vsLabel") self.stm8vsEdit = QtWidgets.QLineEdit(self.centralwidget) self.stm8vsEdit.setGeometry(QtCore.QRect(310, 330, 151, 21)) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(11) self.stm8vsEdit.setFont(font) self.stm8vsEdit.setObjectName("stm8vsEdit") self.dverLabel = QtWidgets.QLabel(self.centralwidget) self.dverLabel.setGeometry(QtCore.QRect(180, 370, 121, 21)) self.dverLabel.setObjectName("dverLabel") self.dverEdit = QtWidgets.QLineEdit(self.centralwidget) self.dverEdit.setGeometry(QtCore.QRect(310, 370, 151, 21)) font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(11) self.dverEdit.setFont(font) self.dverEdit.setObjectName("dverEdit") self.orderLabel = QtWidgets.QLabel(self.centralwidget) self.orderLabel.setGeometry(QtCore.QRect(180, 410, 121, 21)) self.orderLabel.setObjectName("dverLabel") self.orderEdit = QtWidgets.QLineEdit(self.centralwidget) self.orderEdit.setGeometry(QtCore.QRect(310, 410, 151, 21)) # 进度条与按钮 font = QtGui.QFont() font.setFamily("Arial") font.setPointSize(11) self.orderEdit.setFont(font) self.orderEdit.setObjectName("dverEdit") self.pushButton = QtWidgets.QPushButton(self.centralwidget) self.pushButton.setGeometry(QtCore.QRect(300, 520, 111, 41)) self.pushButton.setObjectName("pushButton") self.progressBar = QtWidgets.QProgressBar(self.centralwidget) self.progressBar.setGeometry(QtCore.QRect(130, 540, 491, 31)) self.progressBar.setProperty("value", 0) self.progressBar.setVisible(False) self.progressBar.setObjectName("progressBar") MainWindow.setCentralWidget(self.centralwidget) self.retranslateUi(MainWindow) self.inputFileTool.clicked.connect(self.input_file_path) self.savePathTool.clicked.connect(self.out_save_location) self.pushButton.clicked.connect(self.on_click) QtCore.QMetaObject.connectSlotsByName(MainWindow) def retranslateUi(self, MainWindow): _translate = QtCore.QCoreApplication.translate MainWindow.setWindowTitle(_translate("MainWindow", "杀虫灯质检工具")) self.savePathTool.setText(_translate("MainWindow", "选择文件夹")) self.inputFileTool.setText(_translate("MainWindow", "选择文件")) self.pageTitleLabel.setText(_translate("MainWindow", "

云飞测报灯设备质检工具

")) self.savePathLabel.setText(_translate("MainWindow", "|输出文件位置：")) self.inputFileLabel.setText(_translate("MainWindow", "|输入文件位置：")) self.startTimeLabel.setText(_translate("MainWindow", "|开始时间：")) self.endTimeLabel.setText(_translate("MainWindow", "|结束时间：")) self.platLabel.setText(_translate("MainWindow", "|检验平台：")) self.productLabel.setText(_translate("MainWindow", "|产品名称：")) self.powerSupplyLabel.setText(_translate("MainWindow", "|供电选择：")) self.stm8vsLabel.setText(_translate("MainWindow", "|系统固件版本号：")) self.dverLabel.setText(_translate("MainWindow", "|RTU固件版本号: ")) self.orderLabel.setText(_translate("MainWindow", "|任务单号：")) self.pushButton.setText(_translate("MainWindow", "开始导出")) def input_file_path(self): file_path = QtWidgets.QFileDialog.getOpenFileName(None,"选取文件","./","All Files (*.xlsx;*.xls);;Text Files (*.txt)") self.inputFileEdit.setText(file_path[0]) self.inputFileEdit.setStyleSheet("color:black;") def out_save_location(self): fname = QtWidgets.QFileDialog.getExistingDirectory(None, '选取文件夹', './') self.savePathEdit.setText(fname) self.savePathEdit.setStyleSheet("color:black;") def on_click(self): file_path = self.inputFileEdit.text() save_path = self.savePathEdit.text() start_time = self.startTimeEdit.dateTime().toPyDateTime() end_time = self.endTimeEdit.dateTime().toPyDateTime() set_plat = self.platBox.currentText() set_stm8vs = self.stm8vsEdit.text() set_dver = self.dverEdit.text() set_order = self.orderEdit.text() set_product = self.productBox.currentText() set_power = self.powerSupplyBox.currentText() if not all ([file_path,save_path,start_time,end_time,set_stm8vs,set_dver,set_order]): QtWidgets.QMessageBox.information(None, "提示", "选项未填写完整") elif end_time < start_time: QtWidgets.QMessageBox.information(None, "提示", "结束日期应在开始日期之后") else: xls = open_workbook(file_path) sheet_object = xls.sheets()[0] ncols = sheet_object.ncols read_dict = {} for i in range(ncols): col_value = sheet_object.col_values(i) read_dict[col_value[0]] = col_value[1:] device_list = read_dict.get("设备ID") if device_list: self.inputFileTool.setEnabled(False) self.savePathTool.setEnabled(False) self.startTimeEdit.setEnabled(False) self.endTimeEdit.setEnabled(False) self.platBox.setEnabled(False) self.productBox.setEnabled(False) self.powerSupplyBox.setEnabled(False) self.stm8vsEdit.setEnabled(False) self.dverEdit.setEnabled(False) self.orderEdit.setEnabled(False) self.pushButton.setEnabled(False) self.pushButton.setText("执行中...") self.runThread = CBDThread(device_list,save_path,start_time,end_time,set_plat,set_stm8vs,set_dver,set_order, set_product, set_power) self.runThread.proess_signal.connect(self.set_progressbar_value) self.runThread.start() self.progressBar.setVisible(True) else: QtWidgets.QMessageBox.information(None, "提示", "输入文件无'设备ID'列或该列无数据") def set_progressbar_value(self, value): self.progressBar.setValue(value) if value == 100: QtWidgets.QMessageBox.information(None, "提示", "文件导出完毕！导出文件名:\n{}".format(save_filename)) self.inputFileTool.setEnabled(True) self.inputFileEdit.setText("") self.savePathTool.setEnabled(True) self.savePathEdit.setText("") self.startTimeEdit.setEnabled(True) self.endTimeEdit.setEnabled(True) self.platBox.setEnabled(True) self.productBox.setEnabled(True) self.powerSupplyBox.setEnabled(True) self.stm8vsEdit.setEnabled(True) self.stm8vsEdit.setText("") self.dverEdit.setEnabled(True) self.dverEdit.setText("") self.orderEdit.setEnabled(True) self.orderEdit.setText("") self.pushButton.setEnabled(True) self.pushButton.setText("开始导出") self.progressBar.setVisible(False) self.progressBar.setValue(0) return class CBDThread(QtCore.QThread): """涉及进度条需主界面动态执行GUI,线程执行业务逻辑,避免主页面卡死""" proess_signal = QtCore.pyqtSignal(int) def __init__(self,device_list,save_path,start_time,end_time,set_plat,set_stm8vs,set_dver,set_order, set_product, set_power): super(CBDThread, self).__init__() self.device_list = device_list self.save_path = save_path self.start_time = time.mktime(start_time.timetuple()) self.end_time = time.mktime(end_time.timetuple()) self.start_time_str = start_time.strftime("%y-%m-%d %H:%M:%S") self.end_time_str = end_time.strftime("%y-%m-%d %H:%M:%S") self.set_plat = set_plat self.set_stm8vs = set_stm8vs self.set_dver = set_dver self.set_order = set_order self.set_product = set_product self.set_power = set_power self.user = parse.quote_plus("root") self.passwd = parse.quote_plus("yfkj@6020") self.myclient = pymongo.MongoClient("mongodb://{0}:{1}@8.136.98.49:57017/".format(self.user,self.passwd)) self.db = self.myclient.smartfarming self.device_collection = self.db.sa_device self.cbd_collection = self.db.sa_device_cbd_data self.cbd_photo = self.db.sa_device_cbdphoto self.cbd_alerm = self.db.sa_alarm_record self.config = { 'host': '120.27.222.26', 'port': 3306, 'user': 'yfwlw', 'password': 'sql_yfkj_6019', 'db': 'yfwlw', 'charset': 'utf8mb4', 'cursorclass': pymysql.cursors.DictCursor, } self.connection = pymysql.connect(**self.config) self.cursor = self.connection.cursor() def sim_info(self,iccid): """查询卡信息""" url = "http://8.136.98.49:10001/iotcard/platsimview/inquiries/" try: response = requests.request("POST", url, data={"iccid":iccid}) except: return [0,"查询卡信息异常稍后重试"] else: res_data = json.loads(response.text) if res_data["msg"]=="success" and res_data["data"]: expiry_date = res_data["data"]["expiry_date"] if expiry_date == "未知": return [0,"未查询到卡信息"] elif expiry_date == "未激活": return [3, "未激活"] else: time_difference = self.__time_dif(expiry_date) if time_difference < 30: return [1,"有效期剩余{}天".format(time_difference)] elif time_difference >= 181: return [1,"有效期剩余{}天".format(time_difference)] else: return [1,"有效期剩余{}天".format(time_difference)] else: return [0,"查询无结果"] def __bigdata_iamge_verify(self, device_id): # 对图片质量进行打分并求平均值 scores = [] photo_obj = self.cbd_photo.find({"device_id":str(device_id), 'addtime': {"$gte":self.start_time ,"$lte":self.end_time}}) for p in photo_obj[:5]: photo_addr = p.get("addr") # 把图片下载到本地 local_dir = "org_image" file_name = photo_addr.split("/")[-1] file_name = file_name.replace("?", "") os.makedirs(local_dir) if not os.path.exists(local_dir) else None response = requests.get(f"https://bigdata-image.oss-cn-hangzhou.aliyuncs.com/Basics/cbd/{photo_addr}") with open(os.path.join(local_dir, file_name), 'wb') as f: f.write(response.content) result_score = self.predict(os.path.join(local_dir, file_name)) for r in result_score: scores.append(r) if scores: return sum(scores) / len(scores) else: return 0 def __siqing_image_verify(self, device_id): # 对图片质量进行打分并求平均值 scores = [] sql = "SELECT * FROM AppInfoManage_cbdphoto WHERE equip_id_id='{}' AND upl_time BETWEEN '{}' AND '{}' order by 'alarm_time';".format(device_id,self.start_time_str,self.end_time_str) self.cursor.execute(sql) photo_obj = self.cursor.fetchall() for p in photo_obj[:5]: photo_addr = p.get("addr") # 把图片下载到本地 local_dir = "org_image" file_name = photo_addr.split("/")[-1] os.makedirs(local_dir) if not os.path.exists(local_dir) else None response = requests.get(photo_addr) with open(os.path.join(local_dir, file_name), 'wb') as f: f.write(response.content) result_score = self.predict(os.path.join(local_dir, file_name)) for r in result_score: scores.append(r) if scores: return sum(scores) / len(scores) else: return 0 def mongo_ping(self): """mongo-ping预防连接失效""" try: self.myclient.admin.command('ping') except: # "ConnectionFailure" self.myclient = pymongo.MongoClient("mongodb://{0}:{1}@8.136.98.49:57017/".format(self.user,self.passwd)) self.db = self.myclient.smartfarming self.device_collection = self.db.sa_device self.cbd_collection = self.db.sa_device_cbd_data self.cbd_photo = self.db.sa_device_cbdphoto self.cbd_alerm = self.db.sa_alarm_record def sql_ping(self): """mysql-ping 预防连接失效""" try: self.connection.ping() except: self.connection = pymysql.connect(**self.config) self.cursor = self.connection.cursor() def __time_dif(self,checkdatetime): """计算时间差""" nowdatetime = datetime.datetime.now() checkdatetime = datetime.datetime.strptime(checkdatetime, "%Y-%m-%d %H:%M:%S") timedif = checkdatetime - nowdatetime return timedif.days def device_their_platform(self,shortId): """确定设备所在平台已经完整设备号""" self.mongo_ping() self.sql_ping() regex = re.compile('.*{}$'.format(shortId)) bd_device_dict = self.device_collection.find_one( filter = {"device_id":regex,"device_type_id":3}, projection = {'_id': 0}, sort = [('uptime', pymongo.DESCENDING)] ) device_sql = "SELECT * FROM AppInfoManage_cbdstatus WHERE equip_id_id LIKE '%{}' ORDER BY upl_time DESC LIMIT 1;".format(shortId) self.cursor.execute(device_sql) sq_device_dict = self.cursor.fetchone() if bd_device_dict and sq_device_dict: bd_upltime = bd_device_dict["uptime"] sq_upltime = time.mktime(sq_device_dict["upl_time"].timetuple()) if bd_upltime >= sq_upltime: d_id = bd_device_dict["id"] deviceId = bd_device_dict["device_id"] platform = "大数据平台" else: d_id = "" deviceId = sq_device_dict["equip_id_id"] platform = "四情平台" elif bd_device_dict and not sq_device_dict: d_id = bd_device_dict["id"] deviceId = bd_device_dict["device_id"] platform = "大数据平台" return d_id,deviceId,platform elif not bd_device_dict and sq_device_dict: d_id = "" deviceId = sq_device_dict["equip_id_id"] platform = "四情平台" else: d_id = "" deviceId = "平台无此设备" platform = "未知" return d_id,deviceId,platform def __bigdata_verify(self,data_cursor,data_status,device_id_info, image_score, proess): """ 获取大数据平台查询结果 1. 合格 2. 不合格 """ current_jg = [1,"合格"] # 振动电机工作状态下电流匹配度 up_jg = [1,"合格"] # 上仓门电机电机工作状态 down_jg = [1,"合格"] # 下仓门电机电机工作状态 lig_jg = [1,"合格"] # 灯管工作状态下 machine_jg = [1, "合格"]# 电机工作状态 camera_jg = [1, "合格"] # 相机工作状态 time_jg = [1, "合格"] # 工作时长 for alerm in data_status: a = alerm.get("alarm_desc", "") if a: a = json.loads(a) if a.get("type", "") == "震动电机": if a.get("status") == "电流不足": current_jg = [0,"不合格"] elif a.get("type", "") == "上仓门电机": if a.get("status") in ["打开电流不足", "关闭电流不足"]: up_jg = [0,"不合格"] elif a.get("type", "") == "下仓门电机": if a.get("status") in ["打开电流不足", "关闭电流不足"]: down_jg = [0,"不合格"] elif a.get("type", "") == "灯管": if a.get("status") == "电流不足": lig_jg = [0,"不合格"] elif a.get("type", "") == "相机": if a.get("status") == "电流不足": camera_jg = [0,"不合格"] iccid = "" lng_lst = [] lat_lst = [] work_status = [] # 工作状态切换 elec_vs_lst = [] # 电路板固件版本号 v_vs_lst = [] # 电压 tr_k_jg_lst = [] # 温控 rain_k_jg_lst = []# 雨控 infor_jg_count = 0 # 上报信息条数 heat_temp_lst = []# 加热仓温度 temp_hum = [] # 环境温度 环境湿度 total_time = [] # 总时长 start_time = 0 information_count = 0 dver_lst = [] for index, cur in enumerate(data_cursor): infor_start = 0 start_time = int(cur.get("addtime", 0)) if index == 0: infor_start = start_time else: if start_time < 3600 * 10 + infor_start: lamp = device_data.get("lamp", "") if lamp == "0": information_count += 1 device_data = cur.get("device_data", "") if device_data: device_data = ast.literal_eval(device_data) iccid = device_data.get("iccid", "") lng = device_data.get("lng", "") lng_lst.append(float(lng)) lat = device_data.get("lat", "") lat_lst.append(float(lat)) work_status.append(device_data.get("ws", "")) elec_vs_lst.append(device_data.get("dver", "")) v_vs_lst.append(float(device_data.get("vbat", 0))) tr_k_jg_lst.append(device_data.get("tps", "")) rain_k_jg_lst.append(device_data.get("rps", "")) infor_jg_count += 1 heat_temp_lst.append(float(device_data.get("hrt", 0))) temp_hum.append({"at": device_data.get("at", ""), "ah": device_data.get("ah", "")}) if self.set_power == "AC": total_time.append(device_data.get("tt")) elif self.set_power == "DC": st = device_data.get("st") et = device_data.get("et") st = int(st.split(":")[0]) et = int(et.split(":")[0]) if st > et: total = 24 - st + et else: total = et - st total_time.append(total) dver = device_data.get("dver", "") if dver: dver_lst.append(dver) # 检查卡号及有效期 card_jg = self.sim_info(iccid) # 经度 if max(lng_lst) < 113.77861111 and min(lng_lst) > 113.76194444: lng_jg = [1,"合格"] else: lng_jg = [0,"不合格"] # 纬度 35.02083333,35.0375 if max(lat_lst) < 35.0375 and min(lat_lst) > 35.02083333: lat_jg = [1,"合格"] else: lat_jg = [0,"不合格"] # 工作状态切换 work_status_str = "".join(work_status) wr = work_status_str.count("01") if wr > 4: work_jg = [1, "合格"] else: work_jg = [0, "不合格"] # 电路板输入电压 if max(v_vs_lst) >= 22 and min(v_vs_lst) <= 30: v_vs = [1, "合格"] else: v_vs = [0, "不合格"] # 温控 if "1" in tr_k_jg_lst: tr_k_jg = [1, "合格"] else: tr_k_jg = [0, "不合格"] # 雨控 if "1" in rain_k_jg_lst: rain_k_jg = [1, "合格"] else: rain_k_jg = [0, "不合格"] # 加热仓温度 if max(heat_temp_lst) < 105 and min(heat_temp_lst) > 85: heat_temp = [1, "合格"] else: heat_temp = [0, "不合格"] # 环境温度 if {"at": "25", "ah": "35"} in temp_hum: en_temp = [0,"不合格"] else: en_temp = [1,"合格"] # 环境湿度 if {"at": "35", "ah": "25"} in temp_hum: en_hum = [0,"不合格"] else: en_hum = [1,"合格"] # 图片质量 if image_score > 68: img_jg = [1,"合格"] else: img_jg = [0,"不合格"] # 上报信息条数 if information_count >= 22: infor_jg_count = [1, "合格"] else: infor_jg_count = [0, "不合格"] # 电路板固件版本号 RTU固件版本号 dvers = list(set(dver_lst)) if len(dvers) != 1: gujian_num = [0, "不合格"] rtu_num = [0, "不合格"] else: d = dvers[0] rtu = d.split("-")[-1] print(rtu) gujian = d.replace(rtu, "").replace("-", "") print(gujian) if gujian == self.set_stm8vs: gujian_num = [1, "合格"] else: gujian_num = [0, "不合格"] if rtu == self.set_dver: rtu_num = [1, "合格"] else: rtu_num = [0, "不合格"] return [ device_id_info, [1, self.set_order], card_jg, lng_jg, lat_jg, work_jg, gujian_num, rtu_num, v_vs, tr_k_jg, rain_k_jg, infor_jg_count, heat_temp, en_temp, en_hum, current_jg, up_jg, down_jg, lig_jg, img_jg, machine_jg, camera_jg, time_jg ] def __siqing_verify(self, data_result,status_result, status_all_result, alarm_result, device_id_info, image_score, process): """获取四情平台查询结果""" current_jg = [1,"合格"] # 振动电机工作状态下电流匹配度 up_jg = [1,"合格"] # 上仓门电机电机工作状态 down_jg = [1,"合格"] # 下仓门电机电机工作状态 lig_jg = [1,"合格"] # 灯管工作状态下 machine_jg = [1, "合格"]# 电机工作状态 camera_jg = [1, "合格"] # 相机工作状态 time_jg = [1, "合格"] # 工作时长 if alarm_result: for alerm in alarm_result: a = alerm.get("alarm_desc", "") if a: a = json.loads(a) if a.get("type", "") == "震动电机": if a.get("status") == "电流不足": current_jg = [0,"不合格"] elif a.get("type", "") == "上仓门电机": if a.get("status") in ["打开电流不足", "关闭电流不足"]: up_jg = [0,"不合格"] elif a.get("type", "") == "下仓门电机": if a.get("status") in ["打开电流不足", "关闭电流不足"]: down_jg = [0,"不合格"] elif a.get("type", "") == "灯管": if a.get("status") == "电流不足": lig_jg = [0,"不合格"] elif a.get("type", "") == "相机": if a.get("status") == "电流不足": camera_jg = [0,"不合格"] lng_lst = [] # 经度 lat_lst = [] # 纬度 tr_k_jg_lst = [] # 温控 rain_k_jg_lst = []# 雨控 infor_jg_count = 0 # 上报信息条数 heat_temp_lst = []# 加热仓温度 temp_hum = [] # 环境温度 环境湿度 total_time = [] # 总时长 information_count = 0 for index, cur in enumerate(data_result): upl_time_str = cur.get("upl_time", "0") print(upl_time_str) upl_time_int = int(time.mktime(upl_time_str.timetuple())) infor_start = 0 if index == 0: infor_start = upl_time_int else: if upl_time_int < 3600 * 10 + infor_start: lamp = device_data.get("lamp", "") if lamp == "0": information_count += 1 device_data = cur.get("cbd_data", "") if device_data: device_data = ast.literal_eval(device_data) lng = device_data.get("lng", "") lng_lst.append(float(lng)) lat = device_data.get("lat", "") lat_lst.append(float(lat)) # v_vs_lst.append(float(device_data.get("vbat", 0))) tr_k_jg_lst.append(device_data.get("tps", "")) rain_k_jg_lst.append(device_data.get("rps", "")) infor_jg_count += 1 heat_temp_lst.append(float(device_data.get("hrt", 0))) temp_hum.append({"at": device_data.get("at", ""), "ah": device_data.get("ah", "")}) if self.set_power == "AC": total_time.append(device_data.get("tt")) elif self.set_power == "DC": st = device_data.get("st") et = device_data.get("et") st = int(st.split(":")[0]) et = int(et.split(":")[0]) if st > et: total = 24 - st + et else: total = et - st total_time.append(total) # 经度 if max(lng_lst) < 113.77861111 and min(lng_lst) > 113.76194444: lng_jg = [1,"合格"] else: lng_jg = [0,"不合格"] # 纬度 35.02083333,35.0375 if max(lat_lst) < 35.0375 and min(lat_lst) > 35.02083333: lat_jg = [1,"合格"] else: lat_jg = [0,"不合格"] # # 电路板输入电压 # if max(v_vs_lst) >= 22 and min(v_vs_lst) <= 30: # v_vs = [1, "合格"] # else: # v_vs = [0, "不合格"] # 温控 if "1" in tr_k_jg_lst: tr_k_jg = [1, "合格"] else: tr_k_jg = [0, "不合格"] # 雨控 if "1" in rain_k_jg_lst: rain_k_jg = [1, "合格"] else: rain_k_jg = [0, "不合格"] # 加热仓温度 if max(heat_temp_lst) < 105 and min(heat_temp_lst) > 85: heat_temp = [1, "合格"] else: heat_temp = [0, "不合格"] # 环境温度 if {"at": "25", "ah": "35"} in temp_hum: en_temp = [0,"不合格"] else: en_temp = [1,"合格"] # 环境湿度 if {"at": "35", "ah": "25"} in temp_hum: en_hum = [0,"不合格"] else: en_hum = [1,"合格"] # 图片质量 if image_score > 68: img_jg = [1,"合格"] else: img_jg = [0,"不合格"] # 上报信息条数 if information_count >= 22: infor_jg_count = [1, "合格"] else: infor_jg_count = [0, "不合格"] # rtu_num = [0, "不合格"] # else: # d = dvers[0] # rtu = d.split("-")[-1] # gujian = d.replace(rtu, "") # if gujian == self.set_stm8vs: # gujian_num = [1, "合格"] # else: # gujian_num = [0, "不合格"] # if rtu == self.set_dver: # rtu_num = [1, "合格"] # else: # rtu_num = [0, "不合格"] iccid = "" work_status = [] # 工作状态切换 elec_vs_lst = [] # 电路板固件版本号 for index, cur in enumerate(status_all_result): device_data = cur.get("cbd_status", "") if device_data: device_data = ast.literal_eval(device_data) iccid = device_data.get("iccid", "") work_status.append(device_data.get("ws", "")) elec_vs_lst.append(device_data.get("dver", "")) # 电路板固件版本号 RTU固件版本号 dvers = list(set(elec_vs_lst)) if len(dvers) != 1: gujian_num = [0, "不合格"] else: gujian = dvers[0] if gujian == self.set_stm8vs: gujian_num = [1, "合格"] else: gujian_num = [0, "不合格"] # 工作状态切换 work_status_str = "".join(work_status) wr = work_status_str.count("01") if wr > 4: work_jg = [1, "合格"] else: work_jg = [0, "不合格"] # 检查卡号及有效期 card_jg = self.sim_info(iccid) # rtu 版本号 rtu = "" if status_result: status_result = status_result[0] device_data_status = status_result.get("cbd_status", "") if device_data_status: device_data = ast.literal_eval(device_data) rtu = device_data.get("dver") if rtu == self.set_dver: rtu_num = [1, "合格"] else: rtu_num = [0, "不合格"] else: rtu_num = [0, "不合格"] return [ device_id_info, [1, self.set_order], card_jg, lng_jg, lat_jg, work_jg, gujian_num, rtu_num, [0, "数据不存在"], tr_k_jg, rain_k_jg, infor_jg_count, heat_temp, en_temp, en_hum, current_jg, up_jg, down_jg, lig_jg, img_jg, machine_jg, camera_jg, time_jg ] def run(self): """主业务逻辑,涉及进度条不能模块化,慢慢捋""" proess = 0 now_time = datetime.datetime.now() global save_filename save_filename = self.set_order + "_" + now_time.strftime("%m%d") + "测报灯检验.xlsx" save_path = os.path.join(self.save_path,save_filename) workbook = Workbook(save_path) worksheet = workbook.add_worksheet() worksheet.set_row(0, 50) worksheet.set_row(1, 37.5) worksheet.set_row(2, 50) worksheet.set_row(3, 100) for i in range(len(self.device_list)): worksheet.set_row(i+4, 30) worksheet.set_column(0, 26, 24) toji_style = workbook.add_format(toji_format) title_style = workbook.add_format(title_format) explain_style = workbook.add_format(explain_formal) formal_style = workbook.add_format(formal_format) error_style = workbook.add_format(error_format) common_style = workbook.add_format(common_format) merge_title_style = workbook.add_format(merge_title_format) worksheet.merge_range('A1:Y1',"物联网测报灯在线检验原始表格",merge_title_style) title_data = [ "文档ID","设备ID", "任务单号", "卡号及有效期","经度","纬度","工作状态切换", "电路板固件版本号","RTU固件版本号", "电路板输入电压","温控","雨控","上报信息条数","加热仓温度","环境温度","环境湿度","振动电机工作状态下 \n电流匹配度", "上仓门电机电机工作状态下 \n电流匹配度","下仓门电机工作状态下 \n电流匹配度","灯管工作状态下 \n电流匹配度","图片质量", "清扫电机/转盘电机工作状态下的 \n最大电流","相机工作状态下 \n电流匹配度", "工作时长", "综合判定" ] worksheet.write_row(row = 2 ,col = 0, data = title_data,cell_format=title_style) explain_data = [ "/", "后台显示与输入一致为合格，否则为不合格", "按照输入文件的任务单为依据复制到此列", "有显示卡号及有效期即为合格，否则为不合格", "合格条件(绿色)：\n113°46′13″±30″范围内，转为十进度为：(113.76194444,113.77861111)区间内", "合格条件(绿色)：\n35°1′45″±30″范围内，转为十进度为：(35.02083333,35.0375)区间内", "合格条件(绿色)：\n测试时间段内有待机/工作且出现4次以上循环", "合格条件(绿色)：\n后台上报结果与输入标准对照，相符则合格，其余则不合格", "合格条件(绿色)：\n后台上报结果与输入标准对照，相符则合格，其余则不合格", "合格条件(绿色)：\n电路板输入电压：22~30V", "合格条件(绿色)：\n测试时间段之前至少有一条被温控数据上报", "合格条件(绿色)：\n测试时间段内至少有一条被雨控数据上报", "合格条件(绿色)：\n检测时间大于10h，在灯管状态栏内出现“工作”状态至少大于22条，不允许出现上传间隔小于5分钟的条数2条", "合格条件(绿色)：\n上传的加热温度包含但不局限于85~105°之间的值", "合格条件(绿色)：\n等于35且环境温度等于25的前提下为不合格，其余显示的任何值为合格", "合格条件(绿色)：\n等于25度且环境湿度等于35的前提下为不合格，其余显示的任何值为合格", "合格条件(绿色)：\n在工作时间段内没有接收到“电流不足”的信息为合格，否则不合格", "合格条件(绿色)：\n在工作时间段内没有接收到“电流不足”的信息为合格，否则不合格", "合格条件(绿色)：\n在工作时间段内没有接收到“电流不足”的信息为合格，否则不合格", "合格条件(绿色)：\n在工作时间段内没有接收到“电流不足”的信息为合格，否则不合格", "合格条件(绿色)：\n对自动检查前上传的图片质量进行打分，平均分数大于68分为合格，否则，不合格", "合格条件(绿色)：\n在工作时间段内没有接收到“电流不足”的信息为合格，否则不合格", "合格条件(绿色)：\n在工作时间段内没有接收到“电流不足”的信息为合格，否则不合格", "合格条件(绿色)：\n与供电选择输入对照，（DC为4h，AC为9h）一致判定为合格，否则为不合格", "合格条件(绿色)：\nd:y列全部合格为合格，否则为不合格" ] worksheet.write_row(row = 3 ,col = 0, data = explain_data,cell_format=explain_style) formal_counts = 0 verify_list = [] for index, short_id in enumerate(self.device_list): short_id = str(short_id).strip() # 把device_id转换成字符串并去除空格 d_id,deviceId,platform = self.device_their_platform(short_id) if short_id != deviceId: device_id_info = [0, "不合格"] else: device_id_info = [1, "合格"] worksheet.write(index+4,0,short_id,formal_style) print("ping success", platform) if platform == "大数据平台": try: worksheet.write(index+4,1,deviceId,formal_style) self.mongo_ping() data_status = self.cbd_alerm.find({"equip_id": deviceId, "alarm_time": {"$gte":self.start_time ,"$lte":self.end_time}}).sort([('alarm_time', pymongo.DESCENDING)]) data_cursor = self.cbd_collection.find({"device_id":d_id,'addtime': {"$gte":self.start_time ,"$lte":self.end_time}}).sort([('addtime', pymongo.DESCENDING)]) data_counts = data_cursor.count() image_score = self.__bigdata_iamge_verify(d_id) if data_counts == 0: for i in range(1,24): worksheet.write(index+4,i,"搜索时间内无数据",error_style) worksheet.write(index+4,23,"不合格",error_style) danji_verify = [] for i in range(23): danji_verify.append(0) verify_list.append(danji_verify) else: danji_verify = [] verify_data = self.__bigdata_verify(data_cursor,data_status, device_id_info, image_score, proess) for clo_index, verify_ in enumerate(verify_data): danji_verify.append(verify_[0]) if verify_[0] == 0: worksheet.write(index+4,clo_index+1,verify_[1],error_style) elif verify_[0] == 1: worksheet.write(index+4,clo_index+1,verify_[1],formal_style) else: worksheet.write(index+4,clo_index+1,verify_[1],common_style) if 0 in danji_verify: worksheet.write(index+4,24,"不合格",error_style) else: formal_counts += 1 worksheet.write(index+4,24,"合格",formal_style) verify_list.append(danji_verify) except Exception as e: print(e) elif platform == "四情平台": try: worksheet.write(index+4,1,deviceId,formal_style) self.sql_ping() data_sql = "SELECT * FROM AppInfoManage_cbddata WHERE equip_id_id='{}' AND upl_time BETWEEN '{}' AND '{}' order by upl_time;".format(deviceId,self.start_time_str,self.end_time_str) alarm_sql = "SELECT * FROM AppInfoManage_cbd_alarm_record WHERE equip_id_id='{}' AND alarm_time BETWEEN '{}' AND '{}' order by alarm_time;".format(deviceId,self.start_time_str,self.end_time_str) self.cursor.execute(data_sql) data_result = self.cursor.fetchall() self.cursor.execute(alarm_sql) alarm_result = self.cursor.fetchall() status_sql = "SELECT * FROM AppInfoManage_cbdstatus WHERE equip_id_id='{}' AND upl_time BETWEEN '{}' AND '{}' order by upl_time;".format(deviceId,self.start_time_str,self.end_time_str) self.cursor.execute(status_sql) status_result = self.cursor.fetchall() status_all_sql = "SELECT * FROM AppInfoManage_cbdstatus_all WHERE equip_id_id='{}' AND upl_time BETWEEN '{}' AND '{}' order by upl_time;".format(deviceId,self.start_time_str,self.end_time_str) self.cursor.execute(status_all_sql) status_all_result = self.cursor.fetchall() if len(data_result) == 0: # 空白数据填充 for i in range(1,24): worksheet.write(index+4,i,"搜索时间内无数据",error_style) worksheet.write(index+4,24,"不合格",error_style) danji_verify = [] for i in range(23): danji_verify.append(0) verify_list.append(danji_verify) else: image_score = self.__siqing_image_verify(deviceId) verify_data = self.__siqing_verify(data_result,status_result, status_all_result, alarm_result, device_id_info, image_score, process) danji_verify = [] for clo_index, verify_ in enumerate(verify_data): danji_verify.append(verify_[0]) if verify_[0] == 0: worksheet.write(index+4,clo_index+1,verify_[1],error_style) elif verify_[0] == 1: worksheet.write(index+4,clo_index+1,verify_[1],formal_style) else: worksheet.write(index+4,clo_index+1,verify_[1],common_style) if 0 in danji_verify: worksheet.write(index+4,24,"不合格",error_style) else: formal_counts += 1 worksheet.write(index+4,24,"合格",formal_style) verify_list.append(danji_verify) except Exception as e: print(e) pass proess = (index+1)/len(self.device_list) * 100 if int(proess) == 100: self.proess_signal.emit(99) else: self.proess_signal.emit(int(proess)) toji_data = [ "任务单号",self.set_order, "检验时间",'{}\n{}'.format(self.start_time_str,self.end_time_str), "检验数量",len(self.device_list), "合格数",formal_counts, "检验平台",self.set_plat, "主板版本",self.set_stm8vs, "联网版本",self.set_dver, "产品名称", self.set_product, "供电选择", self.set_power, "报告日期",now_time.strftime("%y-%m-%d %H:%M:%S") ] worksheet.write_row(row = 1 ,col = 0, data = toji_data,cell_format=toji_style) # worksheet.protect(pwd_str) # 保护视图，不允许修改 workbook.close() self.cursor.close() self.connection.close() self.myclient.close() self.proess_signal.emit(100) if __name__== "__main__": QtWidgets.QApplication.setAttribute(QtCore.Qt.AA_EnableHighDpiScaling) QtGui.QGuiApplication.setAttribute(QtCore.Qt.HighDpiScaleFactorRoundingPolicy.PassThrough) app = QtWidgets.QApplication(sys.argv) MainWindow = QtWidgets.QMainWindow() ui = Ui_MainWindow() ui.setupUi(MainWindow) MainWindow.show() sys.exit(app.exec_())