0

我正在尝试使用 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)

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

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

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

任何帮助深表感谢！

4

0 回答 0