wuyuanbiaoba/upload/DataReporter.py

import logging
import queue
import sched
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from time import sleep

import configSer
import models.msg
from upload import mqttHelper
from upload.RateLimiter import RateLimiter

mq_client:mqttHelper.MQTTClient = None

# 启动 MQTT 连接线程
def start_mqtt_connection_thread():
    """
    启动一个后台线程用于 MQTT 连接
    """
    mqtt_thread = threading.Thread(target=connect_mqtt_with_retry, args=())
    mqtt_thread.daemon = True  # 设置为守护线程，随主线程退出而结束
    mqtt_thread.start()

def connect_mqtt_with_retry(max_retries=5, base_delay=10):
    """
    尝试连接 MQTT 并支持重试机制
    :param max_retries: 最大重试次数
    :param base_delay: 初始延迟时间（秒）
    """
    is_mqtt_connected=False
    retry_count = 1
    delay = base_delay

    while not is_mqtt_connected and retry_count <= max_retries:
        try:
            logging.info(f"尝试连接 MQTT... 第 {retry_count} 次")
            if mq_client.start():
                logging.info("MQTT 连接成功！")
                is_mqtt_connected= True
                break
            else:
                logging.info("MQTT 连接失败，稍后重试...")
        except Exception as e:
            logging.info(f"MQTT 通信异常: {e}")

        retry_count += 1
        sleep(delay)

    if not mq_client.is_connected():
        logging.info(f"MQTT 连接失败，已达最大重试次数{max_retries}")

class DataReporter(threading.Thread):
    def __init__(self,data_fps:int,video_fps:int=0):
        super().__init__()
        self.alert_queue = queue.Queue(maxsize=1)  # 数据队列
        self.image_queue = queue.Queue(maxsize=video_fps)  # 图片队列
        self.data_queue = queue.Queue(maxsize=data_fps)  # 数据队列
        self.image_limiter = RateLimiter(max_rate=video_fps, time_window=1)  # 弃用 图片限速: 1张/秒
        self.data_limiter = RateLimiter(max_rate=data_fps, time_window=1)  # 数据限速: 30条/秒
        self.running = True
        self.update = False
        self.daemon = True
        self.tcp_callbacks = []  # 存储多个回调函数
        self.mqtt_callbacks = []  # 存储多个回调函数
        self.callback_executor = ThreadPoolExecutor(max_workers=2)  # 创建线程池
        self.last_data = None  # 存储最后一条数据
        self.last_repeat_counts = 0  # 存储最后一条数据

        self.scheduler = sched.scheduler(time.time)
        self.schedulerEvent = None


    def register_handler(self,handler_fun,conn_type="mqtt"):
        match conn_type:
            case "mqtt":
                self.mqtt_callbacks.append(handler_fun)
            case "tcp":
                self.tcp_callbacks.append(handler_fun)
            case _:
                logging.warn(f"不支持的回调连接类型: {conn_type}")
    def run(self):
        now_timestamp = time.time()
        interval = 1.0 / self.data_limiter.max_rate
        self.schedulerEvent=self.scheduler.enter(interval, 2, self._schedule_data_reporting, (interval,now_timestamp))
        self.scheduler.run()

    def _schedule_data_reporting(self,interval,now_timestamp):
        if self.running:
            if self.update:
                interval = 1.0 / self.data_limiter.max_rate
                self.update=False
            next_timestamp = now_timestamp+interval
            self.scheduler.enterabs(next_timestamp, 2, self._schedule_data_reporting, (interval,next_timestamp))  # 每 x 秒执行一次
            threading.Thread(target=self.upload, daemon=True).start()

    def upload(self):
        self._report_data_if_allowed()
        # 告警
        self._report_alert_if_allowed()
        sleep(0.1)

    def _report_data_if_allowed(self):
        if not self.data_queue.empty():
            data = self.data_queue.get_nowait()
            self.last_data = data
            self.last_repeat_counts = 0
            self._report_data(data, self.last_repeat_counts)
        elif self.last_data is not None and self.last_repeat_counts < 6:
            self.last_repeat_counts += 1
            self._report_data(self.last_data, self.last_repeat_counts)

    def _report_alert_if_allowed(self):
        # alert上报
        if not self.alert_queue.empty():
            try:
                alert_data = self.alert_queue.get_nowait()
                self._report_alert(alert_data)
            except queue.Empty:
                pass

    def _execute_callbacks(self, msg_json):
        """
        执行所有注册的回调函数
        """
        # logging.info(f"will Reporting to {len(self.tcp_callbacks)} tcp client")
        for callback in self.tcp_callbacks:
            try:
                # 提交到线程池异步执行
                self.callback_executor.submit(callback, msg_json)
            except Exception as e:
                print(f"Error executing callback {callback.__name__}: {e}")
    def _execute_mqtt_callbacks(self, msg_json):
        for callback in self.mqtt_callbacks:
            try:
                # 提交到线程池异步执行
                self.callback_executor.submit(callback, msg_json)
            except Exception as e:
                print(f"Error executing callback {callback.__name__}: {e}")
    def _report_alert(self, data):
        print(f"Reporting alert, timestamp: {data[0]}")
        # 这里替换为实际的上报代码
        msg = models.msg.Msg(_from="dev", cmd="alert", values=data[1])
        msg_json = msg.to_json()
        self._execute_mqtt_callbacks(msg_json)
        self._execute_callbacks(msg_json)

    def _report_image(self, data):
        # 实现图片上报逻辑
        print(f"Reporting image, timestamp: {data[0]}")
        # 这里替换为实际的上报代码
        msg = models.msg.Msg(_from="dev", cmd="image", values=data[1])
        msg_json = msg.to_json()
        self._execute_callbacks(msg_json)

    def _report_data(self, data, last_repeat_counts=0):
        # 实际的上报代码,数据结构转换
        msg=models.msg.Msg(_from="dev",cmd="data",values=data[1])
        # 重新格式化时间，数据等间隔
        msg.values.time=datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
        logging.info(f"Reporting [{last_repeat_counts}] data: {data}")
        msg_json=msg.to_json()
        self._execute_mqtt_callbacks(msg_json)
        self._execute_callbacks(msg_json)

    def stop(self):
        self.running = False
        self.join()

    def adjust_rate(self, new_rate, data_type='data'):
        with self.data_limiter.lock:
            self.data_limiter.max_rate = new_rate
            self.data_queue = queue.Queue(maxsize=new_rate)
            self.update=True
            self.data_limiter.update_interval()

    def adjust_mqtt(self, conf: configSer.Upload) -> bool:
        global mq_client
        self.mqtt_callbacks=[]
        if mq_client is not None and mq_client.is_connected():
            mq_client.stop()

        if conf.enable:
            mq_client = mqttHelper.MQTTClient(**conf.mqtt)
            try:
                if mq_client.start():
                    self.register_handler(mq_client.publish,"mqtt")
                else:
                    logging.warn("mq链接失败，不上报")
                    return False
            except Exception as e:
                logging.warn("adjust_mqtt 通讯异常", e)
                sys.exit(0)
        return True