simple pyserial script to receive encoded as weather data over LoRa and decode it locally

lora_weather_decoder.py

import re
import serial
import threading
import time

# Configure the serial port.
# Replace '/dev/ttyUSB0' with your actual port:
# Windows: 'COM3', 'COM4', etc.
# Mac/Linux: '/dev/ttyUSB0', '/dev/ttyAMA0', etc.
SERIAL_PORT = 'COM3'
BAUD_RATE = 115200
target_addr = 10
network_id = 18
network_band = 915000000

try:
    ser = serial.Serial(
        port=SERIAL_PORT,
        baudrate=BAUD_RATE,
        bytesize=serial.EIGHTBITS,
        parity=serial.PARITY_NONE,
        stopbits=serial.STOPBITS_ONE,
        timeout=1.0
    )
except Exception as e:
    print(f"Error opening serial port: {e}")
    exit()


# Simple mapping dictionary to encode hex characters into plausible weather metrics.
# We will use the decimals of different weather readings to encode our payload data.
HEX_MAP = {
    '0': (15.1, 45.1), '1': (15.2, 45.2), '2': (15.3, 45.3), '3': (15.4, 45.4),
    '4': (15.5, 45.5), '5': (15.6, 45.6), '6': (15.7, 45.7), '7': (15.8, 45.8),
    '8': (16.1, 46.1), '9': (16.2, 46.2), 'a': (16.3, 46.3), 'b': (16.4, 46.4),
    'c': (16.5, 46.5), 'd': (16.6, 46.6), 'e': (16.7, 46.7), 'f': (16.8, 46.8)
}

# Reverse mapping for decoding
REV_MAP_TEMP = {v[0]: k for k, v in HEX_MAP.items()}
REV_MAP_HUMID = {v[1]: k for k, v in HEX_MAP.items()}

def decode_from_weather(packet_list):
    """
    Decodes a list of mock weather telemetry packets back into the original hex payload.
    """
    decoded_hex = ""
    for packet in packet_list:
        # Extract the Temperature and Humidity values using regular expressions
        t_match = re.search(r"T:([0-9.]+)C", packet)
        h_match = re.search(r"H:([0-9.]+)%", packet)

        if t_match and h_match:
            t_val = float(t_match.group(1))
            h_val = float(h_match.group(1))

            char1 = REV_MAP_TEMP.get(t_val, '?')
            char2 = REV_MAP_HUMID.get(h_val, '?')

            decoded_hex += char1 + char2

    return decoded_hex

def read_from_port(serial_connection):
    """Continuously listens for incoming data from the RYLR998 module."""
    print("Listening for incoming LoRa messages...")
    while True:
        if serial_connection.in_waiting > 0:
            try:
                # Read the data until a newline character is found
                incoming_data = serial_connection.readline().decode('utf-8').strip()
                if incoming_data:
                    print(f"\n[Received]: {incoming_data}")

                    # Incoming message format is usually: +RCV=ADDRESS,LENGTH,DATA,RSSI,SNR
                    if "+RCV=" in incoming_data:
                        if incoming_data.split(",")[2].startswith("START"):
                            weather_buffer = []
                            buffer_len = int(incoming_data.split(",")[2].split(":")[1])
                            print("Found Start!")
                            print("Resetting buffer...")
                            print("Buffer length: " + str(buffer_len))
                        elif incoming_data.split(",")[2].startswith("END"):
                            print("Found End!")
                            print("Testing for expected length: " + str(buffer_len))
                            print("Weather data received length: " + str(len(weather_buffer)))
                            if len(weather_buffer) ==  buffer_len:
                                extracted_hex = decode_from_weather(weather_buffer)
                                reconstructed_string = bytes.fromhex(extracted_hex).decode('utf-8')

                                print(f"Decoded Hex: {extracted_hex}")
                                print(f"Reconstructed String: {reconstructed_string}")
                            else:
                                print("Transmission error did not receive all {} weather packets!".format(buffer_len))
                        else:
                            weather_buffer.append(incoming_data.split(",")[2])
            except Exception as e:
                print(f"Error reading serial: {e}")
        time.sleep(0.01)


def send_command(command_str):
    """Formats and sends an AT command to the module."""
    full_command = f"{command_str}\r\n"
    ser.write(full_command.encode('utf-8'))
    print(f"[Sent]: {command_str}")


# Basic initialization sequence
time.sleep(1)
send_command("AT")  # Test if the module responds with "+OK"
time.sleep(2)
send_command(f"AT+ADDRESS={target_addr}") #  Set Address
time.sleep(2)
send_command(f"AT+NETWORKID={network_id}") #  Set Network
time.sleep(2)
send_command(f"AT+BAND={network_band}") #  Set Band
time.sleep(2)

# Start a background daemon thread to handle receiving data concurrently
listener_thread = threading.Thread(target=read_from_port, args=(ser,))
listener_thread.start()