• Voraussetzungen
  • Hardware
    • 1-Wire aktivieren
  • Environment
  • Programmstreukur anlegen
• Software
  • Quellcode
  • Webseite

Das Projekt DS18B20 digitaler Temperatursensor muss zuvor vollständig durchgeführt und erfolgreich getestet worden sein. Da hier dieselbe Hardware verwendet wird, sind damit alle Voraussetzungen bereits erfüllt.

Jetzt die ENV aktivieren (falls noch nicht geschehen) und dann erst installieren:

pip install matplotlib

(course_env) pi@raspi88:~/devel/projects/course_temp_curve $ tree
.
├── data
│   ├── temperature.png       # wird automatisch erstellt
│   └── temperature.txt       # wird automatisch erstellt
└── src
    ├── core
    │   ├── hardware.py       # aus dem vorherigen Projekt! 
    │   ├── __init__.py
    │   └── temperature_logger.py  # Program
    └── temp_curve.py

hardware.py

import glob
import time
 
 
# -----------------------------
# API-Funktionen ds18b20
# -----------------------------
SENSOR_TIMEOUT = 1  # Sekunden
 
def is_sensor():
    """Prüft, ob mindestens ein DS18B20 Sensor angeschlossen ist"""
    sensors = glob.glob("/sys/bus/w1/devices/28-*")
    return len(sensors) > 0
 
def get_sensor():
    """Gibt den ersten DS18B20 Sensor zurück oder None"""
    if not is_sensor():
        return None
    sensors = glob.glob("/sys/bus/w1/devices/28-*")
    return sensors[0] + "/w1_slave"
 
def get_temperature():
    """Liest Temperatur vom ersten Sensor aus, None bei Fehler oder Timeout"""
    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

temperature_logger.py

import time
import threading
from datetime import datetime
from pathlib import Path
 
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
 
from core.hardware import get_temperature
 
 
# MODE = "local" oder "web"
MODE = "local"
 
if MODE == "local":
    DATA_DIR = Path("~/devel/projects/course_temp_curve/data").expanduser()
elif MODE == "web":
    DATA_DIR = Path("/var/www/html/temp_curve").expanduser()
else:
    raise ValueError("MODE muss 'local' oder 'web' sein")
 
 
DATA_DIR.mkdir(parents=True, exist_ok=True)
DATA_FILE = DATA_DIR / "temperature.txt"
PLOT_FILE = DATA_DIR / "temperature.png"
 
INTERVAL = 5  # Sekunden (Auflösung / Messintervall)
 
running = False
thread_started = False
lock = threading.Lock()
 
 
def set_interval(seconds: int):
    global INTERVAL
    if seconds < 1:
        seconds = 1
    INTERVAL = seconds
 
 
def generate_plot():
    times = []
    temps = []
 
    if DATA_FILE.exists():
        with open(DATA_FILE, "r") as f:
            for line in f:
                parts = line.strip().split()
                if len(parts) == 2:
                    times.append(parts[0])
                    temps.append(float(parts[1]))
 
    plt.figure()
    plt.title("Temperaturverlauf")
    plt.xlabel("Zeit")
    plt.ylabel("Temperatur (°C)")
 
    if temps:
        plt.plot(times, temps)
        plt.xticks(rotation=45)
    else:
        plt.text(
            0.5,
            0.5,
            "Keine Messdaten",
            ha="center",
            va="center",
            transform=plt.gca().transAxes,
        )
 
    plt.tight_layout()
    plt.savefig(PLOT_FILE)
    plt.close()
 
 
def worker():
    global running
 
    while True:
        with lock:
            active = running
            interval = INTERVAL
 
        if active:
            temp = get_temperature()
            if temp is not None:
                now = datetime.now().strftime("%H:%M:%S")
 
                with open(DATA_FILE, "a") as f:
                    f.write(f"{now} {temp}\n")
 
                generate_plot()
 
            time.sleep(interval)
        else:
            time.sleep(0.2)
 
 
def ensure_thread():
    global thread_started
    if not thread_started:
        t = threading.Thread(target=worker, daemon=True)
        t.start()
        thread_started = True
 
 
def start():
    ensure_thread()
    global running
    with lock:
        running = True
 
 
def stop():
    global running
    with lock:
        running = False
 
 
def reset():
    stop()
    DATA_FILE.write_text("")
    generate_plot()

temp_curve.py

#!/usr/bin/env python3
from core import temperature_logger as temp
 
PROMPT     = "> "
SHORT_HINT = "h = Hilfe"
VERSION    = "v1.0.0"
 
def print_short_banner():
    print("Temperatur-Logger " + VERSION)
    print("Mit 'h' Hilfe anzeigen.\n")
 
def print_help():
    print("""
VERFÜGBARE BEFEHLE
------------------
start | s                 Logger starten
stop  | p                 Logger stoppen
reset | r                 Messwerte zurücksetzen
interval <sek> | i <sek>  Messintervall setzen
status | t                Status anzeigen
help   | h                Diese Hilfe anzeigen
quit   | q                Programm beenden
""")
 
def handle_interval(cmd: str):
    parts = cmd.split()
    if len(parts) == 2 and parts[1].isdigit():
        temp.set_interval(int(parts[1]))
        print("OK")
    else:
        print("Verwendung: interval <sek>")
 
def handle_status():
    state = "RUNNING" if temp.running else "STOPPED"
    print(f"STATUS   : {state}")
    print(f"INTERVALL: {temp.INTERVAL}")
 
def main():
    print_short_banner()
 
    while True:
        try:
            cmd = input(PROMPT).strip().lower()
        except (EOFError, KeyboardInterrupt):
            print()
            temp.stop()
            break
 
        if not cmd:
            continue
 
        if cmd in ("start", "s"):
            temp.start()
 
        elif cmd in ("stop", "p"):
            temp.stop()
 
        elif cmd in ("reset", "r"):
            temp.reset()
 
        elif cmd.startswith("interval") or cmd.startswith("i "):
            handle_interval(cmd)
 
        elif cmd in ("status", "t"):
            handle_status()
 
        elif cmd in ("help", "h"):
            print_help()
 
        elif cmd in ("quit", "q"):
            temp.stop()
            break
 
        else:
            print(f"Unbekannter Befehl ({SHORT_HINT})")
 
if __name__ == "__main__":
    main()