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

Das Projekt DS18B20 digitaler Temperatursensor

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()