☚ zurück - Einstiegskurs Raspberry Pi

In diesem Projekt wird auf dem Raspberry Pi eine Weboberfläche mit FastAPI erstellt, über die eine LED-Ampel geschaltet und die Temperatur eines DS18B20 angezeigt werden kann. Die Anwendung ist im lokalen Netzwerk erreichbar, sodass LEDs und Temperatursensor bequem über einen Webbrowser im LAN gesteuert und überwacht werden können.

• Voraussetzungen
• Software
• Konfiguration

source ~/devel/projects/course_env/bin/activate

Anschließend werden FastAPI und Uvicorn installiert:

pip install fastapi uvicorn

cd ~/devel/projects/

course_web/
|── data/
|   └── temperature.txt
└── src/
    ├── app.py
    ├── core/
    │   ├── __init__.py
    │   └── hardware.py
    └── html/
        ├── led.html
        ├── history.html
        └── temp.html

Die Messwerte werden in einer Textdatei (temperature.txt) gespeichert, damit der Verlauf nachvollziehbar bleibt (auch wenn gerade kein Browser geöffnet ist).

Im folgenden Abschnitt werden die für die Webanwendung benötigten Python- und HTML-Dateien vorgestellt. Dazu gehören die Hardware-Anbindung über GPIO und den Temperatursensor, die HTML-Seiten zur Darstellung im Browser sowie die FastAPI-Anwendung, welche die Routen bereitstellt und die einzelnen Komponenten miteinander verbindet.

Datei: /home/pi/devel/projects/course_web/src/core/hardware.py

import RPi.GPIO as GPIO
import glob
import time
import os
from datetime import datetime
 
# -----------------------------
# API-Funktionen GPIO LED Ampel
# -----------------------------
 
PIN_R = 17
PIN_Y = 27
PIN_G = 22
 
_initialized = False
 
def init():
    global _initialized
    if _initialized:
        return
 
    GPIO.setwarnings(False)
    GPIO.setmode(GPIO.BCM)
 
    GPIO.setup(PIN_R, GPIO.OUT)
    GPIO.setup(PIN_Y, GPIO.OUT)
    GPIO.setup(PIN_G, GPIO.OUT)
 
    _initialized = True
 
 
def setLED(pin, value):
    GPIO.output(pin, GPIO.HIGH if value == 1 else GPIO.LOW)
 
 
def setRedLED(value):
    setLED(PIN_R, value)
 
 
def setYellowLED(value):
    setLED(PIN_Y, value)
 
 
def setGreenLED(value):
    setLED(PIN_G, value)
 
 
def status():
    return (
        int(GPIO.input(PIN_R)),
        int(GPIO.input(PIN_Y)),
        int(GPIO.input(PIN_G)),
    )
 
 
# -----------------------------
# API-Funktionen ds18b20
# -----------------------------
 
SENSOR_TIMEOUT = 1
 
def get_sensor():
    sensors = glob.glob("/sys/bus/w1/devices/28-*")
    if not sensors:
        return None
    return sensors[0] + "/w1_slave"
 
 
def get_temperature():
    sensor_file = get_sensor()
    if sensor_file is None:
        return None
 
    start_time = time.time()
 
    while True:
        with open(sensor_file, "r") as f:
            lines = f.readlines()
 
        if lines[0].strip().endswith("YES"):
            break
 
        if time.time() - start_time > SENSOR_TIMEOUT:
            return None
 
        time.sleep(0.1)
 
    temp_line = lines[1]
    temp_str = temp_line.split("t=")[1]
    return float(temp_str) / 1000.0
 
 
# -----------------------------
# Temperaturverlauf (History)
# -----------------------------
# Datei-Format: YYYY-MM-DD HH:MM:SS;TEMPERATUR
# Beispiel:     2026-02-21 12:00:00;21.437
 
DEFAULT_LOG_FILE = "/home/pi/devel/projects/course_web/data/temperature.txt"
 
def _ensure_parent_dir(path: str):
    parent = os.path.dirname(path)
    if parent and not os.path.exists(parent):
        os.makedirs(parent, exist_ok=True)
 
def history_append(value: float, log_file: str = DEFAULT_LOG_FILE) -> bool:
    """
    Speichert einen Messwert mit Zeitstempel in temperature.txt.
    Rückgabe: True wenn erfolgreich, sonst False.
    """
    try:
        _ensure_parent_dir(log_file)
        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open(log_file, "a") as f:
            f.write(f"{ts};{value:.3f}\n")
        return True
    except Exception:
        return False
 
def history_reset(log_file: str = DEFAULT_LOG_FILE) -> bool:
    """
    Löscht die Messdatei (setzt den Verlauf zurück).
    Rückgabe: True wenn erfolgreich, sonst False.
    """
    try:
        _ensure_parent_dir(log_file)
        with open(log_file, "w") as f:
            f.write("")
        return True
    except Exception:
        return False
 
def history_load(log_file: str = DEFAULT_LOG_FILE, max_lines: int = 2000):
    """
    Lädt Messwerte aus temperature.txt.
    Rückgabe: (labels, values)
      labels: Liste von Zeitstempeln (String)
      values: Liste von Temperaturen (float)
    Hinweis: max_lines begrenzt die Menge (für Browser/Performance).
    """
    labels = []
    values = []
 
    try:
        with open(log_file, "r") as f:
            lines = f.readlines()
    except FileNotFoundError:
        return labels, values
    except Exception:
        return labels, values
 
    # nur die letzten max_lines verwenden
    if max_lines and len(lines) > max_lines:
        lines = lines[-max_lines:]
 
    for line in lines:
        line = line.strip()
        if not line:
            continue
        try:
            ts, val = line.split(";")
            labels.append(ts)
            values.append(float(val))
        except Exception:
            continue
 
    return labels, values

Datei: /home/pi/devel/projects/course_web/src/html/history.html

<!DOCTYPE html>
<html>
<head>
    <title>Temperaturverlauf</title>
    <meta charset="utf-8" />
</head>
<body>
 
<h1>Temperaturverlauf</h1>
 
<h2>Aufzeichnung</h2>
<a href="/history/start">Start</a>
<a href="/history/stop">Stop</a>
<a href="/history/reset">Reset</a>
 
<h2>Intervall (Minuten)</h2>
<a href="/history/interval/1">1</a>
<a href="/history/interval/5">5</a>
<a href="/history/interval/10">10</a>
 
<h2>Diagramm (SVG)</h2>
<img src="/history/plot" alt="Temperaturverlauf" />
 
<br><br>
<a href="/">Zurück</a>
 
</body>
</html>

Datei: /home/pi/devel/projects/course_web/src/app.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from core import hardware
 
import threading
import time
 
app = FastAPI()
 
recording = False
interval_seconds = 60
_collector_started = False
_lock = threading.Lock()
 
 
def _collector_loop():
    global recording
 
    while True:
        if recording:
            t = hardware.get_temperature()
            if t is not None:
                hardware.history_append(t)
        time.sleep(interval_seconds)
 
 
def _ensure_collector():
    global _collector_started
    with _lock:
        if _collector_started:
            return
        th = threading.Thread(target=_collector_loop, daemon=True)
        th.start()
        _collector_started = True
 
 
@app.on_event("startup")
def startup():
    hardware.init()
    _ensure_collector()
 
 
def load_template(name, replacements):
    try:
        with open(f"html/{name}", "r") as f:
            html = f.read()
    except FileNotFoundError:
        raise HTTPException(status_code=500, detail="Template nicht gefunden")
 
    for key, value in replacements.items():
        html = html.replace(key, str(value))
 
    return html
 
 
@app.get("/led", response_class=HTMLResponse)
def led_page():
    r, y, g = hardware.status()
    return HTMLResponse(
        load_template("led.html", {
            "{{R}}": r,
            "{{Y}}": y,
            "{{G}}": g
        })
    )
 
 
@app.get("/led/{color}/{value}")
def set_led(color: str, value: int):
    if value not in (0, 1):
        raise HTTPException(status_code=400)
 
    if color == "r":
        hardware.setRedLED(value)
    elif color == "y":
        hardware.setYellowLED(value)
    elif color == "g":
        hardware.setGreenLED(value)
    else:
        raise HTTPException(status_code=400)
 
    return RedirectResponse(url="/led", status_code=303)
 
 
@app.get("/temp", response_class=HTMLResponse)
def temp_page():
    t = hardware.get_temperature()
 
    if t is None:
        value = "Sensorfehler"
    else:
        value = f"{t:.2f}"
 
    return HTMLResponse(load_template("temp.html", {"{{T}}": value}))
 
 
# -----------------------------
# History (Temperaturverlauf)
# -----------------------------
 
@app.get("/history", response_class=HTMLResponse)
def history_page():
    return HTMLResponse(load_template("history.html", {}))
 
 
@app.get("/history/start")
def history_start():
    global recording
    recording = True
    return RedirectResponse(url="/history", status_code=303)
 
 
@app.get("/history/stop")
def history_stop():
    global recording
    recording = False
    return RedirectResponse(url="/history", status_code=303)
 
 
@app.get("/history/reset")
def history_reset():
    hardware.history_reset()
    return RedirectResponse(url="/history", status_code=303)
 
 
@app.get("/history/interval/{minutes}")
def history_interval(minutes: int):
    global interval_seconds
    if minutes < 1:
        raise HTTPException(status_code=400)
    interval_seconds = minutes * 60
    return RedirectResponse(url="/history", status_code=303)
 
 
def _svg_plot(labels, values, width=900, height=300, pad=30):
    # sehr einfache SVG-Linie (ohne JS, ohne externe Libs)
    if not values:
        return f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}"><text x="20" y="40">Keine Daten</text></svg>'
 
    vmin = min(values)
    vmax = max(values)
    if vmax == vmin:
        vmax = vmin + 1e-6
 
    n = len(values)
    x0 = pad
    y0 = pad
    w = width - 2*pad
    h = height - 2*pad
 
    def x(i):
        return x0 + (i * w / (n - 1 if n > 1 else 1))
 
    def y(v):
        return y0 + (h - (v - vmin) * h / (vmax - vmin))
 
    pts = " ".join([f"{x(i):.2f},{y(values[i]):.2f}" for i in range(n)])
 
    # Achsen + Linie + Min/Max Text
    svg = []
    svg.append(f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}">')
    svg.append(f'<rect x="0" y="0" width="{width}" height="{height}" fill="white" stroke="black" />')
    svg.append(f'<line x1="{pad}" y1="{height-pad}" x2="{width-pad}" y2="{height-pad}" stroke="black" />')
    svg.append(f'<line x1="{pad}" y1="{pad}" x2="{pad}" y2="{height-pad}" stroke="black" />')
    svg.append(f'<polyline fill="none" stroke="black" stroke-width="2" points="{pts}" />')
    svg.append(f'<text x="{pad}" y="{pad-8}" font-size="12">max: {vmax:.2f} °C</text>')
    svg.append(f'<text x="{pad}" y="{height-8}" font-size="12">min: {vmin:.2f} °C</text>')
    svg.append('</svg>')
    return "".join(svg)
 
 
@app.get("/history/plot")
def history_plot():
    labels, values = hardware.history_load(max_lines=500)
    svg = _svg_plot(labels, values)
    return Response(content=svg, media_type="image/svg+xml")

In der Datei /etc/apache2/sites-available/000-default.conf innerhalb von <VirtualHost *:80> ergänzen:

ProxyPreserveHost On
 
ProxyPass        /led      http://127.0.0.1:8000/led
ProxyPassReverse /led      http://127.0.0.1:8000/led
 
ProxyPass        /temp     http://127.0.0.1:8000/temp
ProxyPassReverse /temp     http://127.0.0.1:8000/temp
 
ProxyPass        /history  http://127.0.0.1:8000/history
ProxyPassReverse /history  http://127.0.0.1:8000/history
 
ProxyPass        /history/  http://127.0.0.1:8000/history/
ProxyPassReverse /history/  http://127.0.0.1:8000/history/

sudo a2enmod proxy
sudo a2enmod proxy_http
sudo systemctl restart apache2

/etc/systemd/system/course_web.service

[Unit]
Description=Python Web FastAPI
After=network-online.target
Wants=network-online.target
 
[Service]
User=pi
WorkingDirectory=/home/pi/devel/projects/course_web/src
ExecStart=/home/pi/devel/projects/course_env/bin/uvicorn app:app --host 127.0.0.1 --port 8000
Restart=always
 
[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable course_web
sudo systemctl start course_web
sudo systemctl status course_web

tail -n 20 /home/pi/devel/projects/course_web/data/temperature.txt