billetManager/models/data_checker.py

from utils.s7data import S7data
from utils.logger import Logger
import requests, time, threading

class Checker:
    def __init__(self, data_5: S7data, data_6: S7data, logger: Logger):
        self.data_5 = data_5
        self.data_6 = data_6
        self.logger = logger

        self.already = set()

        self.policy_5 = {
            '定尺看门狗1': [10, '5#机1-4流定尺数据'],
            '定尺看门狗2': [10, '5#机5-8流定尺数据'],
            '推钢机激光': [1800, '5#机组坯、装运等除直轧外所有数据'],
            '天车A1位置': [1800, '5#机堆垛、装车数据'],
            '天车A2位置': [1800, '5#、6#机堆垛、装车、高线数据'],
            '天车A3位置': [1800, '6#机堆垛、装车、高线数据'],
            '车位1视觉看门狗': [10, '5#机车位1到站离站信号'],
            '车位2视觉看门狗': [10, '6#机车位2到站离站信号'],
            '车位3视觉看门狗': [10, '6#机车位3到站离站信号'],
            '车位4视觉看门狗': [10, '6#机车位4到站离站信号'],
            '车位1车牌看门狗': [10, '5#机车位1车牌识别'],
            '车位2车牌看门狗': [10, '6#机车位2车牌识别'],
            '车位3车牌看门狗': [10, '6#机车位3车牌识别'],
            '车位4车牌看门狗': [10, '6#机车位4车牌识别']
        }

        self.policy_6 = {
            '定尺看门狗1': [10, '6#机1-4流定尺数据'],
            '定尺看门狗2': [10, '6#机5-8流定尺数据'],
            '推钢机激光': [300, '6#机钢坯基础信息、组坯等所有数据']
        }

        self.time_5 = {}
        self.time_6 = {}

        self.value_5 = {}
        self.value_6 = {}

        now = time.time()
        for i in self.policy_5.keys():
            self.time_5[i] = now
            self.value_5[i] = 0
        for i in self.policy_6.keys():
            self.time_6[i] = now
            self.value_6[i] = 0

        self.threading_sig = True

    def push(self, name, ccmNo):
        url = 'https://oapi.dingtalk.com/robot/send?access_token=60fbad9923c55730485ead3270621fd0a03fdc28d6a8e78a85a4ce4d0ee727dd'
        tmp = {}
        tmp['msgtype'] = 'markdown'
        tmp['markdown'] = {}
        tmp['markdown']['title'] = '数据异常警告'
        tmp['markdown']['text'] = f'''#### 警告：钢坯数字化项目数据异常！

**异常详情：**
- **异常信号**：{name}
- **异常描述**：{self.policy_5[name][0] if ccmNo == '5' else self.policy_6[name][0]}秒无响应
- **发生时间**：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}
- **影响范围**：{self.policy_5[name][1] if ccmNo == '5' else self.policy_6[name][1]}

---

> 看门狗信号异常请立即检查，其他异常请先检查是否设备停机或短暂停止'''
        response = requests.post(url, json=tmp)
        if response.ok and response.json()['errcode'] == 0:
            self.logger.warning(f"[WARNING]{name}信号异常，已经提交了报告")
        else:
            self.logger.warning(f"[WARNING]{name}信号异常，但提交报告失败")

    def start_check(self):
        while self.threading_sig:
            for i, j in self.policy_5.items():
                newdata = self.data_5.get_value(i)
                if newdata != self.value_5[i]:
                    self.value_5[i] = newdata
                    self.time_5[i] = time.time()
                if '5#'+i not in self.already and time.time() - self.time_5[i] > j[0]:
                    self.push(i, '5')
                    self.already.add('5#'+i)

            for i, j in self.policy_6.items():
                newdata = self.data_6.get_value(i)
                if newdata != self.value_6[i]:
                    self.value_6[i] = newdata
                    self.time_6[i] = time.time()
                if '6#'+i not in self.already and time.time() - self.time_6[i] > j[0]:
                    self.push(i, '6')
                    self.already.add('6#'+i)

            time.sleep(5)

    def async_start_check(self):
        self.thread = threading.Thread(target=self.start_check)
        self.thread.start()

    def clear(self):
        self.already = set()