Commit 2275c039 authored by Lionel's avatar Lionel
Browse files

Suppression test-qualite-donnees et fonctions de verification dans crater_data.py

parent 4d3850d4
import argparse
from crater.crater_logger import log
import crater.politique_fonciere.calculateur as calculateur_politique_fonciere
import crater.population_agricole.calculateur as calculateur_population_agricole
from crater.outils import comparer_valeurs, reinitialiser_dossier
from crater.surface_agricole_utile import collecteur_rpg, calculateur_surface_agricole_utile
from crater.crater_logger import log
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.territoires.calculateur import calculer_territoires, charger_referentiel_territoires
import crater.politique_fonciere.calculateur as calculateur_politique_fonciere
import crater.population_agricole.calculateur as calculateur_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 (
FICHIER_LOG,
CHEMIN_SOURCE_COMMUNES_ET_ARRONDISSEMENTS, CHEMIN_SOURCE_COMMUNES_SIG, CHEMIN_SOURCE_MOUVEMENTS_COMMUNES,
CHEMIN_SOURCE_EPCIS, CHEMIN_SOURCE_DEPARTEMENTS, CHEMIN_SOURCE_REGIONS, CHEMIN_SOURCE_CODES_POSTAUX,
CHEMIN_SOURCE_POPULATION_TOTALE, CHEMIN_SOURCE_POPULATION_TOTALE_HISTORIQUE,
CHEMIN_SOURCE_POPULATION_AGRICOLE,
CHEMIN_SOURCE_ARTIFICIALISATION,
CHEMIN_SOURCE_HVN,
DOSSIER_SOURCE_IGN,
URL_SOURCE_RPG,
NOMS_FICHIERS_RPG,
CHEMIN_OUTPUT_DATA,
......@@ -30,6 +24,8 @@ from crater.reglages import (
CHEMIN_CRATER_DATA_SOURCES,
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 import calculer_territoires, charger_referentiel_territoires
def calculer_population_agricole(
......@@ -38,7 +34,7 @@ def calculer_population_agricole(
chemin_dossier_output,
dossier_source,
dossier_population_agricole
):
) -> None:
sources = calculateur_population_agricole.charger_sources(
chemin_fichier_actifs_agricoles_permanents
)
......@@ -52,8 +48,6 @@ def calculer_population_agricole(
chemin_dossier_output / dossier_population_agricole
)
return axe, sources
def calculer_politique_fonciere(
referentiel_territoires,
......@@ -61,7 +55,7 @@ def calculer_politique_fonciere(
chemin_dossier_output,
dossier_source,
dossier_politique_fonciere
):
) -> None:
sources = calculateur_politique_fonciere.charger_sources(
chemin_fichier_artificialisation
)
......@@ -78,113 +72,6 @@ def calculer_politique_fonciere(
chemin_dossier_output / dossier_politique_fonciere
)
return axe, sources
def verifier_donnees(
referentiel_territoires,
sources_territoires,
population_agricole,
sources_population_agricole,
politique_fonciere,
sources_politique_fonciere,
chemin_fichier_log
):
log.info("\n##### VERIFICATION DES DONNEES #####")
log.info(" => Export des vérifications vers %s", chemin_fichier_log)
with open(chemin_fichier_log, "a") as fichier:
fichier.write("*** INFORMATIONS GENERALES ***\n")
fichier.write(str(referentiel_territoires['communes'].shape[0]) + " communes\n")
fichier.write(str(referentiel_territoires['departements'].shape[0]) + " départements\n")
fichier.write(str(referentiel_territoires['epcis'].shape[0]) + " epcis\n")
fichier.write(str(referentiel_territoires['regions'].shape[0]) + " regions\n")
fichier.write(str(referentiel_territoires['pays'].shape[0]) + " pays\n")
fichier.write("\n*** VERIFICATION DES CODES POSTAUX ***\n")
fichier.write(verification_donnee(
referentiel_territoires,
sources_territoires['codes_postaux'],
'codes_postaux'
))
fichier.write("\n\n*** VERIFICATION DES SAU ***")
fichier.write("\n" + verification_donnee(
referentiel_territoires,
sources_territoires['sau'],
'sau_ha',
True
))
fichier.write("\n\n*** VERIFICATION DES POPULATIONS TOTALES ***")
fichier.write("\n" + verification_donnee(
referentiel_territoires,
sources_territoires['population_totale_historique'],
'population_totale_1990',
True
))
fichier.write("\n" + verification_donnee(
referentiel_territoires,
sources_territoires['population_totale_historique'],
'population_totale_2010',
True
))
fichier.write("\n" + verification_donnee(
referentiel_territoires,
sources_territoires['population_totale'],
'population_totale_2017',
True
))
fichier.write("\n\n*** VERIFICATION POPULATION AGRICOLE ***")
fichier.write("\n" + verification_donnee(
population_agricole['donnees'],
sources_population_agricole['actifs_agricoles_permanents'],
'actifs_agricoles_permanents_2010',
True
))
fichier.write("\n" + verification_donnee(
population_agricole['donnees'],
sources_population_agricole['actifs_agricoles_permanents'],
'actifs_agricoles_permanents_1988',
True
))
fichier.write("\n\n*** VERIFICATION ARTIFICIALISATION ***")
fichier.write("\n" + verification_donnee(
politique_fonciere['donnees'],
sources_politique_fonciere['artificialisation'],
'artificialisation_2011_2016_ha',
True
))
fichier.write("\n" + verification_donnee(
politique_fonciere['donnees'],
sources_politique_fonciere['artificialisation'],
'evolution_menages_2011_2016',
True
))
fichier.write("\n" + verification_donnee(
politique_fonciere['donnees'],
sources_politique_fonciere['artificialisation'],
'evolution_emplois_2011_2016',
True
))
def verification_donnee(territoires, source, colonne, verifier_totaux=False):
communes_sans_donnee = list(territoires['communes'][territoires['communes'][colonne].isnull()]['id_commune'])
communes_donnee_nulle = list(territoires['communes'][territoires['communes'][colonne] == 0]['id_commune'])
r = "\n" + str(len(communes_sans_donnee)) + " commune(s) sans " + colonne + \
"\n" + ", ".join(communes_sans_donnee)
if verifier_totaux:
r = r + comparer_valeurs(
round(source[colonne].abs().sum(), 3),
round(territoires['communes'][colonne].abs().sum(), 3)
)
r = r + "\n" + str(len(communes_donnee_nulle)) + " commune(s) avec " + colonne + " = 0"
return r
def lancer_calcul_surface_agricole_utile():
referentiel_territoires = charger_referentiel_territoires(
......@@ -209,8 +96,7 @@ def lancer_calcul_surface_agricole_utile():
def generer_fichiers_resultats_crater():
open(CHEMIN_OUTPUT_DATA / FICHIER_LOG, "w").close()
referentiel_territoires, sources_territoires = calculer_territoires(
referentiel_territoires = calculer_territoires(
CHEMIN_SOURCE_COMMUNES_ET_ARRONDISSEMENTS,
CHEMIN_SOURCE_MOUVEMENTS_COMMUNES,
CHEMIN_SOURCE_EPCIS,
......@@ -223,21 +109,22 @@ def generer_fichiers_resultats_crater():
CHEMIN_CRATER_DATA_SOURCES / DOSSIER_PARCEL / 'correspondance_territoires_parcel_crater.csv',
CHEMIN_OUTPUT_DATA / DOSSIER_TERRITOIRES
)
population_agricole, sources_population_agricole = calculer_population_agricole(
calculer_population_agricole(
referentiel_territoires,
CHEMIN_SOURCE_POPULATION_AGRICOLE,
CHEMIN_OUTPUT_DATA,
DOSSIER_SOURCES,
DOSSIER_POPULATION_AGRICOLE
)
politique_fonciere, sources_politique_fonciere = calculer_politique_fonciere(
calculer_politique_fonciere(
referentiel_territoires,
CHEMIN_SOURCE_ARTIFICIALISATION,
CHEMIN_OUTPUT_DATA,
DOSSIER_SOURCES,
DOSSIER_POLITIQUE_FONCIERE
)
pratiques_agricoles = calculer_pratiques_agricoles(
calculer_pratiques_agricoles(
referentiel_territoires,
CHEMIN_SOURCE_HVN,
CHEMIN_SOURCE_SAU_BIO,
......@@ -255,16 +142,6 @@ def generer_fichiers_resultats_crater():
CHEMIN_OUTPUT_DATA / DOSSIER_PRODUCTION_BESOINS
)
verifier_donnees(
referentiel_territoires,
sources_territoires,
population_agricole,
sources_population_agricole,
politique_fonciere,
sources_politique_fonciere,
CHEMIN_OUTPUT_DATA / FICHIER_LOG
)
def lancer_generation_territoires():
calculer_territoires(
......@@ -348,26 +225,6 @@ def lancer_calcul_production_besoins():
)
def lancer_verifications():
territoires = charger_referentiel_territoires(
CHEMIN_OUTPUT_DATA / DOSSIER_TERRITOIRES)
# verifier_donnees(
# territoires,
# sources_territoires,
# population_agricole,
# sources_population_agricole,
# politique_fonciere,
# sources_politique_fonciere,
# modes_de_production,
# sources_modes_de_production,
# CHEMIN_OUTPUT_DATA / FICHIER_LOG
# )
# )
reinitialiser_dossier(CHEMIN_OUTPUT_DATA / DOSSIER_PRODUCTION_BESOINS)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser = argparse.ArgumentParser(usage='%(prog)s ETAPE [options]\n'
......@@ -403,7 +260,8 @@ if __name__ == '__main__':
args = parser.parse_args()
if (args.ETAPE == 'parcel-besoins-assiette-actuelle'):
log.info("Génération du fichier des besoins SAUs pour les territoires à partir de l'API PARCEL, assiette actuelle")
log.info(
"Génération du fichier des besoins SAUs pour les territoires à partir de l'API PARCEL, assiette actuelle")
collecteur_besoins_api_parcel.collecter_donnees_besoins_parcel(
ID_ASSIETTE_ACTUELLE,
CHEMIN_CRATER_DATA_SOURCES / DOSSIER_PARCEL / 'correspondance_territoires_parcel_crater.csv',
......@@ -428,7 +286,7 @@ if __name__ == '__main__':
CHEMIN_CRATER_DATA_SOURCES / DOSSIER_PARCEL / 'parcel_produits_et_categories_202008.csv',
CHEMIN_OUTPUT_DATA / DOSSIER_PARCEL / 'reponse_api_parcel_besoins_sau_par_produits_assiette_moins_50p.csv'
)
elif(args.ETAPE == 'rpg'):
elif (args.ETAPE == 'rpg'):
log.info("Téléchargmeent des fichiers source RPG")
collecteur_rpg.telecharger_rpg_toutes_les_regions(
URL_SOURCE_RPG,
......
......@@ -14,7 +14,7 @@ def calculer_pratiques_agricoles(
chemin_fichier_hvn: Path,
chemin_fichier_surfaces_bio: Path,
chemin_dossier_output: Path
) -> Territoires:
) -> None:
log.info("##### CALCUL PRATIQUES AGRICOLES #####")
donnees_pratiques_agricoles = outils_territoires.assembler_dict_territoires_en_df_territoires(
......@@ -42,8 +42,6 @@ def calculer_pratiques_agricoles(
chemin_dossier_output
)
return donnees_pratiques_agricoles
def _exporter(
donnees_pratiques_agricoles: Territoires,
......
......@@ -70,8 +70,6 @@ DOSSIER_POLITIQUE_FONCIERE = Path("politique_fonciere")
DOSSIER_PRATIQUES_AGRICOLES = Path("pratiques_agricoles")
DOSSIER_PRODUCTION_BESOINS = Path("production_besoins")
FICHIER_LOG = Path("log.txt")
URL_SOURCE_RPG = "ftp://RPG_ext:quoojaicaiqu6ahD@ftp3.ign.fr/"
NOMS_FICHIERS_RPG = {
2019: {
......
......@@ -400,7 +400,7 @@ def calculer_territoires(
chemin_fichiers_sau,
chemin_fichier_territoires_parcel,
chemin_dossier_output
):
) -> DictTerritoires:
sources = charger_sources(
chemin_fichier_communes_et_arrondissements,
chemin_fichier_mouvements_communes,
......@@ -422,7 +422,7 @@ def calculer_territoires(
chemin_dossier_output
)
return territoires, sources
return territoires
# TODO : rendre privé, ne plus utiliser cela depuis les autres modules
......
......@@ -24,7 +24,7 @@ class TestTerritoires(unittest.TestCase):
CHEMIN_INPUT_DATA / DOSSIER_SURFACE_AGRICOLE_UTILE / 'sau_par_commune_et_culture',
CHEMIN_INPUT_DATA / DOSSIER_PARCEL / 'correspondance_territoires_parcel_crater.csv',
CHEMIN_OUTPUT_DATA / DOSSIER_TERRITOIRES
)[0]
)
# when
dataframes_territoires_charges: DictTerritoires = calculateur.charger_referentiel_territoires(
CHEMIN_OUTPUT_DATA / DOSSIER_TERRITOIRES
......
import numpy
from pandas import DataFrame, Series
from crater.crater_logger import creer_logger
log = creer_logger("tests_qualite_donnees", 'crater_tests_qualite_donnees.log')
def verifier_ordre_de_grandeur(valeur_indicateur, valeur_attendue, marge_erreur, nom_indicateur):
log.debug(f"VERIFICATION ORDRE DE GRANDEUR pour l'indicateur : {nom_indicateur}")
log.debug \
(f"\tvaleur_indicateur={valeur_indicateur}, valeur_attendue={valeur_attendue}, marge_erreur={marge_erreur}")
ecart_existant = abs(valeur_indicateur - valeur_attendue)
if ecart_existant < marge_erreur:
log.debug(
f"\tRésultat OK : La différence entre la valeur de l'indicateur et la valeur attendue ne dépasse pas la marge d'erreur => {ecart_existant} < {marge_erreur}")
else:
message_erreur = f"\tERREUR : l'écart mesuré dépasse la marge d'erreur => {ecart_existant} > {marge_erreur}"
log.debug(message_erreur)
raise AssertionError(message_erreur)
def verifier_egalite(valeur_indicateur, valeur_attendue, nom_indicateur):
log.debug(f"VERIFICATION EGALITE pour l'indicateur : {nom_indicateur}")
log.debug(f"\tvaleur_indicateur={valeur_indicateur}, valeur_attendue={valeur_attendue}")
if valeur_indicateur == valeur_attendue:
log.debug(f"\tRésultat OK : valeur de l'indicateur ({valeur_indicateur}) = valeur attendue {valeur_attendue}")
else:
message_erreur = f"\tERREUR : valeur de l'indicateur ({valeur_indicateur}) != valeur attendue {valeur_attendue}"
log.debug(message_erreur)
raise AssertionError(message_erreur)
def verifier_seuil_territoires_manquants_dans_df_indicateur(df_territoires: DataFrame, df_donnees: DataFrame,
nom_indicateur, seuil_erreur) -> None:
log.debug(f"VERIFICATION TERRITOIRES MANQUANTS dans df indicateur : {nom_indicateur}")
territoires_absents_du_df_donnees = df_territoires[~df_territoires.iloc[:, 0].isin(df_donnees['id_territoire'])]
id_territoires_manquants_str = ", ".join(list(territoires_absents_du_df_donnees.iloc[:, 0]))
if territoires_absents_du_df_donnees.shape[0] <= seuil_erreur:
log.debug(
f"\tRésultat OK : nb de territoires manquants dans le dataframe de l'indicateur {nom_indicateur} = {territoires_absents_du_df_donnees.shape[0]} (<= au seuil d'erreur {seuil_erreur})")
log.debug(f"\tListe id_territoires manquants : {id_territoires_manquants_str}")
else:
message_erreur = f"\tERREUR : il manque les données de {territoires_absents_du_df_donnees.shape[0]} territoires pour l'indicateur {nom_indicateur} (> au seuil d'erreur {seuil_erreur})\n" \
+ id_territoires_manquants_str
log.debug(message_erreur)
raise AssertionError(message_erreur)
def verifier_donnees_nan_ou_zero(df_donnees: DataFrame, nom_colonne_a_verifier: str, nom_indicateur: str,
seuil_erreur: int) -> None:
log.debug(f"VERIFICATION ZERO et NAN dans df indicateur : {nom_indicateur}")
# nb_valeurs_nan = df_donnees[nom_colonne_a_verifier].isna().sum()
df_valeur_nulle_ou_nan = df_donnees.loc[
(df_donnees[nom_colonne_a_verifier] == 0) | (df_donnees[nom_colonne_a_verifier].isna())]
liste_id_territoires_str = ", ".join(list(df_valeur_nulle_ou_nan['id_territoire']))
if df_valeur_nulle_ou_nan.shape[0] <= seuil_erreur:
log.debug(
f"\tRésultat OK : nb de valeurs nulles ou nan pour l'indicateur {nom_indicateur} = {df_valeur_nulle_ou_nan.shape[0]} (<= au seuil d'erreur {seuil_erreur})")
log.debug(f"\tListe id_territoires concernés : {liste_id_territoires_str}")
else:
message_erreur = f"\tERREUR : il y a {df_valeur_nulle_ou_nan.shape[0]} territoires avec des données nulles ou nan pour l'indicateur {nom_indicateur} (> au seuil d'erreur {seuil_erreur})\n" \
+ liste_id_territoires_str
log.debug(message_erreur)
raise AssertionError(message_erreur)
# def verifier_donnees_nan_ou_zero(df_donnees: DataFrame, nom_colonne_a_verifier: str, nom_indicateur: str,
# seuil_erreur: int) -> None:
# log.debug(f"VERIFICATION ZERO et NAN dans df indicateur : {nom_indicateur}")
# # nb_valeurs_nan = df_donnees[nom_colonne_a_verifier].isna().sum()
# df_valeur_nulle_ou_nan = df_donnees.loc[
# (df_donnees[nom_colonne_a_verifier] == 0) | (df_donnees[nom_colonne_a_verifier].isna())]
# liste_id_territoires_str = ", ".join(list(df_valeur_nulle_ou_nan['id_territoire']))
# if df_valeur_nulle_ou_nan.shape[0] <= seuil_erreur:
# log.debug(
# f"\tRésultat OK : nb de valeurs nulles ou nan pour l'indicateur {nom_indicateur} = {df_valeur_nulle_ou_nan.shape[0]} (<= au seuil d'erreur {seuil_erreur})")
# log.debug(f"\tListe id_territoires concernés : {liste_id_territoires_str}")
# else:
# message_erreur = f"\tERREUR : il y a {df_valeur_nulle_ou_nan.shape[0]} territoires avec des données nulles ou nan pour l'indicateur {nom_indicateur} (> au seuil d'erreur {seuil_erreur})\n" \
# + liste_id_territoires_str
# log.debug(message_erreur)
# raise AssertionError(message_erreur)
def verifier_donnees_nan(df_donnees: DataFrame, nom_colonne_a_verifier: str, seuil_erreur: int) -> None:
df_valeur_nan = df_donnees.loc[df_donnees[nom_colonne_a_verifier].isna()]
_verifier_depassement_seuil("NaN", df_valeur_nan, nom_colonne_a_verifier, seuil_erreur)
def verifier_donnees_nulles(df_donnees: DataFrame, nom_colonne_a_verifier: str, seuil_erreur: int) -> None:
df_valeur_nulle = df_donnees.loc[(df_donnees[nom_colonne_a_verifier] == 0)]
_verifier_depassement_seuil("0", df_valeur_nulle, nom_colonne_a_verifier, seuil_erreur)
def verifier_donnees_inf(df_donnees: DataFrame, nom_colonne_a_verifier: str, seuil_erreur: int) -> None:
df_valeur_inf = df_donnees.loc[numpy.isinf(df_donnees[nom_colonne_a_verifier])]
_verifier_depassement_seuil("inf", df_valeur_inf, nom_colonne_a_verifier, seuil_erreur)
def _verifier_depassement_seuil(nom_valeur_a_verifier, df_valeur_nan, nom_colonne_a_verifier, seuil_erreur):
log.debug(f"Vérification nb de cellules = {nom_valeur_a_verifier} pour la colonne {nom_colonne_a_verifier}")
liste_id_territoires_str = ", ".join(list(df_valeur_nan['id_territoire']))
if df_valeur_nan.shape[0] <= seuil_erreur:
log.debug(
f"\tRésultat OK : {df_valeur_nan.shape[0]} cellules ont une valeur = {nom_valeur_a_verifier} dans la colonne {nom_colonne_a_verifier} (<= au seuil d'erreur {seuil_erreur})")
log.debug(f"\tListe id_territoires concernés : {liste_id_territoires_str}")
else:
message_erreur = f"\tERREUR : {df_valeur_nan.shape[0]} cellulles ont une valeur = {nom_valeur_a_verifier} dans la colonne {nom_colonne_a_verifier} (> au seuil d'erreur {seuil_erreur})\n" \
+ liste_id_territoires_str
log.debug(message_erreur)
raise AssertionError(message_erreur)
def verifier_stats_descriptives(series: Series, nom_indicateur_colonne: str, taille=None, moyenne=None, min=None,
max=None) -> None:
log.debug(f"VERIFICATION STATISTIQUES DESCRIPTIVES pour indicateur : {nom_indicateur_colonne}")
message_erreur = verifier_indicateur_statistique(series, 'count', taille)
message_erreur += verifier_indicateur_statistique(series, 'mean', moyenne)
message_erreur += verifier_indicateur_statistique(series, 'min', min)
message_erreur += verifier_indicateur_statistique(series, 'max', max)
if message_erreur != '':
raise AssertionError(message_erreur)
def verifier_indicateur_statistique(series, indicateur_statistique, valeur_attendue):
message_erreur = ''
statistiques = series.describe()
if (valeur_attendue is None) | (statistiques[indicateur_statistique] == valeur_attendue):
log.debug(f"\tStatistique OK : {indicateur_statistique} = ({statistiques[indicateur_statistique]})")
else:
message_erreur = f"\tERREUR : valeur statistique constatée pour {indicateur_statistique} ({statistiques[indicateur_statistique]}) différente celle attendue ({valeur_attendue})"
log.debug(message_erreur)
message_erreur += '\n'
return message_erreur
# def comparer_valeurs(a, b):
# return \
# "Valeur cible = " + str(a) + \
# " - Valeur testée = " + str(b) + \
# " - Différence absolue = " + str(a - b) + \
# " - Différence relative = " + str(round((b / a - 1) * 100, 1)) + "%"
# def comparer_listes(a, b):
# return \
# "\n" + str(len(set(a) - set(b))) + " éléments manquants d'un côté : " + ", ".join(set(a) - set(b)) + \
# "\n" + str(len(set(b) - set(a))) + " éléments manquants de l'autre côté : " + ", ".join(set(b) - set(a))
import unittest
from pathlib import Path
import pandas
from matplotlib import pyplot
from crater.reglages import DOSSIER_PRODUCTION_BESOINS, DOSSIER_PRATIQUES_AGRICOLES
from tests_qualite_donnees.outils_qualite_donnee import verifier_stats_descriptives
CHEMIN_DONNEES_SOURCES = Path('../../crater-data-sources')
CHEMIN_DONNEES_RESULTATS = Path('../../crater-data-resultats')
class TestGenerationDiagrammes(unittest.TestCase):
def setUp(self):
self.df_production_besoins = pandas.read_csv(
CHEMIN_DONNEES_RESULTATS / DOSSIER_PRODUCTION_BESOINS / 'production_besoins.csv',
sep=";"
)
self.df_pratiques_agricoles = pandas.read_csv(
CHEMIN_DONNEES_RESULTATS / DOSSIER_PRATIQUES_AGRICOLES / 'pratiques_agricoles.csv',
sep=";"
)
def test_histogrammes(self):
# distribution_notes = pandas.DataFrame()
# departements = self.df_production_besoins.loc[self.df_production_besoins['categorie_territoire'] == 'DEPARTEMENT'].copy()
# distribution_notes['departements'] = pandas.cut(departements['note_production_besoins'], bins=10).value_counts().sort_index()
# regions = self.df_production_besoins.loc[self.df_production_besoins['categorie_territoire'] == 'REGION'].copy()
# distribution_notes['regions'] = pandas.cut(regions['note_production_besoins'], bins=10).value_counts().sort_index()
# epcis = self.df_production_besoins.loc[self.df_production_besoins['categorie_territoire'] == 'EPCI'].copy()
# distribution_notes['epcis'] = pandas.cut(epcis['note_production_besoins'], bins=10).value_counts().sort_index()
# hist = distribution_notes.plot.hist(alpha=0.5)
# hist = distribution_notes.plot.area(alpha=0.5, stacked=False)
# df4.plot.hist(alpha=0.5)
hist = self.df_production_besoins.loc[self.df_production_besoins['categorie_territoire'].isin(['COMMUNE'])][
'note_production_besoins'].hist(bins=10)
# fig = hist.get_figure()
# fig.savefig('diagrammes/notes_production_besoins_epci.png')
fig = hist.get_figure()
fig.savefig('diagrammes/notes_production_besoins_pays_region_departement.png')
pyplot.show()
def test_generation_histogrammes_part_sau_bio_valeur_pourcent(self):
dataframe = self.df_pratiques_agricoles[self.df_pratiques_agricoles.part_sau_bio_note == 10]
nom_colonne = 'part_sau_bio_note'
self._generer_histogramme_note(dataframe, nom_colonne)
def _generer_histogramme_note(self, dataframe, nom_colonne):
verifier_stats_descriptives(dataframe[nom_colonne], nom_colonne)
hist = dataframe[nom_colonne].hist(bins=10)
# fig = hist.get_figure()
# fig.savefig('diagrammes/notes_production_besoins_epci.png')
fig = hist.get_figure()
fig.savefig(f'diagrammes/notes_{nom_colonne}.png')
pyplot.show()
import unittest
from pathlib import Path
import pandas
from crater.reglages import DOSSIER_PRATIQUES_AGRICOLES
from tests_qualite_donnees.outils_qualite_donnee import verifier_donnees_nulles, verifier_donnees_nan, \
verifier_donnees_inf
CATEGORIES_TERRITOIRES = ['COMMUNE', 'EPCI', 'DEPARTEMENT', 'REGION', 'PAYS']
CHEMIN_DONNEES_SOURCES = Path('../../crater-data-sources')
CHEMIN_DONNEES_RESULTATS = Path('../../crater-data-resultats')
class TestQualitePratiquesAgricoles(unittest.TestCase):
def setUp(self):
self.df_production_besoins = pandas.read_csv(
CHEMIN_DONNEES_RESULTATS / DOSSIER_PRATIQUES_AGRICOLES / 'pratiques_agricoles.csv',
sep=";"
)
def test_territoires_avec_sau_bio_confidentielle(self):
for categorie in CATEGORIES_TERRITOIRES:
with self.subTest(categorie=categorie):
verifier_donnees_nan(
self.df_production_besoins[self.df_production_besoins.categorie_territoire == categorie],
"surface_bio_ha", 0)
def test_territoires_avec_sau_bio_nulle(self):
for categorie in CATEGORIES_TERRITOIRES:
with self.subTest(categorie=categorie):
verifier_donnees_nulles(
self.df_production_besoins[self.df_production_besoins.categorie_territoire == categorie],
"surface_bio_ha", 0)
def test_territoires_avec_part_sau_bio_infinie(self):
for categorie in CATEGORIES_TERRITOIRES:
with self.subTest(categorie=categorie):
verifier_donnees_inf(
self.df_production_besoins[self.df_production_besoins.categorie_territoire == categorie],
"part_sau_bio_valeur_pourcent", 0)
def test_notes_pratiques_agricoles_nan(self):
for categorie in CATEGORIES_TERRITOIRES:
with self.subTest(categorie=categorie):
verifier_donnees_nan(
self.df_production_besoins[self.df_production_besoins.categorie_territoire == categorie],
"synthese_note", 0)
def test_notes_pratiques_agricoles_nan(self):
for categorie in CATEGORIES_TERRITOIRES:
with self.subTest(categorie=categorie):
verifier_donnees_nan(
self.df_production_besoins[self.df_production_besoins.categorie_territoire == categorie],
"synthese_note", 0)
import unittest
from pathlib import Path
import pandas
from crater.reglages import DOSSIER_PRODUCTION_BESOINS, DOSSIER_TERRITOIRES
from crater.territoires.calculateur import charger_referentiel_territoires
from tests_qualite_donnees.outils_qualite_donnee import verifier_ordre_de_grandeur, \