scripting_tools/bigdata_django/zhijian/qxz/main_test.py

# encoding=utf-8
from PyQt5 import QtCore, QtGui, QtWidgets
import os
import sys
import datetime
from xlrd import open_workbook
from xlsxwriter.workbook import Workbook
import openpyxl


from mongoclient import (
    get_qxz_conf, 
    device_detail_deivce_id, 
    get_sa_qxz_info_record,
    get_sa_qxz_data,
    get_conf_data,
    get_sim_info
)
from utils import (
    merge_title_format, 
    toji_format, 
    formal_format, 
    error_format, 
    common_format, 
    default_formal,
    get_excel_content,
    head_dict,
    sim_info
)

save_filename = ""


class SCDThread(object):
    """涉及进度条需主界面动态执行GUI,线程执行业务逻辑,避免主页面卡死"""
    def __init__(self,device_list,save_path,start_time,end_time,set_plat,set_stm8vs,set_order):

        self.save_path = save_path
        self.device_list = device_list
        self.start_time = start_time
        self.end_time = end_time
        self.set_plat = set_plat
        self.set_stm8vs = set_stm8vs
        self.set_order = set_order


    def run(self):
        start = self.start_time.strftime("%Y-%m-%d %H:%M:%S")
        end = self.end_time.strftime("%Y-%m-%d %H:%M:%S")
        dever_num = self.set_stm8vs
        save_filename = self.set_order + "_" + datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S") + ".xlsx"
        save_path = os.path.join(self.save_path,save_filename)
        workbook = Workbook(save_path)
        device_list = self.device_list
        worksheet = workbook.add_worksheet()
        merge_title_style = workbook.add_format(merge_title_format)
        toji_style = workbook.add_format(toji_format)
        default_style = workbook.add_format(default_formal)
        red_style = workbook.add_format(error_format)
        green_style = workbook.add_format(formal_format)
        yellow_style = workbook.add_format(common_format)
        style_dict = {
            2: red_style,
            1: green_style,
            3: yellow_style,
            4: default_style
        }

        title_name_list = [
            "ID", 
            "检验项目", 
            "电压", 
            "信号强度", 
            "经度", 
            "纬度", 
            "固件版本号", 
            "sim卡信息",
            "位置信息"
        ]
        # 查找第一台设备的配置文件，获取配置项
        device_id, _, _, _, _ = device_detail_deivce_id(device_list[0])
        conf = get_qxz_conf(device_id)
        title_name_list += [v.split("#")[0] for k, v in conf.items()]
        title_name_list.append("单台合格数")
        head_list = [i.split("#")[0] for i in title_name_list]
        worksheet.merge_range(0, 0, 0, len(head_list) - 1, "物联网气象站设备质检表格", merge_title_style)
        toji_data = [
            "任务单号", 
            self.set_order, 
            "检验时间", 
            start, 
            end,
            "报告日期", 
            (datetime.datetime.now()).strftime("%y-%m-%d %H:%M:%S"), 
            "合格数"
        ]
        # 生成Excel表格第2行， 本次数据概要
        for index, k in enumerate(toji_data):
            value = str(k)
            worksheet.write(1, index, value, toji_style)
        # 生成Excel表格第3行， 检测标题
        for index, k in enumerate(title_name_list):
            worksheet.write(2, index, k, default_style)
        # 生成Excel表格第4行， 检测标准
        for index, k in enumerate(title_name_list):
            value = (head_dict.get(k, ["无判定条件", ""]))[0]
            worksheet.write(3, index, value, default_style)
            plus = 12 if index in [0, 1] else 8
            c_n = len(value) / 2 + plus
            worksheet.set_column(index, index, c_n)
        # 输入数据
        is_pass_count = 0
        for index, id in enumerate(device_list):
            device_id, dver_num, lng, lat, position = device_detail_deivce_id(id)
            # 获取电压，信号强度历史数据
            valt_rssi = get_sa_qxz_info_record(device_id, start, end)
            # 获取sim卡信息
            sim = get_sim_info(device_id, start, end)
            sim_in = sim_info(sim)
            # 第一部分拼接 表头及数据
            org_data = {
                "id_1": {"data": id, "explan": "ID"},
                "id_2": {"data": device_id, "explan": "检验项目"},
                "id_3": valt_rssi.get("volt"),  # 电压
                "id_4": valt_rssi.get("rssi"),  # 信号强度
                "id_5": {"data": lng, "explan": "经度"},
                "id_6": {"data": lat, "explan": "纬度"},
                "id_7": {"data": [dver_num.replace("/", "-"), dever_num], "explan": "固件版本号"},
                "id_8": {"data": sim_in, "explan": "sim卡信息"},
                "id_9": {"data": position, "explan": "位置信息"},
            }
            conf = get_qxz_conf(device_id)
            # 获取数据
            qx_data = get_sa_qxz_data(device_id, start, end, conf)
            # 第二部分拼接 表头及数据   获取气象与数据结合
            conf_data = get_conf_data(conf, qx_data)
            org_data.update(conf_data)
            i = 0
            is_right = 0
            for t, v in org_data.items():
                # 执行方法
                name = v.get("explan")
                data = v.get("data")
                func = head_dict.get(name)[1]
                if data:
                    temp = func(data)
                    if temp[0] == 1:
                        is_right += 1
                else:
                    temp = [2, "没有数据"]
                worksheet.write(index + 4, i, temp[1], style_dict[temp[0]])
                i += 1
            last = []
            if is_right == len(title_name_list) - 1:
                last = [1, "合格"]
                is_pass_count += 1
            else:
                last = [2, "不合格"]
            worksheet.write(index + 4, i, last[1], style_dict[last[0]])
        worksheet.write(1, 8, is_pass_count, toji_style)
        worksheet.protect("yfkj")
        workbook.close()


if __name__== "__main__":
    start_time = datetime.datetime.strptime("2023-11-20 18:00:00", "%Y-%m-%d %H:%M:%S")
    end_time = datetime.datetime.strptime("2023-11-21 20:00:00", "%Y-%m-%d %H:%M:%S")
    device_list,save_path,set_plat,set_stm8vs,set_order = ["864865060469831"], "f:/scripting_tools/bigdata_django/zhijian/qxz/", "1", "1", "1"
    qxz = SCDThread(device_list,save_path,start_time,end_time,set_plat,set_stm8vs,set_order)
    qxz.run()