pandas.core.common.PandasError：未正确调用 DataFrame 构造函数

Question

我正在尝试使用 mosquitto 接收数据并使用 python pandas 将其保存为 csv 文件。数据是连续的，直到我停止脚本。

mqtt_pub.py

import paho.mqtt.client as mqtt
import random
import schedule
import time

mqttc = mqtt.Client("python_pub")
mqttc.connect("localhost", 1883)

def job():
    mqttc.publish("hello/world", random.randint(1, 10))

schedule.every(1).seconds.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

mqttc.loop(2)

mqtt_sub.py

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

def on_connect(client, userdata, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("hello/world")

def on_message(client, userdata, msg):
    datas = map(int, msg.payload)
    for num in datas:
        df = pd.DataFrame(data=datas, columns=['the_number'])
        df.to_csv("testing.csv")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

client.connect("localhost", 1883, 60)

client.loop_forever()

从上面的mqtt_sub.py 脚本，我得到testing.csv 看起来像这样

    | the _number
0   | 2

2 是我在停止 mqtt_sub.py 脚本之前收到的最后一个数字

Connected with result code 0
[3]
[9]
[5]
[3]
[7]
[2]
...
...
KeyboardInterrupt

我希望得到这样的testing.csv

    | the_number
0   | 3
1   | 9
2   | 5
...
...
5   | 2

为了实现这一点，我尝试将以下 df = pd.DataFrame(data=datas, columns=['the_number']) 更改为 df = pd.DataFrame(data=num, columns=['the_number']) 并出现以下错误

pandas.core.common.PandasError: DataFrame constructor not properly called!

有人知道如何解决这个错误吗？我也觉得我这里没有正确使用for循环。

感谢您的建议和帮助。

[更新]

我在on_message 方法中添加/更改以下行

def on_message(client, userdata, msg):
    datas = map(int, msg.payload)
    df = pd.DataFrame(data=datas, columns=['the_number'])

    f = open("test.csv", 'a')
    df.to_csv(f)
    f.close()

在 Nulljack 的帮助下，我能够在我的 CSV 文件中得到这样的结果

   | the_number
0  | 3
   | the_number
0  | 9 
   | the_number
0  | 5
   | the_number
0  | 3
   | the_number
0  | 7

我的目标是在 CSV 文件中实现这样的目标

   | the_number
0  | 3
1  | 9
2  | 5 
3  | 3
4  | 7

Answer 1

如果我的理解有误，在此之前我从未使用过 mosquitto。

在我看来，你的 mqtt_sub.py 中的 on_message 方法在你的 mqtt_pub.py 每次发布消息时运行（即每隔一秒），这会导致你的 testing.csv 文件每次发布消息时都会覆盖

为了解决这个问题，我会在你的 on_connect 方法中初始化一个数据帧，然后在 on_message 中通过 df.append 将新值添加到数据帧

至于终止后写入 csv，我不确定。

希望对你有帮助

Answer 2

其他帖子很拥挤，所以我把我的回复移到这里

尝试使用下面的代码

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

# Move df here 
df = pd.DataFrame(columns=['the_number'])

def on_connect(client, userdata, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("hello/world")

def on_message(client, userdata, msg):
    datas = map(int, msg.payload)

    # this adds the data to the dataframe at the correct index
    df.iloc[df.size] = datas

    # I reverted this line back to what you originally had 
    # This will overwrite the testing.csv file every time your subscriber
    # receives a message, but since the dataframe is formatted like you want 
    # it shouldn't matter 
    df.to_csv("testing.csv")


client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

client.connect("localhost", 1883, 60)

client.loop_forever()