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)