68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
from time import time
|
|
from shared.deviceTypes import DEVICE_TYPES
|
|
|
|
|
|
try:
|
|
import ujson as json
|
|
except ImportError:
|
|
import json
|
|
|
|
def as_json(data):
|
|
"""Convert a dictionary to a JSON string."""
|
|
try:
|
|
return json.dumps(data)
|
|
except Exception as e:
|
|
print("[Payloads] Error converting to JSON:", e)
|
|
return "{}" # Return an empty JSON object on error
|
|
|
|
def mqtt_hello(id_microwave):
|
|
return as_json({
|
|
"id_microwave": id_microwave,
|
|
"type": DEVICE_TYPES["MICROWAVE"]
|
|
})
|
|
|
|
def mqtt_hello_ack(id_orchestrator, id_microwave):
|
|
return as_json({
|
|
"id_microwave": id_microwave,
|
|
"id_orchestrator": id_orchestrator
|
|
})
|
|
|
|
def mqtt_cooking_init(id_microwave):
|
|
return as_json({
|
|
"id_microwave": id_microwave
|
|
})
|
|
|
|
def mqtt_sensor_data(id_microwave, dish_temp, ambient_temp):
|
|
return as_json({
|
|
"id_microwave": id_microwave,
|
|
"dish_temp": dish_temp,
|
|
"ambient_temp": ambient_temp
|
|
})
|
|
|
|
def lora_sensor_data(dish_temp, ambient_temp):
|
|
return as_json({
|
|
"dish_temp": dish_temp,
|
|
"ambient_temp": ambient_temp
|
|
})
|
|
|
|
def mqtt_cooking_config(id_microwave, cook_time, power_level, target_temp):
|
|
return as_json({
|
|
"id_microwave": id_microwave,
|
|
"cook_time": cook_time,
|
|
"power_level": power_level,
|
|
"target_temp": target_temp
|
|
})
|
|
|
|
def telemetry_payload(device_id, microwave_states, button_state, cloud_alert, gps_data, ambient_temp, ambient_humidity, connected_components, logs):
|
|
return as_json({
|
|
"device_id": device_id,
|
|
"timestamp": int(time()),
|
|
"states": microwave_states,
|
|
"button_state": button_state,
|
|
"cloud_alert": cloud_alert,
|
|
"gps": gps_data,
|
|
"ambient_temp": ambient_temp,
|
|
"ambient_humidity": ambient_humidity,
|
|
"connected_components": connected_components,
|
|
"logs": logs
|
|
}) |