lora_logger.py

Arduino MKR WAN 1300 · File : lora_logger.py · Language : Python · Use case : Archiver et tracer les trames LoRa reçues sur PC

On the computer side, this pyserial script consumes the LoRa gateway, writes each frame to a timestamped CSV and prints an RSSI bar chart in characters. You follow station range and the effect of moving an antenna live, with no proprietary software.

Hardware

An Arduino MKR WAN 1300 (SAMD21 microcontroller, Semtech SX1272 LoRa modem for point-to-point links, no built-in LoRaWAN stack, 868/915 MHz antenna). PC with Python 3 + pyserial, the MKR WAN gateway plugged over USB, stations transmitting nearby.

Wiring

Nothing beyond the gateway USB cable; the script skips the CSV header emitted by the sketch.

How it works

The for loop over the serial object iterates line by line; semicolon splitting separates metadata from payload. RSSI is mapped to a # bar for instant visual tracking.

Running it

Install Arduino SAMD Boards and the MKRWAN library, select the board, upload. You need TWO LoRa boards (two MKR WAN 1300s, or a 1300 and a 1310) to see exchanges; set the same frequency plan in the code.

Variants

Plot with matplotlib for range curves, or run the script under systemd on a Raspberry Pi for permanent logging.

lora_logger.py

#!/usr/bin/env python3
"""Lit la passerelle LoRa (passerelle-usb.ino) et journalise en CSV.

Usage : python lora_logger.py [port]
"""
import csv
import sys
import time

import serial

PORT = sys.argv[1] if len(sys.argv) > 1 else "/dev/ttyACM0"


def principal() -> None:
    with serial.Serial(PORT, 115200, timeout=1) as carte, \
         open("lora.csv", "a", newline="") as fh:
        ecrivain = csv.writer(fh)
        for ligne in carte:
            l = ligne.decode(errors="replace").strip()
            if l.count(";") < 3 or l.startswith("millis"):
                continue
            t_carte, rssi, snr, *trame = l.split(";")
            horloge = time.strftime("%H:%M:%S")
            ecrivain.writerow([horloge, t_carte, rssi, snr,
                               ";".join(trame)])
            fh.flush()
            barres = "#" * max(0, (int(float(rssi)) + 130) // 2)
            print(f"{horloge} {rssi} dBm {barres} {';'.join(trame)}")


if __name__ == "__main__":
    try:
        principal()
    except KeyboardInterrupt:
        print("\nJournalise a l'arret.")

← Back to the board