325 lines
13 KiB
Python
325 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: UTF-8 -*-.
|
|
|
|
import geopandas as gpd
|
|
from pycen import con,zh
|
|
from os import path,listdir
|
|
from re import findall
|
|
|
|
v_zh = zh().v_zoneshumides()
|
|
PATH = '/media/colas/SRV/FICHIERS/OUTILS/CARTOGRAPHIE/ESPACE DE TRAVAIL/ETUDES/Zones humides'
|
|
lst_layer = [x for x in listdir(PATH) if (x.endswith('.shp') or x.endswith('.gpkg')) and x.startswith('ZHP')]
|
|
|
|
def upper_first_letter(s):
|
|
if s is None or gpd.pd.isna(s):
|
|
return s
|
|
else:
|
|
return s[0].upper() + s[1:]
|
|
|
|
|
|
def manual_replace(s, old, new, index):
|
|
if s is None or gpd.pd.isna(s):
|
|
return s
|
|
else:
|
|
if old == s[index]:
|
|
return s[:index] + new + s[index +1:]
|
|
else :
|
|
return s
|
|
|
|
|
|
def extract_info(df,value):
|
|
|
|
t1 = df[0].str.lower().str.contains(value,na=False)
|
|
t2 = df[1].str.lower().str.contains(value,na=False)
|
|
t3 = df[2].str.lower().str.contains(value,na=False)
|
|
t4 = df[3].str.lower().str.contains(value,na=False)
|
|
t5 = df[4].str.lower().str.contains(value,na=False)
|
|
df[5] = None
|
|
|
|
for i,t in enumerate([t1,t2,t3,t4,t5]):
|
|
if t.any():
|
|
df.loc[t,5] = df[t][i]
|
|
|
|
return df[5].str.strip()
|
|
|
|
|
|
def extract_descripti2(df):
|
|
df['statut'] = extract_info(df,'isit|possible ?|absence zh|non cartographier')
|
|
df['source'] = extract_info(df,'photointe|ortho|scan|information communale')
|
|
return df
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
gdf = gpd.GeoDataFrame()
|
|
for layer in lst_layer:
|
|
print(layer)
|
|
data = gpd.read_file(path.join(PATH,layer))
|
|
gdf = gpd.pd.concat([gdf,data])
|
|
|
|
gdf.DATEMODIF = gpd.pd.to_datetime(gdf.DATEMODIF,format='mixed')
|
|
gdf.columns = gdf.columns.str.lower()
|
|
gdf.sort_values(['datemodif'],inplace=True,ignore_index=True,ascending=False)
|
|
gdf.drop(columns=['X','Y'],inplace=True,errors='ignore')
|
|
gdf.rename_geometry('geom',inplace=True)
|
|
# Suppression des doublons de géometries - temps de calcul long
|
|
gdf.drop_duplicates(subset='geom',inplace=True,ignore_index=True,keep='first')
|
|
|
|
tmp = (gdf.descripti2
|
|
.str.replace('/','|')
|
|
.str.replace('06|19','06/19')
|
|
.str.replace('04|12|2019','04/12/2019')
|
|
.str.replace('vistier','visiter')
|
|
.str.replace('AbsenceZH','Absence ZH')
|
|
.str.replace('Zone humide non cartographiée et','non cartographier | ZH')
|
|
.str.replace('continuité zh de la gère, à visiter','continuité zh de la gère | à visiter')
|
|
.str.split('\\|| - ',expand=True)
|
|
.replace('',None))
|
|
res = extract_descripti2(tmp)
|
|
gdf[['statut','source']] = res[['statut','source']]
|
|
|
|
gdf['comment'] = (gdf.descripti2
|
|
.str.replace('vistier','visiter')
|
|
.str.replace('AbsenceZH','Absence ZH')
|
|
.str.replace('Zone humide non cartographiée et','non cartographier | ZH')
|
|
.str.replace('continuité zh de la gère, à visiter','continuité zh de la gère | à visiter')
|
|
.str.replace('à visiter','A visiter')
|
|
.replace([x for x in gdf.statut.unique() if x is not None],'',regex=True)
|
|
.replace([x for x in gdf.source.unique() if x is not None],'',regex=True)
|
|
.str.strip()
|
|
.replace('',None)
|
|
.apply(lambda x: manual_replace(x,'|','',0)).str.strip()
|
|
.apply(lambda x: manual_replace(x,'|','',0)).str.strip()
|
|
.replace('',None)
|
|
.apply(lambda x: manual_replace(x,'/','',0)).str.strip()
|
|
.apply(lambda x: manual_replace(x,'|','',len(x)-1) if x is not None else None)
|
|
.apply(lambda x: manual_replace(x,',','',len(x)-1) if x is not None else None)
|
|
.replace('',None)
|
|
.apply(lambda x: manual_replace(x,'/','',len(x)-1) if x is not None else None)
|
|
.str.strip()
|
|
.str.replace('| | ','|')
|
|
.replace('',None)
|
|
.apply(lambda x: manual_replace(x,'?','',len(x)-1) if x is not None else None)
|
|
.str.strip()
|
|
.apply(lambda x: manual_replace(x,'|','',len(x)-1) if x is not None else None)
|
|
.str.strip()
|
|
.str.replace('Extension zh amont - peupleraie','Peupleraie - extension zh amont')
|
|
.str.replace("Zone d'effondrement - source","Source - zone d'effondrement")
|
|
)
|
|
|
|
gdf['statut'] = (gdf['statut']
|
|
.apply(lambda x: upper_first_letter(x))
|
|
.str.replace(' ?','?')
|
|
.str.replace('?',' ?')
|
|
.replace('À visiter','A visiter'))
|
|
gdf['source'] = (gdf['source']
|
|
.apply(lambda x: upper_first_letter(x))
|
|
.str.replace('Interprétation','Photointerprétation'))
|
|
|
|
is_inventaire = (gdf.observateu
|
|
.str.contains('Le Pic Vert|LPO Isère|Lo Parvi|Nature Vivante|ONF',case=False,na=False))
|
|
|
|
isna_source = gdf['source'].isna()
|
|
is_cenisere = gdf.observateu.str.contains('cen isère',case=False,na=False)
|
|
gdf.loc[is_inventaire&isna_source,'source'] = 'Inventaire'
|
|
|
|
|
|
gdf.loc[~is_inventaire&isna_source&~is_cenisere].observateu
|
|
|
|
gdf['type'] = (gdf.comment
|
|
.str.replace('prairie','Prairie')
|
|
.str.replace('Prairie humide','Prairiehumide')
|
|
.str.split('\\|| ',n=1)
|
|
.str[0]
|
|
.str.replace('Prairiehumide','Prairie humide')
|
|
.str.strip()
|
|
.replace({
|
|
'Roselières':'Roselière',
|
|
'Ornières':'Ornière',
|
|
'Fontaine':'Source',
|
|
'Lac':'Mare/lac alpin',
|
|
'Culture':'Culture temporairement inondée',
|
|
'Grève':'Alluvion',
|
|
'ZH':'Tourbière',
|
|
})
|
|
.apply(lambda x: upper_first_letter(x))
|
|
)
|
|
gdf['comment'] = (gdf.comment
|
|
.str.replace('prairie','Prairie')
|
|
.replace([x for x in gdf.type.unique() if x is not None or x.contains('INVZH')],'',regex=True)
|
|
.str.replace('Prairie humide','Prairiehumide')
|
|
.str.split('\\|| ',n=1)
|
|
.str[1]
|
|
.str.replace('Prairiehumide','Prairie humide')
|
|
.str.strip()
|
|
.replace('',None)
|
|
.apply(lambda x: manual_replace(x,'-','',0)).str.strip()
|
|
)
|
|
et = gdf.comment.str.startswith('et ',na=False)
|
|
de = gdf.comment.str.startswith('de ',na=False)
|
|
gdf.loc[et,'comment'] = gdf.loc[et,'comment'].str.replace('et ','',n=1)
|
|
gdf.loc[de,'comment'] = gdf.loc[de,'comment'].str.replace('de ','',n=1)
|
|
|
|
invzh = gdf['type'].str.contains('INVZH',na=False)
|
|
gdf.loc[invzh,'comment'] = gdf.loc[invzh,'type']
|
|
gdf.loc[invzh,'type'] = None
|
|
|
|
gdf.comment = (gdf.comment
|
|
.str.strip()
|
|
.str.removeprefix('\|')
|
|
.str.strip()
|
|
.str.removeprefix('?')
|
|
.str.strip()
|
|
.str.removeprefix('-')
|
|
.str.removesuffix('-')
|
|
.str.strip()
|
|
.replace('',None)
|
|
.apply(lambda x: upper_first_letter(x)))
|
|
is_interpret = gdf['source'].str.contains('interprétation',case=False,na=False)
|
|
gdf.loc[is_interpret,'ponctuelle'] = 'Incertain'
|
|
gdf.loc[~is_interpret,'ponctuelle'] = 'Oui'
|
|
|
|
is_jlg = gdf.observateu=='CEN Isère (J.-L. Grossi)'
|
|
type_isna = gdf['type'].isna()
|
|
gdf.loc[is_jlg&type_isna,'type'] = 'Mare'
|
|
gdf.drop(gdf[gdf.observateu.isna()].index,inplace=True)
|
|
|
|
is_balm = gdf.observateu=='CEN Isère (C. Balmain)'
|
|
is_lafo = gdf.observateu=='CEN Isère (A.Lafon)'
|
|
is_nico = gdf.observateu=='CEN Isère (N. Biron)'
|
|
is_obs_inc = gdf.observateu=='Inconnu'
|
|
is_zh = gdf['type']=='Petite zh'
|
|
is_inconnu = gdf['type']=='Inconnu'
|
|
gdf.loc[is_balm&~is_zh&is_inconnu,'type'] = 'Étang'
|
|
gdf.loc[is_balm&is_zh&is_inconnu,'type'] = 'Suintement'
|
|
gdf.loc[is_lafo&is_inconnu,'type'] = 'Boisement'
|
|
gdf.loc[is_obs_inc&is_inconnu,'type'] = 'Étang'
|
|
gdf.loc[is_nico&is_inconnu,'type'] = 'Prairie humide'
|
|
|
|
|
|
is_visite = gdf['statut'].isin(['Possible ?','Groupement fontinal possible ?','Groupement fontinal, possible ?','Roselière et possible mare','Non cartographier'])
|
|
gdf.loc[is_visite,'statut'] = 'A visiter'
|
|
statut_isna = gdf['statut'].isna()
|
|
gdf.loc[is_interpret&statut_isna,'statut'] = 'A visiter'
|
|
|
|
is_zh2 = gdf['statut'].isna()&gdf['comment'].str.startswith('Zh',na=False)
|
|
gdf.loc[is_zh2,'statut'] = 'A décrire'
|
|
|
|
dict_type = {
|
|
**dict.fromkeys(['Barrage','Bassin','Lagunage'],'Artificiel'),
|
|
**dict.fromkeys(['Mégaphorbiaie','Prairie','Prairie humide','Roselière'],'Prairie humide et roselière'),
|
|
**dict.fromkeys(['Source', 'Suintement', 'Talus'],'Source et Suintement'),
|
|
**dict.fromkeys(['Ornière', 'Mare'],'Mare'),
|
|
**dict.fromkeys(['Boisement', 'Peupleraie'],'Boisement'),
|
|
}
|
|
gdf['classe'] = (gdf['type']
|
|
.replace(dict_type))
|
|
|
|
# Harmonisation des observateurs
|
|
dict_obs = {
|
|
'NB':'CEN Isère (N. Biron)',
|
|
'Nico':'CEN Isère (N. Biron)',
|
|
'Alix':'CEN Isère (A. Guedou)',
|
|
'N BIRON':'CEN Isère (N. Biron)',
|
|
'N. Biron (CEN Isère)':'CEN Isère (N. Biron)',
|
|
'Nicolas Biron':'CEN Isère (N. Biron)',
|
|
'M. Juton (CEN Isère)':'CEN Isère (M. Juton)',
|
|
'CEN Isère (J.-L. Grossi)':'CEN Isère (J. L. Grossi)',
|
|
'Lpo Isère (F. Frossard)':'LPO Isère (F. Frossard)',
|
|
'Lpo Isère (H. Coffre)':'LPO Isère (H. Coffre)',
|
|
'Lpo Isère (D. Loose)':'LPO Isère (D. Loose)',
|
|
'J. F. Noblet':'Le Pic Vert (J. F. Noblet)',
|
|
'R. Fonters':'LPO Isère (R. Fonters)',
|
|
'R. Marciau':'CEN Isère (R. Marciau)',
|
|
}
|
|
gdf.replace(dict_obs,inplace=True)
|
|
|
|
# Suppression des ZH taguer Absence
|
|
not_zh = gdf['statut']=='Absence ZH'
|
|
idx_notzh = gdf[not_zh].index
|
|
gdf.drop(idx_notzh,inplace=True)
|
|
|
|
# Intersection des zh_ponctuelles et zh_surfacique - temps de calcul long
|
|
v_zh_union = v_zh.unary_union
|
|
is_intersect = gdf.intersects(v_zh_union)
|
|
|
|
lst_drop = ['Tourbière','Culture','Boisement','Prairie humide','Roselière','Grève','Barrage','Lac']
|
|
drop_prospect_pontc = ~gdf['type'].isin(lst_drop)
|
|
gdf_zh = gdf[is_intersect&drop_prospect_pontc].copy()
|
|
gdf_ponct= gdf[~is_intersect].copy()
|
|
|
|
|
|
# A faire :
|
|
# - Intersecter avec la couche zh surfacique,
|
|
# dissocier les donner d'intersection et conserver
|
|
# dans une autre couche les zones ponctuelles correspondant à des Mares/Etangs/bassin/Lac
|
|
|
|
bdd_table = 'zh_ponctuelles'
|
|
gdf_ponct.to_postgis(bdd_table,con,'zones_humides',if_exists='replace',index=True if 'gid' not in gdf.columns else False,index_label='gid')
|
|
|
|
sql = """
|
|
ALTER TABLE zones_humides.{tab} ADD PRIMARY KEY (gid);
|
|
GRANT ALL ON TABLE zones_humides.{tab} TO grp_admin;
|
|
GRANT SELECT ON TABLE zones_humides.{tab} TO grp_consult;
|
|
""".format(tab=bdd_table)
|
|
with con.begin() as cnx:
|
|
cnx.execute(sql)
|
|
|
|
bdd_table = 'zh_ponctuelles_in_zh'
|
|
gdf_zh.to_postgis(bdd_table,con,'zones_humides',if_exists='replace',index=True if 'gid' not in gdf.columns else False,index_label='gid')
|
|
|
|
sql = """
|
|
ALTER TABLE zones_humides.{tab} ADD PRIMARY KEY (gid);
|
|
GRANT ALL ON TABLE zones_humides.{tab} TO grp_admin;
|
|
GRANT SELECT ON TABLE zones_humides.{tab} TO grp_consult;
|
|
""".format(tab=bdd_table)
|
|
with con.begin() as cnx:
|
|
cnx.execute(sql)
|
|
|
|
|
|
########### POTENTIELLE SURFACIQUE #############
|
|
sql = """
|
|
SELECT * FROM zones_humides.zh_ponctuelles;
|
|
"""
|
|
gdf = gpd.read_postgis(
|
|
sql = sql,
|
|
con = con)
|
|
is_interpret = gdf['source'].str.contains('interprétation',case=False,na=False)
|
|
del_notzh = ~(gdf['statut'] == 'Absence ZH')
|
|
avere = gdf[~is_interpret&del_notzh].copy()
|
|
is_visite = avere['statut'].isin(['Possible ?','Groupement fontinal, possible ?','Roselière et possible mare','Non cartographier'])
|
|
avere.loc[is_visite,'statut'] = 'A visiter'
|
|
dict_type = {
|
|
**dict.fromkeys(['Barrage','Bassin','Lagunage'],'Artificiel'),
|
|
**dict.fromkeys(['Mégaphorbiaie','Prairie','Prairie humide','Roselière'],'Prairie humide et roselière'),
|
|
**dict.fromkeys(['Source', 'Suintement', 'Talus'],'Source et Suintement'),
|
|
**dict.fromkeys(['Ornière', 'Mare'],'Mare'),
|
|
**dict.fromkeys(['Boisement', 'Peupleraie'],'Boisement'),
|
|
}
|
|
avere['classe'] = (gdf['type']
|
|
.replace(dict_type))
|
|
|
|
########### POTENTIELLE #############
|
|
ptent = gdf[is_interpret&del_notzh].copy()
|
|
|
|
# avere['a_visiter'] = None
|
|
avere.to_postgis('zh_ponctuelles_averee',con,'zones_humides',if_exists='replace',index_label='gid')
|
|
|
|
sql = """
|
|
ALTER TABLE zones_humides.zh_ponctuelles_averee ADD PRIMARY KEY (gid);
|
|
GRANT ALL ON TABLE zones_humides.zh_ponctuelles_averee TO grp_admin;
|
|
GRANT SELECT ON TABLE zones_humides.zh_ponctuelles_averee TO grp_consult;
|
|
"""
|
|
with con.begin() as cnx:
|
|
cnx.execute(sql)
|
|
|
|
|
|
|
|
test = """
|
|
SELECT * FROM zones_humides.zh_ponctuelles;
|
|
"""
|
|
data = gpd.read_postgis(
|
|
sql = test,
|
|
con = con)
|
|
|