Python MQTT：在DataFrame中累积消息读取并定期保存

问题描述

我正在尝试使用 MQTT 读取传感器测量值（从另一台设备发布）并将一周的读数存储在 Pandas DataFrame 中，一旦此类数据帧为空，我想将其保存到 .csv 文件并开始填充新的空数据框。此类数据框的示例如下：

                           sensor1  ...  sensorxx
timestamp                           ...                   
2018-11-21 15:15:00-06      0.276   ...   0
2018-11-21 15:30:00-06      0.167   ...   0
2018-11-21 15:45:00-06      0.179   ...   0.1
2018-11-21 16:00:00-06      0.076   ...   0.2
2018-11-21 16:15:00-06      0.064   ...   0

我的代码完全按照我的意图工作，只是在一段时间（数百条消息）后失败（实际上并没有失败，继续运行而没有任何错误消息），好像消息不再流入（这他们是）。

所有这些都发生在一个类中，这是我的代码的简化版本

import pandas as pd
import json
import paho.mqtt.client as mqtt

global bufferDF = None
global counter = 1

class DataSaver():
    def __init__(self,filesfolderpath,sensorslist):
        self.filesfolderpath = filesfolderpath
        self.sensorslist = sensorslist
        self.client = None

    def SaveSensorRead(self,client,userdata,message):
        global bufferDF
        global counter

        message_dict = json.loads(message)
        timestamp = pd.to_datetime(message_dict["timestamp"]) #timestamp message payload
        sensorname = message_dict["sensorname"]
        read = message_dict["read"]

        # creates an empty dataframe over a weekly daterange containg current timestamp 
        #(only for the first call when bufferDF has never been initialized)
        if (bufferDF is None):
            daterange = InitDaterange(timestamp) 
            bufferDF = pd.DataFrame(index=daterange,columns=self.sensorslist)
        
        # checks wether bufferDF is full,if so saves to disk and initializes new one
        if (timestamp > max(bufferDF.index)):
            filename = "week"+str(counter)+".csv"
            bufferDF.to_csv(os.path.join(self.filesfolderpath,filename))
            daterange = InitDaterange(timestamp) 
            bufferDF = pd.DataFrame(index=daterange,columns=self.sensorslist)
            counter += 1
            
        bufferDF.loc[timestamp,sensorname] = read

   def InitComm(self,brokerip,channelname)
       self.client = mqtt.Client("client")
       self.client.on_message = self.SaveSensorRead
       self.client.connect(brokerip,1883)
       self.client.loop_start()
       self.client.subscribe(channelname)


saver = DataSaver(filesfolderpath,sensorslist)
saver.InitComm(brokerip,channelname)

尝试了几件事。在每次迭代时保存数据帧，我可以看到它使用正确的结构进行初始化并正确填充。尝试将数据发布者的频率降低到几秒钟，以便订阅者跟上 as suggested here 并提高服务质量参数，但没有奏效。

这就像一些内存填满了，我的客户在一段时间后无法处理更多这样的消息。我试图保存的每周文件之一大约是 1.5 MB，所以不是真正的 RAM 问题。试图查看 Paho 文档以了解要调整的“缓存”参数，但似乎找不到。

我当然可以减小 DF 的大小，以便用更少的消息填充它，但对我以后不起作用。

非常感谢任何帮助！

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

libmosquitto mosquitto mosquitto mqtt mqtt paho python