# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # air.py : qualité de l'air à l'adresse — RSQAQ (MELCCFP, Données Québec) # # Sources ouvertes officielles du Réseau de surveillance de la qualité de # l'air du Québec : # · stations (rsqaq-stations) : coordonnées, ouverture/fermeture ; # · données horaires continues (PM2.5, NO2, O3, SO2, CO) 2024-2025 ; # · données séquentielles (Concentration PST / PM10 + métaux) 2007-2026 — # les particules en suspension totales demandées (étiquette PST). # `refresh()` télécharge les CSV et réduit tout en moyennes annuelles par # station dans data/air.db (minuscule) ; `lookup(lat, lng)` remonte la # station la plus proche (≤ 60 km) avec ses mesures les plus récentes, # comparées aux repères annuels de l'OMS (2021). # ----------------------------------------------------------------------------- from __future__ import annotations import csv import io import math import sqlite3 import time import urllib.request from pathlib import Path DB_PATH = Path(__file__).resolve().parent.parent / "data" / "air.db" UA = "LouKaBot/1.0 (+https://www.lou-ka.com; contact@spboucher.ai)" URL_STATIONS = ("https://www.donneesquebec.ca/recherche/dataset/" "8656ad05-c174-41c5-9ed7-8c69d308beb9/resource/" "cebea532-a9e0-4a39-8c2d-54f33d937c73/download/" "rsqaq_stations_de_la_qualite_de_lair.csv") URL_HORAIRE = { 2025: ("https://www.donneesquebec.ca/recherche/dataset/" "a80757bd-d442-4d3d-9269-11628330b727/resource/" "370a6be4-1530-4c2b-92f3-0a308224f284/download/" "rsqaq_continues_horaires_2025.csv"), 2024: ("https://www.donneesquebec.ca/recherche/dataset/" "a80757bd-d442-4d3d-9269-11628330b727/resource/" "135146b7-fd4a-4564-b10b-b1ae94257889/download/" "rsqaq_continues_horaires_2024.csv"), } URL_SEQ = ("https://www.donneesquebec.ca/recherche/dataset/" "bff56fad-22c3-450b-aafa-4ccdad6c91f2/resource/" "b83cac3b-6199-4cfd-a903-cdec2369630c/download/" "rsqaq_sequentielles_2007-2026.csv") # colonnes horaires retenues -> polluant canonique _HOURLY = {"PM2.5-T640": "PM2.5", "PM2.5-BAM": "PM2.5", "NO2": "NO2", "O3": "O3", "SO2": "SO2", "CO": "CO"} # contaminants séquentiels retenus (concentrations de particules) _SEQ = {"Concentration PST": "PST", "Concentration PM10": "PM10"} # repères ANNUELS (moyenne) : OMS 2021 pour PM2.5/PM10/NO2 ; PST : norme # annuelle du Règlement sur l'assainissement de l'atmosphère (Québec) REFS = {"PM2.5": ("OMS 2021", 5.0), "PM10": ("OMS 2021", 15.0), "NO2": ("OMS 2021", 10.0), "PST": ("RAA Québec", 60.0)} UNITES = {"PM2.5": "µg/m³", "PM10": "µg/m³", "PST": "µg/m³", "NO2": "ppb", "O3": "ppb", "SO2": "ppb", "CO": "ppm"} def _dl(url: str) -> io.StringIO: req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=300) as r: return io.StringIO(r.read().decode("utf-8-sig")) def refresh() -> None: stations: dict[str, dict] = {} for row in csv.DictReader(_dl(URL_STATIONS)): sid = row["ID_STATION"].strip().lstrip("0") stations[sid] = {"nom": row["NOM_STATION"], "ville": row["MUNICIPALITE"], "lat": float(row["LATITUDE"] or 0), "lng": float(row["LONGITUDE"] or 0)} print(f"[air] {len(stations)} stations") # (station, annee, polluant) -> [somme, n] acc: dict[tuple, list[float]] = {} def _sid(label: str) -> str: return label.split(" - ")[0].strip().lstrip("0") for annee, url in URL_HORAIRE.items(): n = 0 rd = csv.DictReader(_dl(url)) cols = [c for c in (rd.fieldnames or []) if c in _HOURLY] for row in rd: sid = _sid(row["Station"]) for c in cols: v = row.get(c) if not v: continue try: x = float(v) except ValueError: continue if x < 0: continue a = acc.setdefault((sid, annee, _HOURLY[c]), [0.0, 0]) a[0] += x a[1] += 1 n += 1 print(f"[air] horaires {annee} : {n} mesures") n = 0 for row in csv.DictReader(_dl(URL_SEQ)): pol = _SEQ.get(row["Contaminant"]) if not pol: continue annee = int(row["Date"][:4]) if annee < 2022: continue v = row.get("Resultat") if not v: ld = row.get("LD") if not ld: continue try: x = float(ld) / 2.0 # convention < LD -> LD/2 except ValueError: continue else: try: x = float(v) except ValueError: continue a = acc.setdefault((_sid(row["Station"]), annee, pol), [0.0, 0]) a[0] += x a[1] += 1 n += 1 print(f"[air] séquentielles PST/PM10 : {n} mesures (≥ 2022)") con = sqlite3.connect(DB_PATH) con.executescript(""" DROP TABLE IF EXISTS air_stats; CREATE TABLE air_stats (station TEXT, nom TEXT, ville TEXT, lat REAL, lng REAL, annee INTEGER, polluant TEXT, moyenne REAL, n INTEGER); CREATE INDEX idx_air_latlng ON air_stats (lat, lng); """) now = time.strftime("%Y-%m-%d") rows = [] for (sid, annee, pol), (somme, cnt) in acc.items(): st = stations.get(sid) # minimum de couverture : 30 jours d'heures ou 15 échantillons if not st or cnt < (720 if pol in ("PM2.5", "NO2", "O3", "SO2", "CO") else 15): continue rows.append((sid, st["nom"], st["ville"], st["lat"], st["lng"], annee, pol, round(somme / cnt, 2), cnt)) with con: con.executemany("INSERT INTO air_stats VALUES (?,?,?,?,?,?,?,?,?)", rows) con.execute("CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v)") con.execute("INSERT OR REPLACE INTO meta VALUES ('maj', ?)", (now,)) print(f"[air] {len(rows)} agrégats station-année-polluant -> {DB_PATH}") con.close() def _dist_km(lat1, lng1, lat2, lng2) -> float: dlat = math.radians(lat2 - lat1) dlng = math.radians(lng2 - lng1) a = (math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlng / 2) ** 2) return 6371 * 2 * math.asin(math.sqrt(a)) def lookup(lat: float, lng: float, max_km: float = 60.0) -> dict | None: """Mesures de la station RSQAQ la plus proche (année la plus récente).""" if not DB_PATH.exists(): return None con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) con.row_factory = sqlite3.Row d = max_km / 111.0 rows = con.execute( "SELECT * FROM air_stats WHERE lat BETWEEN ? AND ? AND lng BETWEEN " "? AND ?", (lat - d, lat + d, lng - d / max(0.2, math.cos( math.radians(lat))), lng + d / max(0.2, math.cos( math.radians(lat))))).fetchall() con.close() if not rows: return {"station": None} # station la plus proche disposant de données récentes best: dict[str, dict] = {} for r in rows: km = _dist_km(lat, lng, r["lat"], r["lng"]) if km > max_km: continue b = best.setdefault(r["station"], {"km": km, "rows": [], "r": r}) b["rows"].append(r) if not best: return {"station": None} sid, b = min(best.items(), key=lambda kv: kv[1]["km"]) # par polluant : l'année la plus récente mesures: dict[str, dict] = {} for r in sorted(b["rows"], key=lambda r: -r["annee"]): if r["polluant"] in mesures: continue ref = REFS.get(r["polluant"]) mesures[r["polluant"]] = { "moyenne": r["moyenne"], "annee": r["annee"], "n": r["n"], "unite": UNITES.get(r["polluant"], ""), "ref": ref[1] if ref else None, "ref_nom": ref[0] if ref else None} st = b["r"] return {"station": st["nom"], "ville": st["ville"], "distance_km": round(b["km"], 1), "mesures": mesures}