150 lines
5.0 KiB
Python
150 lines
5.0 KiB
Python
import time
|
|
import machine
|
|
import network
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CONFIG
|
|
# ---------------------------------------------------------------------------
|
|
CONFIG = {
|
|
# Wi-Fi du poste de test
|
|
"WIFI_SSID": "",
|
|
"WIFI_PASSWORD": "",
|
|
|
|
# Identité de dispositif (doit exister côté backend)
|
|
"DEVICE_ID": "esp32-01",
|
|
"DEVICE_SECRET": "demo-secret-esp32-01",
|
|
|
|
"BACKEND_URL": "",
|
|
|
|
# Capteur
|
|
"LDR_GPIO": 34,
|
|
|
|
# Test sans envoi : True affiche le payload, n'envoie rien
|
|
"DRY_RUN": False,
|
|
|
|
# Rythme d'envoi en secondes
|
|
"SEND_INTERVAL": 3,
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lecture capteur (LDR sur GPIO34)
|
|
# ---------------------------------------------------------------------------
|
|
def read_ldr():
|
|
"""Retourne la valeur brute de la LDR (0..4095, ADC 12 bits)."""
|
|
pin = machine.Pin(CONFIG["LDR_GPIO"], machine.Pin.IN)
|
|
adc = machine.ADC(pin)
|
|
return adc.read_u16() >> 4
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Payload JSON minimal
|
|
# ---------------------------------------------------------------------------
|
|
def build_payload(raw_value):
|
|
"""Timestamp ISO local au format de base (MicroPython n'a pas datetime)."""
|
|
y, m, d, hh, mm, ss, _, _ = time.localtime()
|
|
ts = "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(y, m, d, hh, mm, ss)
|
|
return {
|
|
"device_id": CONFIG["DEVICE_ID"],
|
|
"timestamp": ts,
|
|
"light_value": int(raw_value),
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Envoi HTTP/HTTPS
|
|
# ---------------------------------------------------------------------------
|
|
def send_measure(payload):
|
|
import urequests # fourni par la plupart des firmwares MicroPython
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Device-ID": CONFIG["DEVICE_ID"],
|
|
"X-Device-Secret": CONFIG["DEVICE_SECRET"],
|
|
}
|
|
try:
|
|
resp = urequests.post(CONFIG["BACKEND_URL"], json=payload, headers=headers)
|
|
status = resp.status_code
|
|
resp.close()
|
|
print("[send] device={} status={} ok={}".format(
|
|
CONFIG["DEVICE_ID"], status, 200 <= status < 300))
|
|
return status, (200 <= status < 300)
|
|
except OSError as exc:
|
|
# Erreur réseau : le errno aide à comprendre (voir debug ci-dessous).
|
|
errno = exc.args[0] if exc.args else None
|
|
print("[http] OSError envoi (device={}) errno={}".format(
|
|
CONFIG["DEVICE_ID"], errno))
|
|
_print_debug_network()
|
|
_print_debug_url()
|
|
return None, False
|
|
except Exception as exc:
|
|
print("[http] erreur envoi (device={}) : {}".format(
|
|
CONFIG["DEVICE_ID"], type(exc).__name__))
|
|
return None, False
|
|
|
|
def _print_debug_network():
|
|
"""Affiche l'IP que l'ESP32 a reçue + l'état Wi-Fi (debug)."""
|
|
try:
|
|
import network
|
|
wlan = network.WLAN(network.STA_IF)
|
|
if wlan.isconnected():
|
|
ip = wlan.ifconfig()[0]
|
|
print("[http] IP locale de l'ESP32 :", ip)
|
|
else:
|
|
print("[http] Wi-Fi NON connecté — l'envoi ne peut pas aboutir")
|
|
except Exception:
|
|
print("[http] état Wi-Fi indisponible")
|
|
|
|
|
|
def _print_debug_url():
|
|
"""Affiche la cible et aide à diagnostiquer l'OSError."""
|
|
url = CONFIG["BACKEND_URL"]
|
|
print("[http] URL cible :", url)
|
|
print("[http] Astuces :")
|
|
print("[http] - OSError negatif (~ -2 / -202) -> probleme DNS : "
|
|
"preferer une IP (192.168.x.y) a un nom d'hote")
|
|
print("[http] - errno ~ -3 / 111 (refus) -> backend injoignable : "
|
|
"verifier IP/port du PC, backend sur 0.0.0.0, firewall")
|
|
print("[http] - Rappel : 127.0.0.1 = l'ESP32 lui-meme, pas votre PC")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Wi-Fi
|
|
# ---------------------------------------------------------------------------
|
|
def connect_wifi():
|
|
if not CONFIG["WIFI_SSID"]:
|
|
print("[wifi] SSID vide, on continue en local")
|
|
return False
|
|
wlan = network.WLAN(network.STA_IF)
|
|
wlan.active(True)
|
|
if not wlan.isconnected():
|
|
wlan.connect(CONFIG["WIFI_SSID"], CONFIG["WIFI_PASSWORD"])
|
|
for _ in range(20):
|
|
if wlan.isconnected():
|
|
break
|
|
time.sleep(0.5)
|
|
if wlan.isconnected():
|
|
print("[wifi] connecté, IP :", wlan.ifconfig()[0])
|
|
else:
|
|
print("[wifi] échec de connexion")
|
|
return wlan.isconnected()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Boucle principale
|
|
# ---------------------------------------------------------------------------
|
|
def main():
|
|
connect_wifi()
|
|
|
|
if CONFIG["DRY_RUN"]:
|
|
print("[dry-run] payload :", build_payload(read_ldr()))
|
|
return
|
|
|
|
while True:
|
|
raw = read_ldr()
|
|
payload = build_payload(raw)
|
|
send_measure(payload)
|
|
time.sleep(CONFIG["SEND_INTERVAL"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|