scripting_tools/bigdata_django/zhijian/qxz/main.py

# encoding=utf-8

from cgi import print_arguments
from multiprocessing.managers import DictProxy
from operator import delitem
from unittest import result
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 uuid
import pymongo
import pymysql
from collections import defaultdict, Counter

from xlsxwriter.workbook import Workbook
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
)


def run():
    start = "2023-08-28 10:00:00"
    end = "2023-08-31 20:00:00"
    dever_num = "111"
    # 获取所有的device_id
    device_list = get_excel_content("F:\\scripting_tools\\热通量增加测试.xlsx")
    workbook = Workbook("F:\\test.xlsx")
    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 = [
        "任务单号", 
        "set_order", 
        "检验时间", 
        "start_time_str", 
        "end_time_str",
        "报告日期", 
        (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:
                print(name)
                print(data)
                print("-"*100)
                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) - 3:
            last = [1, "合格"]
            is_pass_count += 1
        else:
            last = [2, f"不合格: {len(title_name_list) - is_right - 3}"]
        worksheet.write(index + 4, i, last[1], style_dict[last[0]])
    worksheet.write(1, 8, is_pass_count, toji_style)
    workbook.close()


if __name__ == "__main__":
    run()