283 lines
13 KiB
Python
283 lines
13 KiB
Python
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from sqlalchemy import Engine, select
|
|
from sqlalchemy.orm import Session
|
|
from tqdm import tqdm
|
|
|
|
from nupes.cache import get_file
|
|
from nupes.models.geographie import BureauVote, Circonscription, Commune, Departement, Region
|
|
|
|
|
|
def importer_regions(engine: Engine, verbose: bool = False) -> None:
|
|
etag = requests.get(
|
|
"https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-region?select=data_processed").json()['data_processed']
|
|
file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-region/exports/geojson?lang=fr&timezone=Europe%2FParis",
|
|
"georef-france-region.geojson", etag)
|
|
|
|
with file.open('r') as f:
|
|
features = json.load(f)['features']
|
|
|
|
with Session(engine) as session:
|
|
for feature in tqdm(features, desc="Régions", disable=not verbose):
|
|
region_dict = feature['properties']
|
|
code_region = region_dict['reg_code'][0]
|
|
nom_region = region_dict['reg_name'][0]
|
|
|
|
if region := session.execute(select(Region).filter_by(code_insee=code_region)).scalar_one_or_none():
|
|
region.libelle = nom_region
|
|
region.geometry = feature['geometry']
|
|
else:
|
|
region = Region(code_insee=code_region, libelle=nom_region, geometry=feature['geometry'])
|
|
session.add(region)
|
|
|
|
if reg := session.execute(select(Region).filter_by(code_insee="ZZ")).scalar_one_or_none():
|
|
reg.libelle = "Français⋅es de l'étranger"
|
|
else:
|
|
session.add(Region(code_insee="ZZ", libelle="Français⋅es de l'étranger", geometry={}))
|
|
|
|
session.commit()
|
|
|
|
|
|
def importer_departements(engine: Engine, verbose: bool = False) -> None:
|
|
etag = requests.get(
|
|
"https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-departement?select=data_processed").json()['data_processed']
|
|
file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-departement/exports/geojson?lang=fr&timezone=Europe%2FParis",
|
|
"georef-france-departement.geojson", etag)
|
|
|
|
with file.open('r') as f:
|
|
features = json.load(f)['features']
|
|
|
|
with Session(engine) as session:
|
|
for feature in tqdm(features, desc="Départements", disable=not verbose):
|
|
dpt_dict = feature['properties']
|
|
code_dpt = dpt_dict['dep_code'][0]
|
|
nom_dpt = dpt_dict['dep_name'][0]
|
|
|
|
if dpt := session.execute(select(Departement).filter_by(code_insee=code_dpt)).scalar_one_or_none():
|
|
dpt.libelle = nom_dpt
|
|
dpt.region_code = dpt_dict['reg_code'][0]
|
|
dpt.geometry = feature['geometry']
|
|
else:
|
|
dpt = Departement(code_insee=code_dpt, libelle=nom_dpt, region_code=dpt_dict['reg_code'][0],
|
|
geometry=feature['geometry'])
|
|
session.add(dpt)
|
|
|
|
reg_etranger = session.execute(select(Region).filter_by(code_insee="ZZ")).scalar_one_or_none()
|
|
if dpt := session.execute(select(Departement).filter_by(code_insee="ZZ")).scalar_one_or_none():
|
|
dpt.libelle = "Français⋅es de l'étranger"
|
|
dpt.region_code = reg_etranger.code_insee
|
|
else:
|
|
session.add(Departement(code_insee="ZZ", region_code=reg_etranger.code_insee,
|
|
libelle="Français⋅es de l'étranger", geometry={}))
|
|
|
|
session.commit()
|
|
|
|
|
|
def importer_circonscriptions(engine: Engine, verbose: bool = False) -> None:
|
|
file = get_file("https://www.data.gouv.fr/fr/datasets/r/67c0f382-dc8d-4d1f-8a76-1162c53b9dfe",
|
|
"circonscriptions-legislatives-p10.geojson")
|
|
|
|
with file.open('r') as f:
|
|
features = json.load(f)['features']
|
|
|
|
with Session(engine) as session:
|
|
for feature in tqdm(features, desc="Circonscriptions", disable=not verbose):
|
|
circo_dict = feature['properties']
|
|
code_circo = circo_dict['codeCirconscription']
|
|
code_dpt = circo_dict['codeDepartement']
|
|
|
|
match code_dpt:
|
|
case "ZA":
|
|
code_dpt = "971"
|
|
case "ZB":
|
|
code_dpt = "972"
|
|
case "ZC":
|
|
code_dpt = "973"
|
|
case "ZD":
|
|
code_dpt = "974"
|
|
case "ZS":
|
|
code_dpt = "975"
|
|
case "ZM":
|
|
code_dpt = "976"
|
|
|
|
numero_circo = int(code_circo[len(code_dpt):])
|
|
circo_id = f"{code_dpt}-{numero_circo:02d}"
|
|
|
|
if not session.execute(select(Departement).filter_by(code_insee=code_dpt)).scalar_one_or_none():
|
|
print("Département non trouvé avec le code", code_dpt)
|
|
continue
|
|
|
|
if circo := session.execute(select(Circonscription).filter_by(id=circo_id)).scalar_one_or_none():
|
|
circo.departement_code = code_dpt
|
|
circo.numero = numero_circo
|
|
circo.geometry = feature['geometry']
|
|
else:
|
|
circo = Circonscription(id=circo_id, departement_code=code_dpt, numero=numero_circo,
|
|
geometry=feature['geometry'])
|
|
session.add(circo)
|
|
|
|
circos_manquantes = [
|
|
{"id": "977-01", "dpt_id": "977", "numero": 1, "geometry": {}},
|
|
{"id": "986-01", "dpt_id": "986", "numero": 1, "geometry": {}},
|
|
{"id": "987-01", "dpt_id": "987", "numero": 1, "geometry": {}},
|
|
{"id": "987-02", "dpt_id": "987", "numero": 2, "geometry": {}},
|
|
{"id": "987-03", "dpt_id": "987", "numero": 3, "geometry": {}},
|
|
{"id": "988-01", "dpt_id": "988", "numero": 1, "geometry": {}},
|
|
{"id": "988-02", "dpt_id": "988", "numero": 2, "geometry": {}},
|
|
{"id": "ZZ-01", "dpt_id": "ZZ", "numero": 1, "geometry": {}},
|
|
{"id": "ZZ-02", "dpt_id": "ZZ", "numero": 2, "geometry": {}},
|
|
{"id": "ZZ-03", "dpt_id": "ZZ", "numero": 3, "geometry": {}},
|
|
{"id": "ZZ-04", "dpt_id": "ZZ", "numero": 4, "geometry": {}},
|
|
{"id": "ZZ-05", "dpt_id": "ZZ", "numero": 5, "geometry": {}},
|
|
{"id": "ZZ-06", "dpt_id": "ZZ", "numero": 6, "geometry": {}},
|
|
{"id": "ZZ-07", "dpt_id": "ZZ", "numero": 7, "geometry": {}},
|
|
{"id": "ZZ-08", "dpt_id": "ZZ", "numero": 8, "geometry": {}},
|
|
{"id": "ZZ-09", "dpt_id": "ZZ", "numero": 9, "geometry": {}},
|
|
{"id": "ZZ-10", "dpt_id": "ZZ", "numero": 10, "geometry": {}},
|
|
{"id": "ZZ-11", "dpt_id": "ZZ", "numero": 11, "geometry": {}},
|
|
]
|
|
for circo_dict in circos_manquantes:
|
|
circo_geometry_file = Path(__file__).parent.parent / "initial_data" \
|
|
/ f"geometry_circo_{circo_dict['id']}.json"
|
|
if circo_geometry_file.is_file():
|
|
with circo_geometry_file.open('r') as f:
|
|
circo_dict['geometry'] = json.load(f)
|
|
|
|
if not session.execute(select(Departement).filter_by(code_insee=circo_dict['dpt_id'])).scalar_one_or_none():
|
|
print("Département non trouvé avec le code", circo_dict['dpt_id'])
|
|
continue
|
|
|
|
if circo := session.execute(select(Circonscription).filter_by(id=circo_dict['id'])).scalar_one_or_none():
|
|
circo.departement_code = circo_dict['dpt_id']
|
|
circo.numero = circo_dict['numero']
|
|
circo.geometry = circo_dict['geometry']
|
|
else:
|
|
circo = Circonscription(id=circo_dict['id'], departement_code=circo_dict['dpt_id'],
|
|
numero=circo_dict['numero'], geometry=circo_dict['geometry'])
|
|
session.add(circo)
|
|
|
|
session.commit()
|
|
|
|
|
|
def importer_communes(engine: Engine, verbose: bool = False) -> None:
|
|
etag = requests.get(
|
|
"https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-commune-arrondissement-municipal?select=data_processed").json()['data_processed']
|
|
file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-commune-arrondissement-municipal/exports/geojson?lang=fr&timezone=Europe%2FParis",
|
|
"georef-france-commune-arrondissement-municipal.geojson", etag)
|
|
|
|
with file.open('r') as f:
|
|
features = json.load(f)['features']
|
|
|
|
with Session(engine) as session:
|
|
for feature in tqdm(features, desc="Communes", disable=not verbose):
|
|
commune_dict = feature['properties']
|
|
code_commune = commune_dict['com_arm_code'][0]
|
|
nom_commune = commune_dict['com_name'][0]
|
|
|
|
if commune := session.execute(select(Commune).filter_by(code_insee=code_commune)).scalar_one_or_none():
|
|
commune.libelle = nom_commune
|
|
commune.departement_code = commune_dict['dep_code'][0]
|
|
commune.geometry = feature['geometry']
|
|
else:
|
|
commune = Commune(code_insee=code_commune, libelle=nom_commune,
|
|
departement_code=commune_dict['dep_code'][0], geometry=feature['geometry'])
|
|
session.add(commune)
|
|
|
|
geometry_paris_file = Path(__file__).parent.parent / "initial_data" / "geometry_paris.json"
|
|
geometry_lyon_file = Path(__file__).parent.parent / "initial_data" / "geometry_lyon.json"
|
|
geometry_marseille_file = Path(__file__).parent.parent / "initial_data" / "geometry_marseille.json"
|
|
|
|
with geometry_paris_file.open('r') as f:
|
|
geometry_paris = json.load(f)
|
|
with geometry_lyon_file.open('r') as f:
|
|
geometry_lyon = json.load(f)
|
|
with geometry_marseille_file.open('r') as f:
|
|
geometry_marseille = json.load(f)
|
|
|
|
paris_lyon_marseille = [
|
|
{'code_insee': "75056", 'libelle': "Paris", 'departement_code': "75", 'geometry': geometry_paris},
|
|
{'code_insee': "69123", 'libelle': "Lyon", 'departement_code': "69", 'geometry': geometry_lyon},
|
|
{'code_insee': "13055", 'libelle': "Marseille", 'departement_code': "13", 'geometry': geometry_marseille},
|
|
]
|
|
|
|
for commune_dict in paris_lyon_marseille:
|
|
if not session.execute(select(Commune).filter_by(code_insee=commune_dict['code_insee'])) \
|
|
.scalar_one_or_none():
|
|
commune = Commune(code_insee=commune_dict['code_insee'], libelle=commune_dict['libelle'],
|
|
departement_code=commune_dict['departement_code'],
|
|
geometry=commune_dict['geometry'])
|
|
session.add(commune)
|
|
|
|
session.commit()
|
|
|
|
|
|
def importer_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
|
|
file = get_file("https://files.data.gouv.fr/reu/contours-france-entiere-latest-v2.geojson",
|
|
"contours-france-entiere-latest-v2.geojson")
|
|
|
|
with file.open('r') as f:
|
|
features = json.load(f)['features']
|
|
|
|
with Session(engine) as session:
|
|
for feature in tqdm(features, desc="Bureaux de vote", disable=not verbose):
|
|
bv_dict = feature['properties']
|
|
code_commune = bv_dict['id_bv'].split('_')[0]
|
|
code_bv = bv_dict['numeroBureauVote']
|
|
dpt_code = bv_dict['codeDepartement']
|
|
|
|
match dpt_code:
|
|
case "ZA":
|
|
dpt_code = "971"
|
|
case "ZB":
|
|
dpt_code = "972"
|
|
case "ZC":
|
|
dpt_code = "973"
|
|
case "ZD":
|
|
dpt_code = "974"
|
|
case "ZS":
|
|
dpt_code = "975"
|
|
case "ZM":
|
|
dpt_code = "976"
|
|
|
|
numero_circo = int(bv_dict['codeCirconscription'][len(dpt_code):])
|
|
code_circo = f"{dpt_code}-{numero_circo:02d}"
|
|
bv_id = bv_dict['id_bv'].split()[0]
|
|
bv_libelle = f"Bureau {code_bv} de {bv_dict['nomCommune']}"
|
|
|
|
if not session.execute(select(Commune).filter_by(code_insee=code_commune)).scalar_one_or_none():
|
|
print("Commune non trouvée avec le code", code_commune, "et le nom", bv_dict['nomCommune'])
|
|
continue
|
|
|
|
if not session.execute(select(Circonscription).filter_by(id=code_circo)).scalar_one_or_none():
|
|
session.add(Circonscription(id=code_circo, departement_code=dpt_code, numero=numero_circo))
|
|
|
|
if bv := session.execute(select(BureauVote).filter_by(id=bv_id)).scalar_one_or_none():
|
|
bv.commune_code = code_commune
|
|
bv.code_bureau = code_bv
|
|
bv.circo_code = code_circo
|
|
bv.libelle = bv_libelle
|
|
bv.geometry = feature['geometry']
|
|
else:
|
|
bv = BureauVote(id=bv_id, commune_code=code_commune, code_bureau=code_bv, circo_code=code_circo,
|
|
libelle=bv_libelle, adresse="", geometry=feature['geometry'])
|
|
session.add(bv)
|
|
|
|
session.commit()
|
|
|
|
|
|
def run(engine: Engine, verbose: bool = False) -> None:
|
|
importer_regions(engine, verbose)
|
|
importer_departements(engine, verbose)
|
|
importer_circonscriptions(engine, verbose)
|
|
importer_communes(engine, verbose)
|
|
importer_bureaux_vote(engine, verbose)
|