Part II: Biosignal Transmission over LoRa Meshtastic Network

We begin the second part of the project, titled Transmission of Biosignals via LoRa using the Meshtastic Net. This section explains in detail how the system enables the transmission of heart signals, SpO₂ values, and other biomedical measurements through the Meshtastic network. Furthermore, it describes a generalized communication approach that allows the transmission of any type of signal, whether it is a biosignal or another form of continuous analog signal, making the system flexible and adaptable to multiple applications.

Unlike variables such as temperature, humidity, or pressure — which can be transmitted at long intervals (every minute, ten minutes, or even hourly) —biosignals have a distinctive characteristic: they are continuous signals with rapid value changes occurring over very short periods of time. For this reason, it is not sufficient to send isolated samples at long intervals; instead, a transmission scheme that preserves the intrinsic dynamics of these signals is required.

• The first module corresponds to the acquisition device, responsible for capturing the biomedical signal (for example, ECG or SpO₂) using a microcontroller with an integrated ADC (such as ESP32, Arduino, or STM32), or through external protocols like I²C or SPI connected to a dedicated ADC. This module must sample the signal for a defined period (e.g., 30 seconds or 1 minute), store it in memory, and prepare it for transmission.
• The second module is the Meshtastic-compatible device (such as the XIAO LoRa, TTGO T-Beam, or Heltec), whose function is to receive the processed data and transmit it over the LoRa channel while adhering to the protocol’s limitations. The Meshtastic protocol allows a maximum of approximately 228 bytes per binary packet (protobuf payload) but recommends keeping messages below 200 bytes to ensure reliability. If a message exceeds this threshold, it is automaticallyfragmented into multiple packets, which increases the probability of data loss in multi-hop network environments.

First Module

Since biosignals are continuous and rapidly varying, unlike slower variables such as temperature or humidity that can be transmitted once per minute, a block-based transmission scheme is required. In this approach, a segment of the signal is captured, divided into packets of ≤200 bytes, and sent sequentially to be reconstructed at the receiver (for example, on a PC). This results in a well-defined architecture, where the first device focuses on signal acquisition and preprocessing, and the second device handles reliable communication through the mesh network, as illustrated in the figure below:

These segmented signals are processed by transforming each section into a Base64-encoded character string. This is achieved using the following C function:

Function:to_base64 — Sends normalized data through UART2 in Base64-encoded blocks.

SummarySplits the global buffer normalizedData (length max_len_data) into fixed-size blocks (blockSize, default 32 bytes). Each block is Base64-encoded and sent over Serial2, prefixed with a two-digit sequence number (00..99).

Output protocol

• Line 1:len_package;N — number of blocks to be sent (integer).
• Lines 2..N+1:SS<BASE64> — SS is a 2-digit sequence number (00..99) immediately followed by the Base64 string.
• Final line:END...

Example (block #7)

07QUJDRA==

07 → sequence number

• 07 → sequence number
• QUJDRA== → Base64 payload

Preconditions

• normalizedData and max_len_data are defined and contain the data to send.
• Serial2 is initialized at the desired baud rate.
• base64::encode(const uint8_t*, int) is available.

Notes

• If max_len_data is not a multiple of blockSize, the last block is partial.
• Fixed delays are used (2 s before start, 6 s between blocks, 1 s at the end).
• No checksum/ACK is implemented; receivers must handle potential loss.

/* to_base64: Sends normalized data through UART2 in Base64-encoded blocks.

Overview:
- Splits the global buffer `normalizedData` (length `max_len_data`) into fixed-size
  chunks (`blockSize`, default 32 bytes).
- Each chunk is Base64-encoded and sent via Serial2.
- A two-digit sequence number (00..99) is prefixed to every encoded chunk.

Output protocol:
- Line 1: "len_package;N"    // number of blocks to be sent
- Next lines: "SS<BASE64>"   // SS = two-digit sequence number, then Base64
- Final line: "END..."

Example (block 7):
07QUJDRA==      // "07" is the sequence, "QUJDRA==" is the Base64 payload

Preconditions:
- `normalizedData` and `max_len_data` are defined and contain the data to send.
- Serial2 is initialized (baud rate, config).
- base64::encode(const uint8_t*, int) is available.

Notes:
- If max_len_data is not a multiple of blockSize, the last block is partial.
- Uses fixed delays (2s before start, 6s between blocks, 1s at end).
- No checksum/ACK implemented; block loss is possible.
*/
void to_base64() {
  // Send data in Base64-encoded blocks
  const int blockSize = 32;  // Block size
  uint8_t block[blockSize];
  String base64Encoded;
  int conta = 0;

  int len_package = int(max_len_data / blockSize);
  Serial2.print("len_package;");
  Serial2.println(len_package);
  delay(2000);  // Wait before sending

  for (int i = 0; i < max_len_data; i += blockSize) {
    int count = min(blockSize, max_len_data - i);

    // Copy current block
    memcpy(block, &normalizedData[i], count);

    // Add 2-digit sequence number
    if (conta <= 9) {
      base64Encoded = "0" + String(conta) + base64::encode(block, count);
    } else {
      base64Encoded = String(conta) + base64::encode(block, count);
    }

    // Send encoded block over Serial
    Serial2.print(base64Encoded);
    delay(6000);  // Wait between blocks
    Serial.println();  // Debug newline
    conta = conta + 1;
  }
  Serial2.println("END...");
  delay(1000);
}

To execute this function, an M5Atom was used as the microcontroller.

and a BMD101 as the ECG acquisition system.

The system captures the ECG signal, processes it, and divides it into ordered segments for transmission. At the same time, it integrates a lightweight Edge AI model capable of analyzing the captured signal and determining whether it is normal or anomalous.

For training, the dataset from Kaggle was used.

This model assists in providing a preliminary diagnosis of the signal, and the resulting classification is transmitted through the same channel without any additional processing or adaptation. The trained model was later executed using TinyML.

To enable the transmission of biosignals using a device with Meshtastic firmware, it is essential to properly configure the device for serial (UART) communication. This configuration allows the Meshtastic module to act as a bridge between the microcontroller responsible for acquiring the signal (e.g., an external ADC or a biomedical sensor connected to an MCU) and the LoRa Mesh network. Through the serial port, preprocessed or encoded data are sent from the acquisition device to the Meshtastic node, which then encapsulates the information into packets and retransmits them over the network.

In addition, the serial configuration must include fundamental parameters such as baud rate, data bits, parity, and stop bits, ensuring compatibility between both devices. For continuous biosignals, it is recommended to use sufficiently high transmission rates (e.g., 115200 baud) to guarantee that data are transferred to the Meshtastic node without significant loss or delay. In this way, the acquisition microcontroller can focus exclusively on capturing and organizing the signal into blocks, while the Meshtastic module handles the protocol management, fragmentation, and retransmission within the mesh network.

In this connection, the M5Atom communicates with the ESP32 using the RS232 protocol through the pins RX = 4 and TX = 25 of the LILYGO T-LoRa T3-S3 board. This setup enables the direct and continuous transmission of data frames every five minutes, ensuring reliable communication between both devices. The system operates as a continuous ECG signal monitoring device, designed to capture and transmit biosignals periodically.

One of the main advantages of this module is its ability to transmit signals to any device compatible with the Meshtastic firmware. In particular, this system has been tested and validated with several Seeed Studio devices, ensuring stable and reliable communication across the network.

Second Module

The second prototype is a restructured version of a project previously presented in the Machine Builder Competition, where the Intelligent Health Care Beacon was originally designed. In this new version, the system was modified to reuse the nRF9151 DK development board, adapting its functionality to integrate with the capture and transmission of biosignals over the LoRa network.

For the acquisition of ECG signals, the analog-to-digital converters integrated into the SoC were used. At the same time, the nRF9151 DK is responsible for capturing oxygen saturation (SpO₂), heart rate, and the raw photoplethysmography (PPG) signal. These physiological parameters are obtained using a MAX30102 sensor, which operates with a primary power supply of 1.8 V and an additional 3.3 V source for its LEDs. Communication with the microcontroller is carried out via a standard I²C interface, and the module can be powered down through software with virtually zero standby current consumption, allowing the power lines to remain active at all times.

In addition, the system incorporates the AD8232 module, specifically designed for capturing the heart’s electrical activity (ECG). This module integrates a two-pole high-pass filter to remove motion artifacts and electrode half-cell potential, as well as an uncommitted operational amplifier to implement a three-pole low-pass filter, which further reduces noise and improves overall signal quality.

The nRF9151 DK includes a set of commands that enable interaction with the connected sensors. This command list is shown in the following figure.

Using the Meshtastic network, it is possible to send these commands across the nodes. When a device running the Meshtastic firmware receives a command, it forwards it to the nRF9151 DK via the serial port, triggering the data acquisition process.

Once the data capture is complete, the collected information—similarly to the M5Atom case—must be divided into chunks to allow efficient transmission through the Meshtastic network.

This ensures that other nodes can receive the data, particularly those equipped with an interface for subsequent visualization or plotting.

To transmit the data captured by the nRF9151 DK, a Heltec Wireless Tracker V1.1 was used. Running the Meshtastic firmware, this Heltec device performs the same transmission process as a conventional ESP32.

Graphical Interface for Raspberry Pi 5

Up to this point, we have successfully managed to receive the data. The next step is to process it using a Python-based interface, which is designed to run on a Raspberry Pi 5. This interface is structured into seven tabs, each providing a specific functionality:

• Tab 1: Map – Displays the coordinates of the devices running the Meshtastic firmware. It also allows loading and visualizing CSV files previously stored from tracking devices.

• Tab 2: Communications – Enables sending and receiving messages from the nodes connected to the Meshtastic network.

• Tab 3: Bio-Signal Capture – Designed for capturing biosignals, but currently only works on the Raspberry Pi 4 or the SeeedStudio reTerminal. This feature is not supported on the Raspberry Pi 5.

• Tab 4: Robot Control – A dedicated section for controlling the robot. This part is under development and will be delivered in the final stage.

• Tab 5: LoRa Signals – Displays the signals transmitted from our ESP32 and the nRF9151 DK. The system is designed to be robust enough to handle packets arriving out of order. Since Meshtastic uses a multi-hop routing algorithm, data packets may not arrive sequentially. The transmission begins with a header indicating the total number of packets to expect (this mechanism is currently being refactored). Once all packets are received, the Python script takes the Base64-encoded data, reorders them consecutively, decodes the payload, and finally plots the reconstructed signal for analysis.

The following section presents the code used for signal decoding.

# ==========================
#   Tab 5: LoRa Signals (placeholder)
# ==========================
import base64
import tkinter as tk
from tkinter import ttk, messagebox


class LoRaSignalsTab(ttk.Frame):
    """
    Tab 5: Signal reconstruction from Base64 packets with a 2-digit ID.
    - You can pass 'encoded_packages' in the constructor.
    - Or paste them into the text box (one per line, or a Python list).
    - DOES NOT modify the original list: creates a NEW ORDERED list by ID (2 digits).
    """
    def __init__(self, parent, encoded_packages=None):
        super().__init__(parent)
        self.encoded_packages_in = encoded_packages or []

        # --- UI ---
        self.columnconfigure(0, weight=1)
        self.rowconfigure(3, weight=1)

        # Options bar
        opts = ttk.Frame(self, padding=(10, 10))
        opts.grid(row=0, column=0, sticky="ew")
        for i in range(0, 10):
            opts.columnconfigure(i, weight=0)
        opts.columnconfigure(9, weight=1)

        ttk.Label(opts, text="Strict:").grid(row=0, column=0, sticky="w")
        self.var_strict = tk.BooleanVar(value=True)
        ttk.Checkbutton(opts, variable=self.var_strict).grid(row=0, column=1, padx=(4, 12))

        ttk.Label(opts, text="Signed:").grid(row=0, column=2, sticky="w")
        self.var_signed = tk.BooleanVar(value=True)
        ttk.Checkbutton(opts, variable=self.var_signed).grid(row=0, column=3, padx=(4, 12))

        ttk.Label(opts, text="Fs (Hz):").grid(row=0, column=4, sticky="w")
        self.var_fs = tk.StringVar(value="8000")
        ttk.Entry(opts, textvariable=self.var_fs, width=7).grid(row=0, column=5, padx=(4, 12))

        ttk.Label(opts, text="Save PNG (optional):").grid(row=0, column=6, sticky="w")
        self.var_save = tk.StringVar(value="")
        ttk.Entry(opts, textvariable=self.var_save, width=22).grid(row=0, column=7, padx=(4, 8))

        ttk.Button(opts, text="Reconstruct + Plot",
                   command=self._on_reconstruct_and_plot).grid(row=0, column=8, padx=(6, 0))

        # Input area
        ttk.Label(self, text="Packets (one per line with a 2-digit ID at the start):").grid(
            row=1, column=0, sticky="w", padx=10
        )
        self.txt_input = tk.Text(self, height=10)
        self.txt_input.grid(row=2, column=0, sticky="nsew", padx=10, pady=(0, 6))

        # If we received 'encoded_packages', preload them
        if self.encoded_packages_in:
            # Show as simple lines (ID+base64 per line)
            self.txt_input.insert(
                tk.END,
                "\n".join(self.encoded_packages_in) + "\n"
            )

        # Log / results
        ttk.Label(self, text="Log:").grid(row=3, column=0, sticky="w", padx=10)
        self.txt_log = tk.Text(self, height=12)
        self.txt_log.grid(row=4, column=0, sticky="nsew", padx=10, pady=(0, 10))

        # State
        self.bytes_recon = b""
        self.orden_ids = []

    # ======================
    #   RECONSTRUCTION LOGIC (your function, adapted as a method)
    # ======================
    def reconstruir_senal(self, encoded_packages, estricto=True):
        """
        Reconstructs the signal from out-of-order packets.
        - DOES NOT modify 'encoded_packages'.
        - Internally sorts by the first 2 digits (ID).
        - Returns (reconstructed_bytes, ordered_ids).

        estricto=True -> raises ValueError if duplicates or missing IDs are detected.
        """
        pairs = []
        for s in encoded_packages:
            s = s.strip().strip(",")  # trim a possible trailing comma
            if not s:
                continue
            if len(s) < 3 or not s[:2].isdigit():
                raise ValueError(f"Invalid packet: '{s}' (missing 2-digit prefix)")
            pid = int(s[:2])          # ID = 2 digits
            b64 = s[2:]               # rest = base64
            pairs.append((pid, b64))

        # NEW ordered list (do not touch the original)
        ordered = sorted(pairs, key=lambda t: t[0])
        ordered_ids = [pid for pid, _ in ordered]

        # Validations
        ids = [pid for pid, _ in pairs]
        dups = {x for x in ids if ids.count(x) > 1}
        if dups:
            msg = f"Duplicate IDs: {sorted(dups)}"
            if estricto:
                raise ValueError(msg)
            else:
                self._log("[WARN] " + msg)

        if ordered_ids:
            expected = list(range(min(ordered_ids), max(ordered_ids) + 1))
            missing = sorted(set(expected) - set(ordered_ids))
            if missing:
                msg = f"Missing IDs: {missing}"
                if estricto:
                    raise ValueError(msg)
                else:
                    self._log("[WARN] " + msg)

        # Decode and concatenate
        chunks = []
        for pid, b64 in ordered:
            # If needed, normalize padding:
            # b64 = b64 + "=" * ((4 - len(b64) % 4) % 4)
            chunk = base64.b64decode(b64, validate=True)
            chunks.append(chunk)

        return b"".join(chunks), ordered_ids

    # ======================
    #   PLOT (your function, adapted as a method)
    # ======================
    def plot_signal(self, raw_bytes, sample_rate=8000, signed=True,
                    title="Reconstructed signal", show=True, save_path=None):
        import os
        import numpy as np
        import matplotlib
        import matplotlib.pyplot as plt

        # If no display (headless Raspberry), attempt a non-interactive backend
        if not os.environ.get("DISPLAY"):
            try:
                matplotlib.use("Agg")
                show = False
            except Exception:
                pass

        u8 = np.frombuffer(raw_bytes, dtype=np.uint8)
        if signed:
            y = (u8.astype(np.int16) - 128) / 128.0
            ylabel = "Amplitude (signed, ~-1..1)"
        else:
            y = u8.astype(np.float32) / 255.0
            ylabel = "Amplitude (0..1)"

        x = np.arange(len(y)) / float(sample_rate)

        plt.figure(figsize=(10, 3))
        plt.plot(x, y, linewidth=1)
        plt.xlabel("Time (s)")
        plt.ylabel(ylabel)
        plt.title(title)
        plt.tight_layout()
        plt.grid(True)

        # Save if requested (or if headless and show is False)
        if save_path:
            try:
                plt.savefig(save_path, dpi=150)
                self._log(f"[OK] Image saved to: {save_path}")
            except Exception as e:
                self._log(f"[ERROR] Saving image: {e}")

        if show:
            try:
                plt.show()
            except Exception as e:
                # Fallback: save to file if showing failed
                alt = save_path or "signal.png"
                try:
                    plt.savefig(alt, dpi=150)
                    self._log(f"[WARN] No display. Saved to: {alt}")
                except Exception as e2:
                    self._log(f"[ERROR] No display and unable to save: {e2}")
        plt.close()

    # ======================
    #   UI Helpers
    # ======================
    def _log(self, text):
        self.txt_log.insert(tk.END, text + "\n")
        self.txt_log.see(tk.END)

    def _parse_input_packages(self):
        """
        Reads the text box. Supports:
          - A Python list (using ast.literal_eval), or
          - Line-by-line packets (one string per line).
        """
        import ast
        raw = self.txt_input.get("1.0", tk.END).strip()
        if not raw:
            return []
        # Is it a list literal?
        try:
            if raw.startswith("["):
                lst = ast.literal_eval(raw)
                # Sanitize to a list of strings
                return [str(x).strip() for x in lst if str(x).strip()]
        except Exception:
            pass
        # Otherwise, treat as lines
        lines = [ln.strip() for ln in raw.splitlines() if ln.strip()]
        return lines

    def _on_reconstruct_and_plot(self):
        # 1) choose source: from constructor or from text box
        pkgs = self.encoded_packages_in[:] if self.encoded_packages_in else self._parse_input_packages()
        if not pkgs:
            messagebox.showwarning("Attention", "No packets to process.")
            return

        # 2) parameters
        try:
            fs = int(self.var_fs.get())
        except ValueError:
            messagebox.showerror("Error", "Invalid sample rate.")
            return
        signed = bool(self.var_signed.get())
        estricto = bool(self.var_strict.get())
        save_path = self.var_save.get().strip() or None

        # 3) reconstruction
        try:
            data, orden = self.reconstruir_senal(pkgs, estricto=estricto)
            self.bytes_recon = data
            self.orden_ids = orden
            self._log(f"[OK] Order detected: {orden}")
            self._log(f"[OK] Reconstructed bytes: {len(data)}")
        except Exception as e:
            self._log(f"[ERROR] Reconstruction: {e}")
            messagebox.showerror("Error", str(e))
            return

        # 4) plot
        try:
            self.plot_signal(
                self.bytes_recon,
                sample_rate=fs,
                signed=signed,
                title="Reconstructed signal",
                show=True,           # headless mode auto-disables show
                save_path=save_path
            )
        except Exception as e:
            self._log(f"[ERROR] Plot: {e}")
            
def main():
    """
    Launch a Tkinter window with a Notebook and the LoRaSignals tab.
    Optional: --packets PATH  -> file containing either:
        - a JSON list of strings, or
        - one packet per line in the form 'DD<base64>' (DD = 2-digit ID).
    """
    import argparse
    import json
    from pathlib import Path

    parser = argparse.ArgumentParser(description="LoRa Signals GUI")
    parser.add_argument(
        "--packets",
        help="Path to a file with packets (JSON list or lines with 2-digit ID + base64)",
    )
    args = parser.parse_args()

    encoded = []
    if args.packets:
        try:
            txt = Path(args.packets).read_text(encoding="utf-8").strip()
            if txt.startswith("["):
                # JSON list
                data = json.loads(txt)
                encoded = [str(x).strip() for x in data if str(x).strip()]
            else:
                # One per line
                encoded = [ln.strip() for ln in txt.splitlines() if ln.strip()]
        except Exception as e:
            print(f"[WARN] Could not load packets from {args.packets}: {e}")

    root = tk.Tk()
    root.title("LoRa Signals")
    root.minsize(900, 600)

    nb = ttk.Notebook(root)
    nb.pack(expand=True, fill="both")

    tab = LoRaSignalsTab(nb, encoded_packages=encoded or None)
    nb.add(tab, text="LoRa Signals")

    def _on_close():
        root.destroy()

    root.protocol("WM_DELETE_WINDOW", _on_close)
    root.mainloop()


if __name__ == "__main__":
    main()
'''
python your_script.py
# or preloading packets:
python your_script.py --packets packets.txt
'''

This program verifies whether the data sequence is complete or if any signals are missing. If an inconsistency is detected, the system automatically generates an error message, as illustrated in the figure below.

• Tab 6: LoRa Extended – Expands the system to support additional forms of data transmission over LoRa. In this case, a Wio-E5 module is used, enabling communication with other LoRa devices that do not natively support the Meshtastic firmware.

Tab 7: Cameras– A section dedicated to camera integration. This functionality will be explained in a future release.

I apologize for dividing the project into several parts, but due to its complexity, it is not possible to complete it all in a single block.

In the next chapter, the Robot will be presented as an extension of this project. This section may take longer to develop, since it attempts to integrate Vision-Language Models (VLMs), and the training process of these models can take more than a day.