informatics4kids.de

Benutzer-Werkzeuge

Webseiten-Werkzeuge

Zuletzt angesehen:
projekt:python_fastapi1

Dies ist eine alte Version des Dokuments!

OK. MINIMAL-ERWEITERUNG OHNE JAVASCRIPT/CHART.JS: - Verlauf wird als TEXTDATEI gespeichert (course_web/data/temperature.txt) - Aufzeichnung läuft im Hintergrund (Thread) im laufenden FastAPI-Prozess (also auch ohne geöffneten Browser) - Anzeige im Browser: einfaches HTML + vorformatierter Text (letzte N Zeilen), plus Start/Stop/Reset/Intervall - “API” bleibt in hardware.py: dort kommen NUR einfache History-Methoden dazu - app.py bekommt nur neue Routen + startet den Collector-Thread (keine “komische” Logik im HTML)

1) PROJEKTSTRUKTUR (wie bei dir) course_web/

└── src/ 

  ├── app.py
  ├── core/
  │   ├── __init__.py
  │   └── hardware.py
  └── html/
      ├── led.html
      ├── temp.html
      └── history.html

2) DATEI: /home/pi/devel/projects/course_web/src/core/hardware.py ERGÄNZEN (unten dranhängen, bestehender Code bleibt, KEIN Umbau der vorhandenen Funktionen)

- Ziel: simple Methoden:

  1. history_append(temp, file)
  2. history_read(file, max_lines)
  3. history_reset(file)
  4. history_set_recording(True/False)
  5. history_set_interval_minutes(x)
  6. history_start_collector(get_temperature_func)
  7. history_status()
/home/pi/devel/projects/course_web/src/core/hardware.py
# =============================
# Temperaturverlauf (History)
# =============================
import os
import threading
from datetime import datetime
 
HISTORY_FILE_DEFAULT = "/home/pi/devel/projects/course_web/data/temperature.txt"
 
_history_lock = threading.Lock()
_history_recording = False
_history_interval_seconds = 60
_history_thread_started = False
 
 
def history_status():
    with _history_lock:
        return {
            "recording": _history_recording,
            "interval_seconds": _history_interval_seconds,
        }
 
 
def history_set_recording(on: bool):
    global _history_recording
    with _history_lock:
        _history_recording = bool(on)
 
 
def history_set_interval_minutes(minutes: int):
    global _history_interval_seconds
    if minutes < 1:
        raise ValueError("minutes must be >= 1")
    with _history_lock:
        _history_interval_seconds = int(minutes) * 60
 
 
def history_reset(file_path: str = HISTORY_FILE_DEFAULT):
    os.makedirs(os.path.dirname(file_path), exist_ok=True)
    with open(file_path, "w") as f:
        f.write("")
 
 
def history_append(temp_c: float, file_path: str = HISTORY_FILE_DEFAULT):
    os.makedirs(os.path.dirname(file_path), exist_ok=True)
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(file_path, "a") as f:
        f.write(f"{ts};{temp_c:.3f}\n")
 
 
def history_read(file_path: str = HISTORY_FILE_DEFAULT, max_lines: int = 200):
    # gibt Liste von Zeilen (Strings) zurück, ohne Parsing-Zauber
    try:
        with open(file_path, "r") as f:
            lines = [ln.rstrip("\n") for ln in f.readlines() if ln.strip()]
    except FileNotFoundError:
        return []
 
    if max_lines is not None and max_lines > 0:
        return lines[-max_lines:]
    return lines
 
 
def history_start_collector(get_temp_func, file_path: str = HISTORY_FILE_DEFAULT):
    # get_temp_func: callable -> float|None
    global _history_thread_started
 
    def _loop():
        while True:
            with _history_lock:
                rec = _history_recording
                sleep_s = _history_interval_seconds
 
            if rec:
                try:
                    t = get_temp_func()
                    if t is not None:
                        history_append(float(t), file_path=file_path)
                except Exception:
                    # bewusst still, damit Anfänger nicht “zufällig” alles crashen
                    # (wenn gewünscht: print(...) oder logging später)
                    pass
 
            time.sleep(sleep_s)
 
    with _history_lock:
        if _history_thread_started:
            return
        th = threading.Thread(target=_loop, daemon=True)
        th.start()
        _history_thread_started = True

WICHTIG: Oben in hardware.py ist bereits „import time“ vorhanden. Das wird im Collector gebraucht.

3) DATEI: /home/pi/devel/projects/course_web/src/html/history.html OHNE JAVASCRIPT, NUR HTML (Status + Buttons + Textausgabe)

/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>
 
<p><b>Status:</b> {{REC}} | <b>Intervall:</b> {{INT}} s</p>
 
<h2>Steuerung</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>Datei-Auszug (letzte {{N}} Zeilen)</h2>
<pre>{{LINES}}</pre>
 
<br><br>
<a href="/">Zurück</a>
 
</body>
</html>

4) DATEI: /home/pi/devel/projects/course_web/src/app.py ERWEITERN (bestehende Routen bleiben; nur History dazu)

- Startup: hardware.init() bleibt - zusätzlich: hardware.history_start_collector(hardware.get_temperature) - neue Routen:

  1. /history
  2. /history/start
  3. /history/stop
  4. /history/reset
  5. /history/interval/{minutes}
/home/pi/devel/projects/course_web/src/app.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from core import hardware
 
app = FastAPI()
 
 
@app.on_event("startup")
def startup():
    hardware.init()
    hardware.history_start_collector(hardware.get_temperature)
 
 
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
        })
    )
 
 
# =============================
# Temperaturverlauf (History)
# =============================
 
@app.get("/history", response_class=HTMLResponse)
def history_page():
    st = hardware.history_status()
    lines = hardware.history_read(max_lines=200)
 
    rec_txt = "läuft" if st["recording"] else "stopp"
    int_s = st["interval_seconds"]
 
    return HTMLResponse(
        load_template("history.html", {
            "{{REC}}": rec_txt,
            "{{INT}}": int_s,
            "{{N}}": 200,
            "{{LINES}}": "\n".join(lines) if lines else "(keine Daten - Start drücken)",
        })
    )
 
 
@app.get("/history/start")
def history_start():
    hardware.history_set_recording(True)
    return RedirectResponse(url="/history", status_code=303)
 
 
@app.get("/history/stop")
def history_stop():
    hardware.history_set_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):
    try:
        hardware.history_set_interval_minutes(minutes)
    except ValueError:
        raise HTTPException(status_code=400)
    return RedirectResponse(url="/history", status_code=303)

HINWEIS: Das ist “ohne Umbau”: nur ergänzt.

5) APACHE STARTSEITE index.html NUR LINK ERGÄNZEN

- in /var/www/html/index.html ergänzen: 

<li><a href="/history">Temperaturverlauf</a></li>

6) APACHE PROXY (WICHTIG: PREFIX /history/ )

In /etc/apache2/sites-available/000-default.conf innerhalb <VirtualHost *:80>: 

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

(Die 2 letzten Zeilen sind für /history ohne Slash; der Prefix ist für /history/start usw.)

7) SYSTEMD bleibt wie gehabt (course_web.service)

8) TEST (ohne “Sonderweg”)

1) Browser: http:<PI-IP>/history 2) Start klicken 3) 1–2 Minuten warten (oder Intervall 1) 4) Seite neu laden: Zeilen erscheinen 5) Stop → es kommen keine neuen Zeilen 6) Reset → Datei leer CLI-Check: <code bash> tail -n 20 /home/pi/devel/projects/course_web/data/temperature.txt </code> ================================================================================ DAS IST DIE “SIMPLE METHODEN IN hardware.py”-VARIANTE: - kein chart.js - kein javascript - keine daten “nur wenn browser offen” - Start/Stop/Reset/Intervall drin - Anzeige grafisch? NEIN (bewusst), weil sonst JS/PNG/Libs nötig wären Wenn “grafisch” zwingend ist, dann gibt’s genau 2 saubere Anfänger-Optionen: A) serverseitig PNG (matplotlib) + <img src=„/history/plot.png“> (ohne JS) B) JS im Browser (Chart.js) (wie du NICHT willst)

projekt/python_fastapi1.1771674820.txt.gz · Zuletzt geändert: von torsten.roehl