informatics4kids.de

Benutzer-Werkzeuge

Webseiten-Werkzeuge

Zuletzt angesehen: • python_fastapi
projekt:python_fastapi

Dies ist eine alte Version des Dokuments!

Inhaltsverzeichnis

Python FASTAPI

☚ 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.

Überblick

  • Voraussetzungen
  • Software
  • Konfiguration

Details

Voraussetzungen

ENV

Aktivierung der Python-Environment: course_env

Alle weiteren Schritte erfolgen mit der aktivierten Python-Umgebung. 

source ~/devel/projects/course_env/bin/activate  #ndre nichts ohne nachfrage

Anschließend werden FastAPI und Uvicorn installiert: 

pip install fastapi uvicorn
FastAPI / Uvicorn
  • FastAPI
    • stellt das Web-Framework bereit, mit dem die Webseiten und Routen programmiert werden.
  • Uvicorn
    • startet die Anwendung und sorgt dafür, dass sie im Browser erreichbar ist.

Projektstruktur

cd ~/devel/projects/ 

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

Temperaturverlauf: Daten anlegen

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

mkdir -p /home/pi/devel/projects/course_web/data
touch /home/pi/devel/projects/course_web/data/temperature.txt
Format der Datei temperature.txt

Jede Zeile enthält Zeitstempel und Temperatur, getrennt durch ein Semikolon:

YYYY-MM-DD HH:MM:SS;TEMPERATUR

Beispiel: 2026-02-21 12:00:00;21.437

Apache2 Startseite

Im Verzeichnis des Apache2-Webservers (/var/www/html) wird eine index.html bereitgestellt. Beim Aufruf der IP-Adresse des Raspberry Pi im Browser wird diese Startseite geladen, über die anschließend das gewünschte Projekt ausgewählt werden kann.

Tip

Bevor die neue index.html im Verzeichnis /var/www/html erstellt wird, sollte die vorhandene Apache-Standardseite index.html gesichert werden. 

sudo mv /var/www/html/index.html /var/www/html/index_course_backup.html
/var/www/html/index.html
<!DOCTYPE html>
<html>
<head>
    <title>Projekt Auswahl</title>
</head>
<body>
 
<h1>Projekt Auswahl</h1>
 
<ul>
    <li><a href="/wiki">DokuWiki</a></li>
    <li><a href="/led">LED Ampel</a></li>
    <li><a href="/temp">Temperatur</a></li>
    <li><a href="/history">Temperaturverlauf</a></li>
</ul>
 
</body>
</html>

Hardware

Die Hardware, also die LED-Ampel und der Temperatursensor, wurden in den vorausgegangenen Projekten ausführlich behandelt und werden exakt so verwendet wie dort beschrieben.

Software

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.

Temperaturverlauf (History)

Der Temperaturverlauf wird nicht nur beim Aufruf im Browser erzeugt, sondern durch einen Hintergrund-Thread im laufenden FastAPI-Programm. Damit können Messwerte automatisch im Intervall gesammelt, gestoppt und zurückgesetzt werden.

API

/home/pi/devel/projects/course_web/src/core/hardware.py 
import RPi.GPIO as GPIO
import glob
import time
 
# -----------------------------
# 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

HTML

LED

/home/pi/devel/projects/course_web/src/html/led.html
<!DOCTYPE html>
<html>
<head>
    <title>LED Ampel</title>
</head>
<body>
 
<h1>LED Ampel</h1>
 
<h2>Rot</h2>
<p>Status: {{R}}</p>
<a href="/led/r/1">ON</a>
<a href="/led/r/0">OFF</a>
 
<h2>Gelb</h2>
<p>Status: {{Y}}</p>
<a href="/led/y/1">ON</a>
<a href="/led/y/0">OFF</a>
 
<h2>Grün</h2>
<p>Status: {{G}}</p>
<a href="/led/g/1">ON</a>
<a href="/led/g/0">OFF</a>
 
<br><br>
<a href="/">Zurück</a>
 
</body>
</html>

Temperature

/home/pi/devel/projects/course_web/src/html/temp.html
<!DOCTYPE html>
<html>
<head>
    <title>Temperatur</title>
</head>
<body>
 
<h1>Temperatur</h1>
 
<p>Temperatur: {{T}} °C</p>
 
<br>
<a href="/">Zurück</a>
 
</body>
</html>

History

/home/pi/devel/projects/course_web/src/html/history.html
<!DOCTYPE html>
<html>
<head>
    <title>Temperaturverlauf</title>
    <meta charset="utf-8" />
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
</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</h2>
<canvas id="chart" width="900" height="300"></canvas>
 
<script>
let chart = null;
 
async function loadData() {
    const res = await fetch("/history/data");
    return await res.json();
}
 
async function plot() {
    const data = await loadData();
 
    const ctx = document.getElementById("chart").getContext("2d");
    if (chart) chart.destroy();
 
    chart = new Chart(ctx, {
        type: "line",
        data: {
            labels: data.labels,
            datasets: [{
                label: "Temperatur (°C)",
                data: data.values
            }]
        },
        options: {
            animation: false
        }
    });
}
 
plot();
setInterval(plot, 5000);
</script>
 
<br><br>
<a href="/">Zurück</a>
 
</body>
</html>

FASTAPI APP

/home/pi/devel/projects/course_web/src/app.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from core import hardware
 
import threading
import time
from datetime import datetime
 
app = FastAPI()
 
LOG_FILE = "/home/pi/devel/projects/course_web/data/temperature.txt"
 
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:
                ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                try:
                    with open(LOG_FILE, "a") as f:
                        f.write(f"{ts};{t:.3f}\n")
                except Exception:
                    pass
 
        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():
    open(LOG_FILE, "w").close()
    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)
 
 
@app.get("/history/data")
def history_data():
    labels = []
    values = []
 
    try:
        with open(LOG_FILE, "r") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    ts, val = line.split(";")
                    labels.append(ts)          # kompletter Zeitstempel fürs Diagramm
                    values.append(float(val))
                except Exception:
                    continue
    except FileNotFoundError:
        pass
 
    return JSONResponse({"labels": labels, "values": values})

Konfiguration

In diesem Abschnitt wird die Einbindung der FastAPI-Anwendung in den Apache-Webserver sowie die Einrichtung als systemd-Service beschrieben, damit die Anwendung automatisch startet und im lokalen Netzwerk erreichbar ist.

Apache Proxy

Konfiguration

In der Datei /etc/apache2/sites-available/000-default.conf, innerhalb von <VirtualHost *:80> folgendes einfügen:

/etc/apache2/sites-available/000-default.conf
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
Falls /history/start oder /history/data nicht sauber funktionieren, zusätzlich das Prefix weiterleiten:
ProxyPass        /history/  http://127.0.0.1:8000/history/
ProxyPassReverse /history/  http://127.0.0.1:8000/history/

Aktivieren

sudo a2enmod proxy
sudo a2enmod proxy_http
sudo systemctl restart apache2

Systemd

Die Service-Datei muss unter /etc/systemd/system/course_web.service neu erstellt und anschließend bei systemd registriert werden.

Service

/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

Registrierung

sudo systemctl daemon-reload
sudo systemctl enable course_web
sudo systemctl start course_web       # starten
sudo systemctl status course_web      # prüfen!!!

Test Temperaturverlauf

tail -n 20 /home/pi/devel/projects/course_web/data/temperature.txt
projekt/python_fastapi.1771673014.txt.gz · Zuletzt geändert: von torsten.roehl