#!/usr/bin/env python3 from __future__ import annotations import json import csv import time import difflib import math import unicodedata from concurrent.futures import ThreadPoolExecutor, as_completed from io import StringIO from pathlib import Path from typing import Iterable from urllib.parse import urlencode from urllib.request import Request, urlopen ROOT_DIR = Path(__file__).resolve().parents[1] DATA_DIR = ROOT_DIR / "data" TERRITORY_DIR = DATA_DIR / "territories" API_BASE_URL = "https://geo.api.gouv.fr" REGIONS_GEOJSON_URL = ( "https://raw.githubusercontent.com/gregoiredavid/france-geojson/master/" "regions-version-simplifiee.geojson" ) DEPARTMENTS_GEOJSON_URL = ( "https://raw.githubusercontent.com/gregoiredavid/france-geojson/master/" "departements-version-simplifiee.geojson" ) REGION_COMMUNES_GEOJSON_BASE_URL = ( "https://raw.githubusercontent.com/gregoiredavid/france-geojson/master/regions" ) COMMUNE_TERRITORIAL_COMPETENCE_URL = ( "https://www.data.gouv.fr/api/1/datasets/r/c53cd4d4-4623-4772-9b8c-bc72a9cdf4c2" ) GENDARMERIE_PUBLIC_UNITS_URL = ( "https://www.data.gouv.fr/api/1/datasets/r/17320fe6-a896-4686-93e6-502be2ad23f2" ) SERVICE_PUBLIC_ANNUAIRE_API_URL = ( "https://api-lannuaire.service-public.gouv.fr/api/explore/v2.1/catalog/datasets/" "api-lannuaire-administration/records" ) REQUEST_HEADERS = { "User-Agent": "GiePlaces data builder", "Accept": "application/json", } REGION_ENTITY_TYPE = "region" DEPARTMENT_ENTITY_TYPE = "departement" COMMUNE_ENTITY_TYPE = "commune" OVERSEAS_TERRITORIES = [ {"code": "971", "label": "Guadeloupe", "shortLabel": "Guadeloupe"}, {"code": "972", "label": "Martinique", "shortLabel": "Martinique"}, {"code": "973", "label": "Guyane", "shortLabel": "Guyane"}, {"code": "974", "label": "La Réunion", "shortLabel": "La Réunion"}, {"code": "976", "label": "Mayotte", "shortLabel": "Mayotte"}, {"code": "975", "label": "Saint-Pierre-et-Miquelon", "shortLabel": "SPM"}, {"code": "977", "label": "Saint-Barthélemy", "shortLabel": "Saint-Barth"}, {"code": "978", "label": "Saint-Martin", "shortLabel": "Saint-Martin"}, {"code": "986", "label": "Wallis-et-Futuna", "shortLabel": "Wallis"}, {"code": "987", "label": "Polynésie française", "shortLabel": "Polynésie"}, {"code": "988", "label": "Nouvelle-Calédonie", "shortLabel": "Nouvelle-Calédonie"}, ] POLYNESIE_RETAINED_COMMUNES = { "Arue", "Faaa", "Hitiaa O Te Ra", "Mahina", "Moorea-Maiao", "Paea", "Papara", "Papeete", "Pirae", "Punaauia", "Taiarapu-Est", "Taiarapu-Ouest", "Teva I Uta", } POLYNESIE_TRIMMED_COMMUNES = { "Taiarapu-Est", } SPECIAL_TERRITORY_SIMPLIFICATION = { "polynesie-francaise": { "minDistance": 0.0018, "minArea": 0.0000018, }, "nouvelle-caledonie": { "minDistance": 0.0014, "minArea": 0.0000012, }, } SPECIAL_TERRITORY_LAYOUTS = { "polynesie-francaise": { "marquises": {"target": (-151.2, -16.4), "scale": 1.0}, "societe": {"target": (-150.8, -16.8), "scale": 1.0}, "tuamotu-ouest": {"target": (-150.4, -16.8), "scale": 1.0}, "tuamotu-centre": {"target": (-150.0, -16.8), "scale": 1.0}, "tuamotu-est": {"target": (-149.6, -16.8), "scale": 1.0}, "australes": {"target": (-150.2, -17.2), "scale": 1.0}, "gambier": {"target": (-149.8, -17.2), "scale": 1.0}, }, "wallis-et-futuna": { "uvea": {"target": (-176.22, -13.32), "scale": 1.0}, "futuna": {"target": (-176.68, -13.88), "scale": 1.0}, }, } SPECIAL_PLANAR_LAYOUT_ROWS = {} PLANAR_TERRITORY_COMPACTION = { "polynesie-francaise": { "factor": 1.0, }, "nouvelle-caledonie": { "factor": 0.6, }, } SPECIAL_TERRITORY_QUANTIZATION = { "polynesie-francaise": 4, "nouvelle-caledonie": 4, } PREPROJECTED_TERRITORIES = { "polynesie-francaise", "nouvelle-caledonie", } SPECIAL_GROUP_DISTANCE_REDUCTION = {} SPECIAL_GROUP_MIN_GAP_FACTOR = { "polynesie-francaise": 0.0005, } SPECIAL_GROUP_PACK_BY_FEATURE = { "polynesie-francaise": True, } SPECIAL_GROUP_EFFECTIVE_BOUNDS_FACTOR = { "polynesie-francaise": 0.18, } DAY_LABELS = [ ("lundi", "Lun"), ("mardi", "Mar"), ("mercredi", "Mer"), ("jeudi", "Jeu"), ("vendredi", "Ven"), ("samedi", "Sam"), ("dimanche", "Dim"), ("jours_feries", "Fériés"), ] def fetch_json(url: str, retries: int = 3) -> dict | list: last_error = None for attempt in range(1, retries + 1): try: request = Request(url, headers=REQUEST_HEADERS) with urlopen(request, timeout=60) as response: return json.load(response) except Exception as exc: # pragma: no cover - network failures are not deterministic last_error = exc if attempt == retries: raise time.sleep(attempt * 0.75) raise RuntimeError(f"Unable to fetch {url}: {last_error}") def fetch_text(url: str, retries: int = 3) -> str: last_error = None for attempt in range(1, retries + 1): try: request = Request(url, headers=REQUEST_HEADERS) with urlopen(request, timeout=60) as response: return response.read().decode("utf-8-sig") except Exception as exc: # pragma: no cover - network failures are not deterministic last_error = exc if attempt == retries: raise time.sleep(attempt * 0.75) raise RuntimeError(f"Unable to fetch {url}: {last_error}") def slugify(value: str) -> str: normalized = unicodedata.normalize("NFD", str(value)) ascii_value = "".join(char for char in normalized if unicodedata.category(char) != "Mn") chunks = [] for char in ascii_value.lower(): chunks.append(char if char.isalnum() else "-") slug = "".join(chunks) while "--" in slug: slug = slug.replace("--", "-") return slug.strip("-") def normalize_sort_text(value: str) -> str: normalized = unicodedata.normalize("NFD", str(value)) ascii_value = "".join(char for char in normalized if unicodedata.category(char) != "Mn") lowered = ascii_value.lower() cleaned = [] for char in lowered: cleaned.append(char if char.isalnum() else " ") return " ".join("".join(cleaned).split()) def department_territory_key(department_code: str) -> str: return f"departement-{slugify(department_code)}" def region_territory_key(region_name: str) -> str: return f"region-{slugify(region_name)}" def get_department_metadata() -> list[dict]: url = f"{API_BASE_URL}/departements?fields=code,nom,codeRegion,zone&format=json" return fetch_json(url) def get_region_metadata() -> list[dict]: url = f"{API_BASE_URL}/regions?fields=code,nom&format=json" return fetch_json(url) def get_commune_postal_codes_by_code() -> dict[str, list[str]]: url = f"{API_BASE_URL}/communes?fields=code,codesPostaux&format=json" payload = fetch_json(url) return { item["code"]: sorted({str(postal_code) for postal_code in item.get("codesPostaux", []) if postal_code}) for item in payload } def build_service_key(institution: str, service_id: str, service_label: str, commune_code: str) -> str: normalized_institution = (institution or "UNK").upper() normalized_service_id = str(service_id or "").strip() normalized_service_label = str(service_label or "").strip() if normalized_institution == "GN" and normalized_service_id: return f"gn:{normalized_service_id}" if normalized_institution == "PN" and normalized_service_label: return f"pn:{slugify(normalized_service_label)}" if normalized_service_label: return f"{slugify(normalized_institution)}:{slugify(normalized_service_label)}" return f"unk:{commune_code}" def normalize_unit_match_text(value: str) -> str: normalized = unicodedata.normalize("NFD", str(value)) ascii_value = "".join(char for char in normalized if unicodedata.category(char) != "Mn") lowered = ascii_value.lower().replace("'", " ") cleaned = [] for char in lowered: cleaned.append(char if char.isalnum() else " ") return " ".join("".join(cleaned).split()) def get_label_similarity(left: str, right: str) -> float: return difflib.SequenceMatcher(None, normalize_unit_match_text(left), normalize_unit_match_text(right)).ratio() def normalize_service_label(institution: str, service_label: str) -> str: normalized_label = str(service_label or "").strip() normalized_institution = (institution or "").upper() if normalized_label: return normalized_label if normalized_institution == "PN": return "Police nationale" if normalized_institution == "GN": return "Gendarmerie nationale" return "Compétence territoriale non renseignée" def get_commune_territorial_competence_data() -> tuple[dict[str, dict], dict[str, dict], dict[str, set[str]]]: payload = fetch_text(COMMUNE_TERRITORIAL_COMPETENCE_URL) reader = csv.DictReader(StringIO(payload), delimiter=";") rows_by_commune_code: dict[str, list[dict]] = {} service_registry: dict[str, dict] = {} commune_codes_by_service_key: dict[str, set[str]] = {} for row in reader: commune_code = str(row.get("code_commune", "")).strip() if not commune_code: continue institution = str(row.get("institution", "")).strip().upper() service_id = str(row.get("id_service", "")).strip() service_label = normalize_service_label(institution, row.get("service", "")) service_key = build_service_key(institution, service_id, service_label, commune_code) service_registry[service_key] = { "i": institution or "UNK", "l": service_label, } commune_codes_by_service_key.setdefault(service_key, set()).add(commune_code) rows_by_commune_code.setdefault(commune_code, []).append( { "institution": institution or "UNK", "serviceKey": service_key, "serviceLabel": service_label, } ) normalized_competence_by_commune_code: dict[str, dict] = {} for commune_code, rows in rows_by_commune_code.items(): unique_rows = [] seen_rows = set() for row in rows: row_key = (row["institution"], row["serviceKey"], row["serviceLabel"]) if row_key in seen_rows: continue seen_rows.add(row_key) unique_rows.append(row) if len(unique_rows) == 1: normalized_competence_by_commune_code[commune_code] = { "i": unique_rows[0]["institution"], "u": unique_rows[0]["serviceKey"], } continue institutions = {row["institution"] for row in unique_rows} if "GN" in institutions and "PN" in institutions: institution = "MX" elif "PN" in institutions: institution = "PN" elif "GN" in institutions: institution = "GN" else: institution = "UNK" normalized_competence_by_commune_code[commune_code] = { "i": institution, "m": " / ".join(row["serviceLabel"] for row in unique_rows), "v": [row["serviceKey"] for row in unique_rows], } return normalized_competence_by_commune_code, service_registry, commune_codes_by_service_key def format_public_unit_hours(row: dict) -> str: day_chunks = [] for day_key, day_label in DAY_LABELS: ranges = [] for slot in range(1, 4): start = str(row.get(f"{day_key}_plage{slot}_debut", "")).strip() end = str(row.get(f"{day_key}_plage{slot}_fin", "")).strip() if start and end: ranges.append(f"{start}-{end}") if ranges: day_chunks.append(f"{day_label} {' / '.join(ranges)}") return "; ".join(day_chunks) def parse_annuaire_sequence(value: object) -> list[dict]: if isinstance(value, list): return [item for item in value if isinstance(item, dict)] if isinstance(value, str): raw_value = value.strip() if not raw_value: return [] try: parsed = json.loads(raw_value) except json.JSONDecodeError: return [] if isinstance(parsed, list): return [item for item in parsed if isinstance(item, dict)] return [] def fetch_annuaire_records(where: str) -> list[dict]: results = [] offset = 0 limit = 100 while True: query = urlencode({"where": where, "limit": limit, "offset": offset}) payload = fetch_json(f"{SERVICE_PUBLIC_ANNUAIRE_API_URL}?{query}") batch = payload.get("results", []) if not batch: break results.extend(batch) offset += len(batch) if offset >= payload.get("total_count", 0): break return results def format_annuaire_hours(opening_ranges: object) -> str: parsed_ranges = parse_annuaire_sequence(opening_ranges) day_chunks = [] for entry in parsed_ranges: start_day = str(entry.get("nom_jour_debut", "")).strip() end_day = str(entry.get("nom_jour_fin", "")).strip() day_label = start_day if not end_day or end_day == start_day else f"{start_day}-{end_day}" ranges = [] for slot in range(1, 3): start = str(entry.get(f"valeur_heure_debut_{slot}", "")).strip() end = str(entry.get(f"valeur_heure_fin_{slot}", "")).strip() if start and end: ranges.append(f"{start[:5]}-{end[:5]}") if day_label and ranges: day_chunks.append(f"{day_label} {' / '.join(ranges)}") return "; ".join(day_chunks) def get_annuaire_address_value(addresses: object) -> str: parsed_addresses = parse_annuaire_sequence(addresses) if not parsed_addresses: return "" address = parsed_addresses[0] chunks = [ str(address.get("numero_voie", "")).strip(), str(address.get("complement1", "")).strip(), str(address.get("complement2", "")).strip(), str(address.get("service_distribution", "")).strip(), str(address.get("code_postal", "")).strip(), str(address.get("nom_commune", "")).strip(), ] return " ".join(chunk for chunk in chunks if chunk) def get_annuaire_phone_value(phones: object) -> str: parsed_phones = parse_annuaire_sequence(phones) if not parsed_phones: return "" return str(parsed_phones[0].get("valeur", "")).strip() def get_gendarmerie_public_units_by_service_key(service_registry: dict[str, dict]) -> dict[str, dict]: payload = fetch_text(GENDARMERIE_PUBLIC_UNITS_URL) reader = csv.DictReader(StringIO(payload), delimiter=";") units_by_service_key: dict[str, dict] = {} for row in reader: service_id = str(row.get("identifiant_public_unite", "")).strip() if not service_id: continue service_key = f"gn:{service_id}" service = service_registry.get(service_key) if not service or service.get("i") != "GN": continue longitude = str(row.get("geocodage_x_GPS", "")).strip() latitude = str(row.get("geocodage_y_GPS", "")).strip() if not longitude or not latitude: continue units_by_service_key[service_key] = { "s": service_key, "i": "GN", "n": service.get("l") or str(row.get("service", "")).strip() or f"Unité {service_id}", "x": round(float(longitude), 5), "y": round(float(latitude), 5), "c": str(row.get("code_commune_insee", "")).strip(), "m": str(row.get("commune", "")).strip(), "a": str(row.get("adresse_geographique", "")).strip(), "t": str(row.get("telephone", "")).strip(), "h": format_public_unit_hours(row), "u": str(row.get("url", "")).strip(), } return units_by_service_key def get_police_public_units_by_service_key( service_registry: dict[str, dict], commune_codes_by_service_key: dict[str, set[str]], ) -> dict[str, dict]: source_rows = fetch_annuaire_records("partenaire = 'commissariat'") annuaire_records = [ { "serviceKey": build_service_key("PN", "", row.get("nom", ""), ""), "name": str(row.get("nom", "")).strip(), "communeCode": str(row.get("code_insee_commune", "")).strip(), "hostCommune": str((parse_annuaire_sequence(row.get("adresse")) or [{}])[0].get("nom_commune", "")).strip(), "address": get_annuaire_address_value(row.get("adresse")), "phone": get_annuaire_phone_value(row.get("telephone")), "hours": format_annuaire_hours(row.get("plage_ouverture")), "url": str(row.get("url_service_public", "")).strip(), "longitude": str((parse_annuaire_sequence(row.get("adresse")) or [{}])[0].get("longitude", "")).strip(), "latitude": str((parse_annuaire_sequence(row.get("adresse")) or [{}])[0].get("latitude", "")).strip(), } for row in source_rows ] units_by_service_key: dict[str, dict] = {} police_service_keys = [ service_key for service_key, service in service_registry.items() if service.get("i") == "PN" ] for service_key in police_service_keys: service_label = service_registry[service_key]["l"] commune_codes = commune_codes_by_service_key.get(service_key, set()) commune_candidates = [ record for record in annuaire_records if record["communeCode"] and record["communeCode"] in commune_codes ] best_match = None best_ratio = 0.0 for record in commune_candidates or annuaire_records: similarity = get_label_similarity(service_label, record["name"]) if similarity > best_ratio: best_ratio = similarity best_match = record if not best_match: continue minimum_ratio = 0.45 if commune_candidates else 0.84 if best_ratio < minimum_ratio: continue if not best_match["longitude"] or not best_match["latitude"]: continue units_by_service_key[service_key] = { "s": service_key, "i": "PN", "n": service_label, "x": round(float(best_match["longitude"]), 5), "y": round(float(best_match["latitude"]), 5), "c": best_match["communeCode"], "m": best_match["hostCommune"], "a": best_match["address"], "t": best_match["phone"], "h": best_match["hours"], "u": best_match["url"], } return units_by_service_key def get_department_code_from_commune_code(commune_code: str) -> str: code = str(commune_code) if code.startswith("2A") or code.startswith("2B"): return code[:2] if code.startswith("97") or code.startswith("98"): return code[:3] return code[:2] def quantize_pair(pair: Iterable[float], precision: int = 5) -> list[float]: return [round(float(pair[0]), precision), round(float(pair[1]), precision)] def quantize_geometry(geometry: dict, precision: int = 5) -> dict: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates") if geometry_type == "Polygon": return { "type": "Polygon", "coordinates": [ [quantize_pair(point, precision) for point in ring] for ring in coordinates ], } if geometry_type == "MultiPolygon": return { "type": "MultiPolygon", "coordinates": [ [ [quantize_pair(point, precision) for point in ring] for ring in polygon ] for polygon in coordinates ], } raise ValueError(f"Unsupported geometry type: {geometry_type}") def iter_geometry_points(geometry: dict) -> Iterable[list[float]]: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates", []) if geometry_type == "Polygon": for ring in coordinates: for point in ring: yield point return if geometry_type == "MultiPolygon": for polygon in coordinates: for ring in polygon: for point in ring: yield point return raise ValueError(f"Unsupported geometry type: {geometry_type}") def get_geometry_center(geometry: dict) -> tuple[float, float]: min_longitude = float("inf") min_latitude = float("inf") max_longitude = float("-inf") max_latitude = float("-inf") for longitude, latitude in iter_geometry_points(geometry): min_longitude = min(min_longitude, longitude) min_latitude = min(min_latitude, latitude) max_longitude = max(max_longitude, longitude) max_latitude = max(max_latitude, latitude) return ( (min_longitude + max_longitude) / 2, (min_latitude + max_latitude) / 2, ) def simplify_ring(ring: list[list[float]], min_distance: float, min_area: float) -> list[list[float]]: if len(ring) < 4: return ring is_closed = ring[0] == ring[-1] points = ring[:-1] if is_closed else ring[:] if len(points) < 3: return ring filtered = [points[0]] min_distance_sq = min_distance * min_distance for point in points[1:]: delta_longitude = point[0] - filtered[-1][0] delta_latitude = point[1] - filtered[-1][1] if delta_longitude * delta_longitude + delta_latitude * delta_latitude >= min_distance_sq: filtered.append(point) if len(filtered) < 3: filtered = points[:] changed = True while changed and len(filtered) > 3: changed = False reduced = [filtered[0]] for index in range(1, len(filtered) - 1): previous_point = reduced[-1] point = filtered[index] next_point = filtered[index + 1] cross_product = abs( (point[0] - previous_point[0]) * (next_point[1] - previous_point[1]) - (point[1] - previous_point[1]) * (next_point[0] - previous_point[0]) ) if cross_product <= min_area: changed = True continue reduced.append(point) reduced.append(filtered[-1]) filtered = reduced if is_closed: filtered.append(filtered[0]) return filtered def simplify_geometry(geometry: dict, min_distance: float, min_area: float) -> dict: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates", []) if geometry_type == "Polygon": return { "type": "Polygon", "coordinates": [simplify_ring(ring, min_distance, min_area) for ring in coordinates], } if geometry_type == "MultiPolygon": return { "type": "MultiPolygon", "coordinates": [ [simplify_ring(ring, min_distance, min_area) for ring in polygon] for polygon in coordinates ], } raise ValueError(f"Unsupported geometry type: {geometry_type}") def get_ring_area(ring: list[list[float]]) -> float: if len(ring) < 3: return 0.0 area = 0.0 previous_x, previous_y = ring[-1] for point_x, point_y in ring: area += previous_x * point_y - point_x * previous_y previous_x, previous_y = point_x, point_y return abs(area) / 2 def get_polygon_area(polygon: list[list[list[float]]]) -> float: if not polygon: return 0.0 outer_area = get_ring_area(polygon[0]) inner_area = sum(get_ring_area(ring) for ring in polygon[1:]) return max(0.0, outer_area - inner_area) def get_polygon_center(polygon: list[list[list[float]]]) -> tuple[float, float]: min_x = math.inf min_y = math.inf max_x = -math.inf max_y = -math.inf for ring in polygon: for point_x, point_y in ring: min_x = min(min_x, point_x) min_y = min(min_y, point_y) max_x = max(max_x, point_x) max_y = max(max_y, point_y) return (min_x + max_x) / 2, (min_y + max_y) / 2 def trim_special_commune_polygons(territory_key: str, commune_name: str, geometry: dict) -> dict: if ( territory_key != "polynesie-francaise" or commune_name not in POLYNESIE_TRIMMED_COMMUNES or geometry.get("type") != "MultiPolygon" ): return geometry polygons = geometry.get("coordinates", []) if len(polygons) <= 1: return geometry polygon_areas = [get_polygon_area(polygon) for polygon in polygons] main_index = max(range(len(polygons)), key=lambda index: polygon_areas[index]) main_polygon = polygons[main_index] return { "type": "MultiPolygon", "coordinates": [main_polygon], } def get_special_layout_group_id(territory_key: str, longitude: float, latitude: float) -> str | None: if territory_key == "wallis-et-futuna": return "uvea" if longitude > -177.0 else "futuna" if territory_key == "polynesie-francaise": if latitude > -12.0: return "marquises" if latitude <= -21.2: return "gambier" if longitude >= -140.5 else "australes" if longitude < -148.5: return "societe" if longitude < -145.8: return "tuamotu-ouest" if longitude < -143.5: return "tuamotu-centre" return "tuamotu-est" return None def get_layout_source_centers(territory_key: str, features: list[dict]) -> dict[str, tuple[float, float]]: layout_config = SPECIAL_TERRITORY_LAYOUTS.get(territory_key) if not layout_config: return {} grouped_centers: dict[str, list[tuple[float, float]]] = {} for feature in features: center_longitude, center_latitude = get_geometry_center(feature["geometry"]) group_id = get_special_layout_group_id(territory_key, center_longitude, center_latitude) if not group_id: continue grouped_centers.setdefault(group_id, []).append((center_longitude, center_latitude)) source_centers = {} for group_id, centers in grouped_centers.items(): source_centers[group_id] = ( sum(center[0] for center in centers) / len(centers), sum(center[1] for center in centers) / len(centers), ) return source_centers def transform_point_for_layout( territory_key: str, longitude: float, latitude: float, source_centers: dict[str, tuple[float, float]], ) -> list[float]: group_id = get_special_layout_group_id(territory_key, longitude, latitude) layout_config = SPECIAL_TERRITORY_LAYOUTS.get(territory_key, {}) group_layout = layout_config.get(group_id or "") source_center = source_centers.get(group_id or "") if not group_layout or not source_center: return [round(float(longitude), 5), round(float(latitude), 5)] target_longitude, target_latitude = group_layout["target"] scale = group_layout["scale"] return [ round(target_longitude + scale * (longitude - source_center[0]), 5), round(target_latitude + scale * (latitude - source_center[1]), 5), ] def transform_geometry_for_layout( territory_key: str, geometry: dict, source_centers: dict[str, tuple[float, float]], ) -> dict: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates", []) if geometry_type == "Polygon": return { "type": "Polygon", "coordinates": [ [ transform_point_for_layout(territory_key, point[0], point[1], source_centers) for point in ring ] for ring in coordinates ], } if geometry_type == "MultiPolygon": return { "type": "MultiPolygon", "coordinates": [ [ [ transform_point_for_layout(territory_key, point[0], point[1], source_centers) for point in ring ] for ring in polygon ] for polygon in coordinates ], } raise ValueError(f"Unsupported geometry type: {geometry_type}") def project_point_to_planar_mercator(point: list[float]) -> list[float]: longitude = math.radians(float(point[0])) latitude = math.radians(float(point[1])) clamped_latitude = max(min(latitude, math.radians(89.9999)), math.radians(-89.9999)) mercator_y = math.log(math.tan(math.pi / 4 + clamped_latitude / 2)) return [round(longitude, 6), round(-mercator_y, 6)] def project_geometry_to_planar_mercator(geometry: dict) -> dict: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates", []) if geometry_type == "Polygon": return { "type": "Polygon", "coordinates": [ [project_point_to_planar_mercator(point) for point in ring] for ring in coordinates ], } if geometry_type == "MultiPolygon": return { "type": "MultiPolygon", "coordinates": [ [ [project_point_to_planar_mercator(point) for point in ring] for ring in polygon ] for polygon in coordinates ], } raise ValueError(f"Unsupported geometry type: {geometry_type}") def project_units_to_planar_mercator(units: list[dict]) -> list[dict]: projected_units = [] for unit in units: projected_longitude, projected_latitude = project_point_to_planar_mercator([unit["x"], unit["y"]]) projected_unit = dict(unit) projected_unit["x"] = projected_longitude projected_unit["y"] = projected_latitude projected_units.append(projected_unit) return projected_units def compact_planar_point(point_x: float, point_y: float, origin_x: float, origin_y: float, compact_factor: float) -> list[float]: return [ round(origin_x + (point_x - origin_x) * compact_factor, 6), round(origin_y + (point_y - origin_y) * compact_factor, 6), ] def compact_planar_geometry( geometry: dict, origin_x: float, origin_y: float, compact_factor: float, ) -> dict: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates", []) if geometry_type == "Polygon": return { "type": "Polygon", "coordinates": [ [compact_planar_point(point_x, point_y, origin_x, origin_y, compact_factor) for point_x, point_y in ring] for ring in coordinates ], } if geometry_type == "MultiPolygon": return { "type": "MultiPolygon", "coordinates": [ [ [ compact_planar_point(point_x, point_y, origin_x, origin_y, compact_factor) for point_x, point_y in ring ] for ring in polygon ] for polygon in coordinates ], } raise ValueError(f"Unsupported geometry type: {geometry_type}") def compact_planar_features( projected_features: list[dict], compact_factor: float, ) -> tuple[list[dict], tuple[float, float] | None]: if compact_factor >= 1 or not projected_features: return projected_features, None bounds_min_x = math.inf bounds_min_y = math.inf bounds_max_x = -math.inf bounds_max_y = -math.inf for feature in projected_features: min_x, min_y, max_x, max_y = get_geometry_bounds(feature["geometry"]) bounds_min_x = min(bounds_min_x, min_x) bounds_min_y = min(bounds_min_y, min_y) bounds_max_x = max(bounds_max_x, max_x) bounds_max_y = max(bounds_max_y, max_y) if not math.isfinite(bounds_min_x) or not math.isfinite(bounds_max_x): return projected_features, None origin_x = (bounds_min_x + bounds_max_x) / 2 origin_y = (bounds_min_y + bounds_max_y) / 2 compacted_features = [ { "type": "Feature", "properties": dict(feature["properties"]), "geometry": compact_planar_geometry(feature["geometry"], origin_x, origin_y, compact_factor), } for feature in projected_features ] return compacted_features, (origin_x, origin_y) def compact_planar_units(units: list[dict], origin_x: float, origin_y: float, compact_factor: float) -> list[dict]: if compact_factor >= 1 or not units: return units compacted_units = [] for unit in units: compact_unit = dict(unit) compact_unit["x"] = round(origin_x + (float(unit["x"]) - origin_x) * compact_factor, 6) compact_unit["y"] = round(origin_y + (float(unit["y"]) - origin_y) * compact_factor, 6) compacted_units.append(compact_unit) return compacted_units def get_geometry_bounds(geometry: dict) -> tuple[float, float, float, float]: min_x = math.inf min_y = math.inf max_x = -math.inf max_y = -math.inf coordinates = geometry.get("coordinates", []) polygons = coordinates if geometry.get("type") == "MultiPolygon" else [coordinates] for polygon in polygons: for ring in polygon: for point_x, point_y in ring: min_x = min(min_x, point_x) min_y = min(min_y, point_y) max_x = max(max_x, point_x) max_y = max(max_y, point_y) return min_x, min_y, max_x, max_y def translate_geometry(geometry: dict, offset_x: float, offset_y: float) -> dict: geometry_type = geometry.get("type") coordinates = geometry.get("coordinates", []) if geometry_type == "Polygon": return { "type": "Polygon", "coordinates": [ [[round(point_x + offset_x, 6), round(point_y + offset_y, 6)] for point_x, point_y in ring] for ring in coordinates ], } if geometry_type == "MultiPolygon": return { "type": "MultiPolygon", "coordinates": [ [ [[round(point_x + offset_x, 6), round(point_y + offset_y, 6)] for point_x, point_y in ring] for ring in polygon ] for polygon in coordinates ], } raise ValueError(f"Unsupported geometry type: {geometry_type}") def build_planar_group_offsets(territory_key: str, group_bounds: dict[str, tuple[float, float, float, float]]) -> dict[str, tuple[float, float]]: layout_rows = SPECIAL_PLANAR_LAYOUT_ROWS.get(territory_key) if not layout_rows or not group_bounds: return {} widths = [max_x - min_x for min_x, _min_y, max_x, _max_y in group_bounds.values()] heights = [max_y - min_y for _min_x, min_y, _max_x, max_y in group_bounds.values()] gap_x = (max(widths) if widths else 0) * 0.18 gap_y = (max(heights) if heights else 0) * 0.28 remaining_groups = [group_id for group_id in group_bounds if group_id not in {item for row in layout_rows for item in row}] rows = [row[:] for row in layout_rows] if remaining_groups: rows.append(remaining_groups) offsets: dict[str, tuple[float, float]] = {} cursor_y = 0.0 for row in rows: row_groups = [group_id for group_id in row if group_id in group_bounds] if not row_groups: continue row_height = max(group_bounds[group_id][3] - group_bounds[group_id][1] for group_id in row_groups) cursor_x = 0.0 for group_id in row_groups: min_x, min_y, max_x, max_y = group_bounds[group_id] width = max_x - min_x height = max_y - min_y target_min_x = cursor_x target_min_y = cursor_y + (row_height - height) / 2 offsets[group_id] = (target_min_x - min_x, target_min_y - min_y) cursor_x += width + gap_x cursor_y += row_height + gap_y return offsets def apply_special_planar_layout( territory_key: str, raw_features: list[dict], projected_features: list[dict], raw_units: list[dict], projected_units: list[dict], ) -> tuple[list[dict], list[dict]]: if territory_key not in SPECIAL_PLANAR_LAYOUT_ROWS: return projected_features, projected_units projected_group_bounds: dict[str, list[float]] = {} feature_group_by_code: dict[str, str] = {} for raw_feature, projected_feature in zip(raw_features, projected_features): center_longitude, center_latitude = get_geometry_center(raw_feature["geometry"]) group_id = get_special_layout_group_id(territory_key, center_longitude, center_latitude) if not group_id: continue feature_group_by_code[raw_feature["properties"]["c"]] = group_id min_x, min_y, max_x, max_y = get_geometry_bounds(projected_feature["geometry"]) current_bounds = projected_group_bounds.get(group_id) if current_bounds is None: projected_group_bounds[group_id] = [min_x, min_y, max_x, max_y] else: current_bounds[0] = min(current_bounds[0], min_x) current_bounds[1] = min(current_bounds[1], min_y) current_bounds[2] = max(current_bounds[2], max_x) current_bounds[3] = max(current_bounds[3], max_y) offsets = build_planar_group_offsets( territory_key, { group_id: (bounds[0], bounds[1], bounds[2], bounds[3]) for group_id, bounds in projected_group_bounds.items() }, ) if not offsets: return projected_features, projected_units translated_features = [] for raw_feature, projected_feature in zip(raw_features, projected_features): group_id = feature_group_by_code.get(raw_feature["properties"]["c"]) offset_x, offset_y = offsets.get(group_id, (0.0, 0.0)) translated_features.append( { "type": "Feature", "properties": dict(projected_feature["properties"]), "geometry": translate_geometry(projected_feature["geometry"], offset_x, offset_y), } ) translated_units = [] for raw_unit, projected_unit in zip(raw_units, projected_units): group_id = get_special_layout_group_id(territory_key, raw_unit["x"], raw_unit["y"]) offset_x, offset_y = offsets.get(group_id, (0.0, 0.0)) translated_unit = dict(projected_unit) translated_unit["x"] = round(float(translated_unit["x"]) + offset_x, 6) translated_unit["y"] = round(float(translated_unit["y"]) + offset_y, 6) translated_units.append(translated_unit) return translated_features, translated_units def reduce_group_distances_without_overlap( territory_key: str, raw_features: list[dict], projected_features: list[dict], raw_units: list[dict], projected_units: list[dict], reduction_ratio: float, ) -> tuple[list[dict], list[dict]]: if reduction_ratio <= 0 or not projected_features: return projected_features, projected_units clamped_reduction_ratio = min(max(float(reduction_ratio), 0.0), 1.0) distance_factor = 1.0 - clamped_reduction_ratio feature_group_by_code: dict[str, str] = {} group_bounds: dict[str, list[float]] = {} group_centers: dict[str, tuple[float, float]] = {} pack_by_feature = SPECIAL_GROUP_PACK_BY_FEATURE.get(territory_key, False) for raw_feature, projected_feature in zip(raw_features, projected_features): center_longitude, center_latitude = get_geometry_center(raw_feature["geometry"]) feature_code = str(raw_feature["properties"]["c"]) group_id = feature_code if pack_by_feature else get_special_layout_group_id( territory_key, center_longitude, center_latitude ) if not group_id: continue feature_group_by_code[feature_code] = group_id min_x, min_y, max_x, max_y = get_geometry_bounds(projected_feature["geometry"]) current_bounds = group_bounds.get(group_id) if current_bounds is None: group_bounds[group_id] = [min_x, min_y, max_x, max_y] else: current_bounds[0] = min(current_bounds[0], min_x) current_bounds[1] = min(current_bounds[1], min_y) current_bounds[2] = max(current_bounds[2], max_x) current_bounds[3] = max(current_bounds[3], max_y) if not group_bounds: return projected_features, projected_units min_all_x = min(bounds[0] for bounds in group_bounds.values()) min_all_y = min(bounds[1] for bounds in group_bounds.values()) max_all_x = max(bounds[2] for bounds in group_bounds.values()) max_all_y = max(bounds[3] for bounds in group_bounds.values()) global_center_x = (min_all_x + max_all_x) / 2 global_center_y = (min_all_y + max_all_y) / 2 max_dimension = 0.0 group_layout = {} for group_id, bounds in group_bounds.items(): min_x, min_y, max_x, max_y = bounds center_x = (min_x + max_x) / 2 center_y = (min_y + max_y) / 2 width = max_x - min_x height = max_y - min_y max_dimension = max(max_dimension, width, height) target_x = global_center_x + (center_x - global_center_x) * distance_factor target_y = global_center_y + (center_y - global_center_y) * distance_factor group_centers[group_id] = (center_x, center_y) group_layout[group_id] = { "center_x": target_x, "center_y": target_y, "target_x": target_x, "target_y": target_y, "width": width, "height": height, } group_gap_factor = SPECIAL_GROUP_MIN_GAP_FACTOR.get(territory_key, 0.04) effective_bounds_factor = SPECIAL_GROUP_EFFECTIVE_BOUNDS_FACTOR.get(territory_key, 1.0) gap = max_dimension * group_gap_factor group_ids = list(group_layout.keys()) for _ in range(180): moved = False for index in range(len(group_ids)): left_group = group_layout[group_ids[index]] for right_index in range(index + 1, len(group_ids)): right_group = group_layout[group_ids[right_index]] delta_x = right_group["center_x"] - left_group["center_x"] delta_y = right_group["center_y"] - left_group["center_y"] required_x = ( (left_group["width"] + right_group["width"]) * effective_bounds_factor / 2 + gap ) required_y = ( (left_group["height"] + right_group["height"]) * effective_bounds_factor / 2 + gap ) overlap_x = required_x - abs(delta_x) overlap_y = required_y - abs(delta_y) if overlap_x <= 0 or overlap_y <= 0: continue moved = True if overlap_x < overlap_y: direction_x = 1 if delta_x >= 0 else -1 shift = overlap_x / 2 left_group["center_x"] -= direction_x * shift right_group["center_x"] += direction_x * shift else: direction_y = 1 if delta_y >= 0 else -1 shift = overlap_y / 2 left_group["center_y"] -= direction_y * shift right_group["center_y"] += direction_y * shift for group in group_layout.values(): group["center_x"] += (group["target_x"] - group["center_x"]) * 0.08 group["center_y"] += (group["target_y"] - group["center_y"]) * 0.08 if not moved: break for _ in range(120): moved = False for index in range(len(group_ids)): left_group = group_layout[group_ids[index]] for right_index in range(index + 1, len(group_ids)): right_group = group_layout[group_ids[right_index]] delta_x = right_group["center_x"] - left_group["center_x"] delta_y = right_group["center_y"] - left_group["center_y"] required_x = ( (left_group["width"] + right_group["width"]) * effective_bounds_factor / 2 + gap ) required_y = ( (left_group["height"] + right_group["height"]) * effective_bounds_factor / 2 + gap ) overlap_x = required_x - abs(delta_x) overlap_y = required_y - abs(delta_y) if overlap_x <= 0 or overlap_y <= 0: continue moved = True if overlap_x < overlap_y: direction_x = 1 if delta_x >= 0 else -1 shift = overlap_x / 2 left_group["center_x"] -= direction_x * shift right_group["center_x"] += direction_x * shift else: direction_y = 1 if delta_y >= 0 else -1 shift = overlap_y / 2 left_group["center_y"] -= direction_y * shift right_group["center_y"] += direction_y * shift if not moved: break group_offsets = {} for group_id, group in group_layout.items(): center_x, center_y = group_centers[group_id] group_offsets[group_id] = (group["center_x"] - center_x, group["center_y"] - center_y) translated_features = [] for raw_feature, projected_feature in zip(raw_features, projected_features): group_id = feature_group_by_code.get(raw_feature["properties"]["c"]) offset_x, offset_y = group_offsets.get(group_id, (0.0, 0.0)) translated_features.append( { "type": "Feature", "properties": dict(projected_feature["properties"]), "geometry": translate_geometry(projected_feature["geometry"], offset_x, offset_y), } ) translated_units = [] for raw_unit, projected_unit in zip(raw_units, projected_units): unit_commune_code = str(raw_unit.get("c", "")) group_id = unit_commune_code if pack_by_feature and unit_commune_code else get_special_layout_group_id( territory_key, raw_unit["x"], raw_unit["y"] ) offset_x, offset_y = group_offsets.get(group_id, (0.0, 0.0)) translated_unit = dict(projected_unit) translated_unit["x"] = round(float(translated_unit["x"]) + offset_x, 6) translated_unit["y"] = round(float(translated_unit["y"]) + offset_y, 6) translated_units.append(translated_unit) return translated_features, translated_units def optimize_special_territory_features(territory_key: str, features: list[dict]) -> list[dict]: simplification = SPECIAL_TERRITORY_SIMPLIFICATION.get(territory_key) source_centers = get_layout_source_centers(territory_key, features) optimized_features = [] precision = SPECIAL_TERRITORY_QUANTIZATION.get(territory_key, 5) for feature in features: geometry = feature["geometry"] commune_name = feature["properties"].get("n", "") if simplification: geometry = simplify_geometry( geometry, simplification["minDistance"], simplification["minArea"], ) geometry = trim_special_commune_polygons(territory_key, commune_name, geometry) if source_centers: geometry = transform_geometry_for_layout(territory_key, geometry, source_centers) optimized_features.append( { "type": "Feature", "properties": dict(feature["properties"]), "geometry": quantize_geometry(geometry, precision), } ) return optimized_features def optimize_special_territory_units( territory_key: str, units: list[dict], source_centers: dict[str, tuple[float, float]], ) -> list[dict]: if not source_centers: return [dict(unit) for unit in units] optimized_units = [] for unit in units: optimized_longitude, optimized_latitude = transform_point_for_layout( territory_key, unit["x"], unit["y"], source_centers, ) optimized_unit = dict(unit) optimized_unit["x"] = optimized_longitude optimized_unit["y"] = optimized_latitude optimized_units.append(optimized_unit) return optimized_units def fetch_department_commune_features( department_code: str, competence_by_commune_code: dict[str, dict] | None = None, ) -> list[dict]: query = urlencode( { "codeDepartement": department_code, "fields": "nom,code,codeDepartement,codeRegion,centre,contour", "format": "geojson", "geometry": "contour", } ) url = f"{API_BASE_URL}/communes?{query}" payload = fetch_json(url) features = payload.get("features", []) normalized_features = [] for feature in features: properties = feature["properties"] competence = (competence_by_commune_code or {}).get(properties["code"], {}) normalized_properties = { "c": properties["code"], "n": properties["nom"], "d": properties["codeDepartement"], "r": properties["codeRegion"], "k": COMMUNE_ENTITY_TYPE, } if competence.get("i"): normalized_properties["i"] = competence["i"] if competence.get("u"): normalized_properties["u"] = competence["u"] if competence.get("m"): normalized_properties["m"] = competence["m"] if competence.get("v"): normalized_properties["v"] = competence["v"] normalized_features.append( { "type": "Feature", "properties": normalized_properties, "geometry": quantize_geometry(feature["geometry"]), } ) return normalized_features def fetch_region_commune_features( territory: dict, competence_by_commune_code: dict[str, dict] | None = None, ) -> list[dict]: region_slug = territory["key"].removeprefix("region-") url = ( f"{REGION_COMMUNES_GEOJSON_BASE_URL}/{region_slug}/" f"communes-{region_slug}.geojson" ) payload = fetch_json(url) features = payload.get("features", []) normalized_features = [] for feature in features: properties = feature["properties"] code = properties["code"] competence = (competence_by_commune_code or {}).get(code, {}) normalized_properties = { "c": code, "n": properties["nom"], "d": get_department_code_from_commune_code(code), "r": territory["regionCode"], "k": COMMUNE_ENTITY_TYPE, } if competence.get("i"): normalized_properties["i"] = competence["i"] if competence.get("u"): normalized_properties["u"] = competence["u"] if competence.get("m"): normalized_properties["m"] = competence["m"] if competence.get("v"): normalized_properties["v"] = competence["v"] normalized_features.append( { "type": "Feature", "properties": normalized_properties, "geometry": quantize_geometry(feature["geometry"]), } ) return normalized_features def fetch_france_region_features( region_metadata_by_code: dict[str, dict], metro_region_codes: list[str], region_key_by_code: dict[str, str], ) -> list[dict]: payload = fetch_json(REGIONS_GEOJSON_URL) features_by_code = {} for feature in payload.get("features", []): code = feature["properties"]["code"] if code not in region_metadata_by_code or code not in region_key_by_code: continue metadata = region_metadata_by_code[code] features_by_code[code] = { "type": "Feature", "properties": { "c": code, "n": metadata["nom"], "d": "", "r": code, "k": REGION_ENTITY_TYPE, "x": region_key_by_code[code], }, "geometry": quantize_geometry(feature["geometry"]), } return [features_by_code[code] for code in metro_region_codes if code in features_by_code] def fetch_metropolitan_department_features( department_metadata_by_code: dict[str, dict], metro_department_codes: list[str], ) -> list[dict]: payload = fetch_json(DEPARTMENTS_GEOJSON_URL) features_by_code = {} for feature in payload.get("features", []): code = feature["properties"]["code"] if code not in department_metadata_by_code: continue metadata = department_metadata_by_code[code] features_by_code[code] = { "type": "Feature", "properties": { "c": code, "n": metadata["nom"], "d": code, "r": metadata["codeRegion"], "k": DEPARTMENT_ENTITY_TYPE, "x": department_territory_key(code), }, "geometry": quantize_geometry(feature["geometry"]), } return [features_by_code[code] for code in metro_department_codes if code in features_by_code] def build_index_entries( features: list[dict], territory_key: str, commune_postal_codes_by_code: dict[str, list[str]], is_hidden: bool = False, ) -> list[dict]: entries = [] for feature in features: properties = feature["properties"] entry = { "c": properties["c"], "n": properties["n"], "t": territory_key, "k": properties["k"], } if properties["k"] == COMMUNE_ENTITY_TYPE: entry["p"] = commune_postal_codes_by_code.get(properties["c"], []) if is_hidden: entry["h"] = 1 entries.append(entry) return entries def write_json(path: Path, payload: dict | list) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as file: json.dump(payload, file, ensure_ascii=False, separators=(",", ":")) def clear_generated_files() -> None: TERRITORY_DIR.mkdir(parents=True, exist_ok=True) for path in TERRITORY_DIR.glob("*.json"): path.unlink() for path in [DATA_DIR / "communes-index.json", DATA_DIR / "territories.json"]: if path.exists(): path.unlink() def build_france_territory(metro_department_codes: list[str]) -> dict: return { "key": "france", "label": "France", "shortLabel": "France", "entityType": REGION_ENTITY_TYPE, "entityLabelSingular": "région de gendarmerie", "entityLabelPlural": "régions de gendarmerie", "departmentCodes": metro_department_codes, "navVisible": True, } def build_region_territories(metro_regions: list[dict], department_codes_by_region: dict[str, list[str]]) -> list[dict]: territories = [] for region in metro_regions: territories.append( { "key": region_territory_key(region["nom"]), "label": region["nom"], "shortLabel": region["nom"], "entityType": DEPARTMENT_ENTITY_TYPE, "entityLabelSingular": "groupement", "entityLabelPlural": "groupements", "departmentCodes": department_codes_by_region.get(region["code"], []), "regionCode": region["code"], "navVisible": True, } ) return territories def build_department_territories(metro_departments: list[dict], region_key_by_code: dict[str, str]) -> list[dict]: territories = [] for department in metro_departments: territories.append( { "key": department_territory_key(department["code"]), "label": department["nom"], "shortLabel": department["nom"], "entityType": COMMUNE_ENTITY_TYPE, "entityLabelSingular": "commune", "entityLabelPlural": "communes", "departmentCodes": [department["code"]], "regionCode": department["codeRegion"], "parentTerritoryKey": region_key_by_code[department["codeRegion"]], "groupementView": True, "navVisible": False, } ) return territories def build_overseas_territories() -> list[dict]: territories = [] for territory in OVERSEAS_TERRITORIES: territories.append( { "key": slugify(territory["label"]), "label": territory["label"], "shortLabel": territory["shortLabel"], "entityType": COMMUNE_ENTITY_TYPE, "entityLabelSingular": "commune", "entityLabelPlural": "communes", "departmentCodes": [territory["code"]], "groupementView": True, "navVisible": True, } ) return territories def build_data() -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) TERRITORY_DIR.mkdir(parents=True, exist_ok=True) clear_generated_files() department_metadata = get_department_metadata() region_metadata = get_region_metadata() commune_postal_codes_by_code = get_commune_postal_codes_by_code() competence_by_commune_code, service_registry, commune_codes_by_service_key = get_commune_territorial_competence_data() public_units_by_service_key = get_gendarmerie_public_units_by_service_key(service_registry) public_units_by_service_key.update( get_police_public_units_by_service_key(service_registry, commune_codes_by_service_key) ) department_metadata_by_code = {department["code"]: department for department in department_metadata} region_metadata_by_code = {region["code"]: region for region in region_metadata} metro_departments = [department for department in department_metadata if department.get("zone") == "metro"] metro_department_codes = [department["code"] for department in metro_departments] department_codes_by_region: dict[str, list[str]] = {} for department in metro_departments: department_codes_by_region.setdefault(department["codeRegion"], []).append(department["code"]) metro_regions = [ region for region in region_metadata if region["code"] in department_codes_by_region ] metro_region_codes = [region["code"] for region in metro_regions] region_key_by_code = {region["code"]: region_territory_key(region["nom"]) for region in metro_regions} france_territory = build_france_territory(metro_department_codes) region_territories = build_region_territories(metro_regions, department_codes_by_region) department_territories = build_department_territories(metro_departments, region_key_by_code) overseas_territories = build_overseas_territories() territory_definitions = [ france_territory, *region_territories, *overseas_territories, *department_territories, ] france_region_features = fetch_france_region_features( region_metadata_by_code, metro_region_codes, region_key_by_code, ) print(f"France: {len(france_region_features)} régions") metropolitan_department_features = fetch_metropolitan_department_features( department_metadata_by_code, metro_department_codes, ) region_department_features_by_key = {territory["key"]: [] for territory in region_territories} for feature in metropolitan_department_features: region_key = region_key_by_code.get(feature["properties"]["r"]) if region_key: region_department_features_by_key[region_key].append(feature) print(f"Départements métropolitains: {len(metropolitan_department_features)}") department_commune_features_by_key = {territory["key"]: [] for territory in department_territories} with ThreadPoolExecutor(max_workers=8) as executor: future_by_key = { executor.submit(fetch_region_commune_features, territory, competence_by_commune_code): territory["key"] for territory in region_territories } for future in as_completed(future_by_key): region_key = future_by_key[future] features = future.result() for feature in features: department_key = department_territory_key(feature["properties"]["d"]) if department_key in department_commune_features_by_key: department_commune_features_by_key[department_key].append(feature) print(f"{region_key}: {len(features)} communes réparties par département") overseas_commune_features_by_key: dict[str, list[dict]] = {} overseas_masked_commune_features_by_key: dict[str, list[dict]] = {} with ThreadPoolExecutor(max_workers=6) as executor: future_by_key = { executor.submit( fetch_department_commune_features, territory["departmentCodes"][0], competence_by_commune_code, ): territory["key"] for territory in overseas_territories } for future in as_completed(future_by_key): territory_key = future_by_key[future] features = future.result() if territory_key == "polynesie-francaise": hidden_features = [ feature for feature in features if feature["properties"]["n"] not in POLYNESIE_RETAINED_COMMUNES ] features = [ feature for feature in features if feature["properties"]["n"] in POLYNESIE_RETAINED_COMMUNES ] overseas_masked_commune_features_by_key[territory_key] = hidden_features overseas_commune_features_by_key[territory_key] = features print(f"{territory_key}: {len(features)} communes") territories_manifest = [] index_entries = [] for territory in territory_definitions: masked_territory_features: list[dict] = [] if territory["key"] == "france": territory_features = france_region_features elif territory["entityType"] == DEPARTMENT_ENTITY_TYPE: territory_features = region_department_features_by_key[territory["key"]] elif territory.get("parentTerritoryKey"): territory_features = department_commune_features_by_key[territory["key"]] else: territory_features = overseas_commune_features_by_key[territory["key"]] masked_territory_features = overseas_masked_commune_features_by_key.get(territory["key"], []) payload_features = territory_features projected_unit_origin: tuple[float, float] | None = None projected_unit_source: list[dict] = [] special_layout_source_centers: dict[str, tuple[float, float]] = {} planar_compaction = PLANAR_TERRITORY_COMPACTION.get(territory["key"]) compact_factor = planar_compaction["factor"] if planar_compaction else 1.0 group_distance_reduction = SPECIAL_GROUP_DISTANCE_REDUCTION.get(territory["key"], 0.0) if territory["key"] in SPECIAL_TERRITORY_LAYOUTS or territory["key"] in SPECIAL_TERRITORY_SIMPLIFICATION: special_layout_source_centers = get_layout_source_centers(territory["key"], territory_features) payload_features = optimize_special_territory_features(territory["key"], territory_features) if territory["key"] in PREPROJECTED_TERRITORIES: payload_features = [ { "type": "Feature", "properties": dict(feature["properties"]), "geometry": project_geometry_to_planar_mercator(feature["geometry"]), } for feature in payload_features ] if compact_factor < 1: payload_features, projected_unit_origin = compact_planar_features(payload_features, compact_factor) payload = { "type": "FeatureCollection", "features": payload_features, } if territory["key"] in PREPROJECTED_TERRITORIES: payload["coordinateSpace"] = ( "planar-layout" if territory["key"] in SPECIAL_PLANAR_LAYOUT_ROWS or compact_factor < 1 else "planar-mercator" ) service_lookup = {} unit_lookup = {} for feature in territory_features: service_keys = [] if feature["properties"].get("u"): service_keys.append(feature["properties"]["u"]) service_keys.extend(feature["properties"].get("v", [])) for service_key in service_keys: if service_key in service_registry: service_lookup[service_key] = service_registry[service_key] if service_key in public_units_by_service_key and territory.get("groupementView"): unit_lookup[service_key] = public_units_by_service_key[service_key] if service_lookup: payload["services"] = service_lookup if unit_lookup: optimized_units = list(unit_lookup.values()) if special_layout_source_centers: optimized_units = optimize_special_territory_units( territory["key"], optimized_units, special_layout_source_centers, ) if territory["key"] in PREPROJECTED_TERRITORIES: projected_unit_source = optimized_units optimized_units = project_units_to_planar_mercator(optimized_units) if compact_factor < 1 and projected_unit_origin: origin_x, origin_y = projected_unit_origin optimized_units = compact_planar_units(optimized_units, origin_x, origin_y, compact_factor) if territory["key"] in SPECIAL_PLANAR_LAYOUT_ROWS: payload_features, optimized_units = apply_special_planar_layout( territory["key"], territory_features, payload_features, projected_unit_source or [], optimized_units, ) payload["features"] = payload_features if group_distance_reduction > 0 and territory["key"] in PREPROJECTED_TERRITORIES: payload_features, optimized_units = reduce_group_distances_without_overlap( territory["key"], territory_features, payload_features, projected_unit_source or [], optimized_units, group_distance_reduction, ) payload["features"] = payload_features payload["units"] = sorted( optimized_units, key=lambda item: (normalize_sort_text(item["n"]), item["n"].lower(), item["s"]), ) elif territory["key"] in SPECIAL_PLANAR_LAYOUT_ROWS: payload_features, _ = apply_special_planar_layout( territory["key"], territory_features, payload_features, [], [], ) payload["features"] = payload_features if group_distance_reduction > 0 and territory["key"] in PREPROJECTED_TERRITORIES and not unit_lookup: payload_features, _ = reduce_group_distances_without_overlap( territory["key"], territory_features, payload["features"], [], [], group_distance_reduction, ) payload["features"] = payload_features if masked_territory_features: masked_communes = sorted( [ { "c": feature["properties"]["c"], "n": feature["properties"]["n"], "p": commune_postal_codes_by_code.get(feature["properties"]["c"], []), } for feature in masked_territory_features ], key=lambda item: (normalize_sort_text(item["n"]), item["n"].lower(), item["c"]), ) masked_commune_name_by_code = {item["c"]: item["n"] for item in masked_communes} masked_units_by_service_key: dict[str, dict] = {} for feature in masked_territory_features: properties = feature["properties"] service_keys = [] if properties.get("u"): service_keys.append(properties["u"]) service_keys.extend(properties.get("v", [])) for service_key in service_keys: service = service_registry.get(service_key, {}) public_unit = public_units_by_service_key.get(service_key, {}) unit_label = public_unit.get("n") or service.get("l") or service_key institution = public_unit.get("i") or service.get("i") or "UNK" if service_key not in masked_units_by_service_key: masked_units_by_service_key[service_key] = { "s": service_key, "i": institution, "n": unit_label, "c": set(), } masked_units_by_service_key[service_key]["c"].add(properties["c"]) masked_units = [] for unit in masked_units_by_service_key.values(): commune_codes = sorted(unit["c"]) commune_names = sorted( {masked_commune_name_by_code.get(code, code) for code in commune_codes}, key=normalize_sort_text, ) masked_units.append( { "s": unit["s"], "i": unit["i"], "n": unit["n"], "c": commune_codes, "m": commune_names, } ) masked_units.sort(key=lambda item: (normalize_sort_text(item["n"]), item["n"].lower(), item["s"])) payload["masked"] = { "communes": masked_communes, "units": masked_units, } write_json(TERRITORY_DIR / f"{territory['key']}.json", payload) territories_manifest.append( { "key": territory["key"], "label": territory["label"], "shortLabel": territory["shortLabel"], "departmentCodes": territory["departmentCodes"], "entityType": territory["entityType"], "entityLabelSingular": territory["entityLabelSingular"], "entityLabelPlural": territory["entityLabelPlural"], "entityCount": len(territory_features), "file": f"./data/territories/{territory['key']}.json", "navVisible": territory.get("navVisible", True), "parentTerritoryKey": territory.get("parentTerritoryKey"), "groupementView": territory.get("groupementView", False), "regionCode": territory.get("regionCode"), } ) index_entries.extend( build_index_entries( territory_features, territory["key"], commune_postal_codes_by_code, ) ) if masked_territory_features: index_entries.extend( build_index_entries( masked_territory_features, territory["key"], commune_postal_codes_by_code, is_hidden=True, ) ) print( f"{territory['label']}: {len(territory_features)} {territory['entityLabelPlural']}" ) index_entries.sort(key=lambda item: (normalize_sort_text(item["n"]), item["n"].lower(), item["c"], item["t"])) write_json(DATA_DIR / "communes-index.json", index_entries) write_json(DATA_DIR / "territories.json", territories_manifest) print(f"Index written: {len(index_entries)} entrées") if __name__ == "__main__": build_data()