from __future__ import annotations
import io
import os
import logging
import locale
from datetime import date, datetime, timedelta, time
from typing import Dict, Any, Optional
import numpy as np
import pandas as pd
import requests
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
from matplotlib.backends.backend_pdf import PdfPages
from matplotlib.patches import Patch
from flask import Flask, request, send_file, jsonify
from flask_cors import CORS
# --- Configuration -------------------------------------------------
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
ALLOWED_ORIGINS = os.environ.get("ALLOWED_ORIGINS", "https://sglh.nl,https://www.sglh.nl").split(",")
DEFAULT_ROLLUP = os.environ.get("DEFAULT_ROLLUP", "10s")
# Locale
try:
locale.setlocale(locale.LC_TIME, "nl_NL.utf8")
except Exception:
try:
locale.setlocale(locale.LC_TIME, "nl_NL")
except Exception:
logging.getLogger().warning("nl_NL locale not available; falling back to system locale")
logging.basicConfig(level=LOG_LEVEL)
logger = logging.getLogger(__name__)
app = Flask(__name__)
CORS(app, resources={r"/*": {"origins": ALLOWED_ORIGINS}})
# --- Available Kits ---
AVAILABLE_KITS = [
{"id": 19067, "name": "Julianalaan 1"},
{"id": 19068, "name": "de Wiel 14"},
{"id": 19057, "name": "Kanaaldijk N.W. 77"},
{"id": 19185, "name": "Kanaaldijk N.W. 49B"},
{"id": 19062, "name": "Torenstraat 40"},
{"id": 19128, "name": "Wethouder van Wellaan 81"},
{"id": 19146, "name": "Sneeuwheide 20"},
{"id": 19122, "name": "Kerkstraat Zuid 17"},
{"id": 19148, "name": "Meester strikstraat 42"},
{"id": 19127, "name": "Binderseind 47"},
{"id": 19137, "name": "Berkendonk 73"},
{"id": 19139, "name": "Brabanthof 23"},
{"id": 19088, "name": "Zwanebloemsingel 40"},
{"id": 19131, "name": "Gerwenseweg 48"}
]
# --- Helper functions -----------------------------------------------
def iso_week_start_date(year: int, week: int) -> Optional[date]:
try:
return datetime.fromisocalendar(year, week, 1).date()
except Exception:
return None
def safe_get_json(url: str, params: Optional[Dict[str, Any]] = None, timeout: int = 10) -> Dict[str, Any]:
try:
resp = requests.get(url, params=params, timeout=timeout)
resp.raise_for_status()
return resp.json()
except requests.RequestException as e:
logger.exception("HTTP error while requesting %s", url)
raise
except ValueError as e:
logger.exception("Invalid JSON from %s", url)
raise
def extract_sensors_from_meta(meta: Dict[str, Any]) -> Dict[str, int]:
sensors_root = None
if isinstance(meta, dict):
sensors_root = meta.get("data", {}).get("sensors") or meta.get("sensors") or meta.get("data")
result: Dict[str, int] = {}
if not sensors_root:
return result
for s in sensors_root:
name = (s.get("name") or "").lower()
sid = s.get("id")
if not sid:
continue
if "noise level a weighting" in name:
result["Geluid"] = sid
elif "pm1" in name:
result["PM1"] = sid
elif "pm2.5" in name or "pm25" in name:
result["PM2.5"] = sid
elif "pm4.0" in name or "pm40" in name:
result["PM4.0"] = sid
elif "pm10" in name:
result["PM10"] = sid
return result
def haal_readings(device_id: int, sensor_id: int, start_iso: str, end_iso: str, functie: str = "avg", rollup: str = "30m") -> pd.DataFrame:
def ensure_z(ts: str) -> str:
return ts if ts.endswith("Z") else f"{ts}Z"
from_ts = ensure_z(start_iso)
to_ts = ensure_z(end_iso)
url = f"https://api.smartcitizen.me/v0/devices/{device_id}/readings"
params = {"sensor_id": sensor_id, "rollup": rollup, "from": from_ts, "to": to_ts, "function": functie}
try:
data = safe_get_json(url, params=params)
except Exception:
return pd.DataFrame()
rows = []
if isinstance(data, list):
rows = data
elif isinstance(data, dict):
if "readings" in data and isinstance(data["readings"], list):
rows = data["readings"]
elif "results" in data and isinstance(data["results"], list):
rows = data["results"]
else:
try:
rows = [{"recorded_at": k, "waarde": v} for k, v in data.items()]
except Exception:
rows = []
if not rows:
return pd.DataFrame()
df = pd.DataFrame(rows)
if "recorded_at" not in df.columns and "timestamp" in df.columns:
df = df.rename(columns={"timestamp": "recorded_at"})
if "recorded_at" not in df.columns or "waarde" not in df.columns:
for col in df.columns:
if df[col].dtype == object and pd.to_datetime(df[col], errors="coerce").notna().any():
df = df.rename(columns={col: "recorded_at"})
break
if "waarde" not in df.columns:
numcols = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
if numcols:
df = df.rename(columns={numcols[0]: "waarde"})
df = df[[c for c in ["recorded_at", "waarde"] if c in df.columns]]
if df.empty:
return pd.DataFrame()
df["recorded_at"] = pd.to_datetime(df["recorded_at"], errors="coerce")
df = df.dropna(subset=["recorded_at"])
if df.empty:
return pd.DataFrame()
df = df.set_index("recorded_at").sort_index()
return df
def kleur_nachtzones(ax, df: pd.DataFrame) -> None:
if df is None or df.empty:
return
start_date = df.index.min().date()
end_date = df.index.max().date()
for dag in pd.date_range(start_date, end_date):
start = datetime.combine(dag, time(22, 0))
einde = datetime.combine(dag + timedelta(days=1), time(7, 0))
ax.axvspan(start, einde, color="lightgrey", alpha=0.3)
def bereken_nachtstatistiek(boven_extreem: pd.DataFrame) -> Optional[Dict[str, Any]]:
if boven_extreem is None or boven_extreem.empty:
return None
df = boven_extreem.copy()
df["nacht"] = df.index.to_series().apply(lambda x: x.date() if x.time() >= time(22, 0) else (x - timedelta(days=1)).date())
groeps = df.groupby("nacht")["waarde"]
try:
return {
"gemiddeld_aantal": groeps.count().mean(),
"min_waarde": groeps.min().min(),
"max_waarde": groeps.max().max(),
"totaal_pieken": len(df),
}
except Exception:
return None
def bereken_loki_score(groen_percentages: Dict[str, float], weights: Optional[Dict[str, int]] = None) -> float:
if weights is None:
weights = {"Geluid": 2, "PM2.5": 2, "PM10": 1}
relevant = [k for k in weights if k in groen_percentages]
if not relevant:
return 0.0
try:
totaal = sum(groen_percentages[k] * weights[k] for k in relevant)
score = round(totaal / sum(weights[k] for k in relevant) / 10, 2)
except Exception:
score = 0.0
return score
def bereken_loki_week(device_id: int, weeknummer: int, jaar: Optional[int] = None) -> Dict[str, Any]:
jaar = jaar or date.today().year
start_date = iso_week_start_date(jaar, weeknummer)
if not start_date:
return {"error": "Ongeldig weeknummer"}
end_date = start_date + timedelta(days=6)
start_iso, end_iso = start_date.isoformat(), end_date.isoformat()
meta_url = f"https://api.smartcitizen.me/v0/devices/{device_id}"
try:
meta = safe_get_json(meta_url)
except Exception as e:
return {"error": f"Fout bij ophalen device meta: {e}"}
target_sensors = extract_sensors_from_meta(meta)
if not target_sensors:
return {"error": "Geen sensoren gevonden voor dit device"}
groen_percentages: Dict[str, float] = {}
for naam, sid in target_sensors.items():
df_avg = haal_readings(device_id, sid, start_iso, end_iso, "avg")
if df_avg.empty:
groen_percentages[naam] = 0.0
continue
if naam == "Geluid":
normen = {"nacht": 45}
elif naam in ["PM1","PM2.5","PM4.0"]:
normen = {"nacht": 2}
else:
normen = {"nacht": 5}
total = len(df_avg)
groen_percentages[naam] = ((df_avg["waarde"] <= normen["nacht"]).sum() / total) * 100 if total > 0 else 0.0
loki_weights = {"Geluid": 2, "PM2.5": 2, "PM10": 1}
loki_score = bereken_loki_score(groen_percentages, loki_weights)
return {"loki_score": loki_score, "percentages": groen_percentages}
# --- Flask endpoints -----------------------------------------------
@app.route("/get_kits", methods=["GET"])
def get_kits():
return jsonify(AVAILABLE_KITS)
@app.route("/get_weeks", methods=["GET"])
def get_weeks():
jaar = date.today().year
weeknr = date.today().isocalendar().week
week_items = []
for i in range(43, weeknr):
start_date = iso_week_start_date(jaar, i)
if not start_date:
continue
end_date = start_date + timedelta(days=6)
week_items.append({
"weeknummer": i,
"periode": f"{start_date.strftime('%d %B %Y')} - {end_date.strftime('%d %B %Y')}",
})
return jsonify(week_items)
@app.route("/api/readings", methods=["GET"])
def api_readings():
device_id = request.args.get("device_id")
sensor_id = request.args.get("sensor_id")
from_date = request.args.get("from")
to_date = request.args.get("to")
rollup = request.args.get("rollup", DEFAULT_ROLLUP)
if not all([device_id, sensor_id, from_date, to_date]):
return jsonify({"error": "device_id, sensor_id, from en to zijn verplicht"}), 400
df = haal_readings(int(device_id), int(sensor_id), from_date, to_date, rollup=rollup)
if df.empty:
return jsonify({"error": "Geen readings gevonden"}), 404
return jsonify(df.reset_index().rename(columns={"recorded_at": "timestamp", "waarde": "value"}).to_dict(orient="records"))
# --- generate_pdf endpoint -----------------------------------------
@app.route("/generate_pdf", methods=["POST"])
def generate_pdf():
data = request.get_json(silent=True) or {}
try:
weeknummer = int(data.get("weeknummer"))
device_id = int(data.get("device_id"))
rollup = data.get("rollup", DEFAULT_ROLLUP)
except Exception:
return jsonify({"error": "weeknummer en device_id verplicht en moeten integers zijn"}), 400
kit_name = data.get("kit_name") or f"Device_{device_id}"
jaar = date.today().year
start_date = iso_week_start_date(jaar, weeknummer)
if not start_date:
return jsonify({"error": "Ongeldig weeknummer"}), 400
end_date = start_date + timedelta(days=6)
start_iso, end_iso = start_date.isoformat(), end_date.isoformat()
meta_url = f"https://api.smartcitizen.me/v0/devices/{device_id}"
try:
meta = safe_get_json(meta_url)
except Exception as e:
return jsonify({"error": f"Fout bij ophalen device meta: {e}"}), 500
target_sensors = extract_sensors_from_meta(meta)
if not target_sensors:
return jsonify({"error": "Geen sensoren gevonden"}), 400
kleuren = {"groen": "#00b050","geel": "#ffff00","oranje": "#ffc000","rood": "#ff0000","paars": "#7030a0"}
legenda_labels_staaf = {"groen": "goed","geel": "matig","oranje": "onvoldoende","rood": "slecht","paars": "zeer slecht"}
figuren = []
sensor_namen = []
zone_data = {k: [] for k in kleuren.keys()}
groen_percentages_for_loki: Dict[str, float] = {}
for naam, sid in target_sensors.items():
df_avg = haal_readings(device_id, sid, start_iso, end_iso, "avg", rollup=rollup)
if df_avg.empty:
continue
df_max = haal_readings(device_id, sid, start_iso, end_iso, "max", rollup=rollup)
if naam == "Geluid":
normen = {"nacht": 45, "dag": 53, "max": 55, "extreem": 63}
eenheid = "dB(A)"
elif naam in ["PM1","PM2.5","PM4.0"]:
normen = {"nacht": 2, "dag": 5, "max": 10, "extreem": 20}
eenheid = "µg/m³"
else:
normen = {"nacht": 5, "dag": 10, "max": 15, "extreem": 20}
eenheid = "µg/m³"
fig, ax = plt.subplots(figsize=(10, 6))
kleur_nachtzones(ax, df_avg)
total = len(df_avg)
if total == 0:
continue
ax.fill_between(df_avg.index, 0, df_avg["waarde"], where=(df_avg["waarde"] <= normen["nacht"]), color=kleuren["groen"], alpha=0.4)
ax.fill_between(df_avg.index, 0, df_avg["waarde"], where=(df_avg["waarde"] > normen["nacht"]) & (df_avg["waarde"] <= normen["dag"]), color=kleuren["geel"], alpha=0.4)
ax.fill_between(df_avg.index, 0, df_avg["waarde"], where=(df_avg["waarde"] > normen["dag"]) & (df_avg["waarde"] <= normen["max"]), color=kleuren["oranje"], alpha=0.4)
ax.fill_between(df_avg.index, 0, df_avg["waarde"], where=(df_avg["waarde"] > normen["max"]) & (df_avg["waarde"] <= normen["extreem"]), color=kleuren["rood"], alpha=0.4)
ax.fill_between(df_avg.index, 0, df_avg["waarde"], where=(df_avg["waarde"] > normen["extreem"]), color=kleuren["paars"], alpha=0.4)
ax.axhline(normen["nacht"], color='green', linestyle='--', linewidth=1)
ax.axhline(normen["dag"], color='yellow', linestyle='--', linewidth=1)
ax.axhline(normen["max"], color='orange', linestyle='--', linewidth=1)
ax.axhline(normen["extreem"], color='red', linestyle='--', linewidth=1)
ax.plot(df_avg.index, df_avg["waarde"], color="black", linewidth=1)
ax.set_title(f"{naam} ({eenheid}) — {start_date.strftime('%d %B %Y')} t/m {end_date.strftime('%d %B %Y')} — {kit_name}")
ax.set_ylabel(eenheid)
ax.set_xlabel("Datum/tijd")
boven_extreem = df_max[df_max["waarde"] > normen["extreem"]] if not df_max.empty else pd.DataFrame()
stats = bereken_nachtstatistiek(boven_extreem) or {"gemiddeld_aantal": 0.0, "min_waarde": 0.0, "max_waarde": 0.0, "totaal_pieken": 0}
ypos = ax.get_ylim()[1] * 1.02
ax.text(0.01, 0.98, f"Gem. pieken/nacht: {stats['gemiddeld_aantal']:.1f} - Min/Max: {stats['min_waarde']:.1f}/{stats['max_waarde']:.1f} - Totaal: {stats['totaal_pieken']}", transform=ax.transAxes, fontsize=10, verticalalignment='top')
sensor_namen.append(naam)
figuren.append(fig)
groen_percentages_for_loki[naam] = ((df_avg["waarde"] <= normen["nacht"]).sum() / total) * 100 if total > 0 else 0.0
loki_score = bereken_loki_score(groen_percentages_for_loki)
# Generate PDF
pdf_buffer = io.BytesIO()
with PdfPages(pdf_buffer) as pdf:
for fig in figuren:
pdf.savefig(fig)
plt.close(fig)
d = pdf.infodict()
d['Title'] = f"Loki weekrapport {weeknummer}"
d['Author'] = "SG-LH"
pdf_buffer.seek(0)
return send_file(pdf_buffer, mimetype='application/pdf', as_attachment=True, download_name=f"loki_week_{weeknummer}_{kit_name}.pdf")
# --- Run server ----------------------------------------------------
if __name__ == "__main__":
host = os.environ.get("FLASK_RUN_HOST", "0.0.0.0")
port = int(os.environ.get("FLASK_RUN_PORT", "5000"))
debug = os.environ.get("FLASK_DEBUG", "False").lower() in ("1", "true", "yes")
app.run(host=host, port=port, debug=debug)
