Lou·Ka — tous les logements à louer du Québec, un seul endroit.
HTML 98.9%
Python 0.6%
1# -----------------------------------------------------------------------------2# Lou-Ka — Agrégateur de logements à louer (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# kascores.py : les KA Scores — famille de scores maison 0-100 par immeuble.5#6# KA Walk Score marchabilité (besoins quotidiens à pied)7# KA Transit Score desserte en transport collectif8# KA Bike Score praticité du vélo9# KA Calme Score tranquillité estimée du secteur10# KA Services Score richesse des services (1 km / 3 km)11# KA Score global moyenne pondérée (personnalisable côté client)12#13# Sources : OpenStreetMap (© contributeurs OSM, ODbL) via environment.py,14# mesures de proximité StatCan (PMD 2021, quartier.db) pour le volet15# fréquence/qualité du transport collectif. AUCUNE donnée inventée :16# secteur sans données → NULL (« Données insuffisantes »), territoire sans17# arrêt → NULL transit (« Non desservi »), et chaque score expose son18# détail (details JSON) affiché sur la fiche.19#20# Honnêteté méthodologique (affichée sur /ka-scores) :21# · distances à vol d'oiseau × 1,3 (facteur réseau usuel), pas un routage ;22# · le Calme est une estimation d'environnement, pas une mesure sonore ;23# · le Bike ignore le dénivelé (v1) ;24# · barème versionné — tout changement recalcule l'ensemble du parc.25# -----------------------------------------------------------------------------26from __future__ import annotations2728import json29import math30import sqlite331import time32from pathlib import Path3334from . import db35from .environment import load_tile36from .poi import TILE, _haversine_m, _tile_of3738VERSION = "2026.08-v1"39DETOUR = 1.3 # vol d'oiseau → distance de marche estimée40QUARTIER_DB = Path(__file__).resolve().parent.parent / "data" / "quartier.db"4142# --- barème Walk : (catégorie, poids, pleine note ≤ m, zéro au-delà de m) ---43WALK_BAREME = [44 ("epicerie", 3.0, 400, 1600),45 ("pharmacie", 2.0, 400, 1600),46 ("parc", 2.0, 300, 1200),47 ("cafe", 1.5, 300, 1200),48 ("ecole", 1.5, 500, 1600),49 ("clinique", 1.5, 600, 2400),50 ("garderie", 1.0, 500, 1600),51 ("depanneur", 1.0, 250, 1000),52 ("gym", 1.0, 500, 2000),53 ("bibliotheque", 1.0, 500, 2000),54]5556# --- barème Services : famille → (catégories, rayon m, saturation, poids) ---57SERVICES_BAREME = [58 ("commerces", ("epicerie", "depanneur", "cafe"), 1000, 15, 0.35),59 ("sante", ("pharmacie", "clinique", "hopital"), 3000, 8, 0.25),60 ("education", ("ecole", "garderie", "bibliotheque"), 3000, 8, 0.20),61 ("loisirs", ("gym", "parc"), 1000, 6, 0.20),62]6364# --- pondérations par défaut du score global (personnalisables client) ---65GLOBAL_POIDS = {"walk": 0.30, "transit": 0.20, "bike": 0.15,66 "calme": 0.20, "services": 0.15}6768LABELS = [(85, "Exceptionnel"), (70, "Excellent"), (55, "Très bon"),69 (40, "Moyen"), (0, "Faible")]707172def label(score: float | None) -> str | None:73 if score is None:74 return None75 for seuil, lbl in LABELS:76 if score >= seuil:77 return lbl78 return "Faible"798081# ---------------------------------------------------------------------------82# Index spatial en grille (cellules ~0,01° ≈ 1,1 km) — recherches locales83# ---------------------------------------------------------------------------8485class Grid:86 def __init__(self, cell: float = 0.01) -> None:87 self.cell = cell88 self.cells: dict[tuple[int, int], list[tuple[float, float]]] = {}8990 def add(self, lat: float, lng: float) -> None:91 key = (int(lat // self.cell), int(lng // self.cell))92 self.cells.setdefault(key, []).append((lat, lng))9394 def near(self, lat: float, lng: float, radius_m: float):95 """Tous les points à ≤ radius_m (parcours des cellules voisines)."""96 r_lat = radius_m / 111000.097 r_lng = radius_m / (111000.0 * max(0.2, math.cos(math.radians(lat))))98 span = int(max(r_lat, r_lng) // self.cell) + 199 cy, cx = int(lat // self.cell), int(lng // self.cell)100 for dy in range(-span, span + 1):101 for dx in range(-span, span + 1):102 for plat, plng in self.cells.get((cy + dy, cx + dx), ()):103 if abs(plat - lat) > r_lat or abs(plng - lng) > r_lng:104 continue105 d = _haversine_m(lat, lng, plat, plng)106 if d <= radius_m:107 yield d, plat, plng108109 def nearest(self, lat: float, lng: float, radius_m: float) -> float | None:110 best = None111 for d, _la, _ln in self.near(lat, lng, radius_m):112 if best is None or d < best:113 best = d114 return best115116 def count(self, lat: float, lng: float, radius_m: float) -> int:117 return sum(1 for _ in self.near(lat, lng, radius_m))118119120class TileIndex:121 """Grilles par classe pour une tuile d'environnement."""122123 def __init__(self, env: dict) -> None:124 self.pois: dict[str, Grid] = {}125 for cat, pts in env.get("pois", {}).items():126 g = Grid()127 for lat, lng in pts:128 g.add(lat, lng)129 self.pois[cat] = g130 self.points: dict[str, Grid] = {}131 for cls, pts in env.get("points", {}).items():132 g = Grid()133 for lat, lng in pts:134 g.add(lat, lng)135 self.points[cls] = g136 # lignes : nœuds DENSIFIÉS (≤ 60 m) dans une grille — la distance au137 # nœud le plus proche approxime alors la distance à la voie, même138 # sur les longs segments droits (autoroutes rurales)139 self.lines: dict[str, Grid] = {}140 self.cyclable_seglen: dict[tuple[int, int], float] = {}141 for cls, ways in env.get("lines", {}).items():142 g = self.lines.setdefault(cls, Grid())143 for way in ways:144 for i, (lat, lng) in enumerate(way):145 g.add(lat, lng)146 if i == 0:147 continue148 plat, plng = way[i - 1]149 seg = _haversine_m(plat, plng, lat, lng)150 if cls == "cyclable":151 mid_lat, mid_lng = (lat + plat) / 2, (lng + plng) / 2152 key = (int(mid_lat // 0.01), int(mid_lng // 0.01))153 self.cyclable_seglen[key] = (154 self.cyclable_seglen.get(key, 0.0) + seg)155 if seg > 60:156 n = int(seg // 60)157 for j in range(1, n + 1):158 t = j / (n + 1)159 g.add(plat + (lat - plat) * t, plng + (lng - plng) * t)160161 def cyclable_metres(self, lat: float, lng: float, radius_m: float) -> float:162 """Mètres de voies cyclables ~dans le rayon (somme par cellule)."""163 r_cells = int(radius_m / 1100) + 1164 cy, cx = int(lat // 0.01), int(lng // 0.01)165 total = 0.0166 for dy in range(-r_cells, r_cells + 1):167 for dx in range(-r_cells, r_cells + 1):168 total += self.cyclable_seglen.get((cy + dy, cx + dx), 0.0)169 return total170171172# ---------------------------------------------------------------------------173# Calcul des scores pour une coordonnée174# ---------------------------------------------------------------------------175176def _decroissance(dist_m: float, pleine: float, zero: float) -> float:177 """1 en deçà de `pleine`, 0 au-delà de `zero`, linéaire entre les deux."""178 if dist_m <= pleine:179 return 1.0180 if dist_m >= zero:181 return 0.0182 return (zero - dist_m) / (zero - pleine)183184185def score_walk(idx: TileIndex, lat: float, lng: float) -> tuple[float | None, dict]:186 total_poids = sum(p for _c, p, _f, _z in WALK_BAREME)187 acquis = 0.0188 cats = []189 trouvees = 0190 for cat, poids, pleine, zero in WALK_BAREME:191 g = idx.pois.get(cat)192 brut = g.nearest(lat, lng, zero / DETOUR + 200) if g else None193 if brut is None:194 cats.append({"cat": cat, "dist_m": None, "pts": 0.0})195 continue196 marche = brut * DETOUR197 part = _decroissance(marche, pleine, zero)198 acquis += poids * part199 trouvees += 1200 cats.append({"cat": cat, "dist_m": round(marche), "pts": round(part * 100)})201 if trouvees < 2:202 return None, {"cats": cats, "raison": "moins de 2 commodités cartographiées"}203 # bonus de choix : plusieurs épiceries/cafés à ≤ 800 m de marche204 bonus = 0.0205 for cat in ("epicerie", "cafe"):206 g = idx.pois.get(cat)207 if g and g.count(lat, lng, 800 / DETOUR) >= 3:208 bonus += 2.5209 score = min(100.0, 100.0 * acquis / total_poids + bonus)210 return round(score, 1), {"cats": cats, "bonus_choix": bonus}211212213def score_transit(idx: TileIndex, lat: float, lng: float,214 pmd_pct: float | None) -> tuple[float | None, dict]:215 g_bus, g_metro = idx.pois.get("bus"), idx.pois.get("metro")216 d_bus = g_bus.nearest(lat, lng, 900) if g_bus else None217 d_metro = g_metro.nearest(lat, lng, 1600) if g_metro else None218 prox = 0.0219 if d_bus is not None:220 prox = max(prox, 100.0 * _decroissance(d_bus * DETOUR, 200, 900))221 if d_metro is not None:222 prox = max(prox, 100.0 * _decroissance(d_metro * DETOUR, 600, 1800) * 1.15)223 prox = min(100.0, prox)224 detail = {225 "arret_bus_m": round(d_bus * DETOUR) if d_bus is not None else None,226 "station_metro_m": round(d_metro * DETOUR) if d_metro is not None else None,227 "pmd_percentile": round(pmd_pct) if pmd_pct is not None else None,228 }229 if d_bus is None and d_metro is None:230 if pmd_pct is None or pmd_pct <= 1:231 return None, {**detail, "raison": "aucun arrêt à distance de marche"}232 return round(pmd_pct * 0.5, 1), detail # desserte lointaine plausible233 if pmd_pct is None:234 return round(prox * 0.85, 1), detail # proximité seule, prudente235 # proximité de l'arrêt × qualité de desserte du secteur (PMD StatCan)236 return round(0.45 * prox + 0.55 * pmd_pct, 1), detail237238239def score_bike(idx: TileIndex, lat: float, lng: float) -> tuple[float | None, dict]:240 km = idx.cyclable_metres(lat, lng, 1000) / 1000.0241 infra = min(60.0, km * 11.0)242 # accessibilité des besoins quotidiens à vélo (seuils marche × 3)243 total_poids = sum(p for _c, p, _f, _z in WALK_BAREME)244 acquis = 0.0245 trouvees = 0246 for cat, poids, pleine, zero in WALK_BAREME:247 g = idx.pois.get(cat)248 brut = g.nearest(lat, lng, zero * 3 / DETOUR + 400) if g else None249 if brut is None:250 continue251 acquis += poids * _decroissance(brut * DETOUR, pleine * 3, zero * 3)252 trouvees += 1253 if trouvees < 2 and km == 0:254 return None, {"raison": "réseau cyclable et commodités non cartographiés"}255 access = 40.0 * acquis / total_poids256 return round(min(100.0, infra + access), 1), {257 "km_cyclables_1km": round(km, 1),258 "note": "dénivelé non pris en compte (v1)",259 }260261262def score_calme(idx: TileIndex, lat: float, lng: float) -> tuple[float, dict]:263 score = 88.0264 sources = []265266 def penalite(cls: str, rayon: float, poids: float, nom: str,267 lignes: bool = True) -> None:268 nonlocal score269 g = idx.lines.get(cls) if lignes else idx.points.get(cls)270 d = g.nearest(lat, lng, rayon) if g else None271 if d is not None:272 p = poids * _decroissance(d, rayon * 0.08, rayon)273 if p > 0.5:274 score -= p275 sources.append({"source": nom, "dist_m": round(d), "pen": round(p, 1)})276277 penalite("autoroute", 800, 42, "autoroute")278 penalite("artere", 400, 22, "artère principale")279 penalite("rail", 500, 18, "voie ferrée")280 penalite("aeroport", 3000, 25, "aéroport/héliport", lignes=False)281 penalite("industriel", 600, 14, "zone industrielle", lignes=False)282283 g_bar = idx.points.get("bar")284 n_bars = g_bar.count(lat, lng, 250) if g_bar else 0285 if n_bars >= 2:286 p = min(12.0, 4.0 * (n_bars - 1))287 score -= p288 sources.append({"source": f"{n_bars} bars/boîtes à moins de 250 m",289 "dist_m": None, "pen": round(p, 1)})290291 g_parc = idx.pois.get("parc")292 d_parc = g_parc.nearest(lat, lng, 700) if g_parc else None293 bonus = 0.0294 if d_parc is not None:295 bonus = 8.0 if d_parc <= 300 else 4.0296 score += bonus297 return round(max(0.0, min(100.0, score)), 1), {298 "sources_bruit": sources, "bonus_parc": bonus,299 "note": "estimation basée sur l'environnement, pas une mesure sonore",300 }301302303def score_services(idx: TileIndex, lat: float, lng: float) -> tuple[float | None, dict]:304 total = 0.0305 familles = {}306 n_cats = 0307 for fam, cats, rayon, sat, poids in SERVICES_BAREME:308 n = sum((idx.pois.get(c).count(lat, lng, rayon) if idx.pois.get(c) else 0)309 for c in cats)310 n_cats += 1 if n > 0 else 0311 part = min(1.0, math.log1p(n) / math.log1p(sat))312 total += poids * part313 familles[fam] = n314 if n_cats == 0:315 return None, {"familles": familles, "raison": "aucun service cartographié"}316 return round(100.0 * total, 1), {"familles": familles}317318319def score_global(scores: dict[str, float | None]) -> float | None:320 poids_total = 0.0321 acquis = 0.0322 for k, p in GLOBAL_POIDS.items():323 if scores.get(k) is not None:324 poids_total += p325 acquis += p * scores[k] # type: ignore[operator]326 if poids_total < 0.5: # trop peu de composantes fiables327 return None328 return round(acquis / poids_total, 1)329330331# ---------------------------------------------------------------------------332# Recalcul du parc333# ---------------------------------------------------------------------------334335def _pmd_transit_by_dauid() -> dict[str, float]:336 """Percentile PMD « transport collectif » par aire de diffusion."""337 if not QUARTIER_DB.exists():338 return {}339 qcon = sqlite3.connect(QUARTIER_DB)340 qcon.row_factory = sqlite3.Row341 try:342 return {r["dauid"]: r["prox_transport"] for r in qcon.execute(343 "SELECT dauid, prox_transport FROM da_pmd_pct"344 " WHERE prox_transport IS NOT NULL")}345 except sqlite3.OperationalError:346 return {}347 finally:348 qcon.close()349350351def run(recompute_all: bool = False) -> dict:352 """Calcule les KA Scores de tous les immeubles couverts par env_tiles.353354 Incrémental : seules les coordonnées sans score (ou d'une version de355 barème antérieure) sont recalculées, sauf `recompute_all`.356 """357 con = db.connect()358 t0 = time.time()359 pmd = _pmd_transit_by_dauid()360361 rows = con.execute(362 """SELECT ROUND(lat,4) la, ROUND(lng,4) ln,363 MAX(dauid) dauid364 FROM listings365 WHERE active=1 AND lat IS NOT NULL AND lng IS NOT NULL366 GROUP BY la, ln""").fetchall()367368 existants = {r["coord_key"] for r in con.execute(369 "SELECT coord_key FROM kascores WHERE version=?", (VERSION,))}370371 par_tuile: dict[tuple[int, int], list] = {}372 for r in rows:373 key = f"{r['la']},{r['ln']}"374 if not recompute_all and key in existants:375 continue376 par_tuile.setdefault(_tile_of(r["la"], r["ln"]), []).append(377 (key, r["la"], r["ln"], r["dauid"]))378379 calcules = sans_tuile = 0380 for tile, coords in sorted(par_tuile.items()):381 env = load_tile(con, *tile)382 if env is None:383 sans_tuile += len(coords)384 continue385 idx = TileIndex(env)386 for key, lat, lng, dauid in coords:387 pmd_pct = pmd.get(dauid) if dauid and dauid != "hors-zone" else None388 walk, d_walk = score_walk(idx, lat, lng)389 transit, d_transit = score_transit(idx, lat, lng, pmd_pct)390 bike, d_bike = score_bike(idx, lat, lng)391 calme, d_calme = score_calme(idx, lat, lng)392 services, d_services = score_services(idx, lat, lng)393 scores = {"walk": walk, "transit": transit, "bike": bike,394 "calme": calme, "services": services}395 glob = score_global(scores)396 details = {397 "walk": d_walk, "transit": d_transit, "bike": d_bike,398 "calme": d_calme, "services": d_services,399 "labels": {k: label(v) for k, v in scores.items()},400 }401 con.execute(402 """INSERT OR REPLACE INTO kascores403 (coord_key, lat, lng, walk, transit, bike, calme, services,404 global, details, version, computed_at)405 VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",406 (key, lat, lng, walk, transit, bike, calme, services, glob,407 json.dumps(details, ensure_ascii=False), VERSION, time.time()))408 calcules += 1409 con.commit()410 print(f" tuile {tile[0]},{tile[1]} : {len(coords)} immeubles")411412 # clé de jointure sur les annonces (nouvelles incluses)413 con.execute(414 """UPDATE listings415 SET coord_key = ROUND(lat,4) || ',' || ROUND(lng,4)416 WHERE lat IS NOT NULL AND lng IS NOT NULL417 AND (coord_key IS NULL418 OR coord_key != ROUND(lat,4) || ',' || ROUND(lng,4))""")419 con.commit()420421 # calibration : distribution du score global sur le parc422 dist = [r["global"] for r in con.execute(423 "SELECT global FROM kascores WHERE global IS NOT NULL ORDER BY global")]424 deciles = ([round(dist[int(len(dist) * q / 10)]) for q in range(10)] +425 [round(dist[-1])]) if dist else []426 con.close()427 return {"immeubles_calcules": calcules, "sans_tuile_env": sans_tuile,428 "scores_en_base": len(dist), "deciles_global": deciles,429 "version": VERSION, "duree_s": round(time.time() - t0, 1)}430431432def stats() -> dict:433 """Couverture et distributions — page méthodologie + Stats Lou-Ka."""434 con = db.connect()435 total = con.execute(436 """SELECT COUNT(*) c FROM listings437 WHERE active=1 AND published=1 AND dup_of IS NULL""").fetchone()["c"]438 scored = con.execute(439 """SELECT COUNT(*) c FROM listings l JOIN kascores k440 ON k.coord_key = l.coord_key441 WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL442 AND k.global IS NOT NULL""").fetchone()["c"]443 out: dict = {"annonces": total, "avec_score": scored,444 "couverture_pct": round(100.0 * scored / total, 1) if total else 0,445 "version": VERSION, "moyennes": {}}446 for col in ("walk", "transit", "bike", "calme", "services", "global"):447 r = con.execute(448 f"SELECT AVG({col}) m, COUNT({col}) n FROM kascores").fetchone()449 out["moyennes"][col] = {"moyenne": round(r["m"], 1) if r["m"] else None,450 "n": r["n"]}451 con.close()452 return out453