Alle 3 Charts in einem kombinierten Bild zusammengefasst

Ein einziges PNG mit drei gestapelten Subplots und Rahmen um die
gesamte Figure. Mail enthält nur noch einen Anhang.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
rxf
2026-05-26 11:21:18 +02:00
parent 8717d0f8ad
commit e1bda88ab4
+89 -117
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Erstellt Wetterdiagramme für den letzten Kalendermonat und verschickt sie per Mail."""
"""Erstellt einen kombinierten Wetterbericht für den letzten Kalendermonat."""
import os
import smtplib
@@ -11,6 +11,7 @@ from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import matplotlib.dates as mdates
import matplotlib.patches as mpatches
import matplotlib.pyplot as plt
import requests
from dotenv import load_dotenv
@@ -43,10 +44,9 @@ plt.rcParams.update({
def last_month_range() -> tuple[datetime, datetime, str]:
"""Gibt (start, end, Monatsname) für den letzten Kalendermonat zurück."""
today = datetime.now(timezone.utc)
first_of_this = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
end = first_of_this - timedelta(seconds=1) # letzter Moment des Vormonats
end = first_of_this - timedelta(seconds=1)
start = end.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
label = f"{MONTHS_DE[start.month - 1]} {start.year}"
return start, end, label
@@ -63,109 +63,109 @@ def parse_dt(s: str) -> datetime:
return datetime.fromisoformat(s.replace("Z", "+00:00"))
def chart_temp_minmax(start: datetime, end: datetime, label: str) -> bytes:
def _data_temp_minmax(start: datetime, end: datetime):
data = fetch("/weather/daily-aggregated-range", {
"start": start.isoformat(),
"end": end.isoformat(),
"start": start.isoformat(), "end": end.isoformat(),
})
dates = [parse_dt(d["datetime"]) for d in data]
t_min = [d["min_temperature"] for d in data]
t_max = [d["max_temperature"] for d in data]
fig, ax = plt.subplots(figsize=(12, 5))
ax.plot(dates, t_max, color="#e05c2a", linewidth=2, marker="o", markersize=4,
label="Tages-Maximum")
ax.plot(dates, t_min, color="#2a7be0", linewidth=2, marker="o", markersize=4,
label="Tages-Minimum")
ax.fill_between(dates, t_min, t_max, alpha=0.12, color="#888888")
ax.set_title(f"Temperaturverlauf (Tages-Min / Tages-Max) {label}",
fontweight="bold", pad=14)
ax.set_ylabel("Temperatur (°C)")
ax.set_xlim(start, end)
ax.xaxis.set_major_formatter(mdates.DateFormatter("%d.%m."))
ax.xaxis.set_major_locator(mdates.DayLocator(interval=3))
fig.autofmt_xdate(rotation=0, ha="center")
ax.legend(framealpha=0.8)
fig.tight_layout()
return _fig_to_bytes(fig)
return dates, t_min, t_max
def chart_temp_hourly(start: datetime, end: datetime, label: str) -> bytes:
def _data_temp_hourly(start: datetime, end: datetime):
data = fetch("/weather/hourly-aggregated-range", {
"start": start.isoformat(),
"end": end.isoformat(),
"start": start.isoformat(), "end": end.isoformat(),
})
dates = [parse_dt(d["datetime"]) for d in data]
temps = [d["temperature"] for d in data]
return dates, temps
fig, ax = plt.subplots(figsize=(12, 5))
ax.plot(dates, temps, color="#2a7be0", linewidth=1.5, label="Stundenmittel")
ax.set_title(f"Temperaturverlauf (Stundenmittel) {label}",
fontweight="bold", pad=14)
def _data_rain_daily(start: datetime, end: datetime):
data = fetch("/weather/range", {
"start": start.isoformat(), "end": end.isoformat(), "limit": 50000,
})
by_day: dict[str, list[float]] = defaultdict(list)
for d in data:
if d.get("rain") is not None:
by_day[parse_dt(d["datetime"]).strftime("%Y-%m-%d")].append(d["rain"])
dates, rain = [], []
for day_key in sorted(by_day):
vals = by_day[day_key]
daily = max(vals) - min(vals)
if daily < 0:
daily = max(vals)
dates.append(datetime.fromisoformat(day_key).replace(tzinfo=timezone.utc))
rain.append(round(daily, 1))
return dates, rain
def create_combined_chart(start: datetime, end: datetime, label: str) -> bytes:
dates_mm, t_min, t_max = _data_temp_minmax(start, end)
dates_h, temps = _data_temp_hourly(start, end)
dates_r, rain = _data_rain_daily(start, end)
fig, axes = plt.subplots(3, 1, figsize=(13, 16))
fig.suptitle(f"Wetterbericht {label}", fontsize=15, fontweight="bold", y=0.985)
# --- Chart 1: Tages-Min / Tages-Max ---
ax = axes[0]
ax.plot(dates_mm, t_max, color="#e05c2a", linewidth=2, marker="o",
markersize=4, label="Tages-Maximum")
ax.plot(dates_mm, t_min, color="#2a7be0", linewidth=2, marker="o",
markersize=4, label="Tages-Minimum")
ax.fill_between(dates_mm, t_min, t_max, alpha=0.12, color="#888888")
ax.set_title("Temperaturverlauf (Tages-Min / Tages-Max)", fontweight="bold", pad=10)
ax.set_ylabel("Temperatur (°C)")
ax.set_xlim(start, end)
ax.xaxis.set_major_formatter(mdates.DateFormatter("%d.%m."))
ax.xaxis.set_major_locator(mdates.DayLocator(interval=3))
fig.autofmt_xdate(rotation=0, ha="center")
plt.setp(ax.get_xticklabels(), rotation=0, ha="center")
ax.legend(framealpha=0.8)
fig.tight_layout()
return _fig_to_bytes(fig)
# --- Chart 2: Stundenmittel ---
ax = axes[1]
ax.plot(dates_h, temps, color="#2a7be0", linewidth=1.5, label="Stundenmittel")
ax.set_title("Temperaturverlauf (Stundenmittel)", fontweight="bold", pad=10)
ax.set_ylabel("Temperatur (°C)")
ax.set_xlim(start, end)
ax.xaxis.set_major_formatter(mdates.DateFormatter("%d.%m."))
ax.xaxis.set_major_locator(mdates.DayLocator(interval=3))
plt.setp(ax.get_xticklabels(), rotation=0, ha="center")
ax.legend(framealpha=0.8)
def chart_rain_daily(start: datetime, end: datetime, label: str) -> bytes:
# Das rain-Feld ist ein kumulativer Zähler; Tagessumme = max - min je Tag
data = fetch("/weather/range", {
"start": start.isoformat(),
"end": end.isoformat(),
"limit": 50000,
})
by_day: dict[str, list[float]] = defaultdict(list)
for d in data:
if d.get("rain") is not None:
day_key = parse_dt(d["datetime"]).strftime("%Y-%m-%d")
by_day[day_key].append(d["rain"])
dates = []
rain = []
for day_key in sorted(by_day):
vals = by_day[day_key]
daily = max(vals) - min(vals)
if daily < 0: # Zählerreset → nehme max als Tagessumme
daily = max(vals)
dates.append(datetime.fromisoformat(day_key).replace(tzinfo=timezone.utc))
rain.append(round(daily, 1))
fig, ax = plt.subplots(figsize=(12, 5))
# --- Chart 3: Niederschlag ---
ax = axes[2]
colors = ["#2a7be0" if r > 0 else "#ccddee" for r in rain]
ax.bar(dates, rain, color=colors, width=0.7, label="Niederschlag")
ax.bar(dates_r, rain, color=colors, width=0.7, label="Niederschlag")
total = sum(rain)
ax.set_title(
f"Niederschlag pro Tag {label} (Summe: {total:.1f} mm)",
fontweight="bold", pad=14,
)
ax.set_title(f"Niederschlag pro Tag (Summe: {total:.1f} mm)",
fontweight="bold", pad=10)
ax.set_ylabel("Niederschlag (mm)")
ax.set_xlim(start - timedelta(hours=12), end + timedelta(hours=12))
ax.xaxis.set_major_formatter(mdates.DateFormatter("%d.%m."))
ax.xaxis.set_major_locator(mdates.DayLocator(interval=3))
fig.autofmt_xdate(rotation=0, ha="center")
plt.setp(ax.get_xticklabels(), rotation=0, ha="center")
ax.legend(framealpha=0.8)
fig.tight_layout()
return _fig_to_bytes(fig)
# Watermark im letzten Chart
axes[2].text(0.99, 0.03, "Wetterstation der Sternwarte Welzheim",
ha="right", va="bottom", fontsize=9, color="#bbbbbb",
transform=axes[2].transAxes)
fig.tight_layout(rect=[0.015, 0.015, 0.985, 0.975])
# Rahmen um die gesamte Figure
border = mpatches.FancyBboxPatch(
(0.008, 0.008), 0.984, 0.984,
boxstyle="square,pad=0",
linewidth=2, edgecolor="#999999", facecolor="none",
transform=fig.transFigure, clip_on=False,
)
fig.add_artist(border)
def _fig_to_bytes(fig: plt.Figure) -> bytes:
ax = fig.axes[0]
ax.text(0.99, 0.03, "Wetterstation der Sternwarte Welzheim",
ha="right", va="bottom", fontsize=9, color="#bbbbbb",
transform=ax.transAxes)
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
fig.savefig(f.name, dpi=150, bbox_inches="tight")
path = f.name
@@ -176,7 +176,7 @@ def _fig_to_bytes(fig: plt.Figure) -> bytes:
return data
def send_email(charts: dict[str, bytes], label: str) -> None:
def send_email(png: bytes, label: str) -> None:
if not all([MAIL_SERVER, MAIL_USER, MAIL_PASSWORD, MAIL_FROM, MAIL_TO]):
print("E-Mail-Konfiguration unvollständig Mail wird nicht gesendet.")
return
@@ -189,34 +189,16 @@ def send_email(charts: dict[str, bytes], label: str) -> None:
body = MIMEText(
f"<html><body>"
f"<h2>Wetterbericht {label}</h2>"
f"<p>Anbei die Wetterdiagramme für {label}.</p>"
+ "".join(
f'<p><strong>{name}</strong><br>'
f'<img src="cid:{cid}" style="max-width:100%;"/></p>'
for (name, _), cid in zip(
charts.items(),
["chart1@weather", "chart2@weather", "chart3@weather"],
)
)
+ "</body></html>",
f'<img src="cid:chart@weather" style="max-width:100%;"/>'
f"</body></html>",
"html",
)
msg.attach(body)
filenames = [
"temperatur_minmax.png",
"temperatur_stundenmittel.png",
"niederschlag_taglich.png",
]
for (name, png), cid, fname in zip(
charts.items(),
["chart1@weather", "chart2@weather", "chart3@weather"],
filenames,
):
img = MIMEImage(png, name=fname)
img.add_header("Content-ID", f"<{cid}>")
img.add_header("Content-Disposition", "inline", filename=fname)
msg.attach(img)
img = MIMEImage(png, name="wetterbericht.png")
img.add_header("Content-ID", "<chart@weather>")
img.add_header("Content-Disposition", "inline", filename="wetterbericht.png")
msg.attach(img)
with smtplib.SMTP(MAIL_SERVER, MAIL_PORT) as smtp:
smtp.ehlo()
@@ -231,25 +213,15 @@ def main() -> None:
start, end, label = last_month_range()
print(f"Hole Wetterdaten für {label} ({start.date()} {end.date()}) …")
charts = {
"Tages-Min / Tages-Max Temperatur": chart_temp_minmax(start, end, label),
"Temperatur Stundenmittel": chart_temp_hourly(start, end, label),
"Niederschlag pro Tag": chart_rain_daily(start, end, label),
}
print("Diagramme erstellt.")
png = create_combined_chart(start, end, label)
os.makedirs("images", exist_ok=True)
for fname, (_, png) in zip(
["temperatur_minmax.png", "temperatur_stundenmittel.png", "niederschlag_taglich.png"],
charts.items(),
):
path = os.path.join("images", fname)
with open(path, "wb") as f:
f.write(png)
print(f"{path}")
path = os.path.join("images", "wetterbericht.png")
with open(path, "wb") as f:
f.write(png)
print(f"Diagramm erstellt → {path}")
send_email(charts, label)
send_email(png, label)
if __name__ == "__main__":