jikong_backend_code/005_H570/jikong_script/alarm_buzzer.py

#!/usr/bin/env python
# encoding: utf-8
"""
@CreateTime: 2022/06/16 10:48
@Author: lxc
@LastEditTime: 
@Desctiption:根据数据表存储的报警，判断蜂鸣器是否需要报警
"""

from event_storage import EventStorage
from log import OutPutLog
import time
from datetime import datetime
from utility import Utility


class AlarmBuzzer:
    def __init__(self):
        self._storage = EventStorage()
        self._log = OutPutLog()

    def run(self):

        run_time = 0
        while True:
            this_time = time.time()
            if this_time - run_time >5:
                run_time = this_time

                self.judge_buzzer_alarm()


    def judge_buzzer_alarm(self):
        # 蜂鸣器状态：c809
        read_real_list = ["c809"]
        real_data_dict = self._storage.get_real_data(read_real_list)
        buzzer_status = real_data_dict["c809"]          # 蜂鸣器当前状态，0：蜂鸣器关闭，1：蜂鸣器开启
        if buzzer_status is not None:
            buzzer_status = int(buzzer_status)

        get_set_buzzer_status = "SELECT buzzer_status FROM `alarm_set`;"
        set_buzzer_status = self._storage.execute_sql(get_set_buzzer_status)    # 是否开启蜂鸣器报警功能，0=关闭，1=开启
        # print("-------------------", set_buzzer_status)

        if set_buzzer_status[0]["buzzer_status"] == 1:
            # 判断是否有报警点，如果有，则蜂鸣器报警
            get_alarm_sql = "SELECT id,CAST(create_time AS CHAR) AS create_time,CAST(update_time AS CHAR) AS update_time,point_name,data,buzzer_status_time,signal_type FROM alarm_tbl WHERE is_cancel=0;"
            alarm_data = self._storage.execute_sql(get_alarm_sql)

            if alarm_data:
                buzzer_status_time = alarm_data[0]["buzzer_status_time"]
                self.now_time = datetime.now()

                if buzzer_status is None or buzzer_status_time is None:
                    # turn on the buzzer
                    if self.send_command("open"):
                        self.update_alarm_tbl()
                    else:
                        print("turn on the buzzer failed!")

                else:
                    # 计算蜂鸣器保持开启或关闭状态的时长
                    buzzer_hold_time = (self.now_time - buzzer_status_time).total_seconds()
                    if buzzer_status == 0 and buzzer_hold_time > 60 * 10:
                        # buzzer off and turn on the buzzer
                        if self.send_command("open"):
                            self.update_alarm_tbl()
                        else:
                            print("turn on the buzzer failed!")

                    elif buzzer_status == 1 and buzzer_hold_time > 60 * 3:
                        # buzzer on and turn off the buzzer
                        if self.send_command("close"):
                            self.update_alarm_tbl()
                        else:
                            print("turn off the buzzer failed!")
            elif buzzer_status == 1:
                # 没有报警点 判断蜂鸣器状态 如果蜂鸣器在响 关掉蜂鸣器
                if self.send_command("close"):
                    print("turn off the buzzer success!")
                else:
                    print("turn off the buzzer failed！")


        elif buzzer_status == 1:
            # 无论是否有报警，都关闭蜂鸣器
            if self.send_command("close"):
                print("turn off the buzzer success!")
            else:
                print("turn off the buzzer failed！")



    def send_command(self, status):
        '''给蜂鸣器发送指令，并更新数据库'''
        if status == "open":
            # turn on the buzzer
            command = {"device_id": 1, "start_addr": 0, "output_value": 65280, "function_code": 5, "res": 65280}
        elif status == "close":
            # turn off the buzzer
            command = {"device_id": 1, "start_addr": 0, "output_value": 0, "function_code": 5, "res": 0}
        elif status == "status":
            # views the buzzer status
            command = {"device_id": 1, "start_addr": 0, "length": 1, "function_code": 1}

        station_name = "buzzer"
        commend_result = Utility.available_connectors[station_name].send_command(command)
        return commend_result


    def update_alarm_tbl(self):
        '''更新数据库的报警信息'''
        update_sql = f"UPDATE alarm_tbl SET buzzer_status_time=\'{self.now_time}\' WHERE is_cancel=0;"
        update_result = self._storage.execute_update_sql(update_sql)
        # print(update_result)
        return update_result