Commit 2b200897 authored by Benjam's avatar Benjam
Browse files

modification des modules pour qu'ils ingerent le fichier unique des territoires

parent 79409097
......@@ -9,8 +9,8 @@ from crater.outils import reinitialiser_dossier
from crater.parcel import collecteur_besoins_api_parcel
from crater.parcel.collecteur_besoins_api_parcel import ID_ASSIETTE_ACTUELLE, ID_ASSIETTE_MOINS_25_POURCENT, \
ID_ASSIETTE_MOINS_50_POURCENT
from crater.politique_fonciere.calculateur import calculer_politique_fonciere
from crater.population_agricole.calculateur import calculer_population_agricole
from crater.politique_fonciere.calculateur_politique_fonciere import calculer_politique_fonciere
from crater.population_agricole.calculateur_population_agricole import calculer_population_agricole
from crater.pratiques_agricoles.calculateur_pratiques_agricoles import calculer_pratiques_agricoles
from crater.production_besoins.calculateur_production_besoins import calculer_axe_production_besoins
from crater.reglages import (
......@@ -50,7 +50,7 @@ def lancer_calcul_surface_agricole_utile(nombre_traitements_paralleles: int):
)
calculateur_surface_agricole_utile.calculer_surface_agricole_utile_par_commune_et_culture(
2017,
referentiel_territoires['regions'],
referentiel_territoires,
communes_sig,
CHEMIN_CRATER_DATA_SOURCES / DOSSIER_SOURCE_IGN,
NOMS_FICHIERS_RPG,
......@@ -89,8 +89,6 @@ def lancer_calcul_territoires():
calculer_donnees_territoires(
territoires,
CHEMIN_SOURCE_COMMUNES_ET_ARRONDISSEMENTS,
CHEMIN_SOURCE_MOUVEMENTS_COMMUNES,
CHEMIN_SOURCE_POPULATION_TOTALE,
CHEMIN_SOURCE_POPULATION_TOTALE_HISTORIQUE,
CHEMIN_OUTPUT_DATA / DOSSIER_SURFACE_AGRICOLE_UTILE / 'sau_par_commune_et_culture',
......
......@@ -204,4 +204,4 @@ def calculerNoteParInterpolation3Seuils(valeur, seuil1, seuil2, seuil3, note_seu
def produit_cartesien(df_1, df_2):
return df_1.assign(key=1).merge(df_2.assign(key=1), on='key').drop('key', 1)
return df_1.assign(key=1).merge(df_2.assign(key=1), on='key').drop('key', axis=1)
import pandas as pd
from pathlib import Path
from pandas import DataFrame
import numpy as np
import crater.politique_fonciere.chargeur_artificialisation as chargeur_artificialisation
import crater.politique_fonciere.politique_artificialisation as politique_artificialisation
......@@ -6,17 +8,17 @@ import crater.politique_fonciere.rythme_artificialisation_sau as rythme_artifici
import crater.politique_fonciere.sau_par_habitant as sau_par_habitant
import crater.politique_fonciere.synthese_politique_fonciere as synthese_politique_fonciere
from crater.crater_logger import log
from crater.outils import reinitialiser_dossier, DictTerritoires, supprimer_donnees_apres_annee_mouvement_commune
from crater.outils import reinitialiser_dossier, DictDataFrames
from crater.territoires import outils as outils_territoires
from crater.territoires.outils import traduire_code_insee_vers_id_commune_crater
from crater.territoires.outils import traduire_code_insee_vers_id_commune_crater, ids_echelles_geographiques_supracommunales
def calculer_politique_fonciere(
referentiel_territoires,
chemin_fichier_artificialisation,
chemin_dossier_output,
dossier_source,
dossier_politique_fonciere
territoires: DataFrame,
chemin_fichier_artificialisation: Path,
chemin_dossier_output: Path,
dossier_source: Path,
dossier_politique_fonciere: Path
) -> None:
sources = charger_sources(
chemin_fichier_artificialisation
......@@ -26,7 +28,7 @@ def calculer_politique_fonciere(
chemin_dossier_output / dossier_source / dossier_politique_fonciere
)
axe = calculer(
referentiel_territoires,
territoires,
sources
)
exporter(
......@@ -36,112 +38,110 @@ def calculer_politique_fonciere(
def calculer(
referentiel_territoires,
sources_politique_fonciere
):
territoires: DataFrame,
sources_politique_fonciere: DictDataFrames
) -> DictDataFrames:
log.info("##### CALCUL POLITIQUE FONCIERE #####")
donnees_politique_fonciere = calculer_donnees_politique_fonciere(referentiel_territoires,
sources_politique_fonciere)
indicateur_politique_artificialisation = politique_artificialisation.calculer(donnees_politique_fonciere)
indicateur_sau_par_habitant = sau_par_habitant.calculer(donnees_politique_fonciere)
indicateur_rythme_artificialisation_sau = rythme_artificialisation_sau.calculer(donnees_politique_fonciere)
synthese = synthese_politique_fonciere.calculer(
indicateur_sau_par_habitant,
indicateur_rythme_artificialisation_sau
donnees_politique_fonciere = calculer_donnees_politique_fonciere(
territoires,
sources_politique_fonciere
)
donnees_indicateur_politique_artificialisation = politique_artificialisation.calculer(donnees_politique_fonciere)
donnees_indicateur_sau_par_habitant = sau_par_habitant.calculer(donnees_politique_fonciere)
donnees_indicateur_rythme_artificialisation_sau = rythme_artificialisation_sau.calculer(donnees_politique_fonciere)
donnees_synthese = synthese_politique_fonciere.calculer(
donnees_indicateur_sau_par_habitant,
donnees_indicateur_rythme_artificialisation_sau
)
return {
'donnees': donnees_politique_fonciere,
'indicateur_sau_par_habitant': indicateur_sau_par_habitant,
'indicateur_rythme_artificialisation_sau': indicateur_rythme_artificialisation_sau,
'indicateur_politique_artificialisation': indicateur_politique_artificialisation,
'synthese': synthese
'indicateur_sau_par_habitant': donnees_indicateur_sau_par_habitant,
'indicateur_rythme_artificialisation_sau': donnees_indicateur_rythme_artificialisation_sau,
'indicateur_politique_artificialisation': donnees_indicateur_politique_artificialisation,
'synthese': donnees_synthese
}
def calculer_donnees_politique_fonciere(referentiel_territoires, sources_politique_fonciere):
donnees_politique_fonciere: DictTerritoires = outils_territoires.copier_en_gardant_colonnes(
referentiel_territoires,
['sau_productive_ha',
'population_totale_2017'
]
)
donnees_politique_fonciere['communes'] = pd.merge(
donnees_politique_fonciere['communes'],
sources_politique_fonciere['artificialisation'],
how='left',
on=['id_commune'],
suffixes=(False, False)
)
supprimer_donnees_apres_annee_mouvement_commune(referentiel_territoires,
donnees_politique_fonciere,
['artificialisation_2011_2016_ha', 'evolution_menages_2011_2016',
'evolution_emplois_2011_2016'],
2019)
donnees_politique_fonciere['communes'] = donnees_politique_fonciere['communes'].assign(
evolution_menages_emplois_2011_2016=lambda x:
x['evolution_menages_2011_2016'] + x['evolution_emplois_2011_2016']
)
def calculer_donnees_politique_fonciere(
territoires: DataFrame,
sources_politique_fonciere : DictDataFrames
) -> DataFrame:
colonnes = ['nom_territoire', 'categorie_territoire'] + ids_echelles_geographiques_supracommunales + ['annee_dernier_mouvement_commune', 'sau_productive_ha', 'population_totale_2017']
donnees_politique_fonciere = territoires[colonnes]
donnees_artificialisation = sources_politique_fonciere['artificialisation'].rename(
columns={'id_commune':'id_territoire'}).set_index(
'id_territoire')
outils_territoires.ajouter_colonne_par_agregation(
donnees_politique_fonciere = donnees_politique_fonciere.join(donnees_artificialisation)
donnees_politique_fonciere.loc[donnees_politique_fonciere.annee_dernier_mouvement_commune > 2019, 'artificialisation_2011_2016_ha'] = np.nan
donnees_politique_fonciere.loc[donnees_politique_fonciere.annee_dernier_mouvement_commune > 2019, 'evolution_menages_2011_2016'] = np.nan
donnees_politique_fonciere.loc[donnees_politique_fonciere.annee_dernier_mouvement_commune > 2019, 'evolution_emplois_2011_2016'] = np.nan
donnees_politique_fonciere['evolution_menages_emplois_2011_2016'] = \
donnees_politique_fonciere['evolution_menages_2011_2016'] + \
donnees_politique_fonciere['evolution_emplois_2011_2016']
donnees_politique_fonciere = outils_territoires.ajouter_donnees_supracommunales_par_somme_donnees_communales(
donnees_politique_fonciere,
'artificialisation_2011_2016_ha'
)
outils_territoires.ajouter_colonne_par_agregation(
donnees_politique_fonciere = outils_territoires.ajouter_donnees_supracommunales_par_somme_donnees_communales(
donnees_politique_fonciere,
'evolution_menages_2011_2016',
'Int64'
)
outils_territoires.ajouter_colonne_par_agregation(
donnees_politique_fonciere = outils_territoires.ajouter_donnees_supracommunales_par_somme_donnees_communales(
donnees_politique_fonciere,
'evolution_emplois_2011_2016',
'Int64'
)
outils_territoires.ajouter_colonne_par_agregation(
donnees_politique_fonciere = outils_territoires.ajouter_donnees_supracommunales_par_somme_donnees_communales(
donnees_politique_fonciere,
'evolution_menages_emplois_2011_2016',
'Int64'
)
outils_territoires.ajouter_colonne_par_agregation(
donnees_politique_fonciere = outils_territoires.ajouter_donnees_supracommunales_par_somme_donnees_communales(
donnees_politique_fonciere,
'superficie_ha'
'superficie_ha'
)
for t in donnees_politique_fonciere:
donnees_politique_fonciere[t] = donnees_politique_fonciere[t].assign(
artificialisation_2011_2016_sur_superficie_pourcent=lambda x:
x['artificialisation_2011_2016_ha'] / x['superficie_ha'] * 100
)
donnees_politique_fonciere['artificialisation_2011_2016_sur_superficie_pourcent'] = \
donnees_politique_fonciere['artificialisation_2011_2016_ha'] / \
donnees_politique_fonciere['superficie_ha'] * \
100
return donnees_politique_fonciere
def exporter(
politique_fonciere,
chemin_dossier_output
):
donnees_politique_fonciere: DataFrame,
chemin_dossier_output: Path
) -> None:
log.info("##### EXPORT POLITIQUE FONCIERE #####")
reinitialiser_dossier(chemin_dossier_output)
outils_territoires.exporter_fichier_unique(
politique_fonciere['donnees'],
outils_territoires.exporter_donnees_territoires(
donnees_politique_fonciere['donnees'].drop(columns=['annee_dernier_mouvement_commune'] + ids_echelles_geographiques_supracommunales),
(chemin_dossier_output / "donnees.csv")
)
outils_territoires.exporter_fichier_unique(
politique_fonciere['indicateur_sau_par_habitant'],
outils_territoires.exporter_donnees_territoires(
donnees_politique_fonciere['indicateur_sau_par_habitant'],
(chemin_dossier_output / "indicateur_sau_par_habitant.csv")
)
outils_territoires.exporter_fichier_unique(
politique_fonciere['indicateur_rythme_artificialisation_sau'],
outils_territoires.exporter_donnees_territoires(
donnees_politique_fonciere['indicateur_rythme_artificialisation_sau'],
(chemin_dossier_output / "indicateur_rythme_artificialisation_sau.csv")
)
outils_territoires.exporter_fichier_unique(
politique_fonciere['indicateur_politique_artificialisation'],
outils_territoires.exporter_donnees_territoires(
donnees_politique_fonciere['indicateur_politique_artificialisation'],
(chemin_dossier_output / "indicateur_politique_artificialisation.csv")
)
outils_territoires.exporter_fichier_unique(
politique_fonciere['synthese'],
outils_territoires.exporter_donnees_territoires(
donnees_politique_fonciere['synthese'],
(chemin_dossier_output / "synthese.csv")
)
......
import pandas
import crater.territoires.outils as outils_territoires
import pandas as pd
from pandas import DataFrame
from crater.crater_logger import log
def calculer(territoires):
log.info(" => Indicateur efficacité artificialisation")
def calculer(donnees_politique_fonciere: DataFrame) -> DataFrame:
log.info(" => Indicateur politique d'artificialisation")
indicateur = outils_territoires.copier_en_gardant_colonnes(
territoires,
['evolution_menages_emplois_2011_2016',
'artificialisation_2011_2016_ha',
'artificialisation_2011_2016_sur_superficie_pourcent'
]
)
for t in indicateur:
indicateur[t]['politique_artificialisation_valeur'] = indicateur[t].apply(
donnees_indicateur = donnees_politique_fonciere.loc[:,
['nom_territoire',
'categorie_territoire',
'evolution_menages_emplois_2011_2016',
'artificialisation_2011_2016_ha',
'artificialisation_2011_2016_sur_superficie_pourcent'
]]
donnees_indicateur.loc[:, 'politique_artificialisation_valeur'] = (
donnees_indicateur.apply(
lambda x:
valeur(
politique_artificialisation_valeur(
x['evolution_menages_emplois_2011_2016'],
x['artificialisation_2011_2016_ha'],
x['artificialisation_2011_2016_sur_superficie_pourcent']
),
axis=1
)
)
return indicateur
return donnees_indicateur
def valeur(evolution_menages_emplois_2011_2016, artificialisation_2011_2016_ha, artificialisation_2011_2016_sur_superficie_pourcent):
def politique_artificialisation_valeur(evolution_menages_emplois_2011_2016, artificialisation_2011_2016_ha, artificialisation_2011_2016_sur_superficie_pourcent):
if artificialisation_2011_2016_ha <= 0:
return "L'objectif Zéro Artificialisation Nette a été atteint entre 2011 et 2016 !"
elif artificialisation_2011_2016_ha > 0 and (
pandas.isna(evolution_menages_emplois_2011_2016) or evolution_menages_emplois_2011_2016 > 0):
pd.isna(evolution_menages_emplois_2011_2016) or evolution_menages_emplois_2011_2016 > 0):
return ("Attention l'objectif Zéro Artificialisation Nette n'a pas été atteint entre 2011 et 2016 puisque " +
"{:,.0f} ha (soit {:,.1f}% de la superficie totale du territoire) ont été artificialisés.").format(
artificialisation_2011_2016_ha,
......
from pandas import DataFrame
import crater.outils as outils
import crater.territoires.outils as outils_territoires
from crater.crater_logger import log
......@@ -6,49 +8,53 @@ from crater.crater_logger import log
# Dans ce cas (div par 0), on obtient une valeur NaN pour l'indicateur d'artificialisation, et on affiche Données non disponible
# voir si on ne peut pas afficher un messager plus spécifique, ou calculer autrement l'indicateur
def calculer(territoires):
def calculer(donnees_politique_fonciere: DataFrame) -> DataFrame:
log.info(" => Indicateur rythme d'artificialisation de la sau")
indicateur = outils_territoires.copier_en_gardant_colonnes(
territoires,
['artificialisation_2011_2016_ha',
'sau_productive_ha'
]
)
for t in indicateur:
indicateur[t]['rythme_artificialisation_sau_valeur_pourcent'] = indicateur[t].apply(
donnees_indicateur = donnees_politique_fonciere.loc[:,
['nom_territoire',
'categorie_territoire',
'artificialisation_2011_2016_ha',
'sau_productive_ha'
]]
donnees_indicateur.loc[:, 'rythme_artificialisation_sau_valeur_pourcent'] = (
donnees_indicateur.apply(
lambda x:
valeur(
rythme_artificialisation_sau_valeur_pourcent(
x['artificialisation_2011_2016_ha'],
x['sau_productive_ha']
),
axis=1
)
if indicateur['pays'].query(f"id_pays=='{outils_territoires.ID_FRANCE}'")[
)
if donnees_indicateur.query(f"id_territoire=='{outils_territoires.ID_FRANCE}'")[
'rythme_artificialisation_sau_valeur_pourcent'][0] <= 0:
raise ValueError('Attention, la valeur moyenne française pose problème vu la définition de calcul de la note !')
for t in indicateur:
indicateur[t]['rythme_artificialisation_sau_note'] = indicateur[t].apply(
donnees_indicateur.loc[:, 'rythme_artificialisation_sau_note'] = (
donnees_indicateur.apply(
lambda x:
outils.calculerNoteParInterpolation2Seuils(
x['rythme_artificialisation_sau_valeur_pourcent'],
0, indicateur['pays'].query(f"id_pays=='{outils_territoires.ID_FRANCE}'")[
0, donnees_indicateur.query(f"id_territoire=='{outils_territoires.ID_FRANCE}'")[
'rythme_artificialisation_sau_valeur_pourcent'][0],
10, 0
),
axis=1
)
)
return indicateur
return donnees_indicateur
def valeur(artificialisation_5ans_ha, sau_productive_ha):
def rythme_artificialisation_sau_valeur_pourcent(artificialisation_5ans_ha, sau_productive_ha):
try:
x = round(
return round(
artificialisation_5ans_ha / 5 / sau_productive_ha * 100,
2
)
except:
x = ""
return x
return ""
import numpy as np
from pandas import DataFrame
import pandas as pd
import crater.outils as outils
import crater.territoires.outils as outils_territoires
from crater.crater_logger import log
from crater.politique_fonciere.parametres import seuil_regime_tres_vegetal, seuil_regime_actuel
def calculer(territoires):
def calculer(donnees_politique_fonciere: DataFrame) -> DataFrame:
log.info(" => Indicateur sau par habitant")
indicateur = outils_territoires.copier_en_gardant_colonnes(
territoires,
['sau_productive_ha',
'population_totale_2017'
]
)
for t in indicateur:
indicateur[t]['sau_par_habitant_valeur_m2_par_hab'] = indicateur[t].apply(
donnees_indicateur = donnees_politique_fonciere.loc[:,
['nom_territoire',
'categorie_territoire',
'sau_productive_ha',
'population_totale_2017'
]]
donnees_indicateur.loc[:, 'sau_par_habitant_valeur_m2_par_hab'] = (
donnees_indicateur.apply(
lambda x:
valeur(
sau_par_habitant_valeur_m2_par_hab(
x['sau_productive_ha'],
x['population_totale_2017']
),
axis=1
)
for t in indicateur:
indicateur[t]['sau_par_habitant_note'] = indicateur[t].apply(
)
donnees_indicateur.loc[:, 'sau_par_habitant_note'] = (
donnees_indicateur.apply(
lambda x:
outils.calculerNoteParInterpolation3Seuils(
x['sau_par_habitant_valeur_m2_par_hab'],
0, seuil_regime_tres_vegetal, seuil_regime_actuel,
0, 5, 10,
0, 5, 10
),
axis=1
)
)
return indicateur
return donnees_indicateur
def valeur(sau_productive_ha, population_totale):
def sau_par_habitant_valeur_m2_par_hab(sau_productive_ha, population_totale):
try:
x = round(
return round(
sau_productive_ha * 1e4 / population_totale
)
except:
x = ""
return x
return ""
from pandas import DataFrame
import numpy as np
import crater.territoires.outils as outils_territoires
import crater.outils as outils
from crater.crater_logger import log
from crater.politique_fonciere.parametres import seuil_regime_tres_vegetal, seuil_regime_moins_carne, seuil_regime_actuel
def calculer(
indicateur_sau_par_habitant,
indicateur_rythme_artificialisation_sau
):
indicateur_sau_par_habitant: DataFrame,
indicateur_rythme_artificialisation_sau: DataFrame
) -> DataFrame:
log.info(" => Synthèse")
synthese = outils_territoires.joindre(
outils_territoires.copier_en_gardant_colonnes(
indicateur_sau_par_habitant,
['sau_par_habitant_valeur_m2_par_hab', 'sau_par_habitant_note']
),
outils_territoires.copier_en_gardant_colonnes(
indicateur_rythme_artificialisation_sau,
['artificialisation_2011_2016_ha', 'rythme_artificialisation_sau_note']
)
)
calculer_notes_messages(synthese)
indicateur_sau_par_habitant = indicateur_sau_par_habitant.loc[:,
['nom_territoire',
'categorie_territoire',
'sau_par_habitant_valeur_m2_par_hab',
'sau_par_habitant_note'
]]
indicateur_rythme_artificialisation_sau = indicateur_rythme_artificialisation_sau.loc[:,
['artificialisation_2011_2016_ha',
'rythme_artificialisation_sau_note'
]]
return synthese
donnees_synthese = indicateur_sau_par_habitant.join(indicateur_rythme_artificialisation_sau)
donnees_synthese = calculer_notes_messages(donnees_synthese)
def calculer_notes_messages(territoires):
for t in territoires:
territoires[t]['synthese_note'] = territoires[t].apply(
return donnees_synthese
def calculer_notes_messages(donnees_synthese: DataFrame) -> DataFrame:
donnees_synthese.loc[:, 'synthese_note'] = (
donnees_synthese.apply(
lambda x:
note(
x['sau_par_habitant_note'],
......@@ -36,7 +41,10 @@ def calculer_notes_messages(territoires):
),
axis=1
)
territoires[t]['synthese_message'] = territoires[t].apply(
)
donnees_synthese.loc[:, 'synthese_message'] = (
donnees_synthese.apply(
lambda x:
message(
x['sau_par_habitant_valeur_m2_par_hab'],
......@@ -45,7 +53,9 @@ def calculer_notes_messages(territoires):
),
axis=1
)
)
return donnees_synthese
def note(
sau_par_habitant_note,
......
from pathlib import Path
import pandas as pd
from pandas import DataFrame
import numpy as np
import crater.population_agricole.chargeur_actifs_agricoles_permanents as chargeur_population_agricole
import crater.population_agricole.evolution_part_population_agricole as evolution_part_population_agricole
......@@ -8,16 +10,16 @@ import crater.population_agricole.part_population_agricole_2010 as part_populati
import crater.population_agricole.synthese_population_agricole as synthese_population_agricole
import crater.territoires.outils as outils_territoires
from crater.crater_logger import log
from crater.outils import reinitialiser_dossier, DictTerritoires, supprimer_donnees_apres_annee_mouvement_commune
from crater.territoires.outils import traduire_code_insee_vers_id_commune_crater
from crater.outils import reinitialiser_dossier, DictDataFrames
from crater.territoires.outils import traduire_code_insee_vers_id_commune_crater, ids_echelles_geographiques_supracommunales
def calculer_population_agricole(
referentiel_territoires,
chemin_fichier_actifs_agricoles_permanents,
chemin_dossier_output,
dossier_source,
dossier_population_agricole
territoires: DataFrame,
chemin_fichier_actifs_agricoles_permanents: Path,
chemin_dossier_output: Path,
dossier_source: Path,
dossier_population_agricole: Path
) -> None:
sources = charger_sources(
chemin_fichier_actifs_agricoles_permanents
......@@ -26,62 +28,65 @@ def calculer_population_agricole(
sources,
chemin_dossier_output / dossier_source / dossier_population_agricole
)
axe = calculer(referentiel_territoires, sources)
axe = calculer(territoires, sources)
exporter(
axe,
chemin_dossier_output / dossier_population_agricole
)
def calculer(referentiel_territoires, sources_population_agricole):
def calculer(
territoires: DataFrame,
sources_population_agricole: DictDataFrames
) -> DictDataFrames:
log.info("##### CALCUL POPULATION AGRICOLE #####")
donnees_population_agricole: DictTerritoires = calculer_donnees_population_agricole(referentiel_territoires,
sources_population_agricole[
'actifs_agricoles_permanents'],
2011)
indicateur_part_population_agricole_2010 = part_population_agricole_2010.calculer(donnees_population_agricole)
indicateur_part_population_agricole_1988 = part_population_agricole_1988.calculer(donnees_population_agricole)