""" build_noise_zones.py ==================== Builds a noise-proximity layer for Nord (59) using OSM road/rail geometries fetched from the Overpass API. Major roads and railways are stored as LineString features with an estimated LDEN value based on road/rail type. LDEN estimates (simplified, at roadside distance): motorway → 75 dB trunk → 70 dB primary → 65 dB secondary → 60 dB railway → 70 dB These are approximate reference values. Official LDEN cartography (EU Directive 2002/49/CE) for Nord is not available from public WFS/WMS endpoints. Usage: python build_noise_zones.py python build_noise_zones.py --tile-size 0.5 """ from __future__ import annotations import argparse import hashlib import json import logging import urllib.parse import urllib.request from datetime import datetime, timezone import pymysql from domain.core.mysql_db import get_connection, reset_table logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s — %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("build_noise_zones") # Nord (59) bounding box NORD_BBOX = (2.0, 50.0, 3.9, 51.1) # min_lng, min_lat, max_lng, max_lat OVERPASS_URL = "https://overpass-api.de/api/interpreter" # Estimated typical LDEN at roadside per road/rail type LDEN_BY_TYPE: dict[str, int] = { "motorway": 75, "motorway_link": 70, "trunk": 70, "trunk_link": 65, "primary": 65, "primary_link": 60, "secondary": 60, "secondary_link": 55, "railway": 70, } # Human-readable source label stored in source_type column SOURCE_LABEL: dict[str, str] = { "motorway": "routier", "motorway_link": "routier", "trunk": "routier", "trunk_link": "routier", "primary": "routier", "primary_link": "routier", "secondary": "routier", "secondary_link": "routier", "railway": "ferroviaire", } _OVERPASS_QUERY = """\ [out:json][timeout:120][bbox:{min_lat},{min_lng},{max_lat},{max_lng}]; ( way["highway"~"^(motorway|motorway_link|trunk|trunk_link|primary|primary_link|secondary|secondary_link)$"]; way["railway"="rail"]["usage"~"^(main|branch)$"]; ); out geom; """ _DDL = """ CREATE TABLE noise_zones ( id INT NOT NULL AUTO_INCREMENT, source_type VARCHAR(20) NOT NULL, lden_min INT NOT NULL, geom_type VARCHAR(20) NOT NULL, coordinates MEDIUMTEXT NOT NULL, coords_hash VARCHAR(64) NOT NULL, bbox_min_lng DOUBLE NOT NULL, bbox_min_lat DOUBLE NOT NULL, bbox_max_lng DOUBLE NOT NULL, bbox_max_lat DOUBLE NOT NULL, fetched_at VARCHAR(40) NOT NULL, PRIMARY KEY (id), UNIQUE KEY uq_coords_hash (coords_hash) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ _IDX_BBOX = ( "CREATE INDEX idx_noise_bbox ON noise_zones" "(bbox_min_lng, bbox_max_lng, bbox_min_lat, bbox_max_lat)" ) def _fetch() -> list[dict]: min_lng, min_lat, max_lng, max_lat = NORD_BBOX query = _OVERPASS_QUERY.format( min_lat=min_lat, min_lng=min_lng, max_lat=max_lat, max_lng=max_lng ) logger.info("Overpass query — Nord bbox …") body = urllib.parse.urlencode({"data": query}).encode("utf-8") req = urllib.request.Request( OVERPASS_URL, data=body, headers={ "Content-Type": "application/x-www-form-urlencoded", "User-Agent": "agent-immobilier/1.0", }, ) with urllib.request.urlopen(req, timeout=150) as resp: raw = resp.read() elements = json.loads(raw).get("elements", []) logger.info("%d ways reçus", len(elements)) return elements def _parse(elements: list[dict]) -> list[tuple]: rows: list[tuple] = [] fetched_at = datetime.now(timezone.utc).isoformat() skipped = 0 for el in elements: if el.get("type") != "way": skipped += 1 continue tags = el.get("tags") or {} geometry = el.get("geometry") or [] if not geometry: skipped += 1 continue # Determine road/rail type highway = tags.get("highway", "") railway = tags.get("railway", "") osm_type = highway if highway else ("railway" if railway else "") if osm_type not in LDEN_BY_TYPE: skipped += 1 continue # Build LineString coordinates [lng, lat] coords = [[pt["lon"], pt["lat"]] for pt in geometry if "lon" in pt and "lat" in pt] if len(coords) < 2: skipped += 1 continue # Compute bounding box lngs = [c[0] for c in coords] lats = [c[1] for c in coords] coords_json = json.dumps(coords, separators=(",", ":")) coords_hash = hashlib.sha256(coords_json.encode()).hexdigest() rows.append(( SOURCE_LABEL[osm_type], LDEN_BY_TYPE[osm_type], "LineString", coords_json, coords_hash, min(lngs), min(lats), max(lngs), max(lats), fetched_at, )) by_src: dict[str, int] = {} by_lden: dict[int, int] = {} for row in rows: by_src[row[0]] = by_src.get(row[0], 0) + 1 by_lden[row[1]] = by_lden.get(row[1], 0) + 1 logger.info( "%d axes retenus | %d ignorés | sources: %s | niveaux: %s", len(rows), skipped, " ".join(f"{k}:{v}" for k, v in sorted(by_src.items())), " ".join(f"{k}dB:{v}" for k, v in sorted(by_lden.items())), ) return rows def _save(rows: list[tuple]) -> None: conn = get_connection() try: reset_table(conn, "noise_zones", _DDL, indexes=[_IDX_BBOX]) with conn.cursor() as cur: cur.executemany( """INSERT IGNORE INTO noise_zones (source_type, lden_min, geom_type, coordinates, coords_hash, bbox_min_lng, bbox_min_lat, bbox_max_lng, bbox_max_lat, fetched_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", rows, ) conn.commit() finally: conn.close() logger.info("Table noise_zones : %d lignes insérées -> MySQL", len(rows)) def main() -> None: parser = argparse.ArgumentParser( description="Construit la table noise_zones depuis OSM (Overpass API)" ) parser.parse_args() elements = _fetch() rows = _parse(elements) _save(rows) print() print("=" * 52) print(f" Axes insérés : {len(rows):,}") print(f" Base : MySQL (MYSQL_DATABASE env var)") print() print(" Sources : OSM road/rail geometry (Overpass API)") print(" LDEN estimés (valeurs de référence à l'abord)") print(" motorway=75 | trunk=70 | primary=65 | secondary=60") print("=" * 52) if __name__ == "__main__": main()