Lưu luồng dữ liệu (Datastream)

Bây giờ bạn sẽ lấy một luồng dữ liệu MQTT và nối mỗi điểm dữ liệu mới vào danh sách store.

Với thư viện paho.mqtt, bạn có thể đăng ký (subscribe) vào một luồng dữ liệu bằng subscribe.callback().

Mỗi thông điệp mới sẽ kích hoạt một lần gọi tới hàm của bạn, hàm này cần có các tham số sau:

client: thể hiện (instance) client cho callback này
userdata: dữ liệu người dùng riêng được đặt khi tạo instance
message: một instance của MQTTMessage. Trong bài này, payload là thuộc tính duy nhất bạn cần quan tâm.

Bạn cần parse dữ liệu dạng chuỗi JSON bằng json.loads() và nối nó vào danh sách store. Sau đó, bạn sẽ chuyển đổi sang DataFrame và lưu DataFrame thành tệp CSV.

json, pandas dưới tên pd, MQTT_HOST và topic đã có sẵn trong phiên làm việc của bạn.

Bài tập này là một phần của khóa học

Phân tích dữ liệu IoT bằng Python

Xem khóa học

Bài tập tương tác thực hành trực tiếp

Hãy thử làm bài tập này bằng cách hoàn thành đoạn mã mẫu này.

# Define function to call by callback method
def ____(____):
    # Parse the message.payload
    data = ____
    store.append(data)

Chỉnh sửa và Chạy Mã