import sys import time import random IS_MICROPYTHON = sys.implementation.name == 'micropython' if IS_MICROPYTHON: import _thread from machine import Pin, SPI import ubinascii import ujson as json else: import threading import serial import json # --- BASE RELIABLE LORA DEVICE --- class BaseLoraDevice: """Base class providing automatic ACK generation, retries, and duplicate filtering.""" def __init__(self): self.processed_msg_ids = set() self.received_acks = set() self.pending_rx_queue = [] self.default_group = 2 def _generate_msg_id(self): return random.getrandbits(16) def _send_ack(self, ack_id): """Sends an immediate acknowledgement packet back to the sender.""" print(f"[ReliableLoRa] -> Triggering ACK send for msg_id: {ack_id}") if IS_MICROPYTHON: time.sleep_ms(10) else: time.sleep(0.01) ack_payload = {"_type": "_ack", "_ack_id": ack_id} self.send(ack_payload) def _process_incoming_packet(self, packet): """Internal packet processor: handles ACKs and deduplication.""" if not packet or packet.get("raw"): return packet data = packet.get("data") if isinstance(data, dict): # 1. Handle incoming ACK response if data.get("_type") == "_ack": ack_id = data.get("_ack_id") print(f"[ReliableLoRa] <- SUCCESSFULLY MATCHED ACK ID: {ack_id}") if ack_id is not None: self.received_acks.add(ack_id) if len(self.received_acks) > 100: self.received_acks.clear() return None # Drop internal protocol message from user queue # 2. Handle incoming command expecting an ACK msg_id = data.get("_msg_id") if msg_id is not None: print(f"[ReliableLoRa] <- Received packet with msg_id {msg_id}. Queuing ACK.") self._send_ack(msg_id) if msg_id in self.processed_msg_ids: print(f"[ReliableLoRa] Discarding duplicate retry for msg_id {msg_id}") return None # Discard duplicate retry self.processed_msg_ids.add(msg_id) if len(self.processed_msg_ids) > 100: self.processed_msg_ids.clear() return packet def send_reliable(self, payload, max_retries=4, ack_timeout=2.5): """Sends a payload and retries until an ACK is received or max retries are reached.""" lock = getattr(self, 'lock', None) if isinstance(payload, dict): payload = dict(payload) else: payload = {"data": payload} msg_id = self._generate_msg_id() payload["_msg_id"] = msg_id print(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===") for attempt in range(max_retries): print(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}") self.send(payload) start_time = time.time() while (time.time() - start_time) < ack_timeout: if lock: lock.acquire() try: if msg_id in self.received_acks: self.received_acks.remove(msg_id) print(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===") return True finally: if lock: lock.release() packet = self.receive_packet(timeout_ms=500) if packet: print(f"[ReliableLoRa] Received raw packet while waiting for ACK: {packet}") if lock: lock.acquire() try: filtered_packet = self._process_incoming_packet(packet) if filtered_packet: self.pending_rx_queue.append(filtered_packet) finally: if lock: lock.release() if lock: lock.acquire() try: if msg_id in self.received_acks: self.received_acks.remove(msg_id) print(f"[ReliableLoRa] === ACK received for msg_id {msg_id} after poll ===") return True finally: if lock: lock.release() print(f"[ReliableLoRa] Attempt {attempt + 1} timed out waiting for ACK for msg_id {msg_id}") print(f"[ReliableLoRa] ERROR: Failed to receive ACK for msg_id {msg_id} after {max_retries} attempts.") return False def receive_reliable(self, timeout_ms=1000): """Receives a packet, automatically sending ACKs and filtering duplicate retries.""" if len(self.pending_rx_queue) > 0: return self.pending_rx_queue.pop(0) start_time = time.time() timeout_s = timeout_ms / 1000.0 while True: elapsed = time.time() - start_time remaining_ms = int((timeout_s - elapsed) * 1000) if remaining_ms <= 0: break poll_time = max(50, min(remaining_ms, 300)) packet = self.receive_packet(timeout_ms=poll_time) if packet: filtered_packet = self._process_incoming_packet(packet) if filtered_packet: return filtered_packet return None if IS_MICROPYTHON: # --- PILOTE SPI DIRECT (ESP32 / Heltec V3) --- class LoraHardwareSPI(BaseLoraDevice): def __init__(self, spi_bus=1, clk=9, mosi=10, miso=11, cs=8, irq=14, rst=12, gpio=13): super().__init__() self._pins = { "spi_bus": spi_bus, "clk": clk, "mosi": mosi, "miso": miso, "cs": cs, "irq": irq, "rst": rst, "gpio": gpio } self._cfg = {"freq": 868.1, "bw": 125.0, "sf": 7, "cr": 5, "power": 14} self.lock = _thread.allocate_lock() self.lora = None self.reset_hardware() def reset_hardware(self): """Resets SX1262 hardware and recreates driver instance.""" with self.lock: try: irq_pin = Pin(self._pins["irq"], Pin.IN) irq_pin.irq(handler=None) except Exception: pass try: rst_pin = Pin(self._pins["rst"], Pin.OUT) rst_pin.value(0) time.sleep_ms(30) rst_pin.value(1) time.sleep_ms(50) except Exception: pass self.lora = None time.sleep_ms(50) try: from sx1262 import SX1262 new_instance = SX1262(**self._pins) new_instance.begin( freq=self._cfg["freq"], bw=self._cfg["bw"], sf=self._cfg["sf"], cr=self._cfg["cr"], power=self._cfg["power"], useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False ) # SyncWord 0x12 = Decimal 18 new_instance.setSyncWord(0x12) self.lora = new_instance except Exception as e: print(f"[LoRa SPI] Initialization error: {e}") def configure(self, freq=868.1, bw=125.0, sf=7, cr=5, power=14): self._cfg = {"freq": freq, "bw": bw, "sf": sf, "cr": cr, "power": power} if self.lora is None: self.reset_hardware() else: with self.lock: try: self.lora.begin( freq=freq, bw=bw, sf=sf, cr=cr, power=power, useRegulatorLDO=False, crcOn=True, preambleLength=8, implicit=False ) self.lora.setSyncWord(0x12) except Exception: self.reset_hardware() def send(self, payload, group=None): """Encodes payload into JSON and prepends group byte.""" with self.lock: if self.lora is None: return if group is None: group = self.default_group if isinstance(payload, (dict, list)): payload = json.dumps(payload) if isinstance(payload, str): payload = payload.encode('utf-8') paquet_physique = bytes([group]) + payload try: self.lora.send(paquet_physique) except Exception as e: print(f"[LoRa SPI] Send error: {e}") def receive_packet(self, timeout_ms=500): """Listens on SPI bus with auto-detection for JSON vs. Grouped headers.""" with self.lock: if self.lora is None: return None try: data, state = self.lora.recv(len=0, timeout_en=True, timeout_ms=timeout_ms) except Exception as e: print(f"[LoRa SPI] Recv error caught: {e}") return None if state == 0 and data is not None and len(data) > 0: if data[0] in (0x7B, 0x5B): # Starts with '{' or '[' group = self.default_group payload_brute = data.strip(b'\x00 \r\n\t') elif len(data) > 1: group = data[0] payload_brute = data[1:].strip(b'\x00 \r\n\t') else: return None 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: # --- PILOTE SÉRIE (Raspberry Pi / Dragino LA66) --- class LoraSerialAT(BaseLoraDevice): def __init__(self, port): super().__init__() 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() # Initial configuration self.configure(freq=868.1, sf=7, bw=125) def _send_at_cmd(self, cmd, wait_time=0.15): """Helper to send AT command and purge response buffer.""" self.ser.write(f"{cmd}\r\n".encode('utf-8')) time.sleep(wait_time) resp = "" while self.ser.in_waiting > 0: resp += self.ser.readline().decode('utf-8', errors='ignore') return resp def configure(self, freq=868.1, sf=7, bw=125): """Configures LA66 frequency, SF, BW, SyncWord, CRC, and continuous RX mode.""" with self.lock: freq_hz = int(freq * 1000000) bw_code = 0 if bw == 125 else 1 # Parameters: Freq, SF, BW, CR(0=4/5), Preamble(8), Header(1=Explicit), CRC(1=ON), IQ(0=Standard), NetMode(0=P2P), Power(14), SyncWord(18=0x12), Format(0), Type(1) at_cfg_cmd = f"AT+CFG={freq_hz},{sf},{bw_code},0,8,1,1,0,0,14,18,0,1" self._send_at_cmd(at_cfg_cmd, wait_time=0.2) # Fallback standalone commands self._send_at_cmd("AT+SYNCWORD=18", wait_time=0.1) self._send_at_cmd("AT+PRECV=65535", wait_time=0.1) self.ser.reset_input_buffer() def send(self, payload, group=None): """Encodes payload into HEX AT command and re-enables continuous RX.""" with self.lock: if group is None: group = self.default_group if isinstance(payload, (dict, list)): payload = json.dumps(payload) if isinstance(payload, str): payload = payload.encode('utf-8') paquet_physique = bytes([group]) + payload hex_payload = paquet_physique.hex() self.ser.reset_input_buffer() print(f"[RPi LoRa Serial] Transmitting HEX payload: {hex_payload}") cmd = f"AT+PSEND={hex_payload}" resp = self._send_at_cmd(cmd, wait_time=0.25) # Wait for RF TX to finish print(f"[RPi LoRa Serial] AT+PSEND response: {resp}") # Re-enable continuous receive mode after transmission completes self._send_at_cmd("AT+PRECV=65535", wait_time=0.05) def receive_packet(self, timeout_ms=500): """Reads incoming serial lines from LA66 stick with robust format parsing.""" 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 # Robust parsing for LA66 response variants (+RECV:, +RCV=, +DRX:, HEX:, Data:) if "+RECV:" in line: parts = line.split("+RECV:")[1].strip().split(",") hex_str = parts[2].strip() if len(parts) >= 3 else parts[0].strip() try: payload_bytes = bytes.fromhex(hex_str) except ValueError: pass elif "+RCV=" in line: parts = line.split("+RCV=")[1].strip().split(",") if len(parts) >= 4: try: payload_bytes = bytes.fromhex(parts[3].strip()) except ValueError: pass elif "+DRX:" in line: parts = line.split("+DRX:")[1].strip().split(",") if len(parts) >= 2: try: payload_bytes = bytes.fromhex(parts[1].strip()) except ValueError: pass elif "(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) > 0: if payload_bytes[0] in (0x7B, 0x5B): group = self.default_group payload_clean = payload_bytes.strip(b'\x00 \r\n\t') elif len(payload_bytes) > 1: group = payload_bytes[0] payload_clean = payload_bytes[1:].strip(b'\x00 \r\n\t') else: continue 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) class LoraCommands: PING = "ping" COOKING_STATE_UPDATE = "cooking_state_update" TOGGLE_PAUSE = "toggle_pause"