| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264 |
- #!/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"
- )
- COASTLINE_GEOJSON_URL = (
- "https://raw.githubusercontent.com/nvkelso/natural-earth-vector/master/geojson/"
- "ne_50m_coastline.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",
- }
- COASTLINE_GRID_SIZE = 1.0
- COASTLINE_DISTANCE_THRESHOLD = 0.08
- 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",
- "Moorea-Maiao",
- "Arue",
- }
- 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 iter_outer_rings(geometry: dict) -> Iterable[list[list[float]]]:
- geometry_type = geometry.get("type")
- coordinates = geometry.get("coordinates", [])
- if geometry_type == "Polygon":
- if coordinates:
- yield coordinates[0]
- return
- if geometry_type == "MultiPolygon":
- for polygon in coordinates:
- if polygon:
- yield polygon[0]
- def normalize_ring_points(ring: list[list[float]], precision: int = 6) -> list[tuple[float, float]]:
- points = [
- (round(float(point[0]), precision), round(float(point[1]), precision))
- for point in ring
- if isinstance(point, list) and len(point) >= 2
- ]
- if len(points) >= 2 and points[0] == points[-1]:
- points = points[:-1]
- return points
- def build_edge_key(start: tuple[float, float], end: tuple[float, float]) -> tuple[tuple[float, float], tuple[float, float]]:
- return (start, end) if start <= end else (end, start)
- def build_global_edge_counts(features: list[dict], precision: int = 6) -> dict[tuple[tuple[float, float], tuple[float, float]], int]:
- edge_counts: dict[tuple[tuple[float, float], tuple[float, float]], int] = {}
- for feature in features:
- for ring in iter_outer_rings(feature.get("geometry", {})):
- points = normalize_ring_points(ring, precision)
- point_count = len(points)
- if point_count < 2:
- continue
- for index in range(point_count):
- start = points[index]
- end = points[(index + 1) % point_count]
- if start == end:
- continue
- edge_key = build_edge_key(start, end)
- edge_counts[edge_key] = edge_counts.get(edge_key, 0) + 1
- return edge_counts
- def build_maritime_lines_from_features(
- features: list[dict],
- edge_counts: dict[tuple[tuple[float, float], tuple[float, float]], int],
- precision: int = 6,
- ) -> list[list[list[float]]]:
- maritime_lines: list[list[list[float]]] = []
- for feature in features:
- for ring in iter_outer_rings(feature.get("geometry", {})):
- points = normalize_ring_points(ring, precision)
- point_count = len(points)
- if point_count < 2:
- continue
- edge_is_maritime = []
- for index in range(point_count):
- start = points[index]
- end = points[(index + 1) % point_count]
- edge_key = build_edge_key(start, end)
- edge_is_maritime.append(edge_counts.get(edge_key, 0) == 1)
- if not any(edge_is_maritime):
- continue
- start_indexes = [
- index
- for index in range(point_count)
- if edge_is_maritime[index] and not edge_is_maritime[index - 1]
- ]
- if not start_indexes:
- line_points = points + [points[0]]
- maritime_lines.append([[point_x, point_y] for point_x, point_y in line_points])
- continue
- for start_index in start_indexes:
- index = start_index
- line_points = [points[start_index]]
- while edge_is_maritime[index]:
- next_index = (index + 1) % point_count
- line_points.append(points[next_index])
- index = next_index
- if index == start_index:
- break
- if len(line_points) >= 2:
- maritime_lines.append([[point_x, point_y] for point_x, point_y in line_points])
- return maritime_lines
- def iter_line_strings(geometry: dict) -> Iterable[list[list[float]]]:
- geometry_type = geometry.get("type")
- coordinates = geometry.get("coordinates", [])
- if geometry_type == "LineString":
- if coordinates:
- yield coordinates
- return
- if geometry_type == "MultiLineString":
- for line in coordinates:
- if line:
- yield line
- def build_coastline_index() -> dict:
- payload = fetch_json(COASTLINE_GEOJSON_URL)
- segments = []
- grid: dict[tuple[int, int], list[int]] = {}
- for feature in payload.get("features", []):
- for line in iter_line_strings(feature.get("geometry", {})):
- for index in range(len(line) - 1):
- start = line[index]
- end = line[index + 1]
- if len(start) < 2 or len(end) < 2:
- continue
- start_x, start_y = float(start[0]), float(start[1])
- end_x, end_y = float(end[0]), float(end[1])
- if start_x == end_x and start_y == end_y:
- continue
- min_x = min(start_x, end_x)
- min_y = min(start_y, end_y)
- max_x = max(start_x, end_x)
- max_y = max(start_y, end_y)
- segment_index = len(segments)
- segments.append((start_x, start_y, end_x, end_y, min_x, min_y, max_x, max_y))
- min_cell_x = math.floor(min_x / COASTLINE_GRID_SIZE)
- max_cell_x = math.floor(max_x / COASTLINE_GRID_SIZE)
- min_cell_y = math.floor(min_y / COASTLINE_GRID_SIZE)
- max_cell_y = math.floor(max_y / COASTLINE_GRID_SIZE)
- for cell_x in range(min_cell_x, max_cell_x + 1):
- for cell_y in range(min_cell_y, max_cell_y + 1):
- grid.setdefault((cell_x, cell_y), []).append(segment_index)
- return {
- "segments": segments,
- "grid": grid,
- }
- def point_to_segment_distance_squared(
- point_x: float,
- point_y: float,
- start_x: float,
- start_y: float,
- end_x: float,
- end_y: float,
- ) -> float:
- delta_x = end_x - start_x
- delta_y = end_y - start_y
- length_squared = delta_x * delta_x + delta_y * delta_y
- if length_squared <= 1e-18:
- return (point_x - start_x) ** 2 + (point_y - start_y) ** 2
- projection = ((point_x - start_x) * delta_x + (point_y - start_y) * delta_y) / length_squared
- projection = max(0.0, min(1.0, projection))
- nearest_x = start_x + projection * delta_x
- nearest_y = start_y + projection * delta_y
- return (point_x - nearest_x) ** 2 + (point_y - nearest_y) ** 2
- def is_segment_near_coastline(
- start: list[float],
- end: list[float],
- coastline_index: dict,
- threshold_squared: float,
- ) -> bool:
- midpoint_x = (float(start[0]) + float(end[0])) / 2
- midpoint_y = (float(start[1]) + float(end[1])) / 2
- cell_x = math.floor(midpoint_x / COASTLINE_GRID_SIZE)
- cell_y = math.floor(midpoint_y / COASTLINE_GRID_SIZE)
- search_radius = 1
- segments = coastline_index["segments"]
- grid = coastline_index["grid"]
- threshold = math.sqrt(threshold_squared)
- candidate_indexes = set()
- for offset_x in range(-search_radius, search_radius + 1):
- for offset_y in range(-search_radius, search_radius + 1):
- candidate_indexes.update(grid.get((cell_x + offset_x, cell_y + offset_y), []))
- if not candidate_indexes:
- return False
- for segment_index in candidate_indexes:
- start_x, start_y, end_x, end_y, min_x, min_y, max_x, max_y = segments[segment_index]
- if (
- midpoint_x < min_x - threshold
- or midpoint_x > max_x + threshold
- or midpoint_y < min_y - threshold
- or midpoint_y > max_y + threshold
- ):
- continue
- if (
- point_to_segment_distance_squared(
- midpoint_x,
- midpoint_y,
- start_x,
- start_y,
- end_x,
- end_y,
- )
- <= threshold_squared
- ):
- return True
- return False
- def filter_maritime_lines_by_coastline(
- maritime_lines: list[list[list[float]]],
- coastline_index: dict,
- threshold: float = COASTLINE_DISTANCE_THRESHOLD,
- ) -> list[list[list[float]]]:
- threshold_squared = threshold * threshold
- filtered_lines: list[list[list[float]]] = []
- for line in maritime_lines:
- if len(line) < 2:
- continue
- segment_flags = [
- is_segment_near_coastline(line[index], line[index + 1], coastline_index, threshold_squared)
- for index in range(len(line) - 1)
- ]
- current_line: list[list[float]] = []
- for index, is_maritime in enumerate(segment_flags):
- if is_maritime:
- if not current_line:
- current_line = [line[index], line[index + 1]]
- else:
- current_line.append(line[index + 1])
- elif current_line:
- if len(current_line) >= 2:
- filtered_lines.append(current_line)
- current_line = []
- if current_line and len(current_line) >= 2:
- filtered_lines.append(current_line)
- return filtered_lines
- 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,
- )
- coastline_index = build_coastline_index()
- 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,
- }
- territory_edge_counts = build_global_edge_counts(payload["features"])
- maritime_lines = build_maritime_lines_from_features(payload["features"], territory_edge_counts)
- if maritime_lines:
- filtered_maritime_lines = filter_maritime_lines_by_coastline(maritime_lines, coastline_index)
- if filtered_maritime_lines:
- maritime_lines = filtered_maritime_lines
- elif not territory.get("parentTerritoryKey") and territory["entityType"] == COMMUNE_ENTITY_TYPE:
- # Overseas islands can be too small for coarse coastline matching; keep raw maritime edges.
- pass
- else:
- maritime_lines = []
- if maritime_lines:
- payload["maritime"] = maritime_lines
- 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()
|