#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ build_contexte.py — Construit data/staging-contexte.db : tables « proximité et contexte socio-éducatif » pour l'enrichissement des fiches Lou-Ka. Tables produites (schéma contractuel) : 1. da_pmd — mesures de proximité StatCan (PMD 2021, parution 2023) agrégées par aire de diffusion (AD, DAUID 24*), valeurs normalisées 0..1. 2. da_defav — quintiles de défavorisation matérielle/sociale INSPQ 2021 par AD (échelle Québec, 1 = favorisé, 5 = défavorisé). 3. ecoles — écoles primaires/secondaires (public + privé) du Québec avec coordonnées MEQ et déciles IMSE/SFR (public seulement). 4. meta — sources, dates, licences. Sources (téléchargées dans /tmp/louka-ctx si absentes, puis nettoyées avec --clean) : * PMD : Base de données sur les mesures de proximité, StatCan 17-26-0002, édition 2023 (données 2021). CSV national par îlot de diffusion (DBUID). Les mesures prox_idx_* sont DÉJÀ normalisées entre 0 et 1 à l'échelle nationale par StatCan (transformation min-max des mesures brutes de proximité) ; « .. » = valeur indisponible. On agrège par AD (colonne DAUID fournie = 8 premiers caractères du DBUID) en moyenne pondérée par la population de l'îlot (DBPOP) ; si aucun îlot peuplé n'a de valeur, moyenne simple des valeurs non nulles. * INSPQ : Indice de défavorisation matérielle et sociale 2021 — tableau d'équivalence complet « TableCorrespondancesCompleteQuebec2021_fr.xlsx » (le jeu Données Québec ne publie le tableau d'équivalence qu'en XLSX, pas en CSV ; on lit le XLSX en streaming avec openpyxl read_only). Colonnes QuintMat / QuintSoc = quintiles à l'échelle Québec. * MEQ écoles : CSV « Écoles publiques » (pps_public_ecole.csv, 1 ligne par immeuble) + « Installations privées » (pps_prive_installation.csv, 1 ligne par installation) + « Établissements privés » (noms officiels). Coordonnées COORD_X_LL84 (longitude) / COORD_Y_LL84 (latitude), WGS84. * MEQ IMSE/SFR : CSV « Défavorisation - Écoles primaires/secondaires 2025-2026 » — déciles 1 (favorisé) à 10 (défavorisé). Choix de jointure écoles ↔ IMSE/SFR (documenté) : Les fichiers IMSE/SFR sont publiés par CODE D'ORGANISME (Code_Org, 7 chiffres, ex. 711002 = l'école en tant qu'entité administrative), pas par code de bâtiment (CD_IMM). On agrège donc pps_public_ecole.csv par CD_ORGNS (= code d'organisme de l'école) et on joint sur Code_Org == CD_ORGNS. Une école multi-bâtiments donne 1 ligne ; ses coordonnées sont celles du premier immeuble géocodé du fichier MEQ. Une école présente dans les deux fichiers IMSE (primaire ET secondaire) reçoit en priorité les déciles du fichier primaire. Le privé n'a pas d'IMSE/SFR publié → NULL. Usage : .venv/bin/python scripts/build_contexte.py [--clean] (--clean : supprime /tmp/louka-ctx à la fin) """ import csv import glob import io import os import sqlite3 import sys import urllib.request import zipfile from collections import defaultdict from datetime import date # --------------------------------------------------------------------------- # Chemins et URLs # --------------------------------------------------------------------------- REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DB_PATH = os.path.join(REPO, "data", "staging-contexte.db") TMP = "/tmp/louka-ctx" URLS = { # StatCan PMD 2021 (parution 2023) — zip ~25 Mo contenant PMD-en.csv "pmd-eng.zip": "https://www150.statcan.gc.ca/pub/17-26-0002/2023001/csv/pmd-eng.zip", # INSPQ IDMS 2021, échelle Québec — zip contenant le tableau d'équivalence XLSX "inspq2021.zip": "https://www.inspq.qc.ca/sites/default/files/IDMS/Qu%C3%A9bec2021_042024.zip", # MEQ localisation des établissements "pps_public_ecole.csv": "https://www.donneesquebec.ca/recherche/dataset/2d3b5cf8-b347-49c7-ad3b-bd6a9c15e443/resource/c6640a54-bc4b-43ec-864e-6c325dce61bc/download/pps_public_ecole.csv", "pps_prive_installation.csv": "https://www.donneesquebec.ca/recherche/dataset/2d3b5cf8-b347-49c7-ad3b-bd6a9c15e443/resource/e22aa6f1-c4ff-4896-9534-6fa65133b3e9/download/pps_prive_installation.csv", "pps_prive_etablissement.csv": "https://www.donneesquebec.ca/recherche/dataset/2d3b5cf8-b347-49c7-ad3b-bd6a9c15e443/resource/83aae1b5-87b5-4e3d-9074-c0d4778fe812/download/pps_prive_etablissement.csv", # MEQ indices de défavorisation scolaire (IMSE/SFR) "defav_prim.csv": "https://www.donneesquebec.ca/recherche/dataset/004de02c-19f1-4da0-9af8-33f893e41972/resource/6c5d4a5d-ba3b-40a6-b570-916f43ab622c/download/defav_ecole_prim_public.csv", "defav_sec.csv": "https://www.donneesquebec.ca/recherche/dataset/004de02c-19f1-4da0-9af8-33f893e41972/resource/3e9aa43a-c32b-4779-b258-8407db716813/download/defav_ecole_sec_public.csv", } # Correspondance colonne PMD -> colonne du schéma contractuel PMD_COLS = [ ("prox_idx_grocery", "prox_epicerie"), ("prox_idx_pharma", "prox_pharmacie"), ("prox_idx_childcare", "prox_garderie"), ("prox_idx_educpri", "prox_ecole_prim"), ("prox_idx_educsec", "prox_ecole_sec"), ("prox_idx_health", "prox_sante"), ("prox_idx_lib", "prox_bibliotheque"), ("prox_idx_parks", "prox_parc"), ("prox_idx_transit", "prox_transport"), ("prox_idx_emp", "prox_emploi"), ] def telecharger(): """Télécharge les fichiers sources manquants dans /tmp/louka-ctx.""" os.makedirs(TMP, exist_ok=True) for nom, url in URLS.items(): dest = os.path.join(TMP, nom) if os.path.exists(dest) and os.path.getsize(dest) > 0: continue print(f" téléchargement {nom} …") req = urllib.request.Request(url, headers={"User-Agent": "lou-ka/1.0"}) with urllib.request.urlopen(req, timeout=300) as r, open(dest, "wb") as f: while True: bloc = r.read(1 << 20) if not bloc: break f.write(bloc) # dézippage (idempotent) if not glob.glob(os.path.join(TMP, "pmd", "**", "PMD-en.csv"), recursive=True): with zipfile.ZipFile(os.path.join(TMP, "pmd-eng.zip")) as z: z.extractall(os.path.join(TMP, "pmd")) if not glob.glob(os.path.join(TMP, "inspq", "**", "*TableCorrespondances*"), recursive=True): with zipfile.ZipFile(os.path.join(TMP, "inspq2021.zip")) as z: z.extractall(os.path.join(TMP, "inspq")) # --------------------------------------------------------------------------- # 1. da_pmd — agrégation des îlots (DB) vers les aires de diffusion (AD) # --------------------------------------------------------------------------- def construire_da_pmd(cx): chemin = glob.glob(os.path.join(TMP, "pmd", "**", "PMD-en.csv"), recursive=True)[0] # Accumulateurs par DAUID : pour chaque mesure, (somme pondérée, somme des # poids, somme simple, compte) — le fichier national (~500 k lignes) est lu # en streaming, on ne garde en mémoire que les AD du Québec (~13-14 k). acc = defaultdict(lambda: [[0.0, 0.0, 0.0, 0] for _ in PMD_COLS]) # NB : le CSV PMD est encodé en Latin-1 (noms de lieux accentués), pas UTF-8. with open(chemin, newline="", encoding="latin-1") as f: lecteur = csv.DictReader(f) for ligne in lecteur: dauid = ligne["DAUID"].strip() if not dauid.startswith("24"): # Québec seulement (PRUID 24) continue # Poids = population de l'îlot de diffusion (DBPOP) ; vide ou 0 = # îlot non peuplé, il ne contribue qu'à la moyenne simple de repli. try: poids = float(ligne["DBPOP"] or 0.0) except ValueError: poids = 0.0 a = acc[dauid] for i, (col_src, _) in enumerate(PMD_COLS): brut = ligne[col_src].strip() if brut in ("", "..", "F", "x"): # valeur indisponible continue try: v = float(brut) except ValueError: continue cell = a[i] if poids > 0: cell[0] += poids * v cell[1] += poids cell[2] += v cell[3] += 1 lignes = [] for dauid, a in acc.items(): vals = [] for wsum, wtot, ssum, n in a: if wtot > 0: v = wsum / wtot # moyenne pondérée par la population elif n > 0: v = ssum / n # repli : moyenne simple des non-nuls else: v = None # aucune valeur disponible dans l'AD if v is not None: v = min(1.0, max(0.0, v)) # garde-fou 0..1 vals.append(v) lignes.append((dauid, *vals)) cx.executemany( "INSERT INTO da_pmd VALUES (?,?,?,?,?,?,?,?,?,?,?)", lignes) return len(lignes) # --------------------------------------------------------------------------- # 2. da_defav — quintiles INSPQ (échelle Québec) par AD # --------------------------------------------------------------------------- def construire_da_defav(cx): import openpyxl # lecture streaming du XLSX (le tableau d'équivalence # n'est publié qu'en XLSX par l'INSPQ / Données Québec) chemin = glob.glob(os.path.join(TMP, "inspq", "**", "*TableCorrespondances*Quebec2021*.xlsx"), recursive=True)[0] wb = openpyxl.load_workbook(chemin, read_only=True) ws = wb["Données"] it = ws.iter_rows(values_only=True) entete = [str(c) for c in next(it)] i_ad = entete.index("AD") i_qm = entete.index("QuintMat") # quintile matériel, échelle Québec i_qs = entete.index("QuintSoc") # quintile social, échelle Québec vus = {} for ligne in it: ad = str(ligne[i_ad]).strip() if ligne[i_ad] is not None else "" if len(ad) != 8 or not ad.startswith("24"): continue qm, qs = ligne[i_qm], ligne[i_qs] qm = int(qm) if qm not in (None, "") else None qs = int(qs) if qs not in (None, "") else None # Une AD scindée entre plusieurs CLSC apparaît sur plusieurs lignes ; # les quintiles « échelle Québec » y sont identiques -> première ligne. if ad not in vus: vus[ad] = (qm, qs) wb.close() cx.executemany("INSERT INTO da_defav VALUES (?,?,?)", [(ad, qm, qs) for ad, (qm, qs) in vus.items()]) return len(vus) # --------------------------------------------------------------------------- # 3. ecoles — localisation MEQ + IMSE/SFR # --------------------------------------------------------------------------- def _lire_defav_meq(): """Déciles IMSE/SFR par code d'organisme (public seulement). Priorité au fichier primaire si un code figure dans les deux fichiers.""" deciles = {} for nom in ("defav_sec.csv", "defav_prim.csv"): # prim écrase sec with open(os.path.join(TMP, nom), newline="", encoding="utf-8-sig") as f: for ligne in csv.DictReader(f): code = ligne["Code_Org"].strip() imse = ligne["Rang_Decile_IMSE"].strip() sfr = ligne["Rang_Decile_SFR"].strip() imse = int(imse) if imse else None sfr = int(sfr) if sfr else None if imse is None and sfr is None and code in deciles: continue # ne pas écraser une valeur par du vide deciles[code] = (imse, sfr) return deciles def _coord(ligne, cx_col, cy_col): """Extrait (lat, lng) WGS84 ; COORD_X = longitude, COORD_Y = latitude.""" try: lng = float(ligne[cx_col]) lat = float(ligne[cy_col]) except (ValueError, TypeError, KeyError): return None # garde-fou : bornes du Québec if not (-80.0 <= lng <= -55.0 and 44.0 <= lat <= 63.0): return None return (lat, lng) def construire_ecoles(cx): deciles = _lire_defav_meq() # --- Public : 1 ligne par immeuble -> agrégation par code d'organisme --- publics = {} with open(os.path.join(TMP, "pps_public_ecole.csv"), newline="", encoding="utf-8-sig") as f: for ligne in csv.DictReader(f): code = ligne["CD_ORGNS"].strip() if not code: continue e = publics.setdefault(code, { "nom": ligne["NOM_OFFCL_ORGNS"].strip() or ligne["NOM_COURT_ORGNS"].strip(), "coord": None, "prim": False, "sec": False, }) e["prim"] |= ligne["PRIM"].strip() == "1" e["sec"] |= ligne["SEC"].strip() == "1" if e["coord"] is None: # premier immeuble géocodé du fichier e["coord"] = _coord(ligne, "COORD_X_LL84_IMM", "COORD_Y_LL84_IMM") # --- Privé : installations (flags d'ordre + coords) regroupées par # établissement responsable (CD_ORGNS_RESP) --- noms_prives = {} with open(os.path.join(TMP, "pps_prive_etablissement.csv"), newline="", encoding="utf-8-sig") as f: for ligne in csv.DictReader(f): noms_prives[ligne["CD_ORGNS"].strip()] = ( ligne["NOM_OFFCL"].strip() or ligne["NOM_COURT"].strip()) prives = {} with open(os.path.join(TMP, "pps_prive_installation.csv"), newline="", encoding="utf-8-sig") as f: for ligne in csv.DictReader(f): code = ligne["CD_ORGNS_RESP"].strip() or ligne["CD_ORGNS"].strip() if not code: continue e = prives.setdefault(code, { "nom": noms_prives.get(code) or ligne["NOM_OFFCL"].strip(), "coord": None, "prim": False, "sec": False, }) e["prim"] |= ligne["PRIM"].strip() == "1" e["sec"] |= ligne["SEC"].strip() == "1" if e["coord"] is None: e["coord"] = _coord(ligne, "COORD_X_LL84", "COORD_Y_LL84") lignes = [] for reseau, ecoles in (("public", publics), ("privé", prives)): for code, e in ecoles.items(): # On ne garde que le primaire/secondaire (exclut préscolaire seul, # formation professionnelle et éducation des adultes). if not (e["prim"] or e["sec"]): continue ordre = ("primaire-secondaire" if e["prim"] and e["sec"] else "primaire" if e["prim"] else "secondaire") lat, lng = e["coord"] if e["coord"] else (None, None) imse, sfr = deciles.get(code, (None, None)) if reseau == "public" \ else (None, None) lignes.append((code, e["nom"], lat, lng, ordre, reseau, imse, sfr)) cx.executemany("INSERT OR IGNORE INTO ecoles VALUES (?,?,?,?,?,?,?,?)", lignes) return len(lignes) # --------------------------------------------------------------------------- # 4. meta # --------------------------------------------------------------------------- def construire_meta(cx): meta = { "date_construction": date.today().isoformat(), "script": "scripts/build_contexte.py", "source_pmd": ("Statistique Canada, Base de données sur les mesures de " "proximité (PMD) 2021, no 17-26-0002, édition 2023. " "https://www150.statcan.gc.ca/n1/pub/17-26-0002/172600022023001-eng.htm"), "licence_pmd": ("Licence du gouvernement ouvert – Canada " "(https://ouvert.canada.ca/fr/licence-du-gouvernement-ouvert-canada). " "Contient de l'information visée par la Licence du " "gouvernement ouvert – Canada, Statistique Canada."), "note_pmd": ("Mesures prox_idx_* déjà normalisées 0..1 par StatCan à " "l'échelle nationale (min-max des mesures brutes de " "proximité par îlot de diffusion). Agrégées ici par AD " "(DAUID 24*) en moyenne pondérée par la population de " "l'îlot (DBPOP), repli sur la moyenne simple des valeurs " "non nulles si aucun îlot peuplé. NULL = aucune donnée."), "source_defav": ("INSPQ, Indice de défavorisation matérielle et sociale " "2021, tableau d'équivalence complet (échelle Québec), " "version avril 2024. https://www.donneesquebec.ca/recherche/" "dataset/indice-de-defavorisation-du-quebec-2021"), "licence_defav": "Creative Commons Attribution 4.0 (CC-BY 4.0) — INSPQ.", "note_defav": ("QuintMat/QuintSoc, échelle Québec : 1 = plus favorisé, " "5 = plus défavorisé. NULL = AD non cotée (pop. faible, " "institutionnelle, etc.). Tableau publié en XLSX " "seulement (pas de CSV)."), "source_ecoles": ("MEQ-MES, Localisation des établissements d'enseignement " "du réseau scolaire (CSV pps_public_ecole, " "pps_prive_installation, pps_prive_etablissement). " "https://www.donneesquebec.ca/recherche/dataset/" "localisation-des-etablissements-d-enseignement-du-" "reseau-scolaire-au-quebec"), "source_imse_sfr": ("MEQ, Indices de défavorisation (IMSE et SFR), déciles " "1 (favorisé) à 10 (défavorisé), année scolaire " "2025-2026. https://www.donneesquebec.ca/recherche/" "dataset/indices-de-defavorisation"), "licence_ecoles": "Creative Commons Attribution 4.0 (CC-BY 4.0) — Gouvernement du Québec (MEQ).", "note_ecoles": ("code = code d'ORGANISME MEQ (7 chiffres), pas le code " "de bâtiment : c'est la clé des fichiers IMSE/SFR " "(Code_Org). Coordonnées WGS84 du premier immeuble/" "installation géocodé. ordre ∈ {primaire, secondaire, " "primaire-secondaire} ; préscolaire seul, FP et adultes " "exclus. IMSE/SFR NULL pour le privé (non publié) et " "pour les écoles publiques sous seuil de diffusion."), } cx.executemany("INSERT INTO meta VALUES (?,?)", sorted(meta.items())) return len(meta) # --------------------------------------------------------------------------- # Validations # --------------------------------------------------------------------------- def valider(cx): q = lambda s: cx.execute(s).fetchone()[0] print("\n=== VALIDATIONS ===") n_pmd = q("SELECT COUNT(*) FROM da_pmd") print(f"da_pmd : {n_pmd} AD (attendu ~10-13 k)") assert 10000 <= n_pmd <= 15000, "nb d'AD hors plage attendue" for _, col in PMD_COLS: mn, mx, nn = cx.execute( f"SELECT MIN({col}), MAX({col}), SUM({col} IS NOT NULL) FROM da_pmd" ).fetchone() assert mn is None or (0.0 <= mn and mx <= 1.0), f"{col} hors 0..1" print(f" {col:18s} min={mn:.4f} max={mx:.4f} non-null={nn}") n_def = q("SELECT COUNT(*) FROM da_defav") n_cot = q("SELECT COUNT(*) FROM da_defav WHERE quintile_materiel IS NOT NULL") # Couverture des AD « urbaines » : approximée par les AD de la PMD ayant # une mesure de transport en commun (desserte = milieu urbain). couv = q("""SELECT 100.0 * SUM(d.quintile_materiel IS NOT NULL) / COUNT(*) FROM da_pmd p LEFT JOIN da_defav d ON d.dauid = p.dauid WHERE p.prox_transport IS NOT NULL""") print(f"da_defav : {n_def} AD, {n_cot} cotées ; couverture AD urbaines " f"(proxy transport) = {couv:.1f} % (exigé >= 90 %)") assert couv >= 90.0, "couverture urbaine insuffisante" n_ec = q("SELECT COUNT(*) FROM ecoles") n_geo = q("SELECT COUNT(*) FROM ecoles WHERE lat IS NOT NULL") n_imse = q("SELECT COUNT(*) FROM ecoles WHERE imse IS NOT NULL") print(f"ecoles : {n_ec} (géocodées {n_geo}, avec IMSE {n_imse})") assert n_ec >= 2500, "trop peu d'écoles" print("échantillon Limoilou (attendu IMSE élevé) / Sillery (faible) :") for motif in ("%Saint-Fidèle%", "%Saint-Paul-Apôtre%", "%Stadacona%", "%Saint-Michel de Sillery%", "%Les Sources%", "%Rochebelle%"): for row in cx.execute( "SELECT code, nom, ordre, reseau, imse, sfr FROM ecoles " "WHERE nom LIKE ? LIMIT 3", (motif,)): print(" ", row) # --------------------------------------------------------------------------- def main(): print("Téléchargement / extraction des sources …") telecharger() os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) if os.path.exists(DB_PATH): os.remove(DB_PATH) # reconstruction complète cx = sqlite3.connect(DB_PATH) cx.executescript(""" CREATE TABLE da_pmd( dauid TEXT PRIMARY KEY, prox_epicerie REAL, prox_pharmacie REAL, prox_garderie REAL, prox_ecole_prim REAL, prox_ecole_sec REAL, prox_sante REAL, prox_bibliotheque REAL, prox_parc REAL, prox_transport REAL, prox_emploi REAL); CREATE TABLE da_defav( dauid TEXT PRIMARY KEY, quintile_materiel INTEGER, quintile_social INTEGER); CREATE TABLE ecoles( code TEXT PRIMARY KEY, nom TEXT, lat REAL, lng REAL, ordre TEXT, reseau TEXT, imse INTEGER, sfr INTEGER); CREATE TABLE meta(cle TEXT PRIMARY KEY, valeur TEXT); """) print("Construction da_pmd (streaming du CSV national ~500 k lignes) …") print(f" {construire_da_pmd(cx)} AD insérées") print("Construction da_defav …") print(f" {construire_da_defav(cx)} AD insérées") print("Construction ecoles …") print(f" {construire_ecoles(cx)} écoles insérées") construire_meta(cx) cx.commit() valider(cx) cx.close() print(f"\nOK -> {DB_PATH} ({os.path.getsize(DB_PATH)/1e6:.1f} Mo)") if "--clean" in sys.argv: import shutil shutil.rmtree(TMP, ignore_errors=True) print(f"{TMP} supprimé") if __name__ == "__main__": main()