Commit 79409097 authored by Benjam's avatar Benjam
Browse files

modification calculateur_referentiel_territoire.py pour qu'il genere un...

modification calculateur_referentiel_territoire.py pour qu'il genere un fichier unique des terriotires en sortie
parent 71ca886e
......@@ -7,6 +7,7 @@ import pandas as pd
from crater.crater_logger import log
DictDataFrames: type = Dict[str, pd.DataFrame]
DictTerritoires: type = Dict[str, pd.DataFrame]
Territoires: type = pd.DataFrame
......
......@@ -14,111 +14,151 @@ import crater.territoires.chargeur_regions as chargeur_regions
import crater.territoires.chargeur_referentiel_territoires as chargeur_referentiel_territoires
import crater.territoires.outils as outils_territoires
from crater.crater_logger import log
from crater.outils import reinitialiser_dossier, DictTerritoires
from crater.outils import reinitialiser_dossier, DictDataFrames
def _generer_referentiel_territoires(sources: DictTerritoires) -> DictTerritoires:
log.info("##### GENERATION DU REFERENTIEL TERRITOIRES #####")
def calculer_referentiel_territoires(
chemin_fichier_communes_et_arrondissements: Path,
chemin_fichier_mouvements_communes: Path,
chemin_fichier_epcis: Path,
chemin_fichier_departements: Path,
chemin_fichier_regions: Path,
chemin_fichier_codes_postaux: Path,
chemin_dossier_output: Path
) -> None:
sources_territoires = _charger_sources_territoires(
chemin_fichier_communes_et_arrondissements,
chemin_fichier_mouvements_communes,
chemin_fichier_epcis,
chemin_fichier_departements,
chemin_fichier_regions,
chemin_fichier_codes_postaux
)
pays: DataFrame = _creer_base_de_donnees_pays()
territoires: DictTerritoires = outils_territoires.compiler_territoires(
sources['communes'],
sources['epcis'],
sources['departements'],
sources['regions'],
pays
territoires = _generer_referentiel_territoires(
sources_territoires
)
_exporter_referentiel_territoires(
territoires,
chemin_dossier_output
)
def _generer_referentiel_territoires(sources: DictDataFrames) -> DataFrame:
log.info("##### GENERATION DU REFERENTIEL TERRITOIRES #####")
_ajouter_codes_postaux(
territoires = _assembler_differents_territoires(sources)
territoires = _ajouter_id_pays_aux_territoires(territoires)
territoires = _reordonner_colonnes(territoires)
territoires = _ajouter_id_departement_et_region_aux_epcis(territoires)
territoires = _ajouter_codes_postaux_communes(
territoires,
sources['arrondissements'],
sources['codes_postaux']
)
_ajouter_annee_dernier_mouvement_communes(territoires, sources['mouvements_communes'])
territoires = _ajouter_annee_dernier_mouvement_communes(
territoires,
sources['mouvements_communes']
)
return territoires
_ajouter_id_departement_et_region_aux_epcis(territoires)
_ajouter_id_pays_aux_territoires(territoires)
def _assembler_differents_territoires(sources: DictDataFrames) -> DataFrame:
communes = _normaliser_df_territoires(sources['communes'], 'id_commune', 'nom_commune', 'COMMUNE')
epcis = _normaliser_df_territoires(sources['epcis'], 'id_epci', 'nom_epci', "EPCI")
departements = _normaliser_df_territoires(sources['departements'], 'id_departement', 'nom_departement', "DEPARTEMENT")
regions = _normaliser_df_territoires(sources['regions'], 'id_region', 'nom_region', "REGION")
pays = _creer_base_de_donnees_pays()
territoires = pd.concat(
[pays,
regions,
departements,
epcis,
communes
],
ignore_index=True
).set_index('id_territoire')
return territoires
def _normaliser_df_territoires(df_territoires, colonne_id, colonne_nom, categorie):
df = df_territoires.copy()
df.rename(
{colonne_id: 'id_territoire', colonne_nom: 'nom_territoire'},
axis=1,
inplace=True,
errors='raise'
)
df.insert(
loc=2,
column='categorie_territoire',
value=categorie
)
return df
def _ajouter_annee_dernier_mouvement_communes(territoires: DictTerritoires, mouvements_communes: DataFrame) -> None:
synthese_mouvements_communes: DataFrame = \
pd.concat([
mouvements_communes.rename(columns={'id_commune_final': 'id_commune'}),
mouvements_communes.rename(columns={'id_commune_initial': 'id_commune'})]) \
.sort_values(by=['annee_modification']) \
.drop_duplicates(subset=['id_commune'], keep='last') \
.loc[:, ['id_commune', 'annee_modification', 'type']] \
.rename(columns={'annee_modification': 'annee_dernier_mouvement', 'type': 'type_mouvement'})
territoires['communes'] = pd.merge(
territoires['communes'],
synthese_mouvements_communes,
how='left',
on=['id_commune'], suffixes=(False, False))
def _reordonner_colonnes(territoires: DataFrame) -> DataFrame:
territoires = territoires[[
'nom_territoire',
'categorie_territoire',
'id_pays',
'id_region',
'id_departement',
'id_epci',
'categorie_epci'
]]
territoires['communes']['annee_dernier_mouvement'] = territoires['communes']['annee_dernier_mouvement'].astype(
'Int64')
return territoires
def _exporter_referentiel_territoires(
territoires,
chemin_dossier_output
):
log.info("##### EXPORT DU REFERENTIEL DES TERRITOIRES #####")
def _ajouter_annee_dernier_mouvement_communes(
territoires: DataFrame,
mouvements_communes: DataFrame
) -> DataFrame:
synthese_mouvements_communes = pd.concat(
[mouvements_communes.rename(columns={'id_commune_final': 'id_commune'}),
mouvements_communes.rename(columns={'id_commune_initial': 'id_commune'})]
).sort_values(by=['annee_modification']
).drop_duplicates(subset=['id_commune'], keep='last'
).loc[:, ['id_commune', 'annee_modification', 'type']
].rename(columns={'id_commune': 'id_territoire', 'annee_modification': 'annee_dernier_mouvement_commune', 'type': 'type_mouvement_commune'}
).set_index('id_territoire')
reinitialiser_dossier(chemin_dossier_output)
outils_territoires.exporter_fichiers_multiples(territoires, chemin_dossier_output)
territoires_assembles = (outils_territoires
.assembler_dict_territoires_en_df_territoires(territoires)
.loc[:,
['id_territoire', 'nom_territoire', 'categorie_territoire', 'codes_postaux', 'id_epci',
'id_departement', 'id_region', 'id_pays']
]
.rename(columns={'codes_postaux': 'codes_postaux_commune'})
)
# temporaire : pour fixer le bug d'affichage pycharm => affichage KO avec float avec NA de pandas...alors que ca marche avec NaN de numpy
# territoires_assembles.annee_dernier_mouvement = territoires_assembles.annee_dernier_mouvement.fillna(0)
territoires_assembles.to_csv(
chemin_dossier_output / "referentiel_territoires.csv",
sep=";",
encoding="utf-8",
index=False,
float_format='%.2f'
)
territoires = territoires.join(synthese_mouvements_communes)
return territoires
def _creer_base_de_donnees_pays() -> DataFrame:
pays = pd.DataFrame(
{'id_pays': [outils_territoires.ID_FRANCE], 'nom_pays': ['France']},
columns=['id_pays', 'nom_pays']
{'id_territoire': [outils_territoires.ID_FRANCE], 'nom_territoire': ['France'], 'categorie_territoire': ['PAYS']}
)
return pays
def _ajouter_id_pays_aux_territoires(territoires: DictTerritoires) -> None:
index_colonne_region = territoires['communes'].columns.get_loc('id_region')
territoires['communes'].insert(index_colonne_region + 1, 'id_pays', outils_territoires.ID_FRANCE)
territoires['epcis']['id_pays'] = outils_territoires.ID_FRANCE
territoires['departements']['id_pays'] = outils_territoires.ID_FRANCE
territoires['regions']['id_pays'] = outils_territoires.ID_FRANCE
def _ajouter_id_pays_aux_territoires(territoires: DataFrame) -> DataFrame:
territoires = territoires.assign(id_pays=outils_territoires.ID_FRANCE)
territoires.loc[outils_territoires.ID_FRANCE, 'id_pays'] = np.nan
return territoires
def _ajouter_id_departement_et_region_aux_epcis(territoires: DictTerritoires) -> None:
nb_communes_par_epci_et_departement = (territoires['communes']
# .loc[(territoires['communes'].annee_dernier_mouvement.isna()) | (territoires['communes'].annee_dernier_mouvement >= 2020)]
def _ajouter_id_departement_et_region_aux_epcis(territoires: DataFrame) -> DataFrame:
#NB on atrribue le département (et donc la région) qui contient le plus de communes de l'epci
nb_communes_par_epci_et_departement = (territoires.query("categorie_territoire == 'COMMUNE'")
.groupby(['id_epci', 'id_departement']).size()
.reset_index()
.rename(
columns={0: "nb_communes_par_epci_et_departement", "id_departement": "id_departement_pour_epci"})
columns={0: "nb_communes_par_epci_et_departement"})
)
index_lignes_avec_max_nb_communes = (nb_communes_par_epci_et_departement
.reset_index()
......@@ -126,28 +166,41 @@ def _ajouter_id_departement_et_region_aux_epcis(territoires: DictTerritoires) ->
['nb_communes_par_epci_et_departement']
.idxmax())
departement_appartenance_epci = (nb_communes_par_epci_et_departement
departements_appartenance_epcis = (nb_communes_par_epci_et_departement
.loc[index_lignes_avec_max_nb_communes]
.rename(columns={'id_departement_pour_epci': 'id_departement'})
)
departement_appartenance_epci = (departement_appartenance_epci
.merge(territoires['departements'], how='left', on='id_departement')
.loc[:, ['id_epci', 'id_departement', 'id_region']])
territoires['epcis'] = territoires['epcis'].merge(departement_appartenance_epci, how='left',
on=['id_epci'])
departements_regions = territoires.query("categorie_territoire == 'DEPARTEMENT'"
).loc[:, ['id_region']
].reset_index(
).rename(columns={'id_territoire': 'id_departement'})
departements_et_regions_appartenance_epcis = pd.merge(
departements_appartenance_epcis,
departements_regions,
how='left',
on='id_departement'
).set_index('id_epci')
territoires.update(departements_et_regions_appartenance_epcis)
return territoires
def _ajouter_codes_postaux(territoires: DictTerritoires, arrondissements: DataFrame, codes_postaux: DataFrame) -> None:
# les codes postaux sont donnés par arrondissements ou communes quand celle-ci n'a pas d'arrondissements
def _ajouter_codes_postaux_communes(
territoires: DataFrame,
arrondissements: DataFrame,
codes_postaux: DataFrame
) -> DataFrame:
# les codes postaux sont donnés par arrondissement ou commune quand celle-ci n'a pas d'arrondissements
# on doit donc remplacer l'id arrondissement par l'id de sa commune le cas échéant
codes_postaux_finaux = codes_postaux.copy()
codes_postaux_finaux = ajouter_id_commune_de_arrondissement(
arrondissements,
codes_postaux_finaux
codes_postaux_finaux,
arrondissements
)
codes_postaux_finaux = (
codes_postaux_finaux[["id_commune", "code_postal"]]
codes_postaux_finaux.loc[:, ["id_commune", "code_postal"]]
).drop_duplicates(
).sort_values(
by=["id_commune",
......@@ -158,18 +211,17 @@ def _ajouter_codes_postaux(territoires: DictTerritoires, arrondissements: DataFr
).agg(
{'code_postal': '|'.join}
).rename(
columns={'code_postal': 'codes_postaux'}
)
territoires['communes'] = pd.merge(
territoires['communes'],
codes_postaux_finaux,
how='left',
on=['id_commune'],
suffixes=(False, False)
)
columns={
'code_postal': 'codes_postaux_commune',
'id_commune': 'id_territoire'}
).set_index('id_territoire')
territoires = territoires.join(codes_postaux_finaux)
return territoires
def ajouter_id_commune_de_arrondissement(arrondissements: DataFrame, df: DataFrame):
def ajouter_id_commune_de_arrondissement(df: DataFrame, arrondissements: DataFrame):
df = pd.merge(
df,
arrondissements[['id_arrondissement', 'id_commune']],
......@@ -183,49 +235,34 @@ def ajouter_id_commune_de_arrondissement(arrondissements: DataFrame, df: DataFra
df['id_commune_ou_arrondissement'],
df['id_commune']
)
return df
def _exporter_referentiel_territoires(
territoires,
chemin_dossier_output
) -> None:
log.info("##### EXPORT DU REFERENTIEL DES TERRITOIRES #####")
def charger_referentiel_territoires(dossier_territoires: Path) -> DictTerritoires:
reinitialiser_dossier(chemin_dossier_output)
log.info("##### CHARGEMENT DU REFERENTIEL DES TERRITOIRES #####")
territoires.to_csv(
chemin_dossier_output / "referentiel_territoires.csv",
sep=";",
encoding="utf-8",
index=True,
float_format='%.2f'
)
referentiel_territoires = outils_territoires.compiler_territoires(
chargeur_referentiel_territoires.charger_communes(dossier_territoires / "communes.csv"),
chargeur_referentiel_territoires.charger_epcis(dossier_territoires / "epcis.csv"),
chargeur_referentiel_territoires.charger_departements(dossier_territoires / "departements.csv"),
chargeur_referentiel_territoires.charger_regions(dossier_territoires / "regions.csv"),
chargeur_referentiel_territoires.charger_pays(dossier_territoires / "pays.csv"))
return referentiel_territoires
def charger_referentiel_territoires(dossier_territoires: Path) -> DataFrame:
def calculer_referentiel_territoires(
chemin_fichier_communes_et_arrondissements: Path,
chemin_fichier_mouvements_communes: Path,
chemin_fichier_epcis: Path,
chemin_fichier_departements: Path,
chemin_fichier_regions: Path,
chemin_fichier_codes_postaux: Path,
chemin_dossier_output: Path
) -> None:
sources_territoires = _charger_sources_territoires(
chemin_fichier_communes_et_arrondissements,
chemin_fichier_mouvements_communes,
chemin_fichier_epcis,
chemin_fichier_departements,
chemin_fichier_regions,
chemin_fichier_codes_postaux
)
log.info("##### CHARGEMENT DU REFERENTIEL DES TERRITOIRES #####")
territoires = _generer_referentiel_territoires(
sources_territoires
)
_exporter_referentiel_territoires(
territoires,
chemin_dossier_output
)
referentiel_territoires = chargeur_referentiel_territoires.charger_referentiel_territoires(dossier_territoires / "referentiel_territoires.csv")
return referentiel_territoires
def _charger_sources_territoires(
......@@ -235,7 +272,7 @@ def _charger_sources_territoires(
chemin_fichier_departements: Path,
chemin_fichier_regions: Path,
chemin_fichier_codes_postaux: Path
) -> DictTerritoires:
) -> DictDataFrames:
log.info("##### CHARGEMEMENT DES SOURCES DU REFERENTIEL TERRITOIRES #####")
communes = chargeur_communes.charger(chemin_fichier_communes_et_arrondissements)
......
......@@ -7,7 +7,7 @@ from crater.crater_logger import log
from crater.territoires.outils import traduire_code_insee_vers_id_commune_crater
TYPES = {
'annee_modif': 'int',
'annee_modif': 'Int64',
'COM_FIN': 'str',
'COM_INI': 'str',
'LIB_COM_FIN': 'str',
......
......@@ -6,118 +6,16 @@ from pandas import DataFrame
from crater.crater_logger import log
def charger_communes(chemin_fichier: Path) -> DataFrame:
log.info(" => Chargement territoires[communes] depuis %s", chemin_fichier)
def charger_referentiel_territoires(chemin_fichier: Path) -> DataFrame:
log.info(" => Chargement territoires depuis %s", chemin_fichier)
return pd.read_csv(
chemin_fichier,
sep=';',
dtype={
'code_insee': 'str',
'nom': 'str',
'code_insee_epci': 'str',
'code_insee_departement': 'str',
'code_insee_region': 'str',
'code_pays': 'str',
'codes_postaux': 'str',
'annee_dernier_mouvement': 'Int64',
'type_mouvement': 'str'
'annee_dernier_mouvement_commune': 'Int64'
},
na_values=['']
).rename(
columns={
'code_insee': 'id_commune',
'nom': 'nom_commune',
'code_insee_epci': 'id_epci',
'code_insee_departement': 'id_departement',
'code_insee_region': 'id_region',
'code_pays': 'id_pays'
},
errors='raise'
)
def charger_epcis(chemin_fichier: Path) -> DataFrame:
log.info(" => Chargement territoires[epcis] depuis %s", chemin_fichier)
return pd.read_csv(
chemin_fichier,
sep=';',
dtype={
'code_insee': 'str',
'nom': 'str',
'categorie': 'str'
},
na_values=['']
).rename(
columns={
'code_insee': 'id_epci',
'nom': 'nom_epci',
'categorie': 'categorie_epci'
},
errors='raise'
)
def charger_departements(chemin_fichier: Path) -> DataFrame:
log.info(" => Chargement territoires[departements] depuis %s", chemin_fichier)
return pd.read_csv(
chemin_fichier,
sep=';',
dtype={
'code_insee': 'str',
'nom': 'str',
'code_insee_region': 'str'
},
na_values=['']
).rename(
columns={
'code_insee': 'id_departement',
'nom': 'nom_departement',
'code_insee_region': 'id_region'
},
errors='raise'
)
def charger_regions(chemin_fichier: Path) -> DataFrame:
log.info(" => Chargement territoires[regions] depuis %s", chemin_fichier)
return pd.read_csv(
chemin_fichier,
sep=';',
dtype={
'code_insee': 'str',
'nom': 'str'
},
na_values=['']
).rename(
columns={
'code_insee': 'id_region',
'nom': 'nom_region'
},
errors='raise'
)
def charger_pays(chemin_fichier: Path) -> DataFrame:
log.info(" => Chargement territoires[pays] depuis %s", chemin_fichier)
return pd.read_csv(
chemin_fichier,
sep=';',
dtype={
'code': 'str',
'nom': 'str',
},
na_values=['']
).rename(
columns={
'code': 'id_pays',
'nom': 'nom_pays'
},
errors='raise'
)
......@@ -516,75 +516,6 @@ def _normaliser_df_dict_territoire(df_territoires, colonne_id, colonne_nom, cate
)
def exporter_fichiers_multiples(
territoires: DictTerritoires,
dossier_output: Path
) -> None:
log.info(" => Export des données de territoires en plusieurs fichiers vers %s", dossier_output)
territoires['communes'].rename(
columns={'id_commune': 'code_insee',
'nom_commune': 'nom',
'id_epci': 'code_insee_epci',
'id_departement': 'code_insee_departement',
'id_region': 'code_insee_region',
'id_pays': 'code_pays'},
errors='raise'
).to_csv(
(dossier_output / "communes.csv"),
sep=";",
encoding="utf-8",
index=False,
float_format='%.2f'
)
territoires['epcis'].rename(
columns={'id_epci': 'code_insee',
'nom_epci': 'nom',
'categorie_epci': 'categorie'},
errors='raise'
).to_csv(
(dossier_output / "epcis.csv"),
sep=";",
encoding="utf-8",
index=False,
float_format='%.2f'
)
territoires['departements'].rename(
columns={'id_departement': 'code_insee',
'nom_departement': 'nom',
'id_region': 'code_insee_region'},
errors='raise'
).to_csv(
(dossier_output / "departements.csv"),
sep=";",
encoding="utf-8",
index=False,
float_format='%.2f'
)
territoires['regions'].rename(
columns={'id_region': 'code_insee',
'nom_region': 'nom'},
errors='raise'
).to_csv(
(dossier_output / "regions.csv"),
sep=";",
encoding="utf-8",
index=False,
float_format='%.2f'
)
territoires['pays'].rename(
columns={'id_pays': 'code',
'nom_pays': 'nom'},
errors='raise'
).to_csv(
(dossier_output / "pays.csv"),
sep=";",
encoding="utf-8",
index=False,
float_format='%.2f'
)
def differences_communes(
communes1: List[str],
communes2: List[str]
......