#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Construction de data/staging-environnement.db — enrichissement « environnement » des fiches Lou-Ka : îlots de chaleur (INSPQ), criminalité (SPVM), indice de gravité de la criminalité (Statistique Canada). Tables produites (schéma contractuel — ne pas modifier) : heat(coord_key TEXT PRIMARY KEY, classe INTEGER, ecart REAL) coord_key = "lat,lng" avec exactement 4 décimales, équivalent SQLite : printf('%.4f', ROUND(lat,4)) || ',' || printf('%.4f', ROUND(lng,4)) classe : 1 à 9 (INSPQ ICFU 2020-2022). SENS VÉRIFIÉ dans les métadonnées du jeu Données Québec : classes 1-2-3 = îlots de FRAÎCHEUR urbains (1 = le plus frais), classes 8-9 = îlots de CHALEUR urbains (9 = le plus chaud). NULL si hors des centres de population couverts (nodata). ecart : écart de température relatif en °C par rapport à un boisé voisin (raster écoumène 2021). NULL si hors écoumène (nodata). crime_mtl(id INTEGER PRIMARY KEY, lat REAL, lng REAL, ts REAL, categorie TEXT) Actes criminels SPVM des 24 derniers mois, coordonnées valides seulement. ts = epoch UTC (minuit) de la date de l'acte. igc(service TEXT, annee INTEGER, indice REAL, taux REAL) indice = Indice de gravité de la criminalité (StatCan 35-10-0187 par corps de police ; 35-10-0026 pour le Canada). taux = taux de criminalité par 100 000 hab. (infractions au Code criminel sauf délits de la route, StatCan 35-10-0179 / 35-10-0177 via l'API WDS). 5 dernières années disponibles. meta(cle TEXT PRIMARY KEY, valeur TEXT) : sources, dates, attributions. Usage : .venv/bin/python scripts/build_environnement.py # tout reconstruire .venv/bin/python scripts/build_environnement.py --etapes crime # rafraîchir la criminalité seulement (à relancer mensuellement) .venv/bin/python scripts/build_environnement.py --etapes chaleur \ --coords nouvelles.txt # échantillonner UNIQUEMENT une liste de nouvelles coordonnées (une paire "lat,lng" par ligne) et les upserter dans heat .venv/bin/python scripts/build_environnement.py --garder-cache # ne pas supprimer les GeoTIFF téléchargés (~9,6 Go) après l'exécution Dépendances : rasterio, pyproj (pip install rasterio pyproj). Licences : INSPQ/CERFO CC-BY 4.0 ; Ville de Montréal CC-BY 4.0 ; Statistique Canada — licence ouverte. """ import argparse import calendar import csv import io import json import shutil import sqlite3 import sys import urllib.request import zipfile from datetime import date, datetime, timedelta from pathlib import Path RACINE = Path(__file__).resolve().parent.parent DB_LOUKA = RACINE / "data" / "louka.db" DB_STAGING = RACINE / "data" / "staging-environnement.db" CACHE_DEFAUT = Path("/tmp/louka-env") # --- Sources ----------------------------------------------------------------- URL_TIF_CLASSES = ("https://dq-prd-bucket1.s3.ca-central-1.amazonaws.com/inspq/" "ICFU2022_CentresPopulation2021_buf2000m_9cl.tif") URL_TIF_ECART = ("https://dq-prd-bucket1.s3.ca-central-1.amazonaws.com/inspq/" "EcartTemperatureRelatif2022_Ecoumene2021.tif") URL_CRIME_CSV = ("https://donnees.montreal.ca/dataset/" "5829b5b0-ea6f-476f-be94-bc2b8797769a/resource/" "c6f482bf-bf0f-4960-8b2f-9982c211addd/download/actes-criminels.csv") URL_STATCAN_ZIP = "https://www150.statcan.gc.ca/n1/tbl/csv/{pid}-fra.zip" URL_WDS = ("https://www150.statcan.gc.ca/t1/wds/rest/" "getDataFromCubePidCoordAndLatestNPeriods") # Le portail donnees.montreal.ca refuse les clients sans User-Agent navigateur. UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36") MOIS_CRIME = 24 # fenêtre criminalité (24 derniers mois) ANNEES_IGC = 5 # nombre d'années IGC conservées # Corps de police retenus. # - code_geo : identifiant entre crochets dans la colonne GÉO des CSV StatCan # 35-10-0187 (indice) et 35-10-0026 (Canada). # - pid_taux / geo_taux : cube et memberId Géographie pour le taux de # criminalité via l'API WDS (35-10-0179 = services de police au Québec, # 35-10-0177 = Canada/provinces/RMR). Coordonnée = geo.infraction.stat.0*7 # avec infraction 3 = « Total des infractions au Code criminel (sauf les # délits de la route) » et statistique 2 = « Taux pour 100 000 habitants ». SERVICES = [ # (nom_affiché, code_geo, pid_indice, pid_taux, geo_taux) ("SPVQ (Ville de Québec)", "24215", 35100187, 35100179, 107), ("Ville de Lévis", "24219", 35100187, 35100179, 109), ("SPVM (Montréal)", "24175", 35100187, 35100179, 82), ("Laval", "24141", 35100187, 35100179, 59), ("Longueuil (agglomération)", "24231", 35100187, 35100179, 115), ("Québec (province)", "24", 35100187, 35100179, 1), ("Canada", "CAN", 35100026, 35100177, 1), ] # --- Utilitaires --------------------------------------------------------------- def telecharger(url: str, dest: Path, description: str) -> None: """Télécharge url vers dest (flux, pas de chargement en mémoire).""" if dest.exists() and dest.stat().st_size > 0: print(f" [cache] {description} déjà présent : {dest}") return print(f" téléchargement {description} …") req = urllib.request.Request(url, headers={"User-Agent": UA}) tmp = dest.with_suffix(dest.suffix + ".part") with urllib.request.urlopen(req, timeout=120) as rep, open(tmp, "wb") as f: shutil.copyfileobj(rep, f, length=1 << 20) tmp.rename(dest) print(f" ok ({dest.stat().st_size / 1e6:.1f} Mo)") def coord_key(lat: float, lng: float) -> str: """Clé "lat,lng" à 4 décimales — identique à printf('%.4f',ROUND(lat,4))||','||printf('%.4f',ROUND(lng,4)) en SQLite.""" return f"{lat:.4f},{lng:.4f}" def ouvrir_staging() -> sqlite3.Connection: DB_STAGING.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_STAGING) conn.executescript(""" CREATE TABLE IF NOT EXISTS heat( coord_key TEXT PRIMARY KEY, classe INTEGER, ecart REAL); CREATE TABLE IF NOT EXISTS crime_mtl( id INTEGER PRIMARY KEY, lat REAL, lng REAL, ts REAL, categorie TEXT); CREATE INDEX IF NOT EXISTS idx_crime_lat ON crime_mtl(lat); CREATE TABLE IF NOT EXISTS igc( service TEXT, annee INTEGER, indice REAL, taux REAL); CREATE TABLE IF NOT EXISTS meta(cle TEXT PRIMARY KEY, valeur TEXT); """) return conn def poser_meta(conn: sqlite3.Connection, cle: str, valeur: str) -> None: conn.execute("INSERT OR REPLACE INTO meta(cle, valeur) VALUES(?,?)", (cle, valeur)) # --- Étape 1 : îlots de chaleur ------------------------------------------------- def coords_depuis_louka() -> list[tuple[float, float]]: """Coordonnées uniques (arrondies à 4 décimales par SQLite) des immeubles actifs.""" conn = sqlite3.connect(f"file:{DB_LOUKA}?mode=ro", uri=True) lignes = conn.execute( "SELECT DISTINCT ROUND(lat,4), ROUND(lng,4) FROM listings " "WHERE active=1 AND lat IS NOT NULL AND lng IS NOT NULL").fetchall() conn.close() return [(la, lo) for la, lo in lignes] def coords_depuis_fichier(chemin: Path) -> list[tuple[float, float]]: """Fichier texte : une paire « lat,lng » par ligne (les valeurs sont ré-arrondies à 4 décimales).""" paires = [] for ligne in chemin.read_text().splitlines(): ligne = ligne.strip() if not ligne or ligne.startswith("#"): continue la, lo = ligne.split(",") paires.append((round(float(la), 4), round(float(lo), 4))) return paires def echantillonner_raster(chemin_tif: Path, coords: list[tuple[float, float]]): """Échantillonne un GeoTIFF aux points (lat,lng) WGS84, par fenêtres (rasterio.sample), sans jamais charger le raster entier. On reprojette les POINTS vers le CRS du raster (EPSG:32198 Québec Lambert pour les deux rasters INSPQ), jamais le raster. Renvoie une liste de valeurs (None si nodata ou hors emprise).""" import math import rasterio from pyproj import Transformer with rasterio.open(chemin_tif) as ds: transfo = Transformer.from_crs("EPSG:4326", ds.crs, always_xy=True) xy = [transfo.transform(lo, la) for la, lo in coords] valeurs = [] for echantillon in ds.sample(xy, indexes=1, masked=True): v = echantillon[0] if v is None or getattr(v, "mask", False) or math.isnan(float(v)) \ or (ds.nodata is not None and float(v) == float(ds.nodata)): valeurs.append(None) else: valeurs.append(float(v)) return valeurs def etape_chaleur(conn: sqlite3.Connection, cache: Path, coords: list[tuple[float, float]] | None) -> None: print("== Étape chaleur (INSPQ ICFU 2020-2022) ==") if coords is None: coords = coords_depuis_louka() print(f" {len(coords)} coordonnées à échantillonner") if not coords: return tif_classes = cache / "ICFU2022_CentresPopulation2021_buf2000m_9cl.tif" tif_ecart = cache / "EcartTemperatureRelatif2022_Ecoumene2021.tif" telecharger(URL_TIF_CLASSES, tif_classes, "raster classes ICFU (≈34 Mo)") telecharger(URL_TIF_ECART, tif_ecart, "raster écarts de température (≈9,6 Go)") print(" échantillonnage classes (9 niveaux) …") classes = echantillonner_raster(tif_classes, coords) print(" échantillonnage écarts de température …") ecarts = echantillonner_raster(tif_ecart, coords) lignes = [] for (la, lo), cl, ec in zip(coords, classes, ecarts): lignes.append((coord_key(la, lo), int(cl) if cl is not None else None, round(ec, 2) if ec is not None else None)) conn.executemany( "INSERT OR REPLACE INTO heat(coord_key, classe, ecart) VALUES(?,?,?)", lignes) poser_meta(conn, "heat.source", "INSPQ/CERFO — Îlots de chaleur/fraîcheur urbains et écarts de " "température relatifs 2020-2022 (Données Québec)") poser_meta(conn, "heat.url_classes", URL_TIF_CLASSES) poser_meta(conn, "heat.url_ecart", URL_TIF_ECART) poser_meta(conn, "heat.licence", "CC-BY 4.0 — attribution : INSPQ et CERFO") poser_meta(conn, "heat.sens_classes", "Vérifié dans les métadonnées Données Québec : classes 1-2-3 = " "îlots de fraîcheur urbains (1 = le plus frais), classes 8-9 = " "îlots de chaleur urbains (9 = le plus chaud). Classement des " "écarts de température par centre de population (2021), " "buffer 2 km, résolution 15 m, EPSG:32198.") poser_meta(conn, "heat.ecart_unite", "°C, écart relatif à un boisé de proximité (écoumène 2021) ; " "NULL = hors écoumène / hors centres de population (nodata)") poser_meta(conn, "heat.coord_key", "printf('%.4f',ROUND(lat,4))||','||printf('%.4f',ROUND(lng,4))") poser_meta(conn, "heat.telecharge_le", datetime.now().isoformat(timespec="seconds")) conn.commit() total, avec_classe = conn.execute( "SELECT COUNT(*), COUNT(classe) FROM heat").fetchone() print(f" heat : {total} lignes ({avec_classe} avec classe)") # --- Étape 2 : criminalité SPVM -------------------------------------------------- def etape_crime(conn: sqlite3.Connection, cache: Path) -> None: print("== Étape criminalité (actes criminels SPVM) ==") fichier = cache / "actes-criminels.csv" # Toujours rafraîchir : le jeu est mis à jour quotidiennement. if fichier.exists(): fichier.unlink() telecharger(URL_CRIME_CSV, fichier, "CSV actes criminels (≈30 Mo)") seuil = date.today() - timedelta(days=MOIS_CRIME * 365 // 12) lignes = [] ecartees = 0 with open(fichier, encoding="utf-8-sig", newline="") as f: for rang in csv.DictReader(f): try: d = datetime.strptime(rang["DATE"].strip(), "%Y-%m-%d").date() la = float(rang["LATITUDE"]) lo = float(rang["LONGITUDE"]) except (ValueError, KeyError, TypeError): ecartees += 1 continue if d < seuil: continue # bornes de validité : île de Montréal et environs if not (44.5 <= la <= 46.5 and -75.0 <= lo <= -72.5): ecartees += 1 continue ts = float(calendar.timegm(d.timetuple())) # epoch UTC (minuit) lignes.append((la, lo, ts, rang["CATEGORIE"].strip())) conn.execute("DELETE FROM crime_mtl") conn.executemany( "INSERT INTO crime_mtl(lat, lng, ts, categorie) VALUES(?,?,?,?)", lignes) poser_meta(conn, "crime.source", "Ville de Montréal (SPVM) — Actes criminels, " "https://donnees.montreal.ca/dataset/actes-criminels") poser_meta(conn, "crime.licence", "CC-BY 4.0 — attribution : Ville de Montréal") poser_meta(conn, "crime.fenetre", f"{MOIS_CRIME} derniers mois (depuis {seuil})") poser_meta(conn, "crime.telecharge_le", datetime.now().isoformat(timespec="seconds")) conn.commit() n = conn.execute("SELECT COUNT(*) FROM crime_mtl").fetchone()[0] print(f" crime_mtl : {n} actes conservés (fenêtre ≥ {seuil}), " f"{ecartees} lignes écartées (coordonnées/date invalides)") # --- Étape 3 : indice de gravité de la criminalité (StatCan) --------------------- def lire_igc_csv(cache: Path, pid: int) -> dict[tuple[str, int], float]: """Extrait {(code_geo, annee): IGC} du CSV complet d'un tableau StatCan.""" zip_path = cache / f"statcan_{pid}.zip" telecharger(URL_STATCAN_ZIP.format(pid=pid), zip_path, f"tableau StatCan {pid}") valeurs: dict[tuple[str, int], float] = {} with zipfile.ZipFile(zip_path) as z, \ io.TextIOWrapper(z.open(f"{pid}.csv"), encoding="utf-8-sig") as f: for rang in csv.DictReader(f, delimiter=";"): if rang["Statistiques"] != "Indice de gravité de la criminalité": continue if not rang["VALEUR"]: continue geo = rang["GÉO"] code = geo.rsplit("[", 1)[-1].rstrip("]") if "[" in geo else \ ("CAN" if geo == "Canada" else geo) valeurs[(code, int(rang["PÉRIODE DE RÉFÉRENCE"]))] = float(rang["VALEUR"]) return valeurs def taux_via_wds(requetes: list[tuple[int, int]]) -> dict[tuple[int, int], dict[int, float]]: """Taux de criminalité (Code criminel sauf délits de la route, /100 000 hab.) via l'API WDS. requetes = [(pid, geo_member)] ; renvoie {(pid, geo): {annee: taux}}.""" charge = [{"productId": pid, "coordinate": f"{geo}.3.2.0.0.0.0.0.0.0", "latestN": ANNEES_IGC} for pid, geo in requetes] req = urllib.request.Request( URL_WDS, data=json.dumps(charge).encode(), headers={"Content-Type": "application/json", "User-Agent": UA}) with urllib.request.urlopen(req, timeout=120) as rep: reponses = json.load(rep) resultat: dict[tuple[int, int], dict[int, float]] = {} for rep_item in reponses: if rep_item.get("status") != "SUCCESS": continue obj = rep_item["object"] geo = int(obj["coordinate"].split(".")[0]) serie = {} for pt in obj["vectorDataPoint"]: if pt.get("value") is None: continue serie[int(pt["refPer"][:4])] = float(pt["value"]) resultat[(obj["productId"], geo)] = serie return resultat def etape_igc(conn: sqlite3.Connection, cache: Path) -> None: print("== Étape IGC (StatCan 35-10-0187 / 35-10-0026 + taux WDS) ==") igc_187 = lire_igc_csv(cache, 35100187) igc_026 = lire_igc_csv(cache, 35100026) indices = {**igc_026, **igc_187} annees = sorted({a for (_, a) in igc_187})[-ANNEES_IGC:] print(f" années retenues : {annees}") try: taux = taux_via_wds(sorted({(s[3], s[4]) for s in SERVICES})) except Exception as exc: # le taux est optionnel (NULL si indisponible) print(f" avertissement : taux WDS indisponible ({exc}) — taux=NULL") taux = {} conn.execute("DELETE FROM igc") for nom, code_geo, _pid_ind, pid_taux, geo_taux in SERVICES: for annee in annees: indice = indices.get((code_geo, annee)) t = taux.get((pid_taux, geo_taux), {}).get(annee) conn.execute("INSERT INTO igc(service, annee, indice, taux) " "VALUES(?,?,?,?)", (nom, annee, indice, t)) poser_meta(conn, "igc.source", "Statistique Canada, tableaux 35-10-0187 (IGC par service de " "police), 35-10-0026 (IGC Canada/provinces), 35-10-0179 et " "35-10-0177 (taux de criminalité — Code criminel sauf délits " "de la route, par 100 000 hab., via l'API WDS)") poser_meta(conn, "igc.licence", "Licence ouverte de Statistique Canada — adapté de Statistique " "Canada ; cela ne constitue pas une approbation par Statistique " "Canada de ce produit") poser_meta(conn, "igc.telecharge_le", datetime.now().isoformat(timespec="seconds")) conn.commit() n = conn.execute("SELECT COUNT(*) FROM igc").fetchone()[0] print(f" igc : {n} lignes") # --- Point d'entrée --------------------------------------------------------------- def principal() -> int: ap = argparse.ArgumentParser( description="Construit data/staging-environnement.db (chaleur, " "criminalité, IGC).") ap.add_argument("--etapes", default="chaleur,crime,igc", help="étapes à exécuter, séparées par des virgules " "(chaleur, crime, igc) — défaut : toutes") ap.add_argument("--coords", type=Path, default=None, help="fichier de nouvelles coordonnées (« lat,lng » par " "ligne) à échantillonner et upserter dans heat, au " "lieu de relire louka.db") ap.add_argument("--cache", type=Path, default=CACHE_DEFAUT, help=f"répertoire de cache des téléchargements " f"(défaut : {CACHE_DEFAUT})") ap.add_argument("--garder-cache", action="store_true", help="ne pas supprimer les GeoTIFF (~9,6 Go) à la fin") args = ap.parse_args() etapes = {e.strip() for e in args.etapes.split(",")} inconnues = etapes - {"chaleur", "crime", "igc"} if inconnues: ap.error(f"étapes inconnues : {inconnues}") args.cache.mkdir(parents=True, exist_ok=True) conn = ouvrir_staging() poser_meta(conn, "generation.script", "scripts/build_environnement.py") poser_meta(conn, "generation.date", datetime.now().isoformat(timespec="seconds")) poser_meta(conn, "attributions", "Îlots de chaleur : INSPQ et CERFO (CC-BY 4.0) ; Actes " "criminels : Ville de Montréal (CC-BY 4.0) ; IGC et taux de " "criminalité : Statistique Canada (licence ouverte)") coords = coords_depuis_fichier(args.coords) if args.coords else None if "chaleur" in etapes: etape_chaleur(conn, args.cache, coords) if "crime" in etapes: etape_crime(conn, args.cache) if "igc" in etapes: etape_igc(conn, args.cache) conn.commit() conn.close() if not args.garder_cache: for tif in args.cache.glob("*.tif"): print(f" nettoyage : suppression {tif}") tif.unlink() print("Terminé :", DB_STAGING) return 0 if __name__ == "__main__": sys.exit(principal())