Serial Port (Arduino/Raspberry Pi) Data Reader

Serial Port (Arduino/Raspberry Pi) Data Reader

Problem Statement

Embedded devices like Arduinos, ESP32s, and Raspberry Pi Picos constantly stream sensor data, debug logs, or telemetry over a serial (UART/USB) connection. Capturing that stream reliably — without losing lines, blocking forever on a dead port, or corrupting a CSV with partial reads — is surprisingly fiddly. A naive while True: print(ser.readline()) loop will hang indefinitely if the device disconnects, crash on decoding errors, and produce output that's useless for later analysis.

This script solves that by providing a robust, headless-friendly serial reader that: auto-discovers the port if not specified, reconnects on transient failures, timestamps every line, optionally writes to CSV, and shuts down cleanly on SIGINT/SIGTERM. It's the kind of small utility you'd drop into a Raspberry Pi to log sensor data for weeks at a time.

How It Works

The script wraps pyserial's blocking readline() in a resilient loop. Data flows: serial device → pyserial buffer → line decoder → optional timestamp → stdout and/or CSV writer. The main design decisions are (1) using a configurable timeout on the serial port so readline() returns periodically instead of blocking forever, which lets us check for shutdown signals and detect dead ports; (2) treating decode errors as recoverable (errors="replace") so a single corrupt byte doesn't kill a multi-day logging run; (3) writing CSV rows atomically per line and flushing, so a kill -9 loses at most the current line. Signal handlers set a global Event that the loop checks — this gives us graceful shutdown without relying on KeyboardInterrupt propagating through C-level serial reads.

Full Script

#!/usr/bin/env python3
"""
Serial Port (Arduino/Raspberry Pi) Data Reader

Reads newline-delimited data from a serial port, timestamps each line,
and writes it to stdout and/or a CSV file. Handles disconnects, decode
errors, and SIGINT/SIGTERM for clean shutdown.

Usage:
    python3 serial_reader.py --port /dev/ttyUSB0 --baud 9600 --output data.csv
"""

from __future__ import annotations

import argparse
import csv
import logging
import os
import signal
import sys
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import IO, Iterator, Optional

import serial
from serial import SerialException
from serial.tools import list_ports

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_BAUD: int = int(os.environ.get("SERIAL_BAUD", "9600"))
DEFAULT_TIMEOUT: float = float(os.environ.get("SERIAL_TIMEOUT", "1.0"))
RECONNECT_DELAY: float = float(os.environ.get("SERIAL_RECONNECT_DELAY", "3.0"))
MAX_RECONNECTS: int = int(os.environ.get("SERIAL_MAX_RECONNECTS", "0"))  # 0 = infinite
ENCODING: str = os.environ.get("SERIAL_ENCODING", "utf-8")
LOG_LEVEL: str = os.environ.get("LOG_LEVEL", "INFO").upper()

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)-7s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger("serial_reader")

# Set by signal handlers; the read loop polls this to know when to stop.
_shutdown = threading.Event()


def _install_signal_handlers() -> None:
    """Trap SIGINT/SIGTERM so we can flush and close cleanly."""

    def _handler(signum: int, _frame: object) -> None:
        log.info("Received signal %d, shutting down...", signum)
        _shutdown.set()

    signal.signal(signal.SIGINT, _handler)
    signal.signal(signal.SIGTERM, _handler)


def autodetect_port() -> Optional[str]:
    """Return the first plausible USB serial device, or None."""
    candidates = []
    for p in list_ports.comports():
        # Prefer devices that look like USB-serial adapters.
        if p.device and any(
            kw in (p.description or "").lower()
            for kw in ("usb", "uart", "serial", "arduino", "ch340", "cp210", "ftdi")
        ):
            candidates.append(p.device)
    if candidates:
        return candidates[0]
    # Fallback: any port at all.
    ports = [p.device for p in list_ports.comports()]
    return ports[0] if ports else None


def open_serial(port: str, baud: int, timeout: float) -> serial.Serial:
    """Open the serial port with sane defaults. Raises SerialException on failure."""
    ser = serial.Serial(
        port=port,
        baudrate=baud,
        bytesize=serial.EIGHTBITS,
        parity=serial.PARITY_NONE,
        stopbits=serial.STOPBITS_ONE,
        timeout=timeout,
        write_timeout=timeout,
    )
    # Some boards reset when DTR toggles; give them a moment to boot.
    time.sleep(0.2)
    ser.reset_input_buffer()
    return ser


def iter_lines(ser: serial.Serial) -> Iterator[bytes]:
    """Yield raw byte lines from the serial port until shutdown is requested."""
    while not _shutdown.is_set():
        try:
            raw = ser.readline()  # returns b"" on timeout, not an exception
        except SerialException as exc:
            log.warning("Serial read failed: %s", exc)
            raise
        if not raw:
            continue  # timeout tick; loop so we can check _shutdown
        yield raw


def decode_line(raw: bytes) -> str:
    """Decode a raw line, stripping trailing CR/LF and replacing bad bytes."""
    return raw.decode(ENCODING, errors="replace").rstrip("\r\n")


def format_row(ts: str, line: str) -> tuple[str, str]:
    """Return the (timestamp, payload) pair written to CSV."""
    return ts, line


def run(port: str, baud: int, timeout: float, output: Optional[Path]) -> int:
    """Main read loop with reconnect logic. Returns process exit code."""
    csv_file: Optional[IO[str]] = None
    csv_writer: Optional[csv.writer] = None

    if output is not None:
        output.parent.mkdir(parents=True, exist_ok=True)
        new_file = not output.exists() or output.stat().st_size == 0
        csv_file = output.open("a", newline="", encoding="utf-8")
        csv_writer = csv.writer(csv_file)
        if new_file:
            csv_writer.writerow(["timestamp", "value"])
            csv_file.flush()
        log.info("Appending CSV output to %s", output)

    attempts = 0
    try:
        while not _shutdown.is_set():
            try:
                log.info("Opening %s @ %d baud (timeout=%.1fs)", port, baud, timeout)
                ser = open_serial(port, baud, timeout)
            except SerialException as exc:
                attempts += 1
                if MAX_RECONNECTS and attempts >= MAX_RECONNECTS:
                    log.error("Giving up after %d reconnect attempts: %s", attempts, exc)
                    return 2
                log.warning("Open failed (%s); retrying in %.1fs", exc, RECONNECT_DELAY)
                if _shutdown.wait(RECONNECT_DELAY):
                    break
                continue

            attempts = 0  # reset backoff after a successful open
            log.info("Connected. Streaming... (Ctrl-C to stop)")
            try:
                with ser:
                    for raw in iter_lines(ser):
                        line = decode_line(raw)
                        ts = datetime.now(timezone.utc).isoformat(timespec="milliseconds")
                        print(f"{ts}  {line}", flush=True)
                        if csv_writer and csv_file:
                            csv_writer.writerow(format_row(ts, line))
                            csv_file.flush()
            except SerialException as exc:
                log.warning("Connection lost: %s", exc)
                if _shutdown.wait(RECONNECT_DELAY):
                    break
    finally:
        if csv_file:
            csv_file.close()
            log.info("CSV file closed.")
    return 0


def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Read newline-delimited data from a serial port.",
    )
    p.add_argument("--port", help="Serial device (e.g. /dev/ttyUSB0, COM3). "
                                  "Auto-detected if omitted.")
    p.add_argument("--baud", type=int, default=DEFAULT_BAUD,
                   help=f"Baud rate (default: {DEFAULT_BAUD}).")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
                   help=f"Read timeout in seconds (default: {DEFAULT_TIMEOUT}).")
    p.add_argument("--output", type=Path, default=None,
                   help="Optional CSV file to append rows to.")
    p.add_argument("--list", action="store_true",
                   help="List available serial ports and exit.")
    return p.parse_args(argv)


def main(argv: Optional[list[str]] = None) -> int:
    args = parse_args(argv)

    if args.list:
        ports = list(list_ports.comports())
        if not ports:
            print("No serial ports detected.")
            return 0
        for p in ports:
            print(f"{p.device}\t{p.description or 'n/a'}\t{p.hwid or ''}")
        return 0

    port = args.port or autodetect_port()
    if not port:
        log.error("No serial port specified and none could be auto-detected. "
                  "Use --port or run with --list.")
        return 1

    _install_signal_handlers()
    try:
        return run(port, args.baud, args.timeout, args.output)
    except Exception:  # last-resort guard so we always return a code
        log.exception("Unhandled error")
        return 3


if __name__ == "__main__":
    sys.exit(main())

Configuration

Usage Examples

1. Auto-detect the port and stream to the terminal:

$ python3 serial_reader.py
2024-06-01T12:00:00 INFO    Opening /dev/ttyUSB0 @ 9600 baud (timeout=1.0s)
2024-06-01T12:00:00 INFO    Connected. Streaming... (Ctrl-C to stop)
2024-06-01T12:00:00.412+00:00  temp=23.4 humidity=41.2
2024-06-01T12:00:01.418+00:00  temp=23.5 humidity=41.1
^C
2024-06-01T12:00:02 INFO    Received signal 2, shutting down...

2. Log a 115200-baud device to CSV for later analysis:

$ python3 serial_reader.py --port /dev/ttyUSB0 --baud 115200 --output logs/sensor.csv
$ head -3 logs/sensor.csv
timestamp,value
2024-06-01T12:00:00.412+00:00,temp=23.4 humidity=41.2
2024-06-01T12:00:01.418+00:00,temp=23.5 humidity=41.1

3. Discover what's plugged in:

$ python3 serial_reader.py --list
/dev/ttyUSB0  USB-Serial CH340  USB VID:PID=1a86:7523 LOCATION=1-1.2
/dev/ttyACM0  Arduino Uno       USB VID:PID=2341:0043 SERIAL=...

Error Handling

• Port missing or busy (SerialException on open): logged as a warning and retried every SERIAL_RECONNECT_DELAY seconds. Set SERIAL_MAX_RECONNECTS to bound the retries.
• Device unplugged mid-stream: readline() raises SerialException; the inner loop exits, the port is closed via the with block, and the outer loop reconnects.
• Non-UTF-8 bytes: decoded with errors="replace" so binary garbage becomes \ufffd instead of crashing the logger.
• No newline ever arrives: the timeout makes readline() return b"" periodically, which we treat as a no-op tick so _shutdown is checked.
• SIGINT/SIGTERM: signal handler sets _shutdown; loop exits after the current line, CSV is flushed and closed in finally.
• No port found: exits with code 1 and a message pointing at --list.

Extending Ideas

• Parse structured payloads: replace format_row with a regex or json.loads to write typed columns (temp, humidity) instead of a single value string — then a pandas read_csv gives you a DataFrame for free.
• Publish to a message bus: swap the CSV writer for an MQTT (paho-mqtt) or Kafka producer so the data lands in your existing telemetry pipeline.
• Add a --duration flag: stop after N seconds or N lines for bounded test runs, useful in CI where you can't rely on a human pressing Ctrl-C.
• Rotate output files: use logging.handlers.TimedRotatingFileHandler semantics (or RotatingFileHandler plus a size check) to prevent a multi-week log from filling the SD card on a Pi.