CEN38/Python_scripts
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Python_scripts/5_GEONATURE/GN_ZH/MEDWET2Geonat.py

1338 lines
55 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
#Nom : : Medwet_to_Geonature.py
#Description :
#Copyright : 2022, CEN 74
#Auteur : Eric Lafon - largement inspiré du script recup_zh_from_medwet.py de Colas Geier pour le CEN 38 (2021)
#Version : 1.1
'''Script ayant pour vocation de récolter les informations des tables de la base ZHRMC (Données Medwet)
et de les stocker sur des feuilles Csv en vue de leur implémentation dans Géonature.'''
#import des librairies
import pandas as pd # pour manipulation des données de la bd medwet
import geopandas as gpd # pour manipulation des données postgis
import pandas_access as mdb # pour lecture de la bd medwet
import numpy as np
import os # pour écriture des fichiers dans un répertoire
from sqlalchemy import create_engine # pour lecture de la bd Géonature
from sqlalchemy.engine import URL # création url conection Geonature
from datetime import datetime as dt # pour transformation des types date en type datetime
import uuid # pour générer des uuid
from geoalchemy2 import Geometry # pour les insertions de geometry
import sys, inspect
# realpath() will make your script run, even if you symlink it :)
cmd_folder = os.path.realpath(os.path.abspath(os.path.split(inspect.getfile( inspect.currentframe() ))[0]))
if cmd_folder not in sys.path:
sys.path.insert(0, cmd_folder)
from tools.pr_zh import to_tzh,to_t_,to_cor_,add_bib_cb,update_t_ownership
#lecture de la BD ZHRMC et des fichiers shapefile qui y sont liés
# DIR = '/home/colas/Documents/13_CEN74/medwet2gn_ZH/'
DIR = os.path.dirname(cmd_folder+'/')
os.chdir(DIR)
db_file = './bdd/Inventaires ZHRMCvivante.mdb'
shp_siteinfo = 'Cartographie_generee/CarteZHGénérée.shp'
#création d'une connection à la bd de géonature
usr = 'geonatadmin'
pdw='g;gCEN74'
host='178.33.42.38'
bd ='geonature2db'
eng = URL.create('postgresql+psycopg2',username=usr,password=pdw,host=host,database=bd)
conn = create_engine(eng)
#création d'une connection à la bd foncière
# Utile en cas d'utilisation de la fonction update_t_ownership()
usr_f = 'admin_ra'
pdw_f ='adminRAsig'
host_f ='91.134.194.223'
bd_f = 'bd_cen_74'
eng_f = URL.create('postgresql+psycopg2',username=usr_f,password=pdw_f,host=host_f,database=bd_f,)
con_f = create_engine(eng_f)
LB_IGNORE = ['99','9','9_1']
LB_DICT = {
'53_1':'53',
'89.26':'89.23',
}
##############################################
# fonctions essentielles pour requêter ZHRMC #
##############################################
def _aggregate_value__(column1,column2,sep='\n'):
"""
aggregation de la colonne 1 et 2 dans la colonne 1
"""
df = pd.merge(column1,column2,left_index=True,right_index=True,how='left')
df['join'] = [
sep.join([y for y in [row.iloc[0],row.iloc[1]] if not pd.isna(y)])
if not (pd.isna(row.iloc[0]) and pd.isna(row.iloc[1])) else None
for x,row in df.iterrows()
]
return df['join']
def t_nomenclature_ZH(bib_mnemo=None):
"""@bib_mnemo : ['CRIT_DELIM', 'CRIT_DEF_ESP_FCT', 'ACTIV_HUM', 'LOCALISATION',
'IMPACTS', 'EVAL_GLOB_MENACES', 'ENTREE_EAU', 'SORTIE_EAU',
'PERMANENCE_ENTREE', 'PERMANENCE_SORTIE', 'SUBMERSION_FREQ',
'SUBMERSION_ETENDUE', 'TYPE_CONNEXION', 'FONCTIONNALITE_HYDRO',
'FONCTIONNALITE_BIO', 'FONCTIONS_HYDRO', 'FONCTIONS_QUALIF',
'FONCTIONS_CONNAISSANCE', 'FONCTIONS_BIO', 'VAL_SOC_ECO',
'INTERET_PATRIM', 'ETAT_CONSERVATION', 'STATUT_PROPRIETE',
'PLAN_GESTION', 'INSTRU_CONTRAC_FINANC', 'NIVEAU_PRIORITE',
'SDAGE', 'SAGE', 'STATUT_PROTECTION', 'IMPACT_TYP',
'OCCUPATION_SOLS', 'NIV_PROTECTION', 'PROTECTION_TYP',
'TYP_DOC_COMM', 'TYP_CLASS_COMM', 'STRAT_GESTION', 'HIERARCHY']
"""
sch = 'ref_nomenclatures'
tab = ['t_nomenclatures','bib_nomenclatures_types']
sql = """
SELECT
a.id_nomenclature,
a.cd_nomenclature,
a.mnemonique mnemo,
a.label_default as label,
a.definition_default def,
b.mnemonique bib_mnemo,
b.label_default bib_label,
b.definition_default bib_def,
a.active
FROM {sch}.{tab0} a
JOIN {sch}.{tab1} b USING (id_type)
WHERE b."source" = 'ZONES_HUMIDES'
""".format(sch=sch,tab0=tab[0],tab1=tab[1])
if bib_mnemo is not None:
sql += " AND b.mnemonique = '%s'"%bib_mnemo
return pd.read_sql_query(sql,conn).replace({r'[’]':"'"},regex=True)
#Définition de Standardized_t_nomenclature_ZH(): permet de sélectionner les dictionnaires reliés à t_zh, et d'en faire un dataframe virtuel pour la fonction merge_dics().
def Standardized_t_nomenclature_ZH(bib_mnemo=None):
sch = 'ref_nomenclatures'
tab = ['t_nomenclatures','bib_nomenclatures_types']
sql = """
SELECT
a.id_nomenclature,
--a.cd_nomenclature,
a.mnemonique mnemo,
a.label_default as label,
a.definition_default def,
b.mnemonique bib_mnemo,
b.label_default bib_label,
b.definition_default bib_def,
a.active
FROM {sch}.{tab0} a
JOIN {sch}.{tab1} b USING (id_type)
WHERE b."source" = 'ZONES_HUMIDES'
""".format(sch=sch,tab0=tab[0],tab1=tab[1])
if bib_mnemo is not None:
sql += " AND b.mnemonique = '%s'"%bib_mnemo
df = pd.read_sql_query(sql,conn)
df['mnemo'] = df['mnemo'].str.lower()
df['mnemo'] = df['mnemo'].replace({
r'[’]':"'",
r'[àáâãäå]':'a',
r'[èéêë]':'e',
r'[ìíîï]':'i',
r'[òóôõö]':'o',
r'[ùúûü]':'u',
r'[ ]':"",
r'[–]':"-"
},regex=True)
df = df[['mnemo','id_nomenclature']]
return df
#Définition de la fonction merge_dics: permet la jointure d'un dictionnaire medwet donné avec la table t_nomenclature. Tout ce qu'il y a à faire est d'y entrer le nom du dictionnaire de medwet dans les paramètres,
#càd à la place de 'dic'.
#Attention cependant: si cette fonction marche avec la plupart des dictionnaires medwet, quelquefois des erreurs se produisent sans qu'elles vous soient signalées. Pensez à bien contrôler vos résultats issus de
#merge_dics pour être sûr qu'ils soient bons.
def merge_dics(dic,bib_mnemo=None,how='left'):
re_dic = {
r'[’]':"'",
r'[àáâãäå]':'a',
r'[èéêë]':'e',
r'[ìíîï]':'i',
r'[òóôõö]':'o',
r'[ùúûü]':'u',
r'[ ]':"",
}
if dic=='DicGen-SDA':
sd_dic = {r'versant':'',r'bas-fond ':'bas fonds '}
re_dic = {**sd_dic,**re_dic}
df1 = Standardized_t_nomenclature_ZH(bib_mnemo=bib_mnemo)
df2 = mdb.read_table(db_file,dic)
df2['DESCR'] = df2['DESCR']\
.str.lower()\
.replace(re_dic,regex=True)
df = pd.merge(df1,df2,how=how,left_on='mnemo',right_on='DESCR')
if bib_mnemo is None:
df = df[df['DESCR'].notna()]
df.drop(columns=['DESCR'])
return df
#Enlever les colonnes vides des tables Medwet
def _del_na_col(df):
for col in df.columns:
tmp = pd.notna(df[col]).unique()
if len(tmp) == 1 and False in tmp:
del df[col]
return df
#####################################################################
# Récupérer, créer ou recréer des dictionnaires Medwet to Geonature #
#####################################################################
# Création données bib_actions.
# Dictionnaire d'actions réalisé à partir des données homogénéïsées de SITE_RAP4,
# avec l'aide d'Aude Soureillat.
def get_bib_actions():
dic_actions = {"name":["Zone humide à prendre en compte dans les documents d'urbanisme",
"Prospections faune/flore complémentaires nécessaires",
"Cartographie des habitats d'intérêt nécessaire",
"Pointage des espèces à enjeux nécessaire",
"Meilleure connaissance du fonctionnement hydraulique nécessaire",
"Contours de la zone humide à préciser",
"Maintien de la zone en l'état",
"Maintenir une gestion agricole compatible avec les enjeux du site",
"Gérer la végétation par des actions de bûcheronnage",
"Gérer la végétation par des actions de débroussaillage",
"Gérer la végétation par des actions de broyage",
"Gérer la végétation par des actions de fauche",
"Réduire les amendements agricoles en périphérie de la zone humide",
"Tamponner les apports du bassin versant amont en plantant des haies en périphérie et/ou des bandes enherbées",
"Limiter le pâturage et le sur-piétinement",
"Aménager des abreuvoirs déportés",
"Gérer la végétation par du pâturage extensif",
"Ne pas retourner les sols",
"Creuser des mares",
"Curage nécessaire",
"Adapter les pratiques sylvicoles",
"Gérer la fréquentation",
"Valorisation pédagogique à prévoir",
"Retirer les remblais",
"Gérer les espèces invasives",
"Restauration hydraulique nécessaire",
"Mieux gérer les aménagements et activités au sein du bassin versant"]}
df=pd.DataFrame(dic_actions)
return df
########################################################################################
# Remplissage des tables de correspondance de Géonature à partir des données de Medwet #
########################################################################################
#Récuparation données pour_cor_zh_ref
def get_cor_zh_ref():
#lecture SITEREF
dfm = mdb.read_table(db_file,'SITEREF')
#lecture t_references
dfg= pd.read_sql_query("SELECT * FROM pr_zh.t_references",conn)
#jointure des tables t_reference & SITEREF
df = pd.merge(
dfm,
dfg[['ref_number','id_reference']],
how='left',left_on='REF_NO',right_on='ref_number')
return df\
.drop(columns=['REF_NO','ref_number'])\
.rename(columns={'id_reference':'id_ref'})
#Récupération données pour cor_zh_protection
#definition d'un dictionnaire spécial pour DicGen_CNS
def Standardized_dicgenCNS_dic(): # ne se joint pas avec les autres dictionnaires car les étiquettes sont spéciales: nécessite une retransformation individuelle.
df=mdb.read_table(db_file,'DicGen-CNS')
df['temp'] = df['CODE'] + ' - ' + df['DESCR']
df = df[['CODE','temp']]
df['temp']=df['temp'].str.lower()
df['temp']=df['temp'].replace({
r'[’]':"'",
r'[àáâãäå]':'a',
r'[èéêë]':'e',
r'[ìíîï]':'i',
r'[òóôõö]':'o',
r'[ùúûü]':'u',
r'[ ]':"",
},regex=True)
return df
#jointure avec t_nomenclature
def merge_dicgenCNS():
df1 = Standardized_t_nomenclature_ZH()
df2 = Standardized_dicgenCNS_dic()
df = pd.merge(df1,df2,how='left',left_on='mnemo',right_on='temp')
df = df[df['temp'].notna()]
df.drop(columns=['temp'])
return df
#récupération donnnées cor_zh_protection
def get_cor_zh_protection():
#lecture SITECONS
df = mdb.read_table(db_file,'SITECONS')
# Sauvetage de données : rectification d'une erreur utilisateur
# présente dans la colonne status_cod pour sauver la donnée
df['STATUS_COD'] = df['STATUS_COD'].replace('3' , '36')
#lecture dicgen-CNS jointure avec t_nomenclature
dic = merge_dicgenCNS()
#jointure du dictionnaire avec SITEREF
df = pd.merge(df,dic, how='left',left_on='STATUS_COD',right_on='CODE')
#nettoyage colonnes en trop
df.drop(columns=['STATUS_COD', 'DESIGNATIO', 'LEGISLAT', 'COVER',
'STATUS_MEMO', 'mnemo','CODE', 'temp'],inplace=True)
df.rename(columns={'id_nomenclature':'id_protection'}, inplace=True)
cor_ = pd.read_sql_table('cor_protection_level_type',conn,'pr_zh')
dic2 = dict(zip(cor_.id_protection_status,cor_.id_protection))
df.id_protection.replace(dic2,inplace=True)
return df
def ref_habitats(typo=None):
sql = """
SELECT cd_hab,lb_code,lb_hab_fr
FROM ref_habitats.habref
JOIN ref_habitats.typoref USING(cd_typo)
"""
if typo is not None:
sql += " WHERE lb_nom_typo = '%s'"%typo
sql += ';'
return pd.read_sql_query(sql, conn)
def __get_CBiot__():
dicbio = mdb.read_table(db_file,'DicGen-CBio')
df = mdb.read_table(db_file,'CATCHCBiot')
#nettoyage des colonnes inutiles:
df.drop(columns=['MAPINFO_ID'],inplace=True)
df.rename(columns={'CB_COD':'lb_code'}, inplace=True)
return df
def __filter_lb_error__(serie=None,ignore=None,replace=None):
"""
Identifie les codes habitats ne figurant pas
dans le référentiel CORINE biotope présents
dans le table `ref_habitats.habref`
"""
cbio = __get_CBiot__() if serie is None else serie
if serie is not None :
cbio = pd.DataFrame({'lb_code':serie})
if ignore is not None:
cbio = cbio[~cbio.lb_code.isin(ignore)]
if replace is not None:
cbio.lb_code.replace(replace,inplace=True)
ref = ref_habitats('CORINE_biotopes')
not_cbio = cbio[~cbio.lb_code.isin(ref.lb_code)]
is_cbio = cbio[cbio.lb_code.isin(ref.lb_code)]
return {'is_cbio':is_cbio,'not_cbio':not_cbio}
def __filter_lb_code__(serie, join_ch=False):
"""
Identifie les codes CORINE biotope présents
dans le table `pr_zh.bib_cb`
"""
ind = serie.index.name
to_corzhcb = serie.reset_index()
bib_cb = pd.read_sql_table('bib_cb',conn,'pr_zh')
bib_cb.lb_code = bib_cb.lb_code.astype(str)
not_bib = to_corzhcb[~to_corzhcb.lb_code.isin(bib_cb.lb_code)]
to_corzhcb.drop(not_bib.index,inplace=True)
not_bib = not_bib\
.groupby(ind).agg(','.join)
if join_ch:
to_corzhcb = to_corzhcb.merge(
bib_cb.drop(columns='humidity'),on='lb_code',how='left'
)
return to_corzhcb, not_bib
#Récupération données cor_zh_cb
def get_cor_zh_cb(out_notbib=False,ignore=None,replace=None):
cbio_filer = __filter_lb_error__(ignore=ignore,replace=replace)
cbio = cbio_filer['is_cbio'].set_index('SITE_COD')
to_corzhcb, not_bib = __filter_lb_code__(cbio)
to_corzhcb.drop_duplicates(inplace=True)
if out_notbib:
return to_corzhcb,not_bib
else:
return to_corzhcb
#Récupération données cor_zh_corine_cover
def get_cor_zh_corine_cover():
#lecture dicgen-clc
dic = merge_dics('DicGen-CLC')
#lecture catchclc
df = mdb.read_table(db_file,'CATCHCLC')
#jointure du dictionaire et de catchclc
df = pd.merge(df,dic,how="left",left_on="LC_COD",right_on="CODE")
#nettoyage des colonnes inutiles
df.drop(columns=['LC_COD','mnemo', 'CODE',
'DESCR','COVER'],inplace=True)
df.rename(columns={'CATCH_COD':'SITE_COD','id_nomenclature':'id_cover'}, inplace=True)
return df
#Récupération données cor_zh_lim_fs #pas de données dans medwet apparently
def get_cor_zh_lim_fs():
#lecture dictionnaire
dic = merge_dics('DicGen-LIM_1','CRIT_DEF_ESP_FCT')
#lecture sitelim
# df=mdb.read_table(db_file,'SITELIM') Destiné à t_zh
df = mdb.read_table(db_file,'EFLIM')
#correspondance des types des colonnes qui vont servir à la jointure
dic['CODE'] = dic['CODE'].astype(int)
df['LIM1_COD'] = df['LIM1_COD'].astype(int)
#jointure du dictionnaire et SITELIM
df = pd.merge(df,dic,how="left",left_on="LIM1_COD",right_on="CODE")
#nettoyage des colonnes inutiles
df.drop(columns=['LIM1_COD', 'mnemo','CODE', 'DESCR'],inplace=True)
df.rename(columns={'SIT_COD':'SITE_COD','id_nomenclature':'id_lim_fs'},inplace=True)
return df
#########################################################
#Recreate most important tables from Medwet to Geonature#
#########################################################
#Récupération données utilisateurs.bib_organismes
def get_utilisateurs_bib_organismes():
table = 'bib_organismes'
return pd.read_sql_table(table,conn,'utilisateurs')
def to_bib_organismes():
# df = pd.DataFrame({
# 'nom_organisme':['CORA Savoie',
# 'Apollon 74',
# 'CPNS',
# 'AGRESTIS',
# 'SIVM Haut-Giffre',
# 'TEREO',
# 'SAGE Environnement',
# 'ONEMA',
# 'Ecotope',
# 'ONF'
# ],
# 'id_organisme':['3','4','5','6','7','8','9','10','11','12']})
table = 'bib_organismes'
df = pd.DataFrame({
'nom_organisme':[
'CORA Savoie',
'Apollon 74',
'CPNS',
'AGRESTIS',
'SIVM Haut-Giffre',
'TEREO',
'SAGE Environnement',
'ONEMA',
'Ecotope',
'ONF'
]})
bib = get_utilisateurs_bib_organismes()
df = df[~df.nom_organisme.isin(bib.nom_organisme)]
df.to_sql(table,conn,'utilisateurs',if_exists='append',index=False)
#Récupération données utilisateurs.t_roles
def get_utilisateurs_t_roles():
dic1 = {
'id_role' : ['0','1_2','13','16','17','18','19','2_8','21','7','72','73',
'ASTERS0002','ASTERS0003','ASTERS0004','ASTERS0005','ASTERS0006','ASTERS0007','ASTERS0008',
'24','25','27','28','29','30','31', '32','33','ASTERS0001','26','23','22'],
'nom_role' : ['AUTRE','PIENNE','MERY','GUILLEMOT',
'HERVIEUX','MINSSIEUX','DUBOIS','BOURON','ROUX-VOLLONS','CHOLET','BAL&PIENNE','AVRILIER&BAL'
,'JORDAN','GAY','LOUIS','MARCHIONINI','MARTIN','ALTHEN','LOPEZ-PINOT','PUXEDDU','ROLLET'
,'CAILLETTE','BOURGOIN','FAUCON-MOUTON','PERRET','MACCAUD','ONF','CHABERT','BAL','SOUREILLAT','DEGROOT','SOUQUET-BASIEGE'],
'prenom_role' : ['-','Marc','Luc','Alexandre','Mathilde', 'Elise','Cécile','Manuel','Céline',
'Jérémie','Bernard&Marc','Jean-Noël&Bernard','Denis','Nicolas','Jean-Claude',
'Laurence','Thomas','Aurore','Dominique','Mathieu','Olivier','Mariane','Laurent','Philippe',
'Patrick','Alexandre','Agents','Chloé','Bernard','Aude','Marion','Jules']} #liste des personnes présentes dans MWDKC. dic à lui tout seul car, noms et prénoms mis dans la même colonne dans medwet.
df = mdb.read_table(db_file,'MWDKC')
dic = pd.DataFrame(dic1)
df = pd.merge(df,dic,how='left', left_on='CODE', right_on='id_role')
df.drop(columns=['DESCR','id_role'],inplace=True)
dic2 = {
'ORGANISAT':{ #homogénéisation des personnes entrées dans la base medwet.
np.NaN:'Autre',
'ASTERS':'Asters, CEN Haute-Savoie',
'Asters':'Asters, CEN Haute-Savoie',
'ASTERS - CEN74':'Asters, CEN Haute-Savoie',
'CORA Savoie':'CORA Savoie',
'CORA Savoie et ASTERS':'CORA Savoie',
'ASTERS / CORA Savoie':'CORA Savoie',
'Office National des Forêts':'ONF'
}
}
bib_org = get_utilisateurs_bib_organismes()
dic3 = dict(zip(bib_org.nom_organisme,bib_org.id_organisme))
df.replace(dic2,inplace=True)
df.ORGANISAT.replace(dic3,inplace=True)
df.drop(columns=['ADDRESS', 'CITY', 'COUNTRY', 'PHONE', 'FAX'],inplace=True)
df.rename(columns={'ORGANISAT':'id_organisme','E_MAIL':'email'},inplace=True)
return df
def get_bib_oragnismes():
return pd.read_sql_table('bib_organismes',conn,'utilisateurs')
def get_t_role(with_orga=False):
df = pd.read_sql_table('t_roles',conn,'utilisateurs')
if with_orga:
df = pd.merge(df,get_bib_oragnismes(),on='id_organisme')
return df
def keep_insert_role(df):
t_role = get_t_role()
cols = ['nom_role','prenom_role','id_organisme']
df = pd.merge(df,t_role[cols+['id_role']],on=cols,how='left')
return df[df.id_role.isna()].drop(columns=['id_role'])
#Récupération données t_zh
def prepare_SITEINFO():
#ouverture de SITEINFO:
df = mdb.read_table(db_file,'SITEINFO')
#suppression des colonnes vides
df = _del_na_col(df)
df.drop(columns=['DEPT','NUM','CATCH_COD','COMPLE_COD','LENGTH','SITE_LOC','SITE_MANA'],inplace=True)
return df
#ajouter les colonnes qui n'existent pas dans SITEINFO mais qui dépendent d'autres colonnes qui seront implémentées (remark_is_other_inventory par exemple)
def add_t_zh_cols():
df = prepare_SITEINFO()
df['id_diag_hydro'] = 759 # corresp à "pas évalué" dans t_nomenc
df['id_diag_bio'] = 763 # corresp à "pas évalué" dans t_nomenc
return df
#conversion des codes medwet en code geonature (via t_nomenclature ou dictionnaire définit sur-mesure)
# conversion id_frequency
def get_t_zh_id_frequency(freq,on_dict='CODE'):
dic = merge_dics('DicGen-FREQ')
dic_freq = dict(zip(dic[on_dict],dic.id_nomenclature))
return freq.replace(dic_freq)
# conversion id_spread
def get_t_zh_id_spread(spread,on_dict='CODE'):
dic = merge_dics('DicGen-ETEND')
dic_sprea = dict(zip(dic[on_dict],dic.id_nomenclature))
return spread.replace(dic_sprea)
# conversion id_connexion
def get_t_zh_id_connexion(connex,on_dict='CODE'):
dic = merge_dics('DicGen-CONNEX')
dic_conex = dict(zip(dic[on_dict],dic.id_nomenclature))
return connex.replace(dic_conex)
# Récupération données t_zh
def get_t_zh():
table = 't_zh'
t_role = get_t_roles().sort_index()
org = get_id_org_przh()
dicsda = merge_dics('DicGen-SDA',bib_mnemo='SDAGE')
iddefaultsda = dicsda[dicsda.mnemo=='nondetermine'].id_nomenclature.values[0]
dic_sda = dict(zip(dicsda.CODE,dicsda.id_nomenclature))
dic_org = dict(zip(org.abbrevation,org.id_org))
dic_role = merge_dic_role()
#renommage colonnes
dic_cols = {
'SITE_COD':'code',
'SITE_NAME':'main_name',
'OTHER_NAME':'secondary_name',
'SITE_LIM':'remark_lim',
'EF_REM':'remark_lim_fs',
'AREA_WET':'area',
'AREA_EF':'ef_area',
'EDITDATE':'create_date',
'SITE_DIAGNOS':'remark_diag',
'UPDDATE':'update_date',
'SITE_RAP1':'remark_eval_function',
'SITE_RAP2':'remark_eval_heritage',
'SITE_RAP3':'remark_eval_thread',
'SITE_RAP4':'remark_eval_actions',
'MEMO_ACTIV':'global_remark_activity',
'ZH_REMARKS':'remark_pres',
}
siteinfo = prepare_SITEINFO()\
.rename(columns=dic_cols)
crit_delim,_cor_lim_list = cor_lim_list(siteinfo.code)
siteinfo = siteinfo.merge(crit_delim, on='code')
siteinfo['create_date'] = pd.to_datetime(siteinfo['create_date'])
siteinfo['update_date'] = pd.to_datetime(siteinfo['update_date'])
siteinfo.loc[siteinfo.update_date.isna(),'update_date'] = siteinfo[siteinfo.update_date.isna()].create_date
siteinfo['create_author'] = siteinfo.COMPILER.replace(dic_role)
siteinfo['update_author'] = siteinfo.COMPILER.replace(dic_role)
siteinfo['id_org'] = siteinfo.ORG.replace(dic_org)
siteinfo['id_sdage'] = siteinfo.SDAGE_COD.replace(dic_sda)
siteinfo.loc[
~siteinfo.id_sdage.isin(dicsda.id_nomenclature),
'id_sdage'] = iddefaultsda
siteinfo['id_frequency'] = get_t_zh_id_frequency(siteinfo.FREQUENCE)
siteinfo['id_spread'] = get_t_zh_id_spread(siteinfo.ETENDUE)
siteinfo['id_connexion'] = get_t_zh_id_connexion(siteinfo.CONNEX)
# Récupération des "AUTRES INVENTAIRES"
siteinfo.loc[~siteinfo.OTHER_INV.str.startswith('ZNIEFF',na=False),'remark_is_other_inventory'] = siteinfo[
~siteinfo.OTHER_INV.str.startswith('ZNIEFF',na=False)
].OTHER_INV
siteinfo['is_other_inventory'] = siteinfo['remark_is_other_inventory'].notna()
# Récupération nb_hab d'Habitats patrimoniaux
cbio_filer = __filter_lb_error__()
cbio = cbio_filer['is_cbio']
notcbio = cbio_filer['not_cbio']\
.groupby('SITE_COD').agg(','.join)
cbio_ch, not_bib = __filter_lb_code__(cbio.set_index('SITE_COD'),join_ch=True)
siteinfo = pd.merge(
siteinfo,
cbio_ch.groupby('SITE_COD',dropna=False)\
.agg({'is_ch':sum})\
.rename(columns={'is_ch':'nb_hab'}),
left_on='code',
right_index=True,
how='left'
)
# Aggregation des habitats considérés NON-HUMIDE
not_bib.lb_code = 'Autre(s) habitat(s) CORINE BIOTOPE décrit(s) :\n' + not_bib.lb_code
notcbio.lb_code = 'Autre(s) habitat(s) décrit(s) :\n' + notcbio.lb_code
siteinfo.remark_pres = _aggregate_value__(
siteinfo.set_index('code').remark_pres,
not_bib.lb_code,
sep='\n'
)
siteinfo.remark_pres = _aggregate_value__(
siteinfo.set_index('code').remark_pres,
notcbio.lb_code,
sep='\n'
)
# JOINDRE GEOM
carteGen_path = './bdd/CarteZHGénérée.shp'
carteGen = gpd.read_file(carteGen_path)\
.drop(columns=['Site','Surface'])\
.rename_geometry('geom')\
.rename(columns={'NewNumZone':'code'})
# Correction des géométries invalides
if not carteGen.is_valid.all():
carteGen.loc[~carteGen.is_valid,'geom'] = \
carteGen[~carteGen.is_valid].geom.buffer(0)
# Jointure des géometries
df = siteinfo.merge(carteGen,on='code')\
.set_geometry('geom')
df['area'] = round(df.geom.area,2)
# reprojection
if hasattr(df,'crs') and df.crs.srs != 'EPSG:4326':
df.to_crs(4326,inplace=True)
tzh_cols = recup_cols_table(table,conn)
lst_cols = df.columns[df.columns.isin(tzh_cols)]
to_tzh = df[lst_cols].copy()
print('Columns non intégrés : %s'%str([x for x in tzh_cols if x not in lst_cols]))
return to_tzh,_cor_lim_list
def recup_cols_table(table,con,schema='pr_zh'):
lst_cols = con.dialect.get_columns(con,table,schema)
return [x['name'] for x in lst_cols]
def get_SITELIM():
#lecture dictionnaire
dic = merge_dics('DicGen-LIM')
#lecture sitelim
df = mdb.read_table(db_file,'SITELIM')
dic['CODE'] = dic['CODE'].astype(int)
df['LIM_COD']=df['LIM_COD'].astype(int)
#jointure du dictionnaire et SITELIM
df = pd.merge(df,dic,how="left",left_on="LIM_COD",right_on="CODE")
df.drop(columns=['LIM_COD', 'mnemo','CODE', 'DESCR'],inplace=True)
df.rename(columns={'SIT_COD':'code','id_nomenclature':'id_lim'},inplace=True)
return df
def cor_lim_list(SITE_COD):
"""Remplis la table pr_zh.cor_lim_list et retourn les uuid associés
"""
delim = t_nomenclature_ZH(bib_mnemo='CRIT_DELIM')
delim.mnemo = delim.mnemo.str.replace(r'.\(.*\)','',regex=True)
dict_delim = dict(zip(delim.mnemo.str.lower(),delim.id_nomenclature))
crit_delim = get_SITELIM()
app_site = SITE_COD[~SITE_COD.isin(crit_delim.code)]
if not app_site.empty:
app_site = pd.DataFrame({'code':app_site})
app_site['id_lim'] = delim[delim.mnemo=='Non déterminé'].id_nomenclature.values[0]
crit_delim = pd.concat([crit_delim,app_site])
genuuid = crit_delim.groupby('code').agg(list).reset_index()
genuuid['id_lim_list'] = [uuid.uuid4() for x in genuuid.index]
_cor_lim_list = genuuid.drop(columns=['code']).explode('id_lim')
uuidreturn = genuuid.drop(columns=['id_lim'])
return uuidreturn,_cor_lim_list
def get_t_roles(id_role=None):
table = 't_roles'
t_roles = pd.read_sql_table(
table,conn,'utilisateurs',['id_role'],columns=['nom_role','prenom_role','id_organisme']
).replace({r'[’]':"'"},regex=True).sort_index()
if id_role:
t_roles = t_roles.iloc[[id_role]]
return pd.merge(t_roles,get_bib_organismes(),'left',left_on='id_organisme',right_index=True)
def get_bib_organismes(schema='utilisateurs'):
table = 'bib_organismes'
return pd.read_sql_table(
table,conn,schema,['id_organisme'],columns=['nom_organisme']
).replace({r'[’]':"'"},regex=True)
def get_id_org_przh():
return pd.read_sql_table('bib_organismes',conn,'pr_zh')
def remove_special_char(obj,space=False):
dict_char = {
r'[’]':"'",
r'[àáâãäå]':'a',
r'[èéêë]':'e',
r'[ìíîï]':'i',
r'[òóôõö]':'o',
r'[ùúûü]':'u',
# r'[ ]':"",
r'[–]':"-"
}
if space:
dict_char = {**dict_char, **{r'[ ]':""}}
return obj.replace(dict_char,regex=True)
def to_bib_organismes_przh(dicorga,first_time=False):
"""Insertion des organisme dans la table pr_zh
@dicorga : dict. {'abbrev':'nom de la structure', ...}
"""
table = 'bib_organismes'
if first_time:
sql = 'DELETE FROM pr_zh.%s'%table
with conn.begin() as cnx:
cnx.execute(sql)
isin_db = pd.read_sql_table(
table,conn,'pr_zh',['id_org'],columns=['name']
).replace({r'[’]':"'"},regex=True)
insert_from = pd.DataFrame({'nom':dicorga.values(),'abbrev':dicorga.keys()})
to_insert = insert_from[~insert_from.nom.str.lower().isin(isin_db.name.str.lower())]
to_insert.abbrev = remove_special_char(to_insert.abbrev,space=True)\
.str.upper()\
.str[:6]\
.fillna('XXXXXX')
to_insert\
.rename(columns={'nom':'name','abbrev':'abbrevation'})\
.to_sql(name=table,con=conn,schema='pr_zh',if_exists='append',index=False)
#Récupération données t_actions
#definition d'un dictionnaire sur-mesure suite à l'incompatibilité avec merge_dics()
def Standardized_DICACT_dic(): # ne se joint pas avec les autres dictionnaires car les étiquettes sont spéciales: nécessite une retransformation individuelle.
df=mdb.read_table(db_file,'DICACT')
df['temp'] = df['CODE'] + ' - ' + df['DESCR']
df = df[['CODE','temp']]
df['temp']=df['temp'].str.lower()
df['temp']=df['temp'].replace({
r'[’]':"'",
r'[àáâãäå]':'a',
r'[èéêë]':'e',
r'[ìíîï]':'i',
r'[òóôõö]':'o',
r'[ùúûü]':'u',
r'[ ]':"",
},regex=True)
return df
#jointure avec t_nomenclatures
def merge_dicact():
df1 = Standardized_t_nomenclature_ZH()
df2 = Standardized_DICACT_dic()
df = pd.merge(df1,df2,how='left',left_on='mnemo',right_on='temp')
df = df[df['temp'].notna()]
df.drop(columns=['temp'])
return df
def get_cor_impact_types():
return pd.read_sql_table('cor_impact_types',conn,'pr_zh')
def get_t_activity():
#lecture du dictionnaire dicgen-pos
dicpos = merge_dics('DicGen-POS')
dicpos = dict(zip(dicpos.CODE,dicpos.id_nomenclature))
#lecture du dictionnaire dicgen-imp
no_imp = t_nomenclature_ZH('IMPACTS')\
.merge(merge_dics('DicGen-IMP')[['id_nomenclature','CODE']],on='id_nomenclature')\
.rename(columns={'id_nomenclature':'id_impact'})
di_imp = pd.merge(get_cor_impact_types(),no_imp[['id_impact','mnemo','CODE']],on='id_impact')
di_imp.CODE = di_imp.CODE.str.strip()
dicimp = dict(zip(di_imp.CODE,di_imp.id_cor_impact_types))
#lecture du dictionnaire DIDACT
dicact = merge_dicact()
dicact = dict(zip(dicact.CODE,dicact.id_nomenclature))
#jointure avec la table SITEACT
siteimp = mdb.read_table(db_file,'SITEIMP') # IMPACT_COD
# siteimp.reset_index(inplace=True)
siteimp = _del_na_col(siteimp)
normimp = dict(zip([x+'.0' if x.find(r'.') == -1 else x for x in no_imp.CODE],no_imp.CODE))
df = mdb.read_table(db_file,'SITEACT').drop(columns=['COORD_X','COORD_Y'])
df.IMPACT_COD = df.IMPACT_COD.astype(str)
df.IMPACT_COD.replace(normimp,inplace=True)
df = _del_na_col(df)
# Drop SITEIMP_IMPACT_COD who is in SITEACT_IMPACT_COD
siteimp['ISIN'] = [False if df[(df.SITE_COD==siteimp.SITE_COD[x])&(df.ACTIV_COD==siteimp.ACTIV_COD[x])&(df.IMPACT_COD==siteimp.IMPACT_COD[x])].empty else True
for x in siteimp.index
]
siteimp_filter = siteimp[~siteimp.ISIN].drop(columns=['ISIN'])
activ = df.merge(siteimp_filter,on=['SITE_COD','ACTIV_COD'],how='outer',suffixes=('','_y'))\
.sort_values(['SITE_COD','ACTIV_COD'])
# Récupération des codes SITEIMP, précisant parfois SITEACT
activ['IMPACT_COD'] = [activ.IMPACT_COD_y[x] if (activ.IMPACT_COD[x] is np.NaN) else activ.IMPACT_COD[x] for x in activ.index]
activ['IMPACT_COD'] = [activ.IMPACT_COD_y[x] if (activ.IMPACT_COD[x]=='0') and (activ.IMPACT_COD_y[x] is not np.NaN) else activ.IMPACT_COD[x] for x in activ.index]
# Suppression des codes SITEIMP existants
activ['IMPACT_COD_y'] = [None if activ.IMPACT_COD_y[x] == activ.IMPACT_COD[x] else activ.IMPACT_COD_y[x] for x in activ.index]
activ_a = activ.drop(columns=['IMPACT_COD_y']).drop_duplicates()
activ_b = activ.copy().drop(columns=['POSITION','REMARKS','IMPACT_COD'])\
.dropna(subset=['IMPACT_COD_y'])\
.rename(columns={'IMPACT_COD_y':'IMPACT_COD'})
# .drop_duplicates()
activ_all = pd.concat([activ_a,activ_b])
dict_imp = t_nomenclature_ZH('LOCALISATION')
iddefault = dict_imp[dict_imp.label=='Non déterminée'].id_nomenclature.values[0]
if not activ_all[activ_all.POSITION.isna()].empty:
activ_all[activ_all.POSITION.isna()].to_csv(path + '/t_activity_NA.csv',index=False)
activ_all.POSITION.fillna(iddefault,inplace=True)
activ_all.POSITION.replace(dicpos,inplace=True)
activ_all.ACTIV_COD.replace(dicact,inplace=True)
activ_all.IMPACT_COD.replace(dicimp,inplace=True)
t_activ = activ_all\
.groupby(['SITE_COD','ACTIV_COD'],dropna=False)\
.agg(list)\
.reset_index()\
.rename(columns={
'ACTIV_COD': 'id_activity',
'POSITION' : 'id_position',
'IMPACT_COD' : 'id_cor_impact_types',
'REMARKS' : 'remark_activity'
})
t_activ.id_position = [list(set(x)) for x in t_activ.id_position ]
t_activ.id_position = [
x[0] if len(x)==1 else [y for y in x if y != iddefault][0]
if len([y for y in x if y != iddefault]) == 1 else iddefault
for x in t_activ.id_position
]
t_activ['remark_activity'] = [
'\n'.join(
list(dict.fromkeys([item for item in x if not(pd.isna(item)) == True]))
) for x in t_activ['remark_activity']
]
t_activ.remark_activity = t_activ.remark_activity\
.str.strip()\
.replace({' ',' '},regex=True)\
.replace({'':None})
t_activ['id_impact_list'] = [uuid.uuid4() for x in t_activ.index]
t_activity = t_activ[['SITE_COD','id_activity', 'id_position','id_impact_list','remark_activity']]
cor_impact_list = t_activ[['id_impact_list','id_cor_impact_types']]\
.explode('id_cor_impact_types')\
.drop_duplicates()
return t_activity,cor_impact_list
#Récupération données t_fonctions
#dictionnaire fonctions hydrologiques
def Standardized_DicGen_FVI_dic(): # ne se joint pas avec les autres dictionnaires car les étiquettes sont spéciales: nécessite une retransformation individuelle.
df=mdb.read_table(db_file,'DicGen-FVI')
df['temp'] = df['FVI_CODE'] + ' - ' + df['DESCR']
df = df[['FVI_CODE','temp']]
df['temp']=df['temp'].str.lower()
df['temp']=df['temp'].replace({
r'[’]':"'",
r'[àáâãäå]':'a',
r'[èéêë]':'e',
r'[ìíîï]':'i',
r'[òóôõö]':'o',
r'[ùúûü]':'u',
r'[ ]':"",
},regex=True)
return df
#jointure avec t_nomenclature
def merge_DicGen_FVI():
df1 = Standardized_t_nomenclature_ZH()
df2 = Standardized_DicGen_FVI_dic()
df = pd.merge(df1,df2,how='left',left_on='mnemo',right_on='temp')
df = df[df['temp'].notna()]
df.drop(columns=['temp'])
return df
def _get_SITEFVI(func):
#lecture du dictionnaire DicGen-FVI
name_table = 'SITEFVI'+func.upper()
dicgenFVIH = merge_DicGen_FVI()
#jointure avec la table SITEVIH
df = mdb.read_table(db_file,name_table)
df = pd.merge(df,dicgenFVIH,how='left',left_on='FVI_COD',right_on='FVI_CODE')
df = _del_na_col(df)
df.drop(columns=['mnemo','temp','FVI_CODE','FVI_COD'],inplace=True)
#reste à transformer les codes avec les id_zh futurs id de get_t_zh
return df
def get_t_functions(clean=True):
#lecture du dictionnaire DicGen-FVI
fvih = _get_SITEFVI(func="H").rename(columns={'MEMO_HYD':'MEMO'})
fvib = _get_SITEFVI(func="B").rename(columns={'MEMO_BIO':'MEMO'})
fvis = _get_SITEFVI(func="S").rename(columns={'MEMO_SOCIO':'MEMO'})
fvip = _get_SITEFVI(func="P").rename(columns={'MEMO_PATRI':'MEMO'})
fvih['type_func'] = 'HYD'
fvib['type_func'] = 'BIO'
fvis['type_func'] = 'SOCIO'
fvip['type_func'] = 'PATRI'
df = pd.concat([fvih,fvib,fvis,fvip])
qualif = t_nomenclature_ZH('FONCTIONS_QUALIF')
knowle = t_nomenclature_ZH('FONCTIONS_CONNAISSANCE')
df['id_qualification'] = qualif.loc[qualif.mnemo=='Non évaluée','id_nomenclature'].values[0]
df['id_knowledge'] = knowle.loc[knowle.mnemo=='Lacunaire ou nulle','id_nomenclature'].values[0]
df.rename(columns={'MEMO':'justification','id_nomenclature':'id_function'},inplace=True)
if clean:
df.drop(columns=['type_func','MAPINFO_ID'],inplace=True)
return df
#Récupération données t_table_héritage
def get_t_table_heritage():
#lecture dictionnaire dicgen-VIP et jointure avec bib_cb. /!\ ne se merge pas avec utilisteurs.t_nomenclatures, mais avec pr_zh.bib_cb !
bib = pd.read_sql_query('SELECT * FROM pr_zh.bib_cb',conn)
dic = mdb.read_table(db_file,'DicGen-CBio')
dict={'1':'10','2':'20','3':'30','4':'40','5':'50','6':'60','7':'70','8':'80','9':'90'}
dic =dic['CODE'].replace(dict)
dic = pd.merge(bib,dic,how='left',left_on='lb_code',right_on='CODE')
#jointure du dictionnaire avec SITEFVIP
df = mdb.read_table(db_file,'SITEFVIP')
df = pd.merge(df,dic,how='left',left_on='FVI_COD',right_on='CODE')
return df
#Récupération données t_inflow
def get_t_inflow():
#récupération dictionnaire dicgen-in1
dicinf1 = merge_dics('DicGen-IN1',bib_mnemo='ENTREE_EAU')
#problème: se mélange avec dicgen-out1 dans t_nomenclatures. Sélection du bon dictionnaire sur les id_nomenclatures
# dicinf1 = dicinf1.loc[dicinf1['id_nomenclature']<=726]
#récupération dictionnaire dicgen-in2
dicinf2 = merge_dics('DicGen-IN2',bib_mnemo='PERMANENCE_ENTREE')
#problème: se mélange avec dicgen-out2 dans t_nomenclatures. Sélection du bon dictionnaire sur les id_nomenclatures
# dicinf2 = dicinf2.loc[dicinf2['id_nomenclature']<=735]
#jointure des dicttionnaires avec SITEINF
df = mdb.read_table(db_file,'SITEINFL')\
.rename(columns={'TOPOINFL':'topo'})
df = pd.merge(df,dicinf1,how = 'left', left_on='INFLOW1', right_on='CODE')\
.rename(columns={'id_nomenclature':'id_inflow'})
df = pd.merge(df,dicinf2,how='left',left_on='INFLOW2',right_on='CODE')\
.rename(columns={'id_nomenclature':'id_permanance'})
#nettoyage des colonnes en trop
df.drop(columns=['INFLOW1', 'INFLOW2','COORD_X', 'COORD_Y',
'mnemo_x','CODE_x', 'DESCR_x', 'mnemo_y','CODE_y', 'DESCR_y'],inplace=True)
return df
#Récupération données t_instruments
def get_t_instruments():
#récupération dictionnaire dicgen-INS.
dicins = merge_dics('DicGen-INS')
#jointure avec SITEINSTR
df = mdb.read_table(db_file,'SITEINSTR')
df['DATE_STA'] = pd.to_datetime(df['DATE_STA'])
df = pd.merge(df,dicins,how='left', left_on='INSTR_COD',right_on='CODE')
#nettoyage des colonnes en trop
df.drop(columns=['INSTR_COD','mnemo','CODE','DESCR','DATE_END','ORGA_INSTR'],inplace=True)
#renommage des colonnes
df.rename(columns={'id_nomenclature':'id_instrument','DATE_STA':'instrument_date'},inplace=True)
return df
#Récupération données t_management_structures
def get_t_management_structures(): # existe pê, à voir avec Aude demain
#lecture sitegest
df = mdb.read_table(db_file,'SITEGEST')
#remplacement id_structure (manuellement, car ici il n'y a que 2 lignes.)
df['STRUC_COD'].replace('ASTERS0001','1',inplace=True)
df.rename(columns={'STRUC_COD':'id_org'},inplace=True)
return df
#Récupération données t_outflow
def get_t_outflow():
#récupération dictionnaire dicgen-out1
dicout1 = merge_dics('DicGen-OUT',bib_mnemo='SORTIE_EAU')
#problème: se mélange avec dicgen-in1 dans t_nomenclatures. Sélection du bon dictionnaire sur les id_nomenclatures
# dicout1 = dicout1.loc[dicout1['id_nomenclature']<=726]
#récupération dictionnaire dicgen-out2
dicout2 = merge_dics('DicGen-OUT2',bib_mnemo='PERMANENCE_SORTIE')
dicout2.loc[dicout2.mnemo=='nondetermine','CODE'] = '0'
dicout2['CODE'] = dicout2['CODE'].astype(int)
#problème: se mélange avec dicgen-in2 dans t_nomenclatures. Sélection du bon dictionnaire sur les id_nomenclatures
# dicout2 = dicout2.loc[dicout2['id_nomenclature']>=735]
#problème: se mélange aussi avec autre dictionnaire dans t_nomenclatures. Sélection du bon dictionnaire sur les id_nomenclatures
# dicout2 = dicout2.loc[dicout2['id_nomenclature']<=942]
#jointure des dicttionnaires avec SITEOUTF
df = mdb.read_table(db_file,'SITEOUTF')
df['PERMANENCE'].fillna(
dicout2.loc[dicout2.mnemo=='nondetermine','CODE'].values[0],
inplace=True
)
df = pd.merge(df,dicout1,how = 'left', left_on='OUTFLOW', right_on='CODE')
#/!\ attention: Colonne permanence en float64 et non en integer. Conversion en integer
# df['PERMANENCE'] = df['PERMANENCE'].fillna(0).astype(int)
df = pd.merge(df,dicout2,how='left',left_on='PERMANENCE',right_on='CODE')
#nettoyage des colonnes en trop
df.drop(columns=[
'OUTFLOW','COORD_X', 'COORD_Y', 'PERMANENCE',
'mnemo_x','CODE_x', 'DESCR_x', 'mnemo_y','CODE_y','DESCR_y'
],inplace=True)
#renommage colonnes
df.rename(columns={
'TOPOOUTFL':'topo','id_nomenclature_x':'id_outflow','id_nomenclature_y':'id_permanance'
},inplace=True)
return df
#Récupération données t_ownership
def get_t_ownership():
#récupération dictionnaire DicGen-STAT
dicown = merge_dics('DicGen-STAT')
#jointure avec SITESTA
df = mdb.read_table(db_file, 'SITESTA')
df = pd.merge(df,dicown, how='left',left_on='TENUR_COD',right_on='CODE')
#nettoyage des colonnes en trop
df.drop(columns=['DESCR','mnemo','TENUR_COD','CODE'],inplace=True)
df.rename(columns={'id_nomenclature':'id_status','MEMO_TENUR':'remark'},inplace=True)
return df
def merge_dic_role():
df = pd.merge(
get_utilisateurs_t_roles(),
get_t_roles().reset_index(drop=False),
on=['nom_role','prenom_role','id_organisme'],
how='left'
)
return dict(zip(df.CODE,df.id_role))
def to_t_references():
table = 't_references'
dic_col_ref = {
'REF_NO':'ref_number',
'REFERENCE':'reference',
'AUTHOR':'authors',
'TITLE':'title',
'YEAR':'pub_year',
'PUBLISHER':'editor',
'LOCATION':'editor_location',
}
df = mdb.read_table(db_file, 'MWDREF')\
.rename(columns=dic_col_ref)
df.loc[df.title.isna(),'title'] = df[df.title.isna()].reference
df.to_sql(name=table,con=conn,schema='pr_zh',if_exists='append',index=False)
def OTHERINV_to_tref():
table = 't_references'
sitinfo = prepare_SITEINFO().set_index('SITE_COD')
otinv = sitinfo[['OTHER_INV']]\
.dropna()
otinv = otinv.OTHER_INV.str\
.split('//',expand=True)\
.stack().droplevel(-1)\
.str.strip().to_frame()
otinv.columns = ['title']
cren = otinv[otinv.title.str.startswith('CREN')].copy()
cren['authors'] = cren.title.str.split(r' ',1,expand=True)[0]
cren['editor'] = cren.title.str.split(r' ',1,expand=True)[0]
cren['pub_year'] = cren.title.str.split(r'[ |,]',2,expand=True)[1].astype(int)
cren['ref_number'] = cren.title.str.rsplit(r' ',1,expand=True)[1]
cren.drop_duplicates(inplace=True)
znif = otinv[otinv.title.str.startswith('ZNIEFF')].copy()
znif['ref_number'] = znif.title.str.rsplit(r'n°',1,expand=True)[1]
znif.drop_duplicates(inplace=True)
cren.to_sql(name=table,con=conn,schema='pr_zh',if_exists='append',index=False)
znif.to_sql(name=table,con=conn,schema='pr_zh',if_exists='append',index=False)
def update_t_ownership(con_zh,con_fon):
table = 't_ownership'
zh = gpd.read_postgis('SELECT id_zh, geom FROM pr_zh.t_zh',con_zh,crs=4326)
zh.to_crs(2154,inplace=True)
sql_fon = '''
SELECT
pa.par_id,
pr.dnuper,
pr.ddenom,
ccogrm_lib,
pa.geom
FROM cadastre.parcelles_cen pa
JOIN cadastre.lots_cen USING (par_id)
JOIN cadastre.cadastre_cen USING (lot_id)
JOIN cadastre.cptprop_cen USING (dnupro)
JOIN cadastre.r_prop_cptprop_cen USING (dnupro)
JOIN cadastre.proprios_cen pr USING (dnuper)
JOIN cadastre.d_ccogrm USING (ccogrm)
'''
fon = gpd.read_postgis(sql_fon,con_fon)
dic = {
'Commune':'Collectivité territoriale (communal, départemental, régional, syndicat mixte)',
'Personnes morales non remarquables':'Propriété privée',
'Département':'Collectivité territoriale (communal, départemental, régional, syndicat mixte)',
'Personnes morales représentant des sociétés':'Propriété privée',
'État':"Domaine de l'Etat",
'Établissements publics ou organismes assimilés':'Établissement public (conservatoire du littoral, parcs nationaux…)',
'Copropriétaire':'Propriété privée',
}
fon['statut'] = fon.ccogrm_lib.replace(dic)
stat = t_nomenclature_ZH('STATUT_PROPRIETE')
dic2 = dict(zip(stat.mnemo,stat.id_nomenclature))
fon.statut.replace(dic2,inplace=True)
gfon = gpd.sjoin(zh,fon[['geom','statut']])\
.drop(columns=['geom','index_right'])\
.drop_duplicates()\
.rename(columns={'statut':'id_status'})
in_db = pd.read_sql_table(table,con_zh,'pr_zh')
gfon = gfon[~(gfon.id_zh.isin(in_db.id_zh) & gfon.id_status.isin(in_db.id_status))]
gfon.to_sql(table,con=con_zh,schema='pr_zh',if_exists='append',index=False)
print('Insert %s news rows !'%gfon.shape[0])
del_in_db = in_db[in_db.id_zh.isin(gfon.id_zh) & (~in_db.id_status.isin(gfon.id_status))]
if not del_in_db.empty:
print('Possible données obsolète dans la table %s'%table)
return del_in_db
print('Table %s à jour !'%table)
################################################
# Conversion des fonctions de requêtage en CSV #
################################################
if __name__ == "__main__":
# def HARVEST():
Dossier = "HARVEST"
path = os.path.join(DIR,Dossier)
if not os.path.exists(path):
os.mkdir(path)
dicorgaprzh = {
'ASTERS':"Conservatoire D'Espaces Naturels De Haute-Savoie",
'CPNS':"Conservatoire d'espaces naturels de Savoie",
}
# to_bib_organismes_przh(dicorga=dicorgaprzh,first_time=True)
# to_t_references()
#récupération des dataFrames en listes
print("...début de la récupération des données ...")
# utilisateurs_bib_organismes = get_utilisateurs_bib_organismes()
bib_actions = get_bib_actions()
_cor_zh_cb,not_bib = get_cor_zh_cb(ignore=LB_IGNORE,replace=LB_DICT,out_notbib=True) # 609 rows where CB_COD not in bib_cb
_cor_zh_corine_cover = get_cor_zh_corine_cover()
_cor_zh_protection = get_cor_zh_protection()
_cor_zh_lim_fs = get_cor_zh_lim_fs()
t_activity, _cor_impact_list = get_t_activity()
t_functions = get_t_functions()
t_inflow = get_t_inflow()
t_outflow = get_t_outflow()
# t_table_heritage = get_t_table_heritage()
t_instruments = get_t_instruments()
t_management_structures = get_t_management_structures()
t_ownership = get_t_ownership()
t_roles = get_t_roles()
utilisateur = get_utilisateurs_t_roles()
insert_newrole = keep_insert_role(utilisateur.drop(columns=['CODE']))
if not insert_newrole.empty:
insert_newrole.to_sql(
name='t_roles',con=conn,schema='utilisateurs',if_exists='append',index=False
)
t_zh,_cor_lim_list = get_t_zh()
_cor_zh_ref = get_cor_zh_ref()
######### COMPLET / RESOLVE auteur indéterminé
t_zh.loc[t_zh.create_author.isna(),['create_author']] = t_roles[t_roles.nom_role=='AUTRE'].index[0]
t_zh.loc[t_zh.update_author.isna(),['update_author']] = t_roles[t_roles.nom_role=='AUTRE'].index[0]
int_col = ['create_author','update_author','id_sdage','id_connexion']
t_zh[int_col] = t_zh[int_col].astype(int)
######### COMPLET / RESOLVE create_date & update_date
t_zh.loc[t_zh.create_date.isna(),['create_date']] = dt.today().date().isoformat()
t_zh.loc[t_zh.update_date.isna(),['update_date']] = dt.today().date().isoformat()
######### SPEC CEN74
# Ammendement de la table lb_code
# A faire qu'une fois !
to_bibcb = pd.DataFrame({'lb_code':not_bib.lb_code.str.split(',',expand=True).stack().unique().tolist()})
to_bibcb['humidity'] = None
to_bibcb['is_ch'] = None
add_bib_cb(to_bibcb,con=conn,humidity='P',is_ch=False)
# Manip du champ remark_eval_heritage
rmk_her = t_zh[['code','remark_eval_heritage']].copy()
rmk_her.set_index('code',inplace=True)
rmk_her = rmk_her.remark_eval_heritage.str.split('FAUNE',expand=True)
rmk_her.columns = ['flore','faune']
fau = rmk_her.faune.str.split("Espèce animale d'intérêt : |Espèces animales d'intérêt par ordre décroissant : ",expand=True)
flo = rmk_her.flore.str.split("Espèce végétale d'intérêt : |Espèces végétales d'intérêt par ordre décroissant : ",expand=True)
fau.columns = ['rmk_faune','ic_faune_desc']
flo.columns = ['rmk_flore','ic_flore_desc']
rmk_her = pd.merge(flo,fau,right_index=True,left_index=True)
# Suppression des valeurs inutiles
rmk_her.replace({
r'[\r\n]':'',
'FLORE':'',
'Aucune espèce de valeur connue':'',
},regex=True,inplace=True)
for c in rmk_her.columns:
rmk_her[c] = rmk_her[c].str.strip().replace({'':None})
# Replacement de certaines précisions
rmk_her.replace({
'Zone en apparence détruite avant destruction':'Zone en apparence détruite. Espèces observées avant destruction',
'avant destruction':None
},inplace=True)
# Ecriture du tableau pour consultation
rmk_her.to_csv(path + '/remark_eval_heritage_especes.csv',index=True)
# Récupération des remarques pour t_zh
t_zh.drop(columns='remark_eval_heritage',inplace=True)
t_zh = t_zh.merge(rmk_her[['rmk_flore']],right_index=True,left_on='code').rename(columns={'rmk_flore':'remark_eval_heritage'})
####### ERROR #######
# Bloquant pour intégration. len(Remark_..) > 2000 characters
crmk = ['code',*t_zh.columns[t_zh.columns.str.contains('remark')]]
t_zh[crmk].iloc[:,5].str.len().max()
t_zh.loc[t_zh[crmk].iloc[:,5].str.len()>2000,['code','remark_diag']].iloc[:,1].values
t_zh.loc[t_zh[crmk].iloc[:,5].str.len()>2000,['code','remark_diag']].to_csv(path + '/remark_diag_TOO_LONG.csv',index=False)
t_zh.loc[t_zh[crmk].iloc[:,5].str.len()>2000,['remark_diag']] = 'Remarques disponibles auprès du référent ZH ou Géomatique'
t_zh[crmk].iloc[:,9].str.len().max()
t_zh.loc[t_zh[crmk].iloc[:,9].str.len()>2000,['code','remark_eval_heritage']].iloc[:,1].values
t_zh.loc[t_zh[crmk].iloc[:,9].str.len()>2000,['code','remark_eval_heritage']].to_csv(path + '/remark_eval_heritage_TOO_LONG.csv',index=False)
#####################
bib_actions.to_sql('bib_actions',conn,'pr_zh','append',False) # Fait
to_tzh(t_zh,conn)
to_cor_(conn,_cor_lim_list,'cor_lim_list')
to_t_(conn,_cor_zh_lim_fs,'cor_zh_lim_fs')
to_t_(conn,t_activity,'t_activity')
to_cor_(conn,_cor_impact_list,'_cor_impact_list')
to_t_(conn,t_functions,'t_functions')
to_t_(conn,t_inflow,'t_inflow')
to_t_(conn,t_outflow,'t_outflow')
to_t_(conn,t_instruments,'t_instruments')
to_t_(conn,t_management_structures,'t_management_structures')
to_t_(conn,t_ownership,'t_ownership')
to_t_(conn,_cor_zh_cb,'cor_zh_cb')
to_t_(conn,_cor_zh_corine_cover,'cor_zh_corine_cover')
to_t_(conn,_cor_zh_protection,'cor_zh_protection')
to_t_(conn,_cor_zh_ref,'cor_zh_ref')
update_t_ownership(con_zh=conn,con_fon=con_f)
# print("fin de la récupération des données.")
# #création du dossier d'acceuil des csv
# print("...création du dossier parent...")
#écriture des dataframes en csv
print("...écriture des tables en fichier.csv ...")
# get_utilisateurs_bib_organismes().to_csv(path + '/bib_organismes.csv',index=False) # Status : A intégrer dans pr_zh.
# get_bib_actions().to_csv(path + '/bib_actions.csv',index=False) # Status : Prêt à intégrer.
# get_cor_zh_cb().to_csv(path + '/_cor_zh_cb.csv',index=False) # 609 rows where CB_COD not in bib_cb
# get_cor_zh_corine_cover().to_csv(path + '/_cor_zh_corine_cover.csv',index=False) # Status : Prêt à intégrer.
# get_cor_zh_protection().to_csv(path + '/_cor_zh_protection.csv',index=False) # Status : Prêt à intégrer.
# get_cor_zh_lim_fs().to_csv(path + '/_cor_zh_lim_fs.csv',index=False) # Status : Prêt à intégrer.
# t_activity, _cor_impact_list = get_t_activity()
# t_activity.to_csv(path + '/t_activity.csv',index=False) # Status : Prêt à intégrer.
# _cor_impact_list.to_csv(path + '/_cor_impact_list.csv',index=False) # Status : Prêt à intégrer.
# get_t_functions().to_csv(path + '/t_functions.csv',index=False) # Status : Prêt à intégrer.
# get_t_inflow().to_csv(path + '/t_inflow.csv',index=False)
# get_t_outflow().to_csv(path + '/t_outflow.csv',index=False) # Status : Prêt à intégrer.
# get_t_table_heritage().to_csv(path + '/t_table_heritage.csv',index=False) # Status : Incompris # Status : Prêt à intégrer.
# get_t_instruments().to_csv(path + '/t_instruments.csv',index=False) # Status : Prêt à intégrer.
# get_cor_zh_ref().to_csv(path + '/_cor_zh_ref.csv',index=False)
# # liste des structures de gestion des zh
# get_t_management_structures().to_csv(path + '/t_management_structures.csv',index=False) # Status : Insatisfaisant, a revoir !
# get_t_ownership().to_csv(path + '/t_ownership.csv',index=False) # Status : Prêt à intégrer.
# keep_insert_role(get_utilisateurs_t_roles()).to_csv(path + '/t_roles.csv',index=False) # Status : Prêt à intégrer.
# get_t_zh().to_csv(path + '/t_zh.csv',index=False) # Status : Encore à Faire ..
# print("Fin de la transciption des tables en CSV.")
# print('Fin de HARVEST. Veuillez passer à TRANSFORM.')
# # Intégration des données de configs
# dicorgaprzh = {
# 'ASTERS':"Conservatoire D'Espaces Naturels De Haute-Savoie",
# 'CPNS':"Conservatoire d'espaces naturels de Savoie",
# }
# to_bib_organismes_przh(dicorga=dicorgaprzh,first_time=True)
# insert_newrole.to_sql(
# name='t_roles',con=conn,schema='utilisateurs', index=False, if_exists='append')
Reference in New Issue View Git Blame Copy Permalink