机舱采集器——不在使用

1.Software/RK3568/gateway/main.py

@@ -0,0 +1,109 @@
+import daemon
+import time
+import logging
+import logging.handlers
+import multiprocessing
+from multiprocessing import Event
+from src.usb import Usb
+from src.mqtt import Mqtt
+from src.payload import USBdata
+from src.config import Config
+import signal
+from daemon.pidfile import PIDLockFile
+import subprocess
+
+
+
+class App():
+    def __init__(self):
+        # 加载配置文件
+        self.cfg = Config()
+
+    def run(self):
+        # 打开工作指示灯
+        subprocess.run(["/usr/local/gateway/led", "89", "1"])
+        time.sleep(1)
+
+        signal.signal(signal.SIGINT, self.exit)
+        signal.signal(signal.SIGTERM, self.exit)
+        # signal.signal(signal.SIGKILL, self.exit)
+        signal.signal(signal.SIGHUP, self.exit)
+
+        # 初始化日志
+        log = logging.getLogger('gateway')
+        log.setLevel(logging.DEBUG)  # 实际运行一般用info
+
+        # 创建一个StreamHandler对象，用于将日志输出到终端上
+        console = logging.StreamHandler()
+        console.setLevel(logging.DEBUG)  # 终端上只显示DEBUG及以上级别的日志
+
+        # 创建一个RotatingFileHandler对象，用于将日志写入到指定路径的文件中
+        fh = logging.handlers.RotatingFileHandler(self.cfg.Sys.logpath, maxBytes=1000000,
+                                                  backupCount=1)  # 滚动式日志文件位置、大小、数量部署时按需修改
+        fh.setLevel(logging.DEBUG)
+
+        # 为两个Handler对象设置日志记录格式
+        formatter = logging.Formatter(u'%(asctime)s [%(levelname)s] %(message)s')  # 日志格式
+        console.setFormatter(formatter)
+        fh.setFormatter(formatter)
+
+        # 将console和fh两个Handler对象添加到Logger对象log中
+        log.addHandler(console)
+        log.addHandler(fh)
+
+        # 公共队列
+        q1 = multiprocessing.Queue()
+        q2 = multiprocessing.Queue(maxsize=50)
+        # 公共事件
+        self.event = Event()
+      
+        # 处理线程，包括参数计算和MQTT发送
+        mqtt = Mqtt(self.cfg.Mqtt, log)
+        err = mqtt.Connect()
+        # 服务器未连接，重连
+        while err != 0:
+            log.error("正在重连...")
+            err = mqtt.Connect()
+            time.sleep(10)
+
+        USBdata_thread = multiprocessing.Process(target=USBdata, args=(q1, q2, log))
+        mqtt_thread = multiprocessing.Process(target=mqtt.Start, args=(q2, self.event, self.cfg.Mqtt))
+
+       
+        # USB线程
+        usb = Usb(self.cfg.Usb, log)
+        err = usb.Connect()
+        if err != 0:
+            exit(err)
+
+        USBGetInfo = multiprocessing.Process(target=usb.WriteDataWork, args=(q1, self.event))
+        USBKeepAliveWork = multiprocessing.Process(target=usb.KeepAliveWork)
+
+        USBGetInfo.start()
+        USBKeepAliveWork.start()
+        mqtt_thread.start()
+        USBdata_thread.start()
+
+        USBGetInfo.join()
+        USBKeepAliveWork.join()
+        mqtt_thread.join()
+        USBdata_thread.join()
+
+        log.info("program exit!")
+        usb.Disconnect()
+        mqtt.Disconnect()
+
+    def exit(self, signum, frame):
+        self.event.set()
+
+
+app = App()
+
+pidfile = PIDLockFile(app.cfg.Sys.pidpath)
+
+# 判断pid文件是否存在或锁定
+if pidfile.is_locked():
+    pidfile.break_lock()
+
+with daemon.DaemonContext(pidfile=pidfile):
+    app.run()

1.Software/RK3568/gateway/src/config.py

@@ -0,0 +1,86 @@
+import toml
+
+path = "/usr/local/gateway/src/config.toml"
+
+#参数配置，导入配置文件
+class Config:
+    def __init__(self):
+        cfg = toml.load(path)
+        self.Mqtt = MqttConf(cfg)
+        self.Usb = UsbConf(cfg)
+        self.Sys = SysConf(cfg)
+
+
+class MqttConf:
+    def __init__(self, cfg):
+        self.broker = cfg["mqtt"]["broker"]
+        self.port = cfg["mqtt"]["port"]
+        self.client_id = cfg["mqtt"]["client_id"]
+        self.username = cfg["mqtt"]["username"]
+        self.password = cfg["mqtt"]["password"]
+        self.keepalive = cfg["mqtt"]["keepalive"]
+        self.device_id = cfg["mqtt"]["device_id"]
+        self.sample_rate_strategy = cfg["mqtt"]["sample_rate_strategy"]
+        self.long_rate_strategy = cfg["mqtt"]["long_rate_strategy"]
+        self.windfarm = cfg["mqtt"]["windfarm"]
+        self.fansnum = cfg["mqtt"]["fansnum"]
+        self.ch0_measurpoint = cfg["mqtt"]["ch0_measurpoint"]
+        self.ch0_measurpointdirection = cfg["mqtt"]["ch0_measurpointdirection"]
+        self.ch1_measurpoint = cfg["mqtt"]["ch1_measurpoint"]
+        self.ch1_measurpointdirection = cfg["mqtt"]["ch1_measurpointdirection"]
+        self.ch2_measurpoint = cfg["mqtt"]["ch2_measurpoint"]
+        self.ch2_measurpointdirection = cfg["mqtt"]["ch2_measurpointdirection"]
+        self.ch3_measurpoint = cfg["mqtt"]["ch3_measurpoint"]
+        self.ch3_measurpointdirection = cfg["mqtt"]["ch3_measurpointdirection"]
+        self.ch4_measurpoint = cfg["mqtt"]["ch4_measurpoint"]
+        self.ch4_measurpointdirection = cfg["mqtt"]["ch4_measurpointdirection"]
+        self.ch5_measurpoint = cfg["mqtt"]["ch5_measurpoint"]
+        self.ch5_measurpointdirection = cfg["mqtt"]["ch5_measurpointdirection"]
+        self.ch6_measurpoint = cfg["mqtt"]["ch6_measurpoint"]
+        self.ch6_measurpointdirection = cfg["mqtt"]["ch6_measurpointdirection"]
+        self.ch7_measurpoint = cfg["mqtt"]["ch7_measurpoint"]
+        self.ch7_measurpointdirection = cfg["mqtt"]["ch7_measurpointdirection"]
+
+        self.ch0_sensorparameters = int(cfg["mqtt"]["ch0_sensorparameters"])
+        self.ch1_sensorparameters = int(cfg["mqtt"]["ch1_sensorparameters"])
+        self.ch2_sensorparameters = int(cfg["mqtt"]["ch2_sensorparameters"])
+        self.ch3_sensorparameters = int(cfg["mqtt"]["ch3_sensorparameters"])
+        self.ch4_sensorparameters = int(cfg["mqtt"]["ch4_sensorparameters"])
+        self.ch5_sensorparameters = int(cfg["mqtt"]["ch5_sensorparameters"])
+        self.ch6_sensorparameters = int(cfg["mqtt"]["ch6_sensorparameters"])
+        self.ch7_sensorparameters = int(cfg["mqtt"]["ch7_sensorparameters"])
+        
+        self.ch0_samplingtime = int(cfg["mqtt"]["ch0_samplingtime"])
+        self.ch1_samplingtime = int(cfg["mqtt"]["ch1_samplingtime"])
+        self.ch2_samplingtime = int(cfg["mqtt"]["ch2_samplingtime"])
+        self.ch3_samplingtime = int(cfg["mqtt"]["ch3_samplingtime"])
+        self.ch4_samplingtime = int(cfg["mqtt"]["ch4_samplingtime"])
+        self.ch5_samplingtime = int(cfg["mqtt"]["ch5_samplingtime"])
+        self.ch6_samplingtime = int(cfg["mqtt"]["ch6_samplingtime"])
+        self.ch7_samplingtime = int(cfg["mqtt"]["ch7_samplingtime"])
+
+        self.ch0_samplingfrequency = int(cfg["mqtt"]["ch0_samplingfrequency"])
+        self.ch1_samplingfrequency = int(cfg["mqtt"]["ch1_samplingfrequency"])
+        self.ch2_samplingfrequency = int(cfg["mqtt"]["ch2_samplingfrequency"])
+        self.ch3_samplingfrequency = int(cfg["mqtt"]["ch3_samplingfrequency"])
+        self.ch4_samplingfrequency = int(cfg["mqtt"]["ch4_samplingfrequency"])
+        self.ch5_samplingfrequency = int(cfg["mqtt"]["ch5_samplingfrequency"])
+        self.ch6_samplingfrequency = int(cfg["mqtt"]["ch6_samplingfrequency"])
+        self.ch7_samplingfrequency = int(cfg["mqtt"]["ch7_samplingfrequency"])
+
+
+class UsbConf:
+    def __init__(self, cfg):
+        self.vendor = cfg["usb"]["vendor"]
+        self.product = cfg["usb"]["product"]
+        self.sample_rate = cfg["usb"]["sample_rate"]
+        self.sample_time = cfg["usb"]["sample_time"]
+        self.dbpath = cfg["usb"]["dbpath"]
+
+
+class SysConf:
+    def __init__(self, cfg):
+        self.logpath = cfg["sys"]["logpath"]
+        self.pidpath = cfg["sys"]["pidpath"]
+
+

1.Software/RK3568/gateway/src/config.toml

@@ -0,0 +1,67 @@
+
+[mqtt]
+  broker = "192.168.123.200"
+  ch0_measurpoint = "主轴前轴承"
+  ch0_measurpointdirection = "水平"
+  ch0_samplingfrequency = 2000
+  ch0_samplingtime = 40
+  ch0_sensorparameters = 500.0
+  ch1_measurpoint = "主轴后轴承"
+  ch1_measurpointdirection = "水平"
+  ch1_samplingfrequency = 2000
+  ch1_samplingtime = 40
+  ch1_sensorparameters = 500.0
+  ch2_measurpoint = "齿轮箱输入轴承"
+  ch2_measurpointdirection = "水平"
+  ch2_samplingfrequency = 2000
+  ch2_samplingtime = 40
+  ch2_sensorparameters = 500.0
+  ch3_measurpoint = "齿轮箱高速轴叶轮侧"
+  ch3_measurpointdirection = "径向"
+  ch3_samplingfrequency = 4000
+  ch3_samplingtime = 10
+  ch3_sensorparameters = 100.0
+  ch4_measurpoint = "齿轮箱平行级低速级轴承"
+  ch4_measurpointdirection = "轴向"
+  ch4_samplingfrequency = 10000
+  ch4_samplingtime = 6
+  ch4_sensorparameters = 100.0
+  ch5_measurpoint = "齿轮箱平行级低速级轴承"
+  ch5_measurpointdirection = "水平"
+  ch5_samplingfrequency = 10000
+  ch5_samplingtime = 6
+  ch5_sensorparameters = 508.061
+  ch6_measurpoint = "发电机前轴承"
+  ch6_measurpointdirection = "径向"
+  ch6_samplingfrequency = 10000
+  ch6_samplingtime = 3
+  ch6_sensorparameters = 508.061
+  ch7_measurpoint = "发电机后轴承"
+  ch7_measurpointdirection = "径向"
+  ch7_samplingfrequency = 10000
+  ch7_samplingtime = 3
+  ch7_sensorparameters = 100.0
+  client_id = "100"
+  device_id = "10B035041002230110003775"
+  fansnum = "100"
+  keepalive = 150
+  long_rate_strategy = 0
+  password = ""
+  port = 1883
+  sample_rate_strategy = "最大采样频率40KHz*3min"
+  source = "叶片"
+  username = ""
+  windfarm = "安北第四风电场"
+
+[sys]
+  logpath = "/usr/local/gateway/runtime/log/gateway.log"
+  ntp = "192.168.123.200"
+  pidpath = "/usr/local/gateway/runtime/pid/gateway.pid"
+  secret = "e3274be5c857fb42ab72d786e281b4b8"
+
+[usb]
+  dbpath = "/usr/local/gateway/runtime/database/gateway.db"
+  product = 22336
+  sample_rate = 40000
+  sample_time = 180
+  vendor = 1155

1.Software/RK3568/gateway/src/mqtt.py

@@ -0,0 +1,385 @@
+import time
+import datetime
+import psutil
+import numpy as np
+import paho.mqtt.client as mqtt
+from src.config import MqttConf
+from src.config import Config
+from logging import Logger
+from multiprocessing import Queue
+from multiprocessing import Event
+from src.payload import DeviceInfo
+from src.payload import alarmInfo
+from src.payload import Flow
+from src.usb import strategy
+import json
+import gc
+import threading
+import subprocess
+
+
+class Mqtt:
+    def __init__(self, cfg: MqttConf, log: Logger):
+        self.broker = cfg.broker
+        self.port = cfg.port
+        self.client_id = cfg.client_id
+        self.username = cfg.username
+        self.password = cfg.password
+        self.keepalive = cfg.keepalive
+        self.device_id = cfg.device_id
+        self.windfarm = cfg.windfarm
+        self.fansnum = cfg.fansnum
+
+        self.ch0_measurpoint = cfg.ch0_measurpoint
+        self.ch1_measurpoint = cfg.ch1_measurpoint
+        self.ch2_measurpoint = cfg.ch2_measurpoint
+        self.ch3_measurpoint = cfg.ch3_measurpoint
+        self.ch4_measurpoint = cfg.ch4_measurpoint
+        self.ch5_measurpoint = cfg.ch5_measurpoint
+        self.ch6_measurpoint = cfg.ch6_measurpoint
+        self.ch7_measurpoint = cfg.ch7_measurpoint
+
+        self.ch0_measurpointdirection = cfg.ch0_measurpointdirection
+        self.ch1_measurpointdirection = cfg.ch1_measurpointdirection
+        self.ch2_measurpointdirection = cfg.ch2_measurpointdirection
+        self.ch3_measurpointdirection = cfg.ch3_measurpointdirection
+        self.ch4_measurpointdirection = cfg.ch4_measurpointdirection
+        self.ch5_measurpointdirection = cfg.ch5_measurpointdirection
+        self.ch6_measurpointdirection = cfg.ch6_measurpointdirection
+        self.ch7_measurpointdirection = cfg.ch7_measurpointdirection
+
+        self.ch0_samplingtime = cfg.ch0_samplingtime
+        self.ch1_samplingtime = cfg.ch1_samplingtime
+        self.ch2_samplingtime = cfg.ch2_samplingtime
+        self.ch3_samplingtime = cfg.ch3_samplingtime
+        self.ch4_samplingtime = cfg.ch4_samplingtime
+        self.ch5_samplingtime = cfg.ch5_samplingtime
+        self.ch6_samplingtime = cfg.ch6_samplingtime
+        self.ch7_samplingtime = cfg.ch7_samplingtime
+
+        # 确定采样时间
+        max_samplingtime = max(self.ch0_samplingtime, self.ch1_samplingtime, \
+                               self.ch2_samplingtime, self.ch3_samplingtime, \
+                               self.ch4_samplingtime, self.ch5_samplingtime, \
+                               self.ch6_samplingtime, self.ch7_samplingtime)
+        print(max_samplingtime)
+        strategy.sample_time.value = max_samplingtime
+
+        self.ch0_samplingfrequency = cfg.ch0_samplingfrequency
+        self.ch1_samplingfrequency = cfg.ch1_samplingfrequency
+        self.ch2_samplingfrequency = cfg.ch2_samplingfrequency
+        self.ch3_samplingfrequency = cfg.ch3_samplingfrequency
+        self.ch4_samplingfrequency = cfg.ch4_samplingfrequency
+        self.ch5_samplingfrequency = cfg.ch5_samplingfrequency
+        self.ch6_samplingfrequency = cfg.ch6_samplingfrequency
+        self.ch7_samplingfrequency = cfg.ch7_samplingfrequency
+
+        # 确定采样频率
+        max_samplingfrequency = max(self.ch0_samplingfrequency, self.ch1_samplingfrequency, \
+                                    self.ch2_samplingfrequency, self.ch3_samplingfrequency, \
+                                    self.ch4_samplingfrequency, self.ch5_samplingfrequency, \
+                                    self.ch6_samplingfrequency, self.ch7_samplingfrequency)
+        print(max_samplingfrequency)
+        if max_samplingfrequency <= 10000:
+            strategy.sample_rate.value = 10000
+        elif max_samplingfrequency <= 20000:
+            strategy.sample_rate.value = 20000
+        elif max_samplingfrequency <= 40000:
+            strategy.sample_rate.value = 40000
+        else :
+            log.error("最大采样频率超过40K")
+        print(strategy.sample_rate.value)
+        # 确定最大采样数据
+        strategy.FFT_BUF_LEN = strategy.sample_rate.value*strategy.sample_time.value
+
+
+
+        self.log = log
+
+        self.errornum = 0
+
+        client = mqtt.Client(client_id=self.client_id, reconnect_on_failure=True)
+        client.on_connect = self.on_connect
+        client.on_message = self.on_message
+        client.on_disconnect = self.on_disconnect
+        client.on_publish = self.on_publish
+        client.username_pw_set(username=self.username, password=self.password)
+        client.reconnect_delay_set(min_delay=1, max_delay=120)
+
+        self.client = client
+
+    def Connect(self) -> int:
+        try:
+            err = self.client.connect(host=self.broker, port=self.port, keepalive=self.keepalive)
+        except OSError:
+            self.log.error("Network is unreachable. Checking Network or MQTT broker")
+            err = -1
+            #网络断开/服务器关闭红灯闪烁、绿灯长灭
+            RedLedtime1 = time.time()
+            while((time.time()-RedLedtime1)<60):
+                subprocess.run(["/usr/local/gateway/led", "88", "1"])
+                time.sleep(0.5)
+                subprocess.run(["/usr/local/gateway/led", "88", "0"])
+                time.sleep(0.5)
+
+        match err:
+            case 0:
+                self.log.info("Connection broker successful!")
+                # 服务器连接成功，红灯灭，绿灯常亮
+                subprocess.run(["/usr/local/gateway/led", "88", "0"])
+                subprocess.run(["/usr/local/gateway/led", "89", "1"])
+            case 1:
+                self.log.error("Connection broker refused - incorrect protocol version!")
+                return err
+            case 2:
+                self.log.error("Connection broker refused - invalid client identifier!")
+                return err
+            case 3:
+                self.log.error("Connection broker refused - server unavailable!")
+                return err
+            case 4:
+                self.log.error("Connection broker refused - bad username or password!")
+                return err
+            case 5:
+                self.log.error("Connection broker refused - not authorised 6-255: Currently unused!")
+                return err
+
+        return err
+
+    def Disconnect(self):
+        pass
+
+    def Start(self, queue2: Queue, event: Event, cfg: MqttConf):
+        self.client.loop_start()  # 开启一个独立的循环通讯线程。
+        startTime = time.time()
+        alarmTime = time.time()
+        while True:
+
+            # 判断是否继续接受数据
+            while strategy.data_trigger.value:
+                # 退出信号
+                if event.is_set():
+                    break
+                # 如果队列非空        
+                if not queue2.empty():
+                    num = queue2.get()
+                    flow = Flow()
+                    ch_id = num.channel
+                    
+                    # getattr () 函数用于返回一个对象属性值
+                    measurpoint = getattr(self, f"ch{ch_id}_measurpoint")
+                    measurpointdirection = getattr(self, f"ch{ch_id}_measurpointdirection")
+                    samplingtime = getattr(self, f"ch{ch_id}_samplingtime")
+                    samplingfrequency = getattr(self, f"ch{ch_id}_samplingfrequency")
+
+                    # 在edge_compute里面对原始数据进行计算
+                    flow.edge_compute(num, self.windfarm, self.fansnum, measurpoint, measurpointdirection, samplingtime, samplingfrequency)
+                    # 把flow序列化为JSON串
+                    payload = flow.to_json()
+                    # 释放num所引用的内存
+                    num = None
+                    # 确保服务器连接成功，成功就发送数据
+                    if self.client.is_connected():
+                        #发送数据绿灯闪烁
+                        subprocess.run(["/usr/local/gateway/led", "89", "0"])
+                        self.log.info("Transmit %d data to CMS"%ch_id)
+                        # device_no是假的数据，具体怎么来，协议没有约定
+                        device_no = 1
+                        data = json.loads(payload)
+                        topic = f'{self.device_id}/{device_no}/{ch_id}/multivalue'
+                        self.client.publish(topic, payload=payload, qos=1)
+                        subprocess.run(["/usr/local/gateway/led", "89", "1"])
+                        # time.sleep(1)
+                else:
+                    # 触发垃圾回收
+                    # self.log.info("垃圾回收")
+                    gc.collect()
+                    time.sleep(0.1)
+                if (time.time() - startTime) > 15:
+                    # 当mqtt连接断开的时候进行重连
+                    if not self.client.is_connected():
+                        try:
+                            self.Connect()
+                            time.sleep(5)
+                        except OSError as e:
+                            # 处理网络连接异常
+                            self.log.error("Network connection error: %s" % str(e))
+                            time.sleep(1)
+                    startTime = time.time()
+                if (time.time() - alarmTime) > 600:
+                    self.systemAlarminfo()
+                    alarmTime = time.time()
+
+    # 当连接到broker后，进行的回调，中间3个参数暂时保留，视情况可取消，包括subscribe中是否需要除topic外的其他参数，具体待进一步沟通
+    def on_connect(self, client, userdata, flags, rc):
+        self.log.info("连接回调")
+        if rc == 0:
+            self.errornum = 0
+            strategy.mqttconnectstatus.value = True
+            self.log.info("Connected with result code " + str(rc))
+            subprocess.run(["/usr/local/gateway/led", "88", "0"])
+            subprocess.run(["/usr/local/gateway/led", "89", "1"])
+        # 在on_connect中订阅消息，好处是当重连时，订阅仍然会保持，以下是根据文档猜测的需要订阅的消息，实际上并没有用的订阅消息
+        self.client.subscribe(topic=f'{self.device_id}/info', qos=0)
+        self.client.subscribe(topic=f'{self.device_id}/config', qos=0)
+        self.client.subscribe(topic=f'{self.device_id}/sysReset', qos=0)
+        # "{device_id}/{device_no}/{ch_id}/trigger"，表示开始传输数据
+        self.client.subscribe(topic=f'{self.device_id}/trigger', qos=0)
+
+
+    # MQTT断开后执行重连
+    def on_disconnect(self, client, userdata, rc):
+        if rc == 0:
+            self.log.info('MQTT服务连接已关闭')
+        elif 1 <= rc <= 5:
+            self.log.error(f'意外断开：{mqtt.connack_string(rc)}，正在重连 ……')
+            # self.client.reconnect()
+        else:
+            self.log.error(f'未知错误码：{rc}，MQTT服务连接关闭')
+            self.errornum = self.errornum + 1
+            self.log.error("网络故障，准备重启：%d"%(10-self.errornum))
+        # MQTT连接标志位置为0
+        strategy.mqttconnectstatus.value = False
+        
+        if self.errornum >= 10:
+            self.log.info("网络故障：重启")
+            subprocess.run(['sudo','reboot'])
+
+        RedLedtime2 = time.time()
+        while((time.time()-RedLedtime2)<60):
+            # 警报灯（红灯）快闪
+            subprocess.run(["/usr/local/gateway/led", "88", "0"])
+            time.sleep(0.5)
+            subprocess.run(["/usr/local/gateway/led", "88", "1"])
+            time.sleep(0.5)
+
+        try:
+            self.Connect()
+            time.sleep(15)
+        except OSError as e:
+            # 处理网络连接异常
+            self.log.error("Network connection error: %s" % str(e))
+            time.sleep(15)
+        # print("进入了disconnect")
+
+        # time.sleep(15)
+
+    # 当收到订阅消息后
+    def on_message(self, client, userdata, msg):
+        self.log.info(msg.topic + " " + str(msg.payload.decode('utf-8')))
+        a = "this is a test message"
+        b = msg.payload.decode('utf-8')
+        
+        Dinfo = f'{self.device_id}/info'
+        Dconfig = f'{self.device_id}/config'
+        Reset = f'{self.device_id}/sysReset'
+        topic = f'{self.device_id}/trigger'
+
+        if msg.topic == Dinfo:
+            if b == a:
+                self.deviceInfo(msg.payload.decode('utf-8'))
+        elif msg.topic == Dconfig:
+            self.deviceConfig(msg.payload.decode('utf-8'))
+        elif msg.topic == Reset:
+            self.sysReset(msg.payload.decode('utf-8'))
+        elif msg.topic == topic:
+            self.dataTrigger(msg.payload.decode('utf-8'))
+
+
+    # QoS为0时，表示数据离开client，当QoS为1,2时，表示broker已经收到数据，如果要本地保存，那么可以在该处将数据删掉
+    def on_publish(self, client, userdata, flags):
+        pass
+        # self.log.info("data publish successes")
+
+    # 收集硬件信息
+    def deviceInfo(self, payload: bytes | bytearray):
+        deviceInfo = DeviceInfo()
+        deviceInfo = json.dumps(deviceInfo, default=lambda o: o.__dict__, ensure_ascii=False)
+        self.publish("/device/info", deviceInfo)
+        # print('硬件信息发送成功')
+        time.sleep(1)
+
+    # 配置采集信息
+    def deviceConfig(self, payload: bytes | bytearray):
+        deviceConfig = payload
+        deviceConfig = json.loads(deviceConfig)
+        sample_rate = deviceConfig['data']['sample_rate']
+        sample_time = deviceConfig['data']['sample_time']
+        long_rate_strategy = deviceConfig['data']['long_rate_strategy']
+        strategy.sample_rate.value = sample_rate
+        strategy.sample_time.value = sample_time
+        strategy.long_rate_strategy = long_rate_strategy
+
+    # 控制数据传递
+    def dataTrigger(self, payload: bytes | bytearray):
+        dataTrigger = payload
+        dataTrigger = json.loads(dataTrigger)
+        dataTrigger_flag = dataTrigger['data']['send']
+        if dataTrigger_flag == 1:
+            strategy.data_trigger.value = True
+            strategy.write_trigger.value = True
+            self.log.info("开始测量")
+        else:
+            strategy.data_trigger.value = False
+            self.log.info("停止测量")
+            strategy.write_trigger.value = True
+
+    # 重启数采设备
+    def sysReset(self, payload: bytes | bytearray):
+        sysreset = payload
+        sysreset = json.loads(sysreset)
+        sysreset_flag = sysreset['sysreset']
+        if sysreset_flag:
+            self.log.info("进入重启")
+            subprocess.run(['sudo','reboot'])
+
+
+    # 发布消息
+    def publish(self, topic: str, payload: any):
+        self.client.publish(topic=topic, payload=payload)
+
+    # 获取系统状态信息并发送报警信息
+    def systemAlarminfo(self):
+        cfg = Config()
+        # 获取CPU温度信息
+        temp = psutil.sensors_temperatures().get('soc-thermal')[0].current
+        # 获取内存信息
+        mem = psutil.virtual_memory().percent
+        # 获取CPU信息
+        cpu = psutil.cpu_percent(interval=1)
+        # 获取硬盘信息
+        disk = psutil.disk_usage('.').percent
+
+        alarminfo = alarmInfo()
+        now = datetime.datetime.now()
+        milliseconds = round(now.microsecond / 1000)
+
+        formatted_time = now.strftime("%Y-%m-%d %H:%M:%S") + f".{milliseconds:03d}"
+        alarminfo.data.time = formatted_time
+
+        if temp > 80:
+            alarminfo.data.temperate = 1
+        else:
+            alarminfo.data.temperate = 0
+        if mem > 80:
+            alarminfo.data.memory = 1
+        else:
+            alarminfo.data.memory = 0
+        if cpu > 80:
+            alarminfo.data.cpu = 1
+        else:
+            alarminfo.data.cpu = 0
+        if disk > 80:
+            alarminfo.data.disk = 1
+        else:
+            alarminfo.data.disk = 0
+
+        alarminfo.data.dev_id = cfg.Mqtt.device_id
+
+        alarmData = json.dumps(alarminfo, default=lambda o: o.__dict__, ensure_ascii=False)
+        # print("cpu温度：", temp)
+        # print("内存占用：", mem)
+        # print("cpu占用：", cpu)
+        # print("硬盘占用：", disk)
+        # print(alarmData)
+        self.publish('device/alarmInfo', alarmData)

1.Software/RK3568/gateway/src/payload.py

@@ -0,0 +1,503 @@
+import os
+import base64
+import time
+from array import array
+import scipy as sc
+from src.config import Config
+from src.config import MqttConf
+import json
+import gc
+import numpy as np
+import pandas as pd
+from multiprocessing import Queue
+from logging import Logger
+from src.config import UsbConf
+from src.usb import strategy
+import sqlite3
+from datetime import datetime
+import math
+from scipy.signal import resample
+from scipy.stats import kurtosis
+# from scipy.fft import fftfreq
+
+# 内部缓冲区凑齐10000采样点数后，统一将byte转成浮点，并构造为np.array进行后续运算
+# FFT_BUF_LEN = 400000
+
+# 一个字典，缓存8个通道的数据，格式为未转换的16进制，KEY值可根据USB桢结构更换，以方便存取处理
+fft_buf = {0: [], 1: [], 2: [], 3: [], 4: [], 5: [], 6: [], 7: [], 8: []}
+
+
+database_file = '/usr/local/gateway/runtime/database/gateway.db'
+
+class InnerData:
+    def __init__(self, channel: int, data: array, Time: float, RPM: float):
+        self.channel = channel
+        self.data = np.array(data)
+        self.Time = Time
+        self.RPM = RPM
+
+
+# 计算的指标
+class Result:
+    def __init__(self):
+        self.rmsvalue = 0  # 有效值
+        self.indexkur = 0  # 峭度指标
+        self.indexi = 0  # 脉冲指标
+        self.indexk = 0  # 波形指标
+        self.indexl = 0  # 裕度指标
+        self.indexsk = 0  # 偏度指标
+        self.indexc = 0  # 峰值
+        self.indexxr = 0  # 方根幅值
+        self.indexmax = 0  # 最大值
+        self.indexmin = 0  # 最小值
+        self.indexmean = 0  # 均值
+        self.indexeven = 0  # 平均幅值
+        self.band1rms = 0
+        self.band2rms = 0
+        self.band3rms = 0
+        self.band4rms = 0
+        self.band5rms = 0
+        self.band6rms = 0
+
+        # TODO: 补充其他指标
+
+
+# 采样数据
+class SampleData:
+    def __init__(self):
+        self.datatag = 1  # 压缩标签, 不压缩赋值为1
+        self.length = "32.768K"  # 采样长度
+        self.sample_freq = 0  # 采样频率
+        self.datatype = "TIMEWAVE"  # 数据类型
+        self.measuredefine = "加速度"  # 测量参数
+        self.filepath = ""  # 文件名称
+        self.rpm = ""  # 转速
+        self.time = 0  # 采样时刻
+        self.wavesave = ""
+        self.data = ""  # 采样原始数据
+        self.spectrum = ""  # 频谱数据
+        self.envelop_set = ""  # 包络设置频带
+        self.spectrum_envelop = ""  # 包络频谱
+        self.band_value1 = ""
+        self.band_value2 = ""
+        self.band_value3 = ""
+        self.band_value4 = ""
+        self.band_value5 = ""
+        self.band_value6 = ""
+        self.result = Result()  # 计算的指标
+        # TODO: 待补充
+
+
+class BandAlarm:
+    def __init__(self):
+        self.sensor_alarm = "normal"
+        self.band_alarm1 = "normal"
+        self.band_alarm2 = "normal"
+        self.band_alarm3 = "normal"
+        self.band_alarm4 = "normal"
+        self.band_alarm5 = "normal"
+        self.band_alarm6 = "normal"
+
+
+class FlowData:
+    def __init__(self):
+        self.sample_data = SampleData()
+        self.band_alarm = BandAlarm()
+
+
+class NumpyArrayEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, np.ndarray):
+            return obj.tolist()
+            # return {
+            #     '__ndarray__': base64.b64encode(obj.tobytes()).decode('utf-8'),
+            #     'dtype': str(obj.dtype),
+            #     'shape': obj.shape
+            # }
+        elif isinstance(obj, Result):
+            return obj.__dict__
+        elif isinstance(obj, BandAlarm):
+            return obj.__dict__
+        return super().default(obj)
+
+
+# 上报的采样数据
+class Flow:
+    def __init__(self):
+        self.code = 200  # 状态码, 200表示成功，其余表示失败
+        self.message = "推送成功"  # 推送成功/失败，并描述相关原因
+        # self.ch = 0  # 通道号
+        self.data = FlowData()  # 传输内容
+
+    # 将数据转换为字典格式的数据
+    def to_dict(self):
+        data = {
+            "code": self.code,
+            "message": self.message,
+            "data": {
+                "sample_data": self.data.sample_data.__dict__,
+                "band_alarm": self.data.band_alarm.__dict__
+            }
+        }
+        return data
+    # 将字典格式数据转换为JSON格式的数据
+    def to_json(self):
+        data_dict = self.to_dict()
+        # start = time.time()
+        json_data = json.dumps(data_dict, cls=NumpyArrayEncoder, ensure_ascii=False)
+        # print("转换时间：", time.time() - start)
+        return json_data
+
+    # TODO:
+    # 定义算法函数，意思是：输入的是InnerData即原始采样数据，输出的是Flow这样直接可由MQTT发出的数据
+    def edge_compute(self, num, windfarm, fansnum, measurpoint, measurpointdirection, samplingtime, samplingfrequency):
+        # start = time.time()
+        data = num.data
+        Time = num.Time
+        rpm = num.RPM
+        xfft = sc.fft.fftn(data)
+        xfft = np.abs(xfft)
+        xfft = np.trunc(xfft * 10 ** 8) / 10 ** 8
+
+        # 计算特征值
+        mean = np.mean(data ** 2)
+        absdata = np.abs(data)
+        maxabs = np.max(absdata)
+        indexeven = np.mean(absdata)  # 平均幅值
+        rmsvalue = np.sqrt(mean)  # 有效值
+        p = maxabs / indexeven
+        indexi = p  # 脉冲指标
+        indexk = maxabs / rmsvalue  # 波形因子
+        indexc = np.amax(data)  # 峰值
+        indexxr = rmsvalue  # 方根幅值
+        indexmax = np.max(data)  # 最大值
+        indexmin = np.min(data)  # 最小值
+        indexmean = np.mean(data)  # 均值
+        indexl = indexmax / indexmean  # 裕度指标
+        indexsk = pd.Series(data).skew()  # 偏度指标
+        # n = len(data)
+        # mean_value = np.mean(data)
+        # std_value = np.std(data, ddof=1)
+        # indexkur = np.sum((data - mean_value)**4) / ((n-1) * std_value**4)
+        # indexkur -= 3 # 峭度
+        indexkur = kurtosis(data,fisher=False)  # 峭度
+        # end = time.time()
+
+        # 将特征值存储到结果集中
+        # self.ch = num.channel
+        self.data.sample_data.time = Time
+        
+        length = samplingtime * samplingfrequency / 1000
+        self.data.sample_data.length = f"{length}K"
+        self.data.sample_data.rpm = rpm
+        self.data.sample_data.sample_freq = samplingfrequency
+        self.data.sample_data.filepath = f"{windfarm}_风机#{fansnum}_{measurpoint}_{measurpointdirection}_{length}K_{samplingfrequency}Hz_TIMEWAVE_加速度_{rpm}RPM_{Time}"
+        # self.data.sample_data.data = data
+        # self.data.sample_data.spectrum = xfft
+        self.data.sample_data.data = ' '.join(map(str, data))
+        self.data.sample_data.spectrum = ' '.join(map(str, xfft))
+        # self.data.sample_data.spectrum = "0"
+        self.data.sample_data.result.rmsvalue = rmsvalue
+        self.data.sample_data.result.indexkur = indexkur
+        self.data.sample_data.result.indexi = indexi
+        self.data.sample_data.result.indexk = indexk
+        self.data.sample_data.result.indexsk = indexsk
+        self.data.sample_data.result.indexc = indexc
+        self.data.sample_data.result.indexxr = indexxr
+        self.data.sample_data.result.indexmax = indexmax
+        self.data.sample_data.result.indexmin = indexmin
+        self.data.sample_data.result.indexmean = indexmean
+        self.data.sample_data.result.indexl = indexl
+        self.data.sample_data.result.indexeven = indexeven
+        # print("FFT时间：", midle - start)
+        # print("特征值时间：", end - midle)
+        # print("差额时间：", end + start - midle*2)
+        # print("计算时间：", end - start)
+
+# 将usb接收的数据进行转换
+def USBdata(queue1: Queue, queue2: Queue, log: Logger):
+    # 检查数据库文件是否存在
+    if not os.path.exists(database_file):
+        # 文件不存在，执行重新创建数据库的操作
+        conn = sqlite3.connect(database_file)
+        cur = conn.cursor()
+        cur.execute('''CREATE TABLE IF NOT EXISTS Data (id INTEGER PRIMARY KEY, myarray TEXT, timestamp TEXT);''')
+    else:
+        # 文件存在，检查文件是否可读
+        if os.access(database_file, os.R_OK):
+            try:
+                # 尝试连接数据库
+                conn = sqlite3.connect(database_file)
+                cur = conn.cursor()
+                cur.execute('''CREATE TABLE IF NOT EXISTS Data (id INTEGER PRIMARY KEY, myarray TEXT, timestamp TEXT);''')
+            except sqlite3.DatabaseError:
+                # 捕获数据库错误，执行重新创建数据库的操作
+                os.remove(database_file)
+                conn = sqlite3.connect(database_file)
+                cur = conn.cursor()
+                cur.execute('''CREATE TABLE IF NOT EXISTS Data (id INTEGER PRIMARY KEY, myarray TEXT, timestamp TEXT);''')
+        else:
+            # 文件不可读，执行重新创建数据库的操作
+            os.remove(database_file)
+            conn = sqlite3.connect(database_file)
+            cur = conn.cursor()
+            cur.execute('''CREATE TABLE IF NOT EXISTS Data (id INTEGER PRIMARY KEY, myarray TEXT, timestamp TEXT);''')
+
+    data_count = 0  # 记录数据库中的数据数量
+
+    global fft_buf
+
+    while True:
+        while strategy.data_trigger.value:
+            if strategy.mqttconnectstatus.value:     
+                # 查询数据库中是否存在数据
+                cur.execute("SELECT COUNT(*) FROM Data")
+                total_rows = cur.fetchone()[0]
+
+                if total_rows > 0:
+                    # 执行分页查询
+                    page_size = 3000
+                    total_pages = math.ceil(total_rows / page_size)
+
+                    for page in range(total_pages):
+                        offset = page * page_size
+                        cur.execute("SELECT * FROM Data LIMIT ? OFFSET ?", (page_size, offset))
+                        rows = cur.fetchall()
+
+                        if len(rows) > 0:
+                            log.info("从SQlite数据库读取数据")
+                            for row in rows:
+                                num = row[1]  # 取出元组中的第二个元素，即字符串数据
+                                timestamp = row[2]
+                                #timestamp = datetime.strptime(timestamp_str, "%Y%m%d%H%M%S")
+                                my_array = json.loads(num)
+                                Dataprocess(my_array, queue2, log, timestamp)
+                                # 删除数据库中的该条数据
+                                cur.execute("DELETE FROM Data WHERE id=?", (row[0],))
+                                conn.commit()
+
+                    # 释放数据库
+                    try:
+                        # 清空列表中的数据，释放内存
+                        rows.clear() 
+                        log.info("释放数据库")
+                        cur.execute('VACUUM')
+                        conn.commit()
+                        time.sleep(0.5)
+
+                    except sqlite3.Error as error:
+                        log.error('SQLite Release error: %s' % (' '.join(error.args)))
+                        time.sleep(1)
+                
+                if not queue1.empty():
+                    item = queue1.get()
+                    num, timestamp = item
+                    # log.info("数据转换")
+                    Dataprocess(num, queue2, log, timestamp)
+                    item = None
+
+                else:
+                    # 触发垃圾回收
+                    # queue1.task_done()
+                    gc.collect()
+                    time.sleep(0.1)
+
+
+            elif not strategy.mqttconnectstatus.value:
+                # 执行读取操作
+                cur.execute("SELECT COUNT(*) FROM Data")
+                data_count = cur.fetchone()[0]
+                log.debug("data_count:%s"%data_count)
+                while (not queue1.empty()) and (not strategy.mqttconnectstatus.value):
+                    item = queue1.get()
+                    num, timestamp = item
+                    daData = json.dumps(list(num))
+
+                    # 将数据保存到数据库
+                    cur.execute("INSERT INTO Data (myarray, timestamp) VALUES (?, ?)", (daData, timestamp))
+                    conn.commit()
+                    item = None
+                    data_count += 1  # 数据数量加1
+                    # 如果数据数量超过14062条，删除最旧的数据
+                    if data_count > 14062:
+                        cur.execute("SELECT id FROM Data ORDER BY id ASC LIMIT 4062")
+                        oldest_ids = cur.fetchall()
+
+                        for oldest_id in oldest_ids:
+                            cur.execute("DELETE FROM Data WHERE id=?", (oldest_id[0],))
+
+                        conn.commit()
+
+                        data_count -= 4062  # 数据数量减1000
+                        
+                        # 释放数据库
+                        try:
+                            log.info("释放数据库")
+                            cur.execute('VACUUM')
+                            conn.commit()
+                            time.sleep(1)
+
+                        except sqlite3.Error as error:
+                            log.error('SQLite Release error: %s' % (' '.join(error.args)))
+                            time.sleep(1)
+            
+                while queue1.empty() and not strategy.mqttconnectstatus.value:
+                    time.sleep(5)
+
+# 将USB接收到的数据进行处理
+def Dataprocess(num, queue2, log, timestamp):
+    cfg = Config()
+    cfg = cfg.Mqtt
+    # 每1028个字节是一个通道数据
+    for i in range(0, len(num), 1028):
+        num_list = num[i:i + 1028]
+        # 检查每段数据帧头是否正确
+        if num_list[0] == num_list[1] == num_list[2] == num_list[3]:
+            # 前八个通道是振动数据段
+            if num_list[0] < 8:
+                # 处理振动数据
+                DataProcess(num_list, num_list[0])
+            elif num_list[0] == 8:
+                # 处理转速数据
+                usb_buf = num_list[4: len(num_list)]
+                Speed_numbers = [Speed_num for i, Speed_num in enumerate(usb_buf) if i % 2 == 0]
+                fft_buf[8].extend(Speed_numbers)
+        else:
+            log.error("帧结构错误:%s", num_list[0:4])
+
+    for k, v in fft_buf.items():
+        # 计算转速 
+        if len(fft_buf[8]) >= strategy.FFT_BUF_LEN:
+            Seppd_date = fft_buf[8]
+            # 找到方波的上升沿的个数
+            num_rising_edges = 0
+            for i in range(1, len(Seppd_date)):
+                if Seppd_date[i] == 1 and Seppd_date[i-1] == 0:
+                    num_rising_edges += 1
+            # 计算周期
+            rpm = num_rising_edges / strategy.sample_time.value
+            # print("风机在线振动检测系统采集装置转速测量：%f"%rpm)
+            # print("风机在线振动检测系统采集装置转速测量：%f"%rpm)
+            # print("风机在线振动检测系统采集装置转速测量：%f"%rpm)
+            fft_buf[8].clear()
+
+        # 如果有通道的fft_buf超过了最大采样数据,那么就做一次算法
+        if len(v) >= strategy.FFT_BUF_LEN and k < 8:
+            original_data = np.around(v, 8)
+            original_data = original_data[-strategy.FFT_BUF_LEN:]
+            match k:
+                case 0:
+                    resampled_data = resampledata(original_data, cfg.ch0_samplingtime, cfg.ch0_samplingfrequency)
+                case 1:
+                    resampled_data = resampledata(original_data, cfg.ch1_samplingtime, cfg.ch1_samplingfrequency)
+                case 2:
+                    resampled_data = resampledata(original_data, cfg.ch2_samplingtime, cfg.ch2_samplingfrequency)
+                case 3:
+                    resampled_data = resampledata(original_data, cfg.ch3_samplingtime, cfg.ch3_samplingfrequency)
+                case 4:
+                    resampled_data = resampledata(original_data, cfg.ch4_samplingtime, cfg.ch4_samplingfrequency)
+                case 5:
+                    resampled_data = resampledata(original_data, cfg.ch5_samplingtime, cfg.ch5_samplingfrequency)
+                case 6:
+                    resampled_data = resampledata(original_data, cfg.ch6_samplingtime, cfg.ch6_samplingfrequency)
+                case 7:
+                    resampled_data = resampledata(original_data, cfg.ch7_samplingtime, cfg.ch7_samplingfrequency)
+            Time = timestamp
+            # rpm = 0
+            resampled_data = np.around(resampled_data, 8)
+            payload = InnerData(k, resampled_data, Time, rpm)
+            queue2.put(payload)  # 尝试向队列中添加数据
+            # 清空当前通道对应的列表
+            fft_buf[k].clear()
+
+
+
+
+def DataProcess(num_list, j):
+    cfg = Config()
+    cfg = cfg.Mqtt
+    usb_buf = num_list[4: len(num_list)]
+    float_list = []
+    for i in range(0, len(usb_buf), 2):
+        # 将2个8位数据合并成一个16位无符号整数
+        int_data = (usb_buf[i] << 8) | usb_buf[i + 1]
+        if int_data & 0x8000:  # 最高位为1，表示负数
+            int_data = -((int_data ^ 0xFFFF) + 1)  # 取反加一得到负数
+        float_data = int_data * 5.0 / 32768.0
+        # 将电压值转为加速度值
+        match j:
+            case 0:
+                float_data = float_data * 1000 / cfg.ch0_sensorparameters
+            case 1:
+                float_data = float_data * 1000 / cfg.ch1_sensorparameters
+            case 2:
+                float_data = float_data * 1000 / cfg.ch2_sensorparameters
+            case 3:
+                float_data = float_data * 1000 / cfg.ch3_sensorparameters
+            case 4:
+                float_data = float_data * 1000 / cfg.ch4_sensorparameters
+            case 5:
+                float_data = float_data * 1000 / cfg.ch5_sensorparameters
+            case 6:
+                float_data = float_data * 1000 / cfg.ch6_sensorparameters
+            case 7:
+                float_data = float_data * 1000 / cfg.ch7_sensorparameters
+        float_list.append(float_data)
+    fft_buf[j].extend(float_list)
+
+# 对数据进行降采样处理
+def resampledata(original_data, Samplingtime, Samplingfrequency):
+    if Samplingtime <= strategy.sample_time.value:
+        target_duration = Samplingtime  # 目标时长
+        # 计算目标样本数
+        target_samples = int(target_duration * strategy.sample_rate.value)
+        # 截取最后时长的数据
+        original_data = original_data[-target_samples:]
+        resampled_data = original_data
+        Samplingtime = strategy.sample_time.value
+
+    if Samplingfrequency < strategy.sample_rate.value:
+        # 计算降采样的比例
+        resample_ratio = Samplingfrequency / strategy.sample_rate.value
+        # 计算降采样后的目标长度
+        target_length = int(len(original_data) * resample_ratio)
+        # 进行降采样
+        resampled_data = resample(original_data, target_length)
+        Samplingfrequency = strategy.sample_rate.value       
+    
+    return resampled_data
+
+# 上报的硬件信息
+class DeviceInfo:
+    def __init__(self):
+        cfg = Config()
+        self.code = 200  # 状态码，200表示成功，其余表示失败
+        self.message = "推送成功"
+        self.data = DeviceInfo_Data(cfg.Mqtt)  # 传输内容
+
+
+class DeviceInfo_Data:
+    def __init__(self, cfg):
+        self.source = cfg.source  # 所属模块
+        self.dev_id = cfg.device_id  # 设备ID
+        self.sample_rate_strategy = cfg.sample_rate_strategy  # 采样策略
+        self.long_rate_strategy = cfg.long_rate_strategy
+
+
+# 报警信息上报
+class alarmInfo:
+    def __init__(self):
+        cfg = Config()
+        self.code = 200  # 状态码，200表示成功，其余表示失败
+        self.message = "推送成功"
+        self.data = alarmInfodata(cfg.Mqtt)  # 传输内容
+
+
+class alarmInfodata:
+    def __init__(self, cfg):
+        times = ''  # 采样时间
+        temperate = 0  # 温度告警
+        memory = 0  # 内存告警
+        cpu = 0  # cpu告警
+        disk = 0  # 硬盘告警
+        dev_id = cfg.device_id  # 采样设备序列号

1.Software/RK3568/gateway/src/usb.py

@@ -0,0 +1,184 @@
+import subprocess
+import time
+from src.config import UsbConf
+from logging import Logger
+from multiprocessing import Queue
+from multiprocessing import Value
+from multiprocessing import Event
+import json
+import usb.core
+import usb.util
+from usb.core import USBError
+import sqlite3
+import datetime
+import gc
+import math
+
+# 这个类用来存储给的的采样策略
+class Usb_sample_strategy:
+    def __init__(self):
+        self.sample_rate = Value('i', 10000)
+        self.FFT_BUF_LEN = Value('i', 400000)
+        self.sample_time = Value('i', 40)
+        self.data_trigger = Value('b', True)
+        self.write_trigger = Value('b', True)
+        self.rate_interval = Value('i', 3510)
+        self.mqttconnectstatus = Value('b', False)
+        self.usbalivestatus = Value('b', False)
+
+# 创建实例对象
+strategy = Usb_sample_strategy()
+
+# 一次从USB读取1000（暂定，具体根据单片机调整，同时也没考虑头尾、校验等多出字节）个字节，放到内部缓冲区
+USB_BUF_LEN = 9252
+
+
+class Usb:
+    def __init__(self, cfg: UsbConf, log: Logger):
+        self.vendor = cfg.vendor
+        self.product = cfg.product
+        self.log = log
+        strategy.rate_interval.value = cfg.sample_time
+
+    # 连接USB
+    def Connect(self) -> int:
+        dev = usb.core.find(idVendor=self.vendor, idProduct=self.product)
+        if dev is None:
+            self.log.error("could not find the usb device!")
+            return -1
+
+        if not isinstance(dev, usb.core.Device):
+            self.log.error("usb dev type is not usb.core.Device!")
+            return -2
+
+        self.dev: usb.core.Device = dev
+
+        # 这几行代码用来解除usb被占用
+
+        if dev.is_kernel_driver_active(0):
+            dev.detach_kernel_driver(0)
+
+        dev.reset()
+        dev.set_configuration()
+        self.log.info("Find the usb device success!")
+        subprocess.run(["/usr/local/gateway/led", "89", "1"])
+        return 0
+
+    def Disconnect(self):
+        pass
+
+   
+    def WriteDataWork(self, queue: Queue, event: Event):
+        while True:
+            # 确定每次采样的时间
+            rateTimes = math.ceil(strategy.sample_rate.value  * strategy.sample_time.value / 512)
+            # print(rateTimes)
+
+            while strategy.data_trigger.value and not strategy.write_trigger.value:
+                if event.is_set():
+                    break
+
+                try:
+                    # usb读取数据
+                    num = self.dev.read(0x81, USB_BUF_LEN, 1000)  
+                    timestamp = datetime.datetime.now()
+                    # 添加时间戳
+                    timestamp = timestamp.strftime('%Y%m%d%H%M%S') 
+                    queue.put((num, timestamp))
+
+                    # self.log.info("采集数据")
+                    rateTimes = rateTimes - 1  
+                    num = None
+                    if rateTimes == 0:
+                        Stoptime = time.time()
+                        self.log.info("开始暂停")
+                        # time.sleep(strategy.rate_interval.value)
+                        while (time.time() - Stoptime) < (strategy.rate_interval.value):
+                            self.log.info("发送心跳")
+                            self.keepAlive()
+                            time.sleep(60)
+                            # 触发垃圾回收
+                            gc.collect()
+                        rateTimes = math.ceil(strategy.sample_rate.value * strategy.sample_time.value / 512)
+                        strategy.usbalivestatus.value = True
+                        self.log.info("暂停结束")
+                        # 触发垃圾回收
+                        gc.collect()
+
+
+                except USBError as e:
+                    self.log.error("Usb Read Data error: %s" % str(e))
+                    if not str(e) == "[Errno 110] Operation timed out": 
+                        try:
+                            self.log.info("重连USB")
+                            self.Connect()
+                            RedLedtime3 = time.time()
+                            while((time.time()-RedLedtime3)<60):
+                                #绿灯慢闪
+                                subprocess.run(["/usr/local/gateway/led", "89", "0"])
+                                time.sleep(3)
+                                subprocess.run(["/usr/local/gateway/led", "89", "1"])
+                                time.sleep(3)
+                            # time.sleep(10)
+                        except USBError as e:
+                            self.log.error("Usb reconnect error: %s" % str(e))
+
+            time.sleep(0.5)
+
+    # 看门狗程序
+    def KeepAliveWork(self):
+        startTime = time.time()
+        aliveTime = time.time()
+        self.log.info("喂狗")
+        subprocess.run(["/usr/local/gateway/watchdog"])
+        while True:       
+            # 发送标志位为True时，向单片机传递频率、启停信息
+            if strategy.write_trigger.value:
+                self.writeData()
+                self.log.info("读标志位：%s", strategy.write_trigger.value)
+                
+            # 每60秒喂一次狗
+            if (time.time() - startTime) > 60:
+                self.log.info("喂狗")
+                subprocess.run(["/usr/local/gateway/watchdog"])
+                startTime = time.time()
+            
+            # 防止usb进程死机
+            # if (time.time() - aliveTime) > 1200 and strategy.data_trigger.value:
+            #     if strategy.usbalivestatus.value:
+            #         strategy.usbalivestatus.value = False
+                
+            #     else:
+            #         time.sleep(10)
+                
+            #     aliveTime = time.time()
+
+            time.sleep(1)
+
+    # usb发送控制数据
+    def writeData(self):
+        listData = [0] * 8
+        listData[:4] = [1] * 4
+        if strategy.data_trigger.value:
+            listData[4] = 1
+            listData[5] = 1
+        else:
+            listData[4] = 0
+            listData[5] = 0
+        if strategy.sample_rate.value == 40000:
+            listData[6] = 0
+            listData[7] = 0
+        elif strategy.sample_rate.value == 20000:
+            listData[6] = 1
+            listData[7] = 1
+        elif strategy.sample_rate.value == 10000:
+            listData[6] = 2
+            listData[7] = 2
+        self.dev.write(0x01, listData)
+        strategy.write_trigger.value = False
+        time.sleep(1)
+
+    # 发送心跳包
+    def keepAlive(self):
+        self.dev.write(0x01, '0000')
+        strategy.write_trigger.value = False

1.Software/RK3568/gateway/watchdog.c

@@ -0,0 +1,36 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <sys/ioctl.h>
+#include <linux/types.h>
+#include <linux/watchdog.h>
+
+int main() {
+    int fd, timeout;
+
+    // 打开看门狗设备文件
+    fd = open("/dev/watchdog", O_WRONLY | O_NONBLOCK);
+    if (fd == -1) {
+        perror("看门狗 ");
+        exit(EXIT_FAILURE);
+    }
+
+    // 设置超时时间为10秒
+    timeout = 60;
+
+    // 设置超时时间并获取实际超时时间
+    ioctl(fd, WDIOC_SETTIMEOUT, &timeout);
+    ioctl(fd, WDIOC_GETTIMEOUT, &timeout);
+    // printf("设置的看门狗超时时间为 %d\n", timeout);
+
+    // 喂狗并等待一段时间
+    ioctl(fd, WDIOC_KEEPALIVE, &timeout);
+    // printf("喂狗了\n");
+    sleep(0.1);
+
+    // 关闭看门狗设备文件
+    close(fd);
+    return 0;
+}

1.Software/RK3568/gateway/更新说明.txt

@@ -0,0 +1,7 @@
+改变了led等闪烁状态
+1、绿灯常亮：服务已经启动
+2、绿灯微闪：正在传输数据
+3、绿灯慢闪：USB传输出现问题
+4、红灯快闪：网络/MQTT连接出现问题
+
+如果MQTT传输的数据从浮点改为整型大约可以降低12%的数据大小