from time import time from shared.deviceTypes import DEVICE_TYPES from shared.alerts import Alert, AlertType import shared.config as config try: from shared.lora_device import LoraCommands except ImportError: pass # No need 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 }) def lora_new_alert(alert: Alert): return { "action": LoraCommands.NEW_ALERT, "alert_type": alert.alertType }