LoRa, MQTT and Oled
This commit is contained in:
@@ -0,0 +1,183 @@
|
||||
import sys
|
||||
import time
|
||||
|
||||
IS_MICROPYTHON = sys.implementation.name == 'micropython'
|
||||
|
||||
if IS_MICROPYTHON:
|
||||
import _thread
|
||||
from machine import Pin, SPI
|
||||
import ubinascii
|
||||
import ujson as json
|
||||
|
||||
# --- PILOTE SPI DIRECT (ESP32 / Heltec V3) ---
|
||||
class LoraHardwareSPI:
|
||||
def __init__(self, spi_bus=1, clk=9, mosi=10, miso=11, cs=8, irq=14, rst=12, gpio=13):
|
||||
from sx1262 import SX1262
|
||||
self.lora = SX1262(
|
||||
spi_bus=spi_bus, clk=clk, mosi=mosi, miso=miso,
|
||||
cs=cs, irq=irq, rst=rst, gpio=gpio
|
||||
)
|
||||
self.default_group = 2 # On définit le groupe par défaut ici
|
||||
self.lock = _thread.allocate_lock() # Création du verrou
|
||||
|
||||
def configure(self, freq=868.1, bw=125.0, sf=7, cr=5, power=14):
|
||||
self.lora.begin(
|
||||
freq=freq, bw=bw, sf=sf, cr=cr, power=power,
|
||||
useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False
|
||||
)
|
||||
self.lora.setSyncWord(0x14)
|
||||
|
||||
def send(self, payload, group=None):
|
||||
"""Encode la payload en JSON si nécessaire, et injecte automatiquement l'octet de groupe."""
|
||||
with self.lock:
|
||||
if group is None:
|
||||
group = self.default_group
|
||||
|
||||
# Si c'est un dictionnaire ou une liste, on le convertit en JSON textuel
|
||||
if isinstance(payload, (dict, list)):
|
||||
payload = json.dumps(payload)
|
||||
|
||||
if isinstance(payload, str):
|
||||
payload = payload.encode('utf-8')
|
||||
|
||||
# Insertion automatique de l'octet de groupe au tout début de la trame physique
|
||||
paquet_physique = bytes([group]) + payload
|
||||
self.lora.send(paquet_physique)
|
||||
|
||||
def receive_packet(self, timeout_ms=1000):
|
||||
"""Écoute, nettoie, extrait le groupe, gère le HEX et parse le JSON."""
|
||||
with self.lock:
|
||||
data, state = self.lora.recv(len=0, timeout_en=True, timeout_ms=timeout_ms)
|
||||
if state == 0 and len(data) > 1:
|
||||
group = data[0]
|
||||
payload_brute = data[1:].strip(b'\x00 \r\n\t')
|
||||
|
||||
try:
|
||||
text = payload_brute.decode('utf-8').strip('\x00 \r\n\t')
|
||||
except UnicodeError:
|
||||
return None
|
||||
|
||||
if text.startswith('{') or text.startswith('['):
|
||||
decoded_text = text
|
||||
elif text.lower().startswith('7b') or text.lower().startswith('5b'):
|
||||
try:
|
||||
decoded_text = ubinascii.unhexlify(text).decode('utf-8').strip('\x00 \r\n\t')
|
||||
except Exception:
|
||||
decoded_text = text
|
||||
else:
|
||||
decoded_text = text
|
||||
|
||||
try:
|
||||
parsed_json = json.loads(decoded_text)
|
||||
return {"group": group, "data": parsed_json, "raw": False}
|
||||
except ValueError:
|
||||
return {"group": group, "data": decoded_text, "raw": True}
|
||||
|
||||
return None
|
||||
|
||||
else:
|
||||
import threading
|
||||
import serial
|
||||
import json
|
||||
|
||||
# --- PILOTE SÉRIE (Raspberry Pi / Dragino LA66) ---
|
||||
class LoraSerialAT:
|
||||
def __init__(self, port):
|
||||
self.port = port
|
||||
self.ser = serial.Serial(
|
||||
port=self.port,
|
||||
baudrate=9600,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=0.1
|
||||
)
|
||||
self.ser.reset_input_buffer()
|
||||
self.ser.reset_output_buffer()
|
||||
self.lock = threading.Lock()
|
||||
def configure(self, **kwargs):
|
||||
pass
|
||||
|
||||
def send(self, payload):
|
||||
"""Encode automatiquement la payload en HEX pour l'envoi via la clé."""
|
||||
with self.lock:
|
||||
if isinstance(payload, (dict, list)):
|
||||
payload = json.dumps(payload)
|
||||
|
||||
if isinstance(payload, str):
|
||||
payload = payload.encode('utf-8')
|
||||
|
||||
hex_payload = payload.hex()
|
||||
self.ser.reset_input_buffer()
|
||||
|
||||
# La clé ajoute d'elle-même l'octet de groupe configuré dans ses registres
|
||||
cmd = f"AT+SEND=1,{hex_payload},1,3\r\n"
|
||||
print(f"RPI : Envoi de la commande HEX -> AT+SEND=1,[HEX_DATA],1,3")
|
||||
self.ser.write(cmd.encode('utf-8'))
|
||||
|
||||
time.sleep(0.2)
|
||||
response = ""
|
||||
start_wait = time.time()
|
||||
while (time.time() - start_wait) < 1.5:
|
||||
if self.ser.in_waiting > 0:
|
||||
response += self.ser.readline().decode('utf-8', errors='ignore')
|
||||
time.sleep(0.05)
|
||||
|
||||
print(f"[RPI LA66 TX STATUS] :\n{response.strip()}")
|
||||
|
||||
def receive_packet(self, timeout_ms=5000):
|
||||
with self.lock:
|
||||
start_time = time.time()
|
||||
timeout_s = timeout_ms / 1000.0
|
||||
|
||||
while (time.time() - start_time) < timeout_s:
|
||||
if self.ser.in_waiting > 0:
|
||||
line = self.ser.readline().decode('utf-8', errors='ignore').strip()
|
||||
if line:
|
||||
payload_bytes = None
|
||||
|
||||
if "(HEX:)" in line:
|
||||
hex_part = line.split("(HEX:)")[1].strip().replace(" ", "")
|
||||
try:
|
||||
payload_bytes = bytes.fromhex(hex_part)
|
||||
except ValueError:
|
||||
pass
|
||||
elif "Data:" in line:
|
||||
payload_bytes = line.split("Data:")[1].strip().encode('utf-8')
|
||||
|
||||
if payload_bytes and len(payload_bytes) > 1:
|
||||
group = payload_bytes[0]
|
||||
payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t')
|
||||
|
||||
try:
|
||||
text = payload_clean.decode('utf-8').strip('\x00 \r\n\t')
|
||||
except UnicodeError:
|
||||
continue
|
||||
|
||||
if text.startswith('{') or text.startswith('['):
|
||||
decoded_text = text
|
||||
elif text.lower().startswith('7b') or text.lower().startswith('5b'):
|
||||
try:
|
||||
decoded_text = bytes.fromhex(text).decode('utf-8').strip('\x00 \r\n\t')
|
||||
except Exception:
|
||||
decoded_text = text
|
||||
else:
|
||||
decoded_text = text
|
||||
|
||||
try:
|
||||
parsed_json = json.loads(decoded_text)
|
||||
return {"group": group, "data": parsed_json, "raw": False}
|
||||
except json.JSONDecodeError:
|
||||
return {"group": group, "data": decoded_text, "raw": True}
|
||||
|
||||
time.sleep(0.01)
|
||||
return None
|
||||
|
||||
|
||||
def get_lora_device(port_or_pins=None):
|
||||
if IS_MICROPYTHON:
|
||||
pins = port_or_pins if port_or_pins else {}
|
||||
return LoraHardwareSPI(**pins)
|
||||
else:
|
||||
port = port_or_pins if port_or_pins else "/dev/serial/by-id/usb-Silicon_Labs_CP2102_USB_to_UART_Bridge_Controller_0001-if00-port0"
|
||||
return LoraSerialAT(port)
|
||||
Reference in New Issue
Block a user