air-Q-pro Messdaten am LCD1602 Display visualisieren
Quellcode
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Titel: LuftqualitĂ€t live: air-Q pro Daten via REST abrufen, entschlĂŒsseln und weiterverarbeiten
Autor: Stefan Draeger
Blogartikel: https://draeger-it.blog/luftqualitaet-live-air-q-pro-daten-via-rest-abrufen-und-anzeigen/
Beschreibung:
Dieses Skript lĂ€dt Sensordaten vom air-Q pro, entschlĂŒsselt sie (AES-256-CBC) und liefert ein
klar strukturiertes Objekt (AirQMessung) zurĂŒck. Die Anzeige/Weiterverarbeitung (CSV, LCD)
ist bewusst entkoppelt, damit du das Objekt flexibel verwenden kannst.
"""
from __future__ import annotations
import argparse
import base64
import csv
import http.client
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional
import urllib.request
from typing import Tuple
from Crypto.Cipher import AES # pycryptodome
# --------------------------------------------------------------------
# Konfiguration (Defaults fĂŒr CLI; produktiv per .env/Umgebungsvariablen)
# --------------------------------------------------------------------
STANDARD_AIRQ_IP = "192.168.178.108"
STANDARD_AIRQ_PASSWORT = "draeger1980"
AIRQ_HTTP_PORT = 80
AIRQ_HTTP_PFAD = "/data"
AIRQ_HTTP_TIMEOUT_S = 8
AIRQ_HTTP_RETRIES = 3
AES_SCHLUESSEL_LAENGE = 32
AES_BLOCKGROESSE = 16
# --------------------------------------------------------------------
# DomÀnenmodell
# --------------------------------------------------------------------
@dataclass(frozen=True)
class AirQMessung:
"""
ReprÀsentiert eine Messung des air-Q.
- zeitpunkt_local / zeitpunkt_utc: Zeitpunkt (aus 'timestamp' in ms)
- device_id, uptime_s: GerÀtemetadaten
- rohwerte: Original-JSON (entschlĂŒsselt)
- werte: deutsch beschriftete, einheitenklare Basiswerte (fĂŒr CSV/LCD)
- status: Warmup-/Hinweistexte je Sensor
"""
zeitpunkt_local: datetime
zeitpunkt_utc: datetime
device_id: Optional
uptime_s: Optional
rohwerte: Dict
werte: Dict
status: Dict
# --------------------------------------------------------------------
# Datenabruf & EntschlĂŒsselung
# --------------------------------------------------------------------
def _entferne_pkcs7_padding(daten: bytes) -> bytes:
"""Entfernt PKCS#7-Padding sicher im Byte-Kontext."""
return daten]
def _erzeuge_aes_schluessel(passwort: str) -> bytes:
"""Erzeugt einen 32-Byte-AES-SchlĂŒssel aus dem Passwort (auffĂŒllen/abschneiden)."""
key = passwort.encode("utf-8")
return key + b"0" * (AES_SCHLUESSEL_LAENGE - len(key)) if len(key) Dict:
"""EntschlĂŒsselt den base64-kodierten 'content' des air-Q und liefert das JSON als Dict."""
verschluesselt = base64.b64decode(base64_text)
iv, payload = verschluesselt, verschluesselt
cipher = AES.new(key=_erzeuge_aes_schluessel(passwort), mode=AES.MODE_CBC, IV=iv)
entschluesselt = cipher.decrypt(payload)
json_text = _entferne_pkcs7_padding(entschluesselt).decode("utf-8")
return json.loads(json_text)
def _hole_rohdaten(ip_adresse: str) -> Dict:
"""
Holt das Ă€uĂere JSON (mit 'content') fest ĂŒber http:///data (Port 80).
Baut Retries und aussagekrÀftige Fehler ein.
"""
last_err: Optional = None
for attempt in range(1, AIRQ_HTTP_RETRIES + 1):
conn = http.client.HTTPConnection(ip_adresse, port=AIRQ_HTTP_PORT, timeout=AIRQ_HTTP_TIMEOUT_S)
try:
conn.request("GET", AIRQ_HTTP_PFAD)
resp = conn.getresponse()
body = resp.read()
if resp.status == 200:
return json.loads(body)
raise RuntimeError(f"HTTP {resp.status} {resp.reason} auf {AIRQ_HTTP_PFAD}")
except (socket.timeout, TimeoutError, OSError, RuntimeError, json.JSONDecodeError) as e:
last_err = e
# kleiner Backoff zwischen den Versuchen
time.sleep(0.5 * attempt)
finally:
try:
conn.close()
except Exception:
pass
raise ConnectionError(
f"air-Q unter http://{ip_adresse}{AIRQ_HTTP_PFAD} nicht erreichbar (Port 80). "
f"Letzter Fehler: {last_err}"
)
def hole_airq_objekt(ip_adresse: str, passwort: str) -> Dict:
"""
EINSTIEGSPUNKT fĂŒr Datenzugriff:
- Holt die Rohantwort
- EntschlĂŒsselt 'content'
- Gibt das entschlĂŒsselte JSON (als Dict) zurĂŒck
"""
rohdaten = _hole_rohdaten(ip_adresse)
return _entschluessle_content(rohdaten, passwort)
# --------------------------------------------------------------------
# Normierung & Beschriftung
# --------------------------------------------------------------------
# Mapping: JSON-Key -> (deutscher Name, Einheit)
MAPPING: Dict]] = {
"pm1": ("PM1.0", "”g/m³"),
"pm2_5": ("PM2.5", "”g/m³"),
"pm10": ("PM10", "”g/m³"),
"TypPS": ("Ă PartikelgröĂe", "”m"),
"dewpt": ("Taupunkt", "°C"),
"temperature": ("Temperatur", "°C"),
"humidity": ("Relative Luftfeuchte", "%"),
"humidity_abs":("Absolute Luftfeuchte", "g/mÂł"),
"pressure": ("Luftdruck", "hPa"),
"sound": ("LĂ€rmpegel", "dB(A)"),
"sound_max": ("LĂ€rmpegel max (2 min)", "dB(A)"),
"tvoc": ("TVOC (flĂŒchtige organische Verbindungen)", "ppb"),
"co": ("CO (Kohlenmonoxid)", "ppm"), # korrigiert
"co2": ("COâ (Kohlendioxid)", "ppm"), # korrigiert
"health": ("Gesundheitsindex", None),
"performance": ("Leistungsindex", None),
"dHdt": ("Î absolute LF", "mg/mÂł/s"),
"dCO2dt": ("Î COâ", "ppb/s"),
"uptime": ("Betriebszeit", "s"),
"measuretime": ("Messdauer letzter Durchlauf", "ms"),
"DeviceID": ("GerÀte-ID", None),
"timestamp": ("Zeitstempel (ms)", "ms"),
}
def _wert_aus_array_oder_skalar(v: Any) -> Tuple]:
"""
air-Q liefert hÀufig .
Gibt (basiswert, unsicherheit) zurĂŒck.
"""
if isinstance(v, (list, tuple)) and len(v) >= 1:
basis = v
unsicherheit = v if len(v) > 1 else None
return basis, unsicherheit
return v, None
def _normalisiere_status(raw) -> Dict:
"""
Wandelt das 'Status'-Feld in ein Dict um.
- Dict: SchlĂŒssel/Werte als Strings
- String/Number/sonstiges: {"status": ""}
- Liste/Tuple: {"status_0": "...", "status_1": "...", ...}
- None: {}
"""
if raw is None:
return {}
if isinstance(raw, dict):
return {str(k): str(v) for k, v in raw.items()}
if isinstance(raw, (list, tuple)):
return {f"status_{i}": str(v) for i, v in enumerate(raw)}
# alles andere (z. B. "OK")
return {"status": str(raw)}
def normiere_messung(entschluesselt: Dict) -> AirQMessung:
"""
Wandelt das entschlĂŒsselte Dict in eine AirQMessung:
- Zeit aus 'timestamp' (ms) â UTC + lokale Zeit
- deutsch beschriftete, einheitenklare Basiswerte
- Statusmeldungen (Warmup etc.) unter 'Status'
"""
# Zeitstempel (ms) â UTC/local
ts_ms = entschluesselt.get("timestamp")
if isinstance(ts_ms, (int, float)):
ts_utc = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
else:
ts_utc = datetime.now(tz=timezone.utc)
ts_local = ts_utc.astimezone()
# Status-Hinweise
status_norm = _normalisiere_status(entschluesselt.get("Status"))
# Beschriftete Werte
werte: Dict = {}
for key, (name_de, einheit) in MAPPING.items():
if key in entschluesselt:
basiswert, _unsicherheit = _wert_aus_array_oder_skalar(entschluesselt)
label = f"{name_de} ({einheit})" if einheit else name_de
werte = basiswert
return AirQMessung(
zeitpunkt_local=ts_local,
zeitpunkt_utc=ts_utc,
device_id=str(entschluesselt.get("DeviceID")) if entschluesselt.get("DeviceID") else None,
uptime_s=int(entschluesselt.get("uptime")) if entschluesselt.get("uptime") is not None else None,
rohwerte=entschluesselt,
werte=werte,
status=status_norm,
)
# --------------------------------------------------------------------
# Weiterverarbeitung (CSV / LCD-Stub) â optional
# --------------------------------------------------------------------
def schreibe_csv_zeile(datei: Path, messung: AirQMessung, erstelle_header: bool = True) -> None:
"""
Schreibt eine Zeile in eine CSV-Datei (Delimiter ';').
- Erstellt optional beim ersten Mal einen Header.
- 'zeitpunkt' wird als ISO-String vorangestellt.
"""
datei.parent.mkdir(parents=True, exist_ok=True)
spalten = + sorted(messung.werte.keys())
schreibe_header = (not datei.exists()) and erstelle_header
with datei.open("a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=spalten, delimiter=";")
if schreibe_header:
writer.writeheader()
zeile = {"zeitpunkt": messung.zeitpunkt_local.isoformat(timespec="seconds")}
zeile.update({k: messung.werte.get(k, "") for k in spalten if k != "zeitpunkt"})
writer.writerow(zeile)
# ------------------------------------------------------------
# 1) Anzeigezeilen aus Messung erzeugen
# ------------------------------------------------------------
def formatiere_lcd_zeilen(messung: AirQMessung) -> Tuple:
"""
Baut zwei Textzeilen fĂŒr ein 16x2 LCD.
Da dein ESP 'scroll' unterstĂŒtzt, dĂŒrfen die Zeilen lĂ€nger sein;
wir halten sie trotzdem kompakt & gut lesbar.
"""
# Werte robust lesen (falls mal etwas fehlt)
co2 = messung.werte.get("COâ (Kohlendioxid) (ppm)")
tvoc = messung.werte.get("TVOC (flĂŒchtige organische Verbindungen) (ppb)")
temp = messung.werte.get("Temperatur (°C)")
rh = messung.werte.get("Relative Luftfeuchte (%)")
# Schonend runden/formatieren
def fmt(v, nd=0):
if v is None or v == "-":
return "-"
try:
return f"{round(float(v), nd)}"
except Exception:
return str(v)
co2_s = fmt(co2, 0)
tvoc_s = fmt(tvoc, 0)
temp_s = fmt(temp, 1)
rh_s = fmt(rh, 1)
# Zeile 1 & 2 kompakt â darf lĂ€nger sein (scroll)
zeile1 = f"CO2:{co2_s}ppm TVOC:{tvoc_s}ppb"
zeile2 = f"T:{temp_s}C rH:{rh_s}%"
return zeile1, zeile2
# ------------------------------------------------------------
# 2) HTTP-POST an den ESP-Endpoint senden
# ------------------------------------------------------------
def sende_an_lcd_http(lcd_url: str, zeile1: str, zeile2: str,
modus: str = "scroll", anzeigedauer: int = 15,
timeout_s: int = 5) -> None:
"""
Sendet die gewĂŒnschten LCD-Zeilen an den ESP (JSON-POST).
Erwartet einen Endpoint wie: http://192.168.178.96/anzeige
"""
payload = {
"zeile1": zeile1,
"zeile2": zeile2,
"modus": modus, # z.B. "scroll", "static", ...
"anzeigedauer": anzeigedauer # Sekunden
}
daten = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
url=lcd_url,
data=daten,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout_s) as resp:
# Antwort bei Bedarf auswerten:
_ = resp.read()
except HTTPError as e:
# Hier gerne auf logging umstellen
print(f"HTTP-Fehler beim Senden an LCD ({e.code}): {e.reason}")
except URLError as e:
print(f"Verbindungsfehler zum LCD: {e.reason}")
except Exception as e:
print(f"Unerwarteter Fehler beim LCD-POST: {e}")
# ------------------------------------------------------------
# 3) Ăffentliche Anzeige-Funktion (ersetzt den Stub)
# ------------------------------------------------------------
def zeige_auf_lcd_http(messung: AirQMessung, lcd_url: str,
modus: str = "scroll", anzeigedauer: int = 15) -> None:
"""
Erzeugt zwei Anzeigezeilen aus der Messung und sendet sie per HTTP an den ESP-LCD-Service.
Read the full article
