# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # kascores.py : les KA Scores — famille de scores maison 0-100 par immeuble. # # KA Walk Score marchabilité (besoins quotidiens à pied) # KA Transit Score desserte en transport collectif # KA Bike Score praticité du vélo # KA Calme Score tranquillité estimée du secteur # KA Services Score richesse des services (1 km / 3 km) # KA Score global moyenne pondérée (personnalisable côté client) # # Sources : OpenStreetMap (© contributeurs OSM, ODbL) via environment.py, # mesures de proximité StatCan (PMD 2021, quartier.db) pour le volet # fréquence/qualité du transport collectif. AUCUNE donnée inventée : # secteur sans données → NULL (« Données insuffisantes »), territoire sans # arrêt → NULL transit (« Non desservi »), et chaque score expose son # détail (details JSON) affiché sur la fiche. # # Honnêteté méthodologique (affichée sur /ka-scores) : # · distances à vol d'oiseau × 1,3 (facteur réseau usuel), pas un routage ; # · le Calme est une estimation d'environnement, pas une mesure sonore ; # · le Bike ignore le dénivelé (v1) ; # · barème versionné — tout changement recalcule l'ensemble du parc. # ----------------------------------------------------------------------------- from __future__ import annotations import json import math import sqlite3 import time from pathlib import Path from . import db from .environment import load_tile from .poi import TILE, _haversine_m, _tile_of VERSION = "2026.08-v1" DETOUR = 1.3 # vol d'oiseau → distance de marche estimée QUARTIER_DB = Path(__file__).resolve().parent.parent / "data" / "quartier.db" # --- barème Walk : (catégorie, poids, pleine note ≤ m, zéro au-delà de m) --- WALK_BAREME = [ ("epicerie", 3.0, 400, 1600), ("pharmacie", 2.0, 400, 1600), ("parc", 2.0, 300, 1200), ("cafe", 1.5, 300, 1200), ("ecole", 1.5, 500, 1600), ("clinique", 1.5, 600, 2400), ("garderie", 1.0, 500, 1600), ("depanneur", 1.0, 250, 1000), ("gym", 1.0, 500, 2000), ("bibliotheque", 1.0, 500, 2000), ] # --- barème Services : famille → (catégories, rayon m, saturation, poids) --- SERVICES_BAREME = [ ("commerces", ("epicerie", "depanneur", "cafe"), 1000, 15, 0.35), ("sante", ("pharmacie", "clinique", "hopital"), 3000, 8, 0.25), ("education", ("ecole", "garderie", "bibliotheque"), 3000, 8, 0.20), ("loisirs", ("gym", "parc"), 1000, 6, 0.20), ] # --- pondérations par défaut du score global (personnalisables client) --- GLOBAL_POIDS = {"walk": 0.30, "transit": 0.20, "bike": 0.15, "calme": 0.20, "services": 0.15} LABELS = [(85, "Exceptionnel"), (70, "Excellent"), (55, "Très bon"), (40, "Moyen"), (0, "Faible")] def label(score: float | None) -> str | None: if score is None: return None for seuil, lbl in LABELS: if score >= seuil: return lbl return "Faible" # --------------------------------------------------------------------------- # Index spatial en grille (cellules ~0,01° ≈ 1,1 km) — recherches locales # --------------------------------------------------------------------------- class Grid: def __init__(self, cell: float = 0.01) -> None: self.cell = cell self.cells: dict[tuple[int, int], list[tuple[float, float]]] = {} def add(self, lat: float, lng: float) -> None: key = (int(lat // self.cell), int(lng // self.cell)) self.cells.setdefault(key, []).append((lat, lng)) def near(self, lat: float, lng: float, radius_m: float): """Tous les points à ≤ radius_m (parcours des cellules voisines).""" r_lat = radius_m / 111000.0 r_lng = radius_m / (111000.0 * max(0.2, math.cos(math.radians(lat)))) span = int(max(r_lat, r_lng) // self.cell) + 1 cy, cx = int(lat // self.cell), int(lng // self.cell) for dy in range(-span, span + 1): for dx in range(-span, span + 1): for plat, plng in self.cells.get((cy + dy, cx + dx), ()): if abs(plat - lat) > r_lat or abs(plng - lng) > r_lng: continue d = _haversine_m(lat, lng, plat, plng) if d <= radius_m: yield d, plat, plng def nearest(self, lat: float, lng: float, radius_m: float) -> float | None: best = None for d, _la, _ln in self.near(lat, lng, radius_m): if best is None or d < best: best = d return best def count(self, lat: float, lng: float, radius_m: float) -> int: return sum(1 for _ in self.near(lat, lng, radius_m)) class TileIndex: """Grilles par classe pour une tuile d'environnement.""" def __init__(self, env: dict) -> None: self.pois: dict[str, Grid] = {} for cat, pts in env.get("pois", {}).items(): g = Grid() for lat, lng in pts: g.add(lat, lng) self.pois[cat] = g self.points: dict[str, Grid] = {} for cls, pts in env.get("points", {}).items(): g = Grid() for lat, lng in pts: g.add(lat, lng) self.points[cls] = g # lignes : nœuds DENSIFIÉS (≤ 60 m) dans une grille — la distance au # nœud le plus proche approxime alors la distance à la voie, même # sur les longs segments droits (autoroutes rurales) self.lines: dict[str, Grid] = {} self.cyclable_seglen: dict[tuple[int, int], float] = {} for cls, ways in env.get("lines", {}).items(): g = self.lines.setdefault(cls, Grid()) for way in ways: for i, (lat, lng) in enumerate(way): g.add(lat, lng) if i == 0: continue plat, plng = way[i - 1] seg = _haversine_m(plat, plng, lat, lng) if cls == "cyclable": mid_lat, mid_lng = (lat + plat) / 2, (lng + plng) / 2 key = (int(mid_lat // 0.01), int(mid_lng // 0.01)) self.cyclable_seglen[key] = ( self.cyclable_seglen.get(key, 0.0) + seg) if seg > 60: n = int(seg // 60) for j in range(1, n + 1): t = j / (n + 1) g.add(plat + (lat - plat) * t, plng + (lng - plng) * t) def cyclable_metres(self, lat: float, lng: float, radius_m: float) -> float: """Mètres de voies cyclables ~dans le rayon (somme par cellule).""" r_cells = int(radius_m / 1100) + 1 cy, cx = int(lat // 0.01), int(lng // 0.01) total = 0.0 for dy in range(-r_cells, r_cells + 1): for dx in range(-r_cells, r_cells + 1): total += self.cyclable_seglen.get((cy + dy, cx + dx), 0.0) return total # --------------------------------------------------------------------------- # Calcul des scores pour une coordonnée # --------------------------------------------------------------------------- def _decroissance(dist_m: float, pleine: float, zero: float) -> float: """1 en deçà de `pleine`, 0 au-delà de `zero`, linéaire entre les deux.""" if dist_m <= pleine: return 1.0 if dist_m >= zero: return 0.0 return (zero - dist_m) / (zero - pleine) def score_walk(idx: TileIndex, lat: float, lng: float) -> tuple[float | None, dict]: total_poids = sum(p for _c, p, _f, _z in WALK_BAREME) acquis = 0.0 cats = [] trouvees = 0 for cat, poids, pleine, zero in WALK_BAREME: g = idx.pois.get(cat) brut = g.nearest(lat, lng, zero / DETOUR + 200) if g else None if brut is None: cats.append({"cat": cat, "dist_m": None, "pts": 0.0}) continue marche = brut * DETOUR part = _decroissance(marche, pleine, zero) acquis += poids * part trouvees += 1 cats.append({"cat": cat, "dist_m": round(marche), "pts": round(part * 100)}) if trouvees < 2: return None, {"cats": cats, "raison": "moins de 2 commodités cartographiées"} # bonus de choix : plusieurs épiceries/cafés à ≤ 800 m de marche bonus = 0.0 for cat in ("epicerie", "cafe"): g = idx.pois.get(cat) if g and g.count(lat, lng, 800 / DETOUR) >= 3: bonus += 2.5 score = min(100.0, 100.0 * acquis / total_poids + bonus) return round(score, 1), {"cats": cats, "bonus_choix": bonus} def score_transit(idx: TileIndex, lat: float, lng: float, pmd_pct: float | None) -> tuple[float | None, dict]: g_bus, g_metro = idx.pois.get("bus"), idx.pois.get("metro") d_bus = g_bus.nearest(lat, lng, 900) if g_bus else None d_metro = g_metro.nearest(lat, lng, 1600) if g_metro else None prox = 0.0 if d_bus is not None: prox = max(prox, 100.0 * _decroissance(d_bus * DETOUR, 200, 900)) if d_metro is not None: prox = max(prox, 100.0 * _decroissance(d_metro * DETOUR, 600, 1800) * 1.15) prox = min(100.0, prox) detail = { "arret_bus_m": round(d_bus * DETOUR) if d_bus is not None else None, "station_metro_m": round(d_metro * DETOUR) if d_metro is not None else None, "pmd_percentile": round(pmd_pct) if pmd_pct is not None else None, } if d_bus is None and d_metro is None: if pmd_pct is None or pmd_pct <= 1: return None, {**detail, "raison": "aucun arrêt à distance de marche"} return round(pmd_pct * 0.5, 1), detail # desserte lointaine plausible if pmd_pct is None: return round(prox * 0.85, 1), detail # proximité seule, prudente # proximité de l'arrêt × qualité de desserte du secteur (PMD StatCan) return round(0.45 * prox + 0.55 * pmd_pct, 1), detail def score_bike(idx: TileIndex, lat: float, lng: float) -> tuple[float | None, dict]: km = idx.cyclable_metres(lat, lng, 1000) / 1000.0 infra = min(60.0, km * 11.0) # accessibilité des besoins quotidiens à vélo (seuils marche × 3) total_poids = sum(p for _c, p, _f, _z in WALK_BAREME) acquis = 0.0 trouvees = 0 for cat, poids, pleine, zero in WALK_BAREME: g = idx.pois.get(cat) brut = g.nearest(lat, lng, zero * 3 / DETOUR + 400) if g else None if brut is None: continue acquis += poids * _decroissance(brut * DETOUR, pleine * 3, zero * 3) trouvees += 1 if trouvees < 2 and km == 0: return None, {"raison": "réseau cyclable et commodités non cartographiés"} access = 40.0 * acquis / total_poids return round(min(100.0, infra + access), 1), { "km_cyclables_1km": round(km, 1), "note": "dénivelé non pris en compte (v1)", } def score_calme(idx: TileIndex, lat: float, lng: float) -> tuple[float, dict]: score = 88.0 sources = [] def penalite(cls: str, rayon: float, poids: float, nom: str, lignes: bool = True) -> None: nonlocal score g = idx.lines.get(cls) if lignes else idx.points.get(cls) d = g.nearest(lat, lng, rayon) if g else None if d is not None: p = poids * _decroissance(d, rayon * 0.08, rayon) if p > 0.5: score -= p sources.append({"source": nom, "dist_m": round(d), "pen": round(p, 1)}) penalite("autoroute", 800, 42, "autoroute") penalite("artere", 400, 22, "artère principale") penalite("rail", 500, 18, "voie ferrée") penalite("aeroport", 3000, 25, "aéroport/héliport", lignes=False) penalite("industriel", 600, 14, "zone industrielle", lignes=False) g_bar = idx.points.get("bar") n_bars = g_bar.count(lat, lng, 250) if g_bar else 0 if n_bars >= 2: p = min(12.0, 4.0 * (n_bars - 1)) score -= p sources.append({"source": f"{n_bars} bars/boîtes à moins de 250 m", "dist_m": None, "pen": round(p, 1)}) g_parc = idx.pois.get("parc") d_parc = g_parc.nearest(lat, lng, 700) if g_parc else None bonus = 0.0 if d_parc is not None: bonus = 8.0 if d_parc <= 300 else 4.0 score += bonus return round(max(0.0, min(100.0, score)), 1), { "sources_bruit": sources, "bonus_parc": bonus, "note": "estimation basée sur l'environnement, pas une mesure sonore", } def score_services(idx: TileIndex, lat: float, lng: float) -> tuple[float | None, dict]: total = 0.0 familles = {} n_cats = 0 for fam, cats, rayon, sat, poids in SERVICES_BAREME: n = sum((idx.pois.get(c).count(lat, lng, rayon) if idx.pois.get(c) else 0) for c in cats) n_cats += 1 if n > 0 else 0 part = min(1.0, math.log1p(n) / math.log1p(sat)) total += poids * part familles[fam] = n if n_cats == 0: return None, {"familles": familles, "raison": "aucun service cartographié"} return round(100.0 * total, 1), {"familles": familles} def score_global(scores: dict[str, float | None]) -> float | None: poids_total = 0.0 acquis = 0.0 for k, p in GLOBAL_POIDS.items(): if scores.get(k) is not None: poids_total += p acquis += p * scores[k] # type: ignore[operator] if poids_total < 0.5: # trop peu de composantes fiables return None return round(acquis / poids_total, 1) # --------------------------------------------------------------------------- # Recalcul du parc # --------------------------------------------------------------------------- def _pmd_transit_by_dauid() -> dict[str, float]: """Percentile PMD « transport collectif » par aire de diffusion.""" if not QUARTIER_DB.exists(): return {} qcon = sqlite3.connect(QUARTIER_DB) qcon.row_factory = sqlite3.Row try: return {r["dauid"]: r["prox_transport"] for r in qcon.execute( "SELECT dauid, prox_transport FROM da_pmd_pct" " WHERE prox_transport IS NOT NULL")} except sqlite3.OperationalError: return {} finally: qcon.close() def run(recompute_all: bool = False) -> dict: """Calcule les KA Scores de tous les immeubles couverts par env_tiles. Incrémental : seules les coordonnées sans score (ou d'une version de barème antérieure) sont recalculées, sauf `recompute_all`. """ con = db.connect() t0 = time.time() pmd = _pmd_transit_by_dauid() rows = con.execute( """SELECT ROUND(lat,4) la, ROUND(lng,4) ln, MAX(dauid) dauid FROM listings WHERE active=1 AND lat IS NOT NULL AND lng IS NOT NULL GROUP BY la, ln""").fetchall() existants = {r["coord_key"] for r in con.execute( "SELECT coord_key FROM kascores WHERE version=?", (VERSION,))} par_tuile: dict[tuple[int, int], list] = {} for r in rows: key = f"{r['la']},{r['ln']}" if not recompute_all and key in existants: continue par_tuile.setdefault(_tile_of(r["la"], r["ln"]), []).append( (key, r["la"], r["ln"], r["dauid"])) calcules = sans_tuile = 0 for tile, coords in sorted(par_tuile.items()): env = load_tile(con, *tile) if env is None: sans_tuile += len(coords) continue idx = TileIndex(env) for key, lat, lng, dauid in coords: pmd_pct = pmd.get(dauid) if dauid and dauid != "hors-zone" else None walk, d_walk = score_walk(idx, lat, lng) transit, d_transit = score_transit(idx, lat, lng, pmd_pct) bike, d_bike = score_bike(idx, lat, lng) calme, d_calme = score_calme(idx, lat, lng) services, d_services = score_services(idx, lat, lng) scores = {"walk": walk, "transit": transit, "bike": bike, "calme": calme, "services": services} glob = score_global(scores) details = { "walk": d_walk, "transit": d_transit, "bike": d_bike, "calme": d_calme, "services": d_services, "labels": {k: label(v) for k, v in scores.items()}, } con.execute( """INSERT OR REPLACE INTO kascores (coord_key, lat, lng, walk, transit, bike, calme, services, global, details, version, computed_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", (key, lat, lng, walk, transit, bike, calme, services, glob, json.dumps(details, ensure_ascii=False), VERSION, time.time())) calcules += 1 con.commit() print(f" tuile {tile[0]},{tile[1]} : {len(coords)} immeubles") # clé de jointure sur les annonces (nouvelles incluses) con.execute( """UPDATE listings SET coord_key = ROUND(lat,4) || ',' || ROUND(lng,4) WHERE lat IS NOT NULL AND lng IS NOT NULL AND (coord_key IS NULL OR coord_key != ROUND(lat,4) || ',' || ROUND(lng,4))""") con.commit() # calibration : distribution du score global sur le parc dist = [r["global"] for r in con.execute( "SELECT global FROM kascores WHERE global IS NOT NULL ORDER BY global")] deciles = ([round(dist[int(len(dist) * q / 10)]) for q in range(10)] + [round(dist[-1])]) if dist else [] con.close() return {"immeubles_calcules": calcules, "sans_tuile_env": sans_tuile, "scores_en_base": len(dist), "deciles_global": deciles, "version": VERSION, "duree_s": round(time.time() - t0, 1)} def stats() -> dict: """Couverture et distributions — page méthodologie + Stats Lou-Ka.""" con = db.connect() total = con.execute( """SELECT COUNT(*) c FROM listings WHERE active=1 AND published=1 AND dup_of IS NULL""").fetchone()["c"] scored = con.execute( """SELECT COUNT(*) c FROM listings l JOIN kascores k ON k.coord_key = l.coord_key WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL AND k.global IS NOT NULL""").fetchone()["c"] out: dict = {"annonces": total, "avec_score": scored, "couverture_pct": round(100.0 * scored / total, 1) if total else 0, "version": VERSION, "moyennes": {}} for col in ("walk", "transit", "bike", "calme", "services", "global"): r = con.execute( f"SELECT AVG({col}) m, COUNT({col}) n FROM kascores").fetchone() out["moyennes"][col] = {"moyenne": round(r["m"], 1) if r["m"] else None, "n": r["n"]} con.close() return out