import random import sys from datetime import datetime, timezone try: import requests except ImportError: sys.exit("Erreur : install requests (pip install requests)") # --------------------------------------------------------------------------- # CONFIG — à adapter # --------------------------------------------------------------------------- CONFIG = { "BACKEND_URL": "http://127.0.0.1:5000/api/ingest", "DEVICE_ID": "esp32-01", "DEVICE_SECRET": "demo-secret-esp32-01", # correspond au backend # Simulation de la LDR "LIGHT_MIN": 0, "LIGHT_MAX": 4095, "SEND_INTERVAL": 60, # secondes entre envois "DRY_RUN": False, # True : affiche le payload sans l'envoyer } # --------------------------------------------------------------------------- # Lecture simulée # --------------------------------------------------------------------------- def read_light(): """Retourne une valeur brute simulée, bornée (0..4095 par défaut).""" return random.randint(CONFIG["LIGHT_MIN"], CONFIG["LIGHT_MAX"]) # --------------------------------------------------------------------------- # Payload JSON # --------------------------------------------------------------------------- def build_payload(raw_value): return { "device_id": CONFIG["DEVICE_ID"], "timestamp": datetime.now(timezone.utc).isoformat(), "light_value": int(raw_value), } # --------------------------------------------------------------------------- # Envoi # --------------------------------------------------------------------------- def send_measure(payload): headers = { "Content-Type": "application/json", "X-Device-ID": CONFIG["DEVICE_ID"], "X-Device-Secret": CONFIG["DEVICE_SECRET"], } try: resp = requests.post(CONFIG["BACKEND_URL"], json=payload, headers=headers, timeout=10) print("[send] device={} status={} ok={}".format( CONFIG["DEVICE_ID"], resp.status_code, 200 <= resp.status_code < 300)) return resp.status_code, (200 <= resp.status_code < 300) except requests.exceptions.RequestException as exc: print("[http] erreur envoi (device={}) : {}".format( CONFIG["DEVICE_ID"], type(exc).__name__)) return None, False # --------------------------------------------------------------------------- # Boucle # --------------------------------------------------------------------------- def main(): dry_run = CONFIG["DRY_RUN"] or "--dry-run" in sys.argv if dry_run: print("[dry-run] payload :", build_payload(read_light())) return while True: raw = read_light() send_measure(build_payload(raw)) print(" (prochain envoi dans {} s)".format(CONFIG["SEND_INTERVAL"])) time.sleep(CONFIG["SEND_INTERVAL"]) if __name__ == "__main__": import time main()