Set it up right. Then make it report.

Pick a board, power supply, storage and adapters that match the job — a headless monitoring box does not need a desktop kit.

Check yourself

Answer out loud first, then open the card.

Imager downloads the OS, writes the card, and pre-configures hostname, user, Wi-Fi and SSH so the Pi boots straight onto the network.

Check yourself

Answer out loud first, then open the card.

Power on, give it a minute, then connect from your laptop by hostname or IP.

# from your laptop (mDNS name you set in Imager)
ssh <username>@pi-lab-01.local

# or by IP address
ssh <username>@192.168.1.50

# first time: confirm the host key fingerprint, then you get a prompt
<username>@pi-lab-01:~ $

How it breaks, and what to check

Check yourself

Answer out loud first, then open the card.

Keep the OS patched with APT and keep project libraries inside a virtual environment.

sudo apt update          # refresh package lists
sudo apt full-upgrade    # install updates (Raspberry Pi recommends full-upgrade)
df -h                    # check free space if the upgrade asks for more
sudo reboot              # after kernel/firmware updates

mkdir -p ~/lab && cd ~/lab
python3 -m venv .venv
source .venv/bin/activate
pip install pymodbus pyserial paho-mqtt flask
pip freeze > requirements.txt   # pin what you tested

Check yourself

Answer out loud first, then open the card.

Give the Pi a predictable address, then learn the handful of commands that prove where a network problem lives.

nmcli connection show                  # find the connection NAME for eth0
sudo nmcli connection modify "<NAME>" \
  ipv4.method manual \
  ipv4.addresses 192.168.1.50/24 \
  ipv4.gateway 192.168.1.1 \
  ipv4.dns "192.168.1.1"
sudo nmcli connection up "<NAME>"      # apply (you may lose SSH if wrong)

How it breaks, and what to check

Check yourself

Answer out loud first, then open the card.

Small, boring steps that stop most problems: personal accounts, key-only SSH, a default-deny firewall, and regular updates.

sudo adduser <name>                    # create a personal account
sudo usermod -aG sudo,dialout <name>   # admin + serial ports (USB-RS485)
passwd                                 # change your own password
sudo deluser --remove-home <oldname>   # remove accounts nobody should use

ssh-keygen -t ed25519                  # creates ~/.ssh/id_ed25519(.pub)
ssh-copy-id <name>@pi-lab-01.local     # installs the public key on the Pi
ssh <name>@pi-lab-01.local             # should log in without a password

AllowUsers <name>
PasswordAuthentication no

sudo sshd -t                 # syntax check before restarting
sudo systemctl restart ssh   # keep your current session open while you test a new one

sudo apt install ufw
sudo ufw default deny incoming
sudo ufw allow ssh                      # or: sudo ufw limit ssh/tcp
sudo ufw allow from 192.168.1.0/24 to any port 5000 proto tcp   # LAN-only dashboard
sudo ufw enable
sudo ufw status verbose

Check yourself

Answer out loud first, then open the card.

A script in a terminal dies when you log out. A systemd service starts at boot, restarts on failure, and logs to the journal.

[Unit]
Description=Lab Modbus reader (synthetic data)
After=network-online.target
Wants=network-online.target

[Service]
User=<name>
WorkingDirectory=/home/<name>/lab
Environment=MODBUS_PORT=/dev/ttyUSB0
ExecStart=/home/<name>/lab/.venv/bin/python reader.py
Restart=on-failure
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload              # after creating/editing a unit
sudo systemctl enable --now meter-reader  # start now + at every boot
systemctl status meter-reader             # state, PID, last log lines
journalctl -u meter-reader -f             # follow its logs live
journalctl -u meter-reader --since "1 hour ago"
sudo systemctl restart meter-reader
systemctl --failed                        # anything crashed?

Check yourself

Answer out loud first, then open the card.

Work bottom-up: power, storage, network, service, application. Write down each result.

The ladder

Check yourself

Answer out loud first, then open the card.

Poll a meter-style device over a USB-RS485 adapter and print readings — the same modem → logger → RS-485 → meter path a monitoring desk supports.

Code — reader.py

# reader.py - poll one Modbus RTU device over a USB-RS485 adapter
# Register addresses are PLACEHOLDERS for the lab simulator.
# On real hardware, take addresses, scaling and word order from the
# meter's own register map - they differ between vendors and models.
import os, sys, time
from pymodbus.client import ModbusSerialClient
from pymodbus.exceptions import ModbusException

PORT = os.getenv("MODBUS_PORT", "/dev/ttyUSB0")
UNIT = int(os.getenv("MODBUS_UNIT", "1"))

client = ModbusSerialClient(port=PORT, baudrate=9600, bytesize=8,
                            parity="N", stopbits=1, timeout=1, retries=2)

def read_once():
    rr = client.read_holding_registers(0, count=4, device_id=UNIT)
    if rr.isError():
        raise IOError(f"Modbus error from unit {UNIT}: {rr}")
    regs = rr.registers
    energy = client.convert_from_registers(
        regs[1:3], data_type=client.DATATYPE.UINT32, word_order="big")
    return {"ts": int(time.time()), "power_w": regs[0],
            "energy_wh": energy, "voltage_v": regs[3] / 10}

if __name__ == "__main__":
    if not client.connect():
        sys.exit(f"cannot open {PORT} - check cable, permissions (dialout), port name")
    try:
        while True:
            try:
                print(read_once(), flush=True)
            except (ModbusException, IOError) as exc:
                print(f"WARN {exc}", file=sys.stderr, flush=True)
            time.sleep(5)
    finally:
        client.close()

Bench simulator — no meter needed

A tiny synthetic “meter” that answers function 03 with random values. Run it on a second adapter, or on Linux use a virtual serial pair:

socat -d pty,raw,echo=0,link=/tmp/ttyLAB0 pty,raw,echo=0,link=/tmp/ttyLAB1 &
python lab_sim.py /tmp/ttyLAB1 &
MODBUS_PORT=/tmp/ttyLAB0 python reader.py
# {'ts': 1791246786, 'power_w': 3638, 'energy_wh': 1250005, 'voltage_v': 240.5}

# lab_sim.py - tiny synthetic Modbus RTU "meter" (function 03 only) for bench practice.
# Pair it with a virtual serial link (socat) or a second USB-RS485 adapter.
import random, struct, sys, serial

def crc16(frame: bytes) -> bytes:
    crc = 0xFFFF
    for b in frame:
        crc ^= b
        for _ in range(8):
            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
    return struct.pack("<H", crc)                 # CRC is sent low byte first

UNIT, energy_wh = 1, 1_250_000
port = serial.Serial(sys.argv[1] if len(sys.argv) > 1 else "/dev/ttyUSB1",
                     9600, bytesize=8, parity="N", stopbits=1, timeout=0.05)
buf = b""
while True:
    buf = (buf + port.read(64))[-8:]
    if len(buf) < 8 or crc16(buf[:6]) != buf[6:]:
        continue                                  # wait for a complete, valid request
    unit, func, start, count = struct.unpack(">BBHH", buf[:6]); buf = b""
    if unit != UNIT or func != 3:
        continue                                  # not for us: stay silent, like a real slave
    power_w = random.randint(3200, 4100)          # synthetic values only
    energy_wh += power_w // 720
    regs = [power_w, energy_wh >> 16, energy_wh & 0xFFFF, 2405] + [0] * 60
    data = struct.pack(f">{count}H", *regs[start:start + count])
    reply = struct.pack(">BBB", UNIT, 3, len(data)) + data
    port.write(reply + crc16(reply))

Support / troubleshooting notes

Store every reading locally first, forward later. When the modem or uplink drops, nothing is lost — the backlog just grows and drains when the link returns.

Code — buffer.py

# buffer.py - store-and-forward telemetry buffer (SQLite, survives reboots and outages)
import json, sqlite3, time

db = sqlite3.connect("telemetry.db")
db.execute("PRAGMA journal_mode=WAL")            # safer concurrent reads/writes
db.execute("""CREATE TABLE IF NOT EXISTS outbox (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ts INTEGER NOT NULL,
                payload TEXT NOT NULL,
                sent INTEGER NOT NULL DEFAULT 0)""")

def store(reading: dict):
    with db:                                     # commit (or roll back) atomically
        db.execute("INSERT INTO outbox (ts, payload) VALUES (?, ?)",
                   (reading["ts"], json.dumps(reading)))

def forward(send, batch=100) -> int:
    """Send oldest unsent rows first; mark sent only after send() succeeds."""
    rows = db.execute("SELECT id, payload FROM outbox WHERE sent = 0 "
                      "ORDER BY id LIMIT ?", (batch,)).fetchall()
    if not rows:
        return 0
    send([json.loads(p) for _, p in rows])       # raises on failure -> rows stay unsent
    with db:
        db.executemany("UPDATE outbox SET sent = 1 WHERE id = ?",
                       [(i,) for i, _ in rows])
    return len(rows)

def backlog() -> int:
    return db.execute("SELECT COUNT(*) FROM outbox WHERE sent = 0").fetchone()[0]

def prune(days=7):
    with db:                                     # keep the SD card from filling up
        db.execute("DELETE FROM outbox WHERE sent = 1 AND ts < ?",
                   (int(time.time()) - days * 86400,))

Test: simulate an outage

link_up = False
def send(batch):
    if not link_up:
        raise ConnectionError("uplink down")
    print("sent", len(batch))

for i in range(5):
    buffer.store({"ts": 1000 + i, "power_w": 3500 + i})
    try:
        buffer.forward(send)
    except ConnectionError as e:
        print("forward failed:", e, "backlog", buffer.backlog())

link_up = True
print("forwarded", buffer.forward(send), "backlog", buffer.backlog())

forward failed: uplink down backlog 1
...
forward failed: uplink down backlog 5
sent 5
forwarded 5 backlog 0

Support / troubleshooting notes

Ping the gateway, two public IPs, and resolve a name every minute. The first layer that fails tells you who to call.

Code — netwatch.py

# netwatch.py - layered reachability log: gateway -> modem/router -> internet -> DNS
# Targets are examples. Replace with your lab's gateway / modem LAN IP.
import csv, socket, subprocess, time
from datetime import datetime, timezone

TARGETS = [("gateway", "192.168.1.1"),       # first hop (ip route | grep default)
           ("internet", "1.1.1.1"),          # public IP: proves routing past the modem
           ("internet2", "8.8.8.8")]         # second public IP: rules out one bad host
DNS_NAME = "example.com"                     # proves name resolution separately

def ping(host: str) -> float | None:
    """Return round-trip ms, or None on loss. One packet, 2 s deadline (Linux iputils)."""
    r = subprocess.run(["ping", "-c", "1", "-W", "2", host],
                       capture_output=True, text=True)
    if r.returncode != 0:
        return None
    for part in r.stdout.split():
        if part.startswith("time="):
            return float(part[5:])
    return None

def dns_ok(name: str) -> bool:
    try:
        socket.getaddrinfo(name, 443)
        return True
    except socket.gaierror:
        return False

def verdict(res: dict, dns: bool) -> str:
    if res["gateway"] is None:
        return "LAN/gateway down"            # cable, Wi-Fi, switch, router power
    if res["internet"] is None and res["internet2"] is None:
        return "WAN down past gateway"       # modem, SIM/signal, ISP
    if not dns:
        return "DNS failure"                 # IP works, names don't
    return "OK"

with open("netwatch.csv", "a", newline="") as f:
    log = csv.writer(f)
    while True:
        res = {name: ping(ip) for name, ip in TARGETS}
        dns = dns_ok(DNS_NAME)
        row = [datetime.now(timezone.utc).isoformat(timespec="seconds"),
               *[res[n] if res[n] is not None else "" for n, _ in TARGETS],
               int(dns), verdict(res, dns)]
        log.writerow(row); f.flush()
        print(row, flush=True)
        time.sleep(60)

Reading the log

Support / troubleshooting notes

Publish readings to a broker with QoS 1 and a retained online/offline status, so a dashboard knows when the device disappears.

sudo apt install mosquitto mosquitto-clients
mosquitto_sub -h localhost -t 'lab/#' -v      # watch everything under lab/
# in another terminal:
MQTT_HOST=localhost python publisher.py

lab/lab-site-01/status online
lab/lab-site-01/meter1/telemetry {"ts": 1791246877, "power_w": 3933}
lab/lab-site-01/status offline

Code — publisher.py

# publisher.py - publish synthetic readings over MQTT (paho-mqtt 2.x)
import json, os, random, time
import paho.mqtt.client as mqtt

BROKER = os.getenv("MQTT_HOST", "localhost")
SITE = "lab-site-01"                               # synthetic site id
TOPIC = f"lab/{SITE}/meter1/telemetry"
STATUS = f"lab/{SITE}/status"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{SITE}-pub")
client.will_set(STATUS, "offline", qos=1, retain=True)   # broker announces if we vanish

def on_connect(c, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    c.publish(STATUS, "online", qos=1, retain=True)

client.on_connect = on_connect
client.connect(BROKER, 1883, keepalive=30)
client.loop_start()                                # network loop + auto-reconnect thread

try:
    while True:
        reading = {"ts": int(time.time()), "power_w": random.randint(3200, 4100)}
        info = client.publish(TOPIC, json.dumps(reading), qos=1)
        if info.rc != mqtt.MQTT_ERR_SUCCESS:
            print("publish not queued:", mqtt.error_string(info.rc))  # keep it in the SQLite buffer
        time.sleep(10)
finally:
    client.publish(STATUS, "offline", qos=1, retain=True).wait_for_publish(2)
    client.loop_stop(); client.disconnect()

Support / troubleshooting notes

A read-only LAN page plus a /health endpoint that returns 503 when data goes stale — easy for people and monitors to check.

Code — dashboard.py

# dashboard.py - small read-only LAN dashboard over the SQLite buffer
import json, sqlite3, time
from flask import Flask, jsonify, render_template_string

app = Flask(__name__)
DB = "telemetry.db"
STALE_AFTER = 300                                   # seconds without data = stale

PAGE = """<!doctype html><meta name=viewport content="width=device-width">
<title>Lab telemetry</title><meta http-equiv=refresh content=30>
<h1>Lab telemetry <small>(synthetic)</small></h1>
<p>Status: <b>{{ s.state }}</b> · last reading {{ s.age_s }} s ago</p>
<p>Power: {{ s.latest.power_w }} W · Unsent backlog: {{ s.backlog }}</p>"""

def summary():
    db = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)   # read-only: can't corrupt the buffer
    try:
        row = db.execute("SELECT ts, payload FROM outbox ORDER BY id DESC LIMIT 1").fetchone()
        backlog = db.execute("SELECT COUNT(*) FROM outbox WHERE sent = 0").fetchone()[0]
    finally:
        db.close()
    if row is None:
        return {"state": "NO DATA", "age_s": None, "latest": {}, "backlog": backlog}
    age = int(time.time()) - row[0]
    return {"state": "OK" if age <= STALE_AFTER else "STALE", "age_s": age,
            "latest": json.loads(row[1]), "backlog": backlog}

@app.get("/")
def index():
    return render_template_string(PAGE, s=summary())

@app.get("/health")
def health():
    s = summary()
    return jsonify(s), (200 if s["state"] == "OK" else 503)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)              # LAN only; firewall it (ufw) to your subnet

Run

python dashboard.py                              # dev server on port 5000
sudo ufw allow from 192.168.1.0/24 to any port 5000 proto tcp
curl -i http://pi-lab-01.local:5000/health
# HTTP/1.1 200 OK
# {"age_s":0,"backlog":1,"latest":{"power_w":3777,"ts":...},"state":"OK"}

Flask’s built-in server is for development. For anything longer-lived, run it under systemd (section 7) behind a production WSGI server.

Support / troubleshooting notes

Sources