Better LoRa
This commit is contained in:
+109
-89
@@ -27,6 +27,12 @@ class BaseLoraDevice:
|
||||
self.received_acks = set()
|
||||
self.pending_rx_queue = []
|
||||
self.default_group = 2
|
||||
|
||||
# NEW: Global TX lock to prevent concurrent overlapping transmissions
|
||||
if IS_MICROPYTHON:
|
||||
self.tx_lock = _thread.allocate_lock()
|
||||
else:
|
||||
self.tx_lock = threading.Lock()
|
||||
|
||||
def _generate_msg_id(self):
|
||||
return random.getrandbits(16)
|
||||
@@ -77,51 +83,49 @@ class BaseLoraDevice:
|
||||
|
||||
def send_reliable(self, payload, max_retries=4, ack_timeout=3.0):
|
||||
"""Sends a payload and listens in a single continuous RX window for the ACK."""
|
||||
lock = getattr(self, 'lock', None)
|
||||
|
||||
if isinstance(payload, dict):
|
||||
payload = dict(payload)
|
||||
else:
|
||||
payload = {"data": payload}
|
||||
with self.tx_lock:
|
||||
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
|
||||
msg_id = self._generate_msg_id()
|
||||
payload["_msg_id"] = msg_id
|
||||
|
||||
log(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===")
|
||||
log(f"\n[ReliableLoRa] === Starting send_reliable for msg_id {msg_id} ===")
|
||||
|
||||
for attempt in range(max_retries):
|
||||
log(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}")
|
||||
self.send(payload)
|
||||
for attempt in range(max_retries):
|
||||
log(f"[ReliableLoRa] Attempt {attempt + 1}/{max_retries} transmitting msg_id {msg_id}")
|
||||
self.send(payload)
|
||||
|
||||
# 1. Open a single continuous RX window for the full timeout duration
|
||||
# (Defaulted to 3.0s to account for LA66 UART + transmission time)
|
||||
timeout_ms = int(ack_timeout * 1000)
|
||||
packet = self.receive_packet(timeout_ms=timeout_ms)
|
||||
timeout_ms = int(ack_timeout * 1000)
|
||||
packet = self.receive_packet(timeout_ms=timeout_ms)
|
||||
|
||||
if 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()
|
||||
|
||||
# 2. Process incoming packet if received
|
||||
if packet:
|
||||
if lock: lock.acquire()
|
||||
try:
|
||||
filtered_packet = self._process_incoming_packet(packet)
|
||||
if filtered_packet:
|
||||
self.pending_rx_queue.append(filtered_packet)
|
||||
if msg_id in self.received_acks:
|
||||
self.received_acks.remove(msg_id)
|
||||
log(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===")
|
||||
return True
|
||||
finally:
|
||||
if lock: lock.release()
|
||||
|
||||
# 3. Check if matching ACK was received
|
||||
if lock: lock.acquire()
|
||||
try:
|
||||
if msg_id in self.received_acks:
|
||||
self.received_acks.remove(msg_id)
|
||||
log(f"[ReliableLoRa] === ACK received for msg_id {msg_id} on attempt {attempt + 1} ===")
|
||||
return True
|
||||
finally:
|
||||
if lock: lock.release()
|
||||
log(f"[ReliableLoRa] Attempt {attempt + 1} timed out waiting for ACK for msg_id {msg_id}")
|
||||
|
||||
log(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
|
||||
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."""
|
||||
@@ -296,19 +300,28 @@ else:
|
||||
self.ser.reset_output_buffer()
|
||||
self.lock = threading.Lock()
|
||||
|
||||
# Initial configuration
|
||||
self._emergency_rx_buffer = []
|
||||
|
||||
self.configure(freq=868.1, sf=7, bw=125)
|
||||
|
||||
def _send_at_cmd(self, cmd, timeout=1.5):
|
||||
"""Sends AT command and reads response until line received or timeout."""
|
||||
"""Sends AT command and safely handles async incoming packets."""
|
||||
self.ser.write(f"{cmd}\r\n".encode('utf-8'))
|
||||
|
||||
start = time.time()
|
||||
resp = ""
|
||||
while (time.time() - start) < timeout:
|
||||
if self.ser.in_waiting > 0:
|
||||
line = self.ser.readline().decode('utf-8', errors='ignore')
|
||||
resp += line
|
||||
line = self.ser.readline().decode('utf-8', errors='ignore').strip()
|
||||
|
||||
if not line:
|
||||
continue
|
||||
|
||||
if any(marker in line for marker in ["+RECV:", "+RCV=", "+DRX:", "(HEX:)", "Data:"]):
|
||||
self._emergency_rx_buffer.append(line)
|
||||
continue
|
||||
|
||||
resp += line + "\n"
|
||||
if "OK" in line or "ERROR" in line or "AT_BUSY" in line:
|
||||
break
|
||||
time.sleep(0.02)
|
||||
@@ -364,64 +377,70 @@ else:
|
||||
timeout_s = timeout_ms / 1000.0
|
||||
|
||||
while (time.time() - start_time) < timeout_s:
|
||||
if self.ser.in_waiting > 0:
|
||||
line = ""
|
||||
|
||||
# NEW: Process emergency buffer first before checking serial
|
||||
if self._emergency_rx_buffer:
|
||||
line = self._emergency_rx_buffer.pop(0)
|
||||
elif 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)
|
||||
|
||||
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 "+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)
|
||||
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 "Data:" in line:
|
||||
payload_bytes = line.split("Data:")[1].strip().encode('utf-8')
|
||||
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
|
||||
|
||||
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:
|
||||
text = payload_clean.decode('utf-8').strip('\x00 \r\n\t')
|
||||
except UnicodeError:
|
||||
continue
|
||||
|
||||
if text.startswith('{') or text.startswith('['):
|
||||
decoded_text = bytes.fromhex(text).decode('utf-8').strip('\x00 \r\n\t')
|
||||
except Exception:
|
||||
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}
|
||||
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
|
||||
@@ -438,6 +457,7 @@ def get_lora_device(port_or_pins=None):
|
||||
|
||||
class LoraCommands:
|
||||
PING = "ping"
|
||||
MICROVAVE_STATE_UPDATE = "microwave_state_update"
|
||||
COOKING_STATE_UPDATE = "cooking_state_update"
|
||||
COOKING_UPDATE = "cooking_update"
|
||||
TOGGLE_PAUSE = "toggle_pause"
|
||||
|
||||
Reference in New Issue
Block a user