Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
reads and parses sensor data of JQ-200 / JQ-300 from ypyt cloud for given device token
#!/usr/bin/env python3
""" reads and parses sensor data from ypyt cloud for given device token """
# see for more context
import json
import os
import sys
import paho.mqtt.client as mqtt
import time as t
if len(sys.argv) == 2:
TOPIC = sys.argv[1]
print("usage: {} DEVICE_TOKEN_TOPIC".format(sys.argv[0]))
PORT = 55450
def handle_message_values(content):
for item in content:
measurement = decode_sensor_seq_item(item.get("seq"), item.get("content"))
if measurement:
SEQ_TYPES = [ None, None, None, None, "temp", "hum", "PM2.5", "HCHO", "TVOC", "eCO2", None ]
SEQ_UNITS = [ None, None, None, None, "C", "%", "ug/m^3", "mg/m^3", "mg/m^3", "ppm", None ]
def decode_sensor_seq_item(seq, content):
meas_type = SEQ_TYPES[seq]
meas_unit = SEQ_UNITS[seq]
if meas_type and meas_unit:
item = "{} \t{} \t{}".format(meas_type, content, meas_unit)
return item
return None
# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
print("Connected with result code "+str(rc))
# Subscribing in on_connect() means that if we lose the connection and
# reconnect then subscriptions will be renewed.
# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
payload = msg.payload
jsondata = json.loads(payload.decode('utf-8'))
msg_type = jsondata.get("type")
msg_content = json.loads(jsondata.get("content"))
if msg_type == "C":
print("message: connection status: {}".format(msg_content))
elif msg_type == "V":
print("message: sensor values")
print("message: unknown: {}".format(msg_content))
#print(json.dumps(sensordata, indent=2))
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(SERVER, PORT, 60)
# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a
# manual interface.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment