Commit 2a8ed44b authored by Lionel's avatar Lionel
Browse files

Séparation en 2 fichiers calculateur_territoires.py et calculateur_donnees_territoires.py

parent fa78c6e1
......@@ -26,9 +26,9 @@ from crater.reglages import (
DOSSIER_PARCEL, DOSSIER_CRATER, DOSSIER_PRODUCTION_BESOINS, DOSSIER_RPG
)
from crater.surface_agricole_utile import collecteur_rpg, calculateur_surface_agricole_utile
from crater.territoires.calculateur_territoires import calculer_referentiel_territoires, \
charger_referentiel_territoires_et_donnees_complementaires, charger_referentiel_territoires, \
calculer_donnees_complementaires_territoires
from crater.territoires.calculateur_donnees_territoires import calculer_donnees_complementaires_territoires, \
charger_referentiel_territoires_et_donnees_complementaires
from crater.territoires.calculateur_territoires import calculer_referentiel_territoires, charger_referentiel_territoires
def lancer_calcul_surface_agricole_utile():
......
from pathlib import Path
import pandas
from pandas import DataFrame
from crater.outils import DictTerritoires
import crater.territoires.outils as outils_territoires
from crater.surface_agricole_utile import chargeur_sau_par_commune_et_culture
from crater.territoires import chargeur_territoires, chargeur_arrondissements, chargeur_mouvements_communes, \
chargeur_population_totale, chargeur_population_totale_historique, chargeur_territoires_parcel
from crater.territoires.calculateur_territoires import charger_referentiel_territoires, _ajouter_id_commune
from crater.crater_logger import log
def calculer_donnees_complementaires_territoires(
territoires: DictTerritoires,
chemin_fichier_communes_et_arrondissements: Path,
chemin_fichier_mouvements_communes: Path,
chemin_fichier_population_totale: Path,
chemin_fichier_population_totale_historique: Path,
chemin_fichiers_sau: Path,
chemin_fichier_territoires_parcel: Path,
chemin_dossier_output: Path
) -> None:
# TODO : supprimer l'utilisation du fichier des mouvements communs pour utiliser les données dans le DF territoires
# TODO : Voir si on peut pas faire pareil avec les arrondissements
sources_territoires_donnees_complementaires = _charger_sources_territoires_donnees_complementaires(
chemin_fichier_communes_et_arrondissements,
chemin_fichier_mouvements_communes,
chemin_fichier_population_totale,
chemin_fichier_population_totale_historique,
chemin_fichiers_sau,
chemin_fichier_territoires_parcel
)
territoires_enrichi = _ajouter_donnees_complementaires_referentiel_territoires(
territoires,
sources_territoires_donnees_complementaires['arrondissements'],
sources_territoires_donnees_complementaires['mouvements_communes'],
sources_territoires_donnees_complementaires
)
outils_territoires.exporter_fichier_unique(
outils_territoires.copier_en_gardant_colonnes(
territoires_enrichi,
['sau_ha',
'population_totale_2017',
'population_totale_2010',
'population_totale_1990',
'id_territoire_parcel']
),
chemin_dossier_output / "donnees.csv"
)
def charger_referentiel_territoires_et_donnees_complementaires(dossier_territoires: Path) -> DictTerritoires:
referentiel_territoires = charger_referentiel_territoires(dossier_territoires)
log.info("##### CHARGEMENT DES DONNEES COMPLEMENTAIRES DES TERRITOIRES #####")
return outils_territoires.joindre_donnees(
referentiel_territoires,
chargeur_territoires.charger_donnees(dossier_territoires / "donnees.csv")
)
def _ajouter_donnees_complementaires_referentiel_territoires(territoires: DictTerritoires,
source_arrondissements: DataFrame,
source_mouvements_communes: DataFrame,
sources: DictTerritoires) -> DictTerritoires:
log.info("##### AJOUT DES DONNEES COMPLEMENTAIRES (population, sau,...) AUX TERRITOIRES #####")
_ajouter_sau(
territoires,
sources['sau'],
source_mouvements_communes,
2019
)
_ajouter_population_totale(
territoires,
source_arrondissements,
sources['population_totale'],
source_mouvements_communes,
2019
)
_ajouter_population_totale_historique(
territoires,
sources['population_totale_historique'],
source_mouvements_communes,
2017
)
_ajouter_id_territoire_parcel(
territoires,
sources['territoires_parcel']
)
return territoires
def _ajouter_population_totale(
territoires: DictTerritoires,
arrondissements: DataFrame,
population_totale: DataFrame,
mouvements_communes: DataFrame,
annee: int
) -> None:
# les populations totales sont donnés par arrondissements ou communes 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
population_totale_final = population_totale.copy()
population_totale_final = _ajouter_id_commune(
arrondissements,
population_totale_final
).groupby(
['id_commune'],
as_index=False
).agg(
{'population_totale_2017': 'sum'}
)
outils_territoires.traiter_donnees_communes_remodelees(
population_totale_final,
mouvements_communes,
annee
)
territoires['communes'] = pandas.merge(
territoires['communes'],
population_totale_final,
how='left',
on=['id_commune'],
suffixes=(False, False)
)
outils_territoires.ajouter_colonne_par_agregation(
territoires,
"population_totale_2017",
"somme",
"Int64"
)
def _ajouter_population_totale_historique(
territoires: DictTerritoires,
population_totale_historique: DataFrame,
mvts_communes: DataFrame,
annee: int
) -> None:
population_totale_historique_final = population_totale_historique.copy()
outils_territoires.traiter_donnees_communes_remodelees(
population_totale_historique_final,
mvts_communes,
annee
)
territoires['communes'] = pandas.merge(
territoires['communes'],
population_totale_historique_final[['id_commune', 'population_totale_2010', 'population_totale_1990']],
how='left',
on=['id_commune'],
suffixes=(False, False)
)
outils_territoires.ajouter_multiples_colonnes_par_agregation(
territoires,
["population_totale_2010",
"population_totale_1990"
],
"somme",
"Int64"
)
def _ajouter_id_territoire_parcel(
territoires: DictTerritoires,
territoires_parcel: DataFrame
) -> None:
id_territoires_parcel = territoires_parcel.loc[:, ['id_territoire', 'id_territoire_parcel']]
territoires['communes'] = (territoires['communes']
.merge(id_territoires_parcel, how='left',
left_on='id_commune', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['epcis'] = (territoires['epcis']
.merge(id_territoires_parcel, how='left',
left_on='id_epci', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['departements'] = (territoires['departements']
.merge(id_territoires_parcel, how='left',
left_on='id_departement',
right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['regions'] = (territoires['regions']
.merge(id_territoires_parcel, how='left',
left_on='id_region', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['pays'] = (territoires['pays']
.merge(id_territoires_parcel, how='left',
left_on='id_pays', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
def _ajouter_sau(
territoires: DictTerritoires,
sau: DataFrame,
mouvements_communes: DataFrame,
annee: int
) -> None:
sau_final = sau.copy()
outils_territoires.traiter_donnees_communes_remodelees(
sau_final,
mouvements_communes,
annee
)
territoires['communes'] = pandas.merge(
territoires['communes'],
sau_final.groupby(
['id_commune'],
as_index=False
).agg(
{'sau_ha': 'sum'}
),
how='left',
on=['id_commune'],
suffixes=(False, False)
)
outils_territoires.ajouter_colonne_par_agregation(
territoires,
"sau_ha",
"somme"
)
def _charger_sources_territoires_donnees_complementaires(
chemin_fichier_communes_et_arrondissements: Path,
chemin_fichier_mouvements_communes: Path,
chemin_fichier_population_totale: Path,
chemin_fichier_population_totale_historique: Path,
chemin_fichiers_sau: Path,
chemin_fichier_territoires_parcel: Path
) -> DictTerritoires:
log.info(
"##### CHARGEMEMENT DES DONNEES SOURCES COMPLEMENTAIRES RATTACHEES AUX TERRITOIRES (population, sau,...) #####")
arrondissements = chargeur_arrondissements.charger(chemin_fichier_communes_et_arrondissements)
mouvements_communes = chargeur_mouvements_communes.charger_mouvements(chemin_fichier_mouvements_communes)
chargeur_arrondissements.traduire_code_insee_vers_id_crater(arrondissements)
chargeur_mouvements_communes.traduire_code_insee_vers_id_crater(mouvements_communes)
sau = chargeur_sau_par_commune_et_culture.charger(chemin_fichiers_sau)
population_totale = chargeur_population_totale.charger(chemin_fichier_population_totale)
population_totale_historique = chargeur_population_totale_historique.charger(
chemin_fichier_population_totale_historique)
territoires_parcel = chargeur_territoires_parcel.charger(chemin_fichier_territoires_parcel)
chargeur_population_totale.traduire_code_insee_vers_id_crater(population_totale)
chargeur_population_totale_historique.traduire_code_insee_vers_id_crater(population_totale_historique)
return {
'mouvements_communes': mouvements_communes,
'arrondissements': arrondissements,
'population_totale': population_totale,
'population_totale_historique': population_totale_historique,
'sau': sau,
'territoires_parcel': territoires_parcel
}
......@@ -4,21 +4,17 @@ import numpy as np
import pandas as pd
from pandas import DataFrame
import crater.surface_agricole_utile.chargeur_sau_par_commune_et_culture as chargeur_sau_par_commune_et_culture
import crater.territoires.chargeur_arrondissements as chargeur_arrondissements
import crater.territoires.chargeur_codes_postaux as chargeur_codes_postaux
import crater.territoires.chargeur_communes as chargeur_communes
import crater.territoires.chargeur_departements as chargeur_departements
import crater.territoires.chargeur_epcis as chargeur_epcis
import crater.territoires.chargeur_mouvements_communes as chargeur_mouvements_communes
import crater.territoires.chargeur_population_totale as chargeur_population_totale
import crater.territoires.chargeur_population_totale_historique as chargeur_population_totale_historique
import crater.territoires.chargeur_regions as chargeur_regions
import crater.territoires.chargeur_territoires as chargeur_territoires
import crater.territoires.outils as outils_territoires
from crater.crater_logger import log
from crater.outils import reinitialiser_dossier, DictTerritoires
from crater.territoires import chargeur_territoires_parcel
def _generer_referentiel_territoires(sources: DictTerritoires) -> DictTerritoires:
......@@ -49,37 +45,6 @@ def _generer_referentiel_territoires(sources: DictTerritoires) -> DictTerritoire
return territoires
def _ajouter_donnees_complementaires_referentiel_territoires(territoires: DictTerritoires,
source_arrondissements: DataFrame,
source_mouvements_communes: DataFrame,
sources: DictTerritoires) -> DictTerritoires:
log.info("##### AJOUT DES DONNEES COMPLEMENTAIRES (population, sau,...) AUX TERRITOIRES #####")
_ajouter_sau(
territoires,
sources['sau'],
source_mouvements_communes,
2019
)
_ajouter_population_totale(
territoires,
source_arrondissements,
sources['population_totale'],
source_mouvements_communes,
2019
)
_ajouter_population_totale_historique(
territoires,
sources['population_totale_historique'],
source_mouvements_communes,
2017
)
_ajouter_id_territoire_parcel(
territoires,
sources['territoires_parcel']
)
return territoires
def _ajouter_mouvements_communes(territoires: DictTerritoires, mouvements_communes: DataFrame) -> None:
......@@ -204,143 +169,6 @@ def _ajouter_codes_postaux(territoires: DictTerritoires, arrondissements: DataFr
)
def _ajouter_sau(
territoires: DictTerritoires,
sau: DataFrame,
mouvements_communes: DataFrame,
annee: int
) -> None:
sau_final = sau.copy()
outils_territoires.traiter_donnees_communes_remodelees(
sau_final,
mouvements_communes,
annee
)
territoires['communes'] = pd.merge(
territoires['communes'],
sau_final.groupby(
['id_commune'],
as_index=False
).agg(
{'sau_ha': 'sum'}
),
how='left',
on=['id_commune'],
suffixes=(False, False)
)
outils_territoires.ajouter_colonne_par_agregation(
territoires,
"sau_ha",
"somme"
)
def _ajouter_population_totale(
territoires: DictTerritoires,
arrondissements: DataFrame,
population_totale: DataFrame,
mouvements_communes: DataFrame,
annee: int
) -> None:
# les populations totales sont donnés par arrondissements ou communes 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
population_totale_final = population_totale.copy()
population_totale_final = _ajouter_id_commune(
arrondissements,
population_totale_final
).groupby(
['id_commune'],
as_index=False
).agg(
{'population_totale_2017': 'sum'}
)
outils_territoires.traiter_donnees_communes_remodelees(
population_totale_final,
mouvements_communes,
annee
)
territoires['communes'] = pd.merge(
territoires['communes'],
population_totale_final,
how='left',
on=['id_commune'],
suffixes=(False, False)
)
outils_territoires.ajouter_colonne_par_agregation(
territoires,
"population_totale_2017",
"somme",
"Int64"
)
def _ajouter_population_totale_historique(
territoires: DictTerritoires,
population_totale_historique: DataFrame,
mvts_communes: DataFrame,
annee: int
) -> None:
population_totale_historique_final = population_totale_historique.copy()
outils_territoires.traiter_donnees_communes_remodelees(
population_totale_historique_final,
mvts_communes,
annee
)
territoires['communes'] = pd.merge(
territoires['communes'],
population_totale_historique_final[['id_commune', 'population_totale_2010', 'population_totale_1990']],
how='left',
on=['id_commune'],
suffixes=(False, False)
)
outils_territoires.ajouter_multiples_colonnes_par_agregation(
territoires,
["population_totale_2010",
"population_totale_1990"
],
"somme",
"Int64"
)
def _ajouter_id_territoire_parcel(
territoires: DictTerritoires,
territoires_parcel: DataFrame
) -> None:
id_territoires_parcel = territoires_parcel.loc[:, ['id_territoire', 'id_territoire_parcel']]
territoires['communes'] = (territoires['communes']
.merge(id_territoires_parcel, how='left',
left_on='id_commune', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['epcis'] = (territoires['epcis']
.merge(id_territoires_parcel, how='left',
left_on='id_epci', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['departements'] = (territoires['departements']
.merge(id_territoires_parcel, how='left',
left_on='id_departement',
right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['regions'] = (territoires['regions']
.merge(id_territoires_parcel, how='left',
left_on='id_region', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
territoires['pays'] = (territoires['pays']
.merge(id_territoires_parcel, how='left',
left_on='id_pays', right_on='id_territoire')
.drop(columns=['id_territoire'])
)
def _ajouter_id_commune(arrondissements: DataFrame, df: DataFrame):
df = pd.merge(
df,
......@@ -358,16 +186,6 @@ def _ajouter_id_commune(arrondissements: DataFrame, df: DataFrame):
return df
def charger_referentiel_territoires_et_donnees_complementaires(dossier_territoires: Path) -> DictTerritoires:
referentiel_territoires = charger_referentiel_territoires(dossier_territoires)
log.info("##### CHARGEMENT DES DONNEES COMPLEMENTAIRES DES TERRITOIRES #####")
return outils_territoires.joindre_donnees(
referentiel_territoires,
chargeur_territoires.charger_donnees(dossier_territoires / "donnees.csv")
)
def charger_referentiel_territoires(dossier_territoires: Path) -> DictTerritoires:
......@@ -409,46 +227,6 @@ def calculer_referentiel_territoires(
)
def calculer_donnees_complementaires_territoires(
territoires: DictTerritoires,
chemin_fichier_communes_et_arrondissements: Path,
chemin_fichier_mouvements_communes: Path,
chemin_fichier_population_totale: Path,
chemin_fichier_population_totale_historique: Path,
chemin_fichiers_sau: Path,
chemin_fichier_territoires_parcel: Path,
chemin_dossier_output: Path
) -> None:
# TODO : supprimer l'utilisation du fichier des mouvements communs pour utiliser les données dans le DF territoires
# TODO : Voir si on peut pas faire pareil avec les arrondissements
sources_territoires_donnees_complementaires = _charger_sources_territoires_donnees_complementaires(
chemin_fichier_communes_et_arrondissements,
chemin_fichier_mouvements_communes,
chemin_fichier_population_totale,
chemin_fichier_population_totale_historique,
chemin_fichiers_sau,
chemin_fichier_territoires_parcel
)
territoires_enrichi = _ajouter_donnees_complementaires_referentiel_territoires(
territoires,
sources_territoires_donnees_complementaires['arrondissements'],
sources_territoires_donnees_complementaires['mouvements_communes'],
sources_territoires_donnees_complementaires
)
outils_territoires.exporter_fichier_unique(
outils_territoires.copier_en_gardant_colonnes(
territoires_enrichi,
['sau_ha',
'population_totale_2017',
'population_totale_2010',
'population_totale_1990',
'id_territoire_parcel']
),
chemin_dossier_output / "donnees.csv"
)
def _charger_sources_territoires(
chemin_fichier_communes_et_arrondissements: Path,
......@@ -486,39 +264,3 @@ def _charger_sources_territoires(
'codes_postaux': codes_postaux,
}
def _charger_sources_territoires_donnees_complementaires(
chemin_fichier_communes_et_arrondissements: Path,
chemin_fichier_mouvements_communes: Path,
chemin_fichier_population_totale: Path,
chemin_fichier_population_totale_historique: Path,
chemin_fichiers_sau: Path,
chemin_fichier_territoires_parcel: Path
) -> DictTerritoires:
log.info(
"##### CHARGEMEMENT DES DONNEES SOURCES COMPLEMENTAIRES RATTACHEES AUX TERRITOIRES (population, sau,...) #####")
arrondissements = chargeur_arrondissements.charger(chemin_fichier_communes_et_arrondissements)
mouvements_communes = chargeur_mouvements_communes.charger_mouvements(chemin_fichier_mouvements_communes)
chargeur_arrondissements.traduire_code_insee_vers_id_crater(arrondissements)
chargeur_mouvements_communes.traduire_code_insee_vers_id_crater(mouvements_communes)
sau = chargeur_sau_par_commune_et_culture.charger(chemin_fichiers_sau)
population_totale = chargeur_population_totale.charger(chemin_fichier_population_totale)
population_totale_historique = chargeur_population_totale_historique.charger(
chemin_fichier_population_totale_historique)
territoires_parcel = chargeur_territoires_parcel.charger(chemin_fichier_territoires_parcel)
chargeur_population_totale.traduire_code_insee_vers_id_crater(population_totale)
chargeur_population_totale_historique.traduire_code_insee_vers_id_crater(population_totale_historique)
return {
'mouvements_communes': mouvements_communes,
'arrondissements': arrondissements,
'population_totale': population_totale,
'population_totale_historique': population_totale_historique,
'sau': sau,
'territoires_parcel': territoires_parcel
}
......@@ -2,8 +2,8 @@ import unittest
from crater.reglages import DOSSIER_TERRITOIRES, DOSSIER_SURFACE_AGRICOLE_UTILE, DOSSIER_PARCEL
from crater.territoires import calculateur_territoires
from crater.territoires.calculateur_territoires import charger_referentiel_territoires, \
calculer_donnees_complementaires_territoires
from crater.territoires.calculateur_territoires import charger_referentiel_territoires
from crater.territoires.calculateur_donnees_territoires import calculer_donnees_complementaires_territoires
from tests.test_outils import assert_list_of_csv_files_are_equals, \
CHEMIN_INPUT_DATA, CHEMIN_OUTPUT_DATA, CHEMIN_EXPECTED_DATA
......
<