#!/usr/bin/env python3 # -*- coding: UTF-8 -*- # import requests # import numpy as np # import pandas as pd # import os # Standard import json from qgis.core import QgsGeometry, QgsProject, QgsJsonUtils, QgsVectorLayer # PyQt from qgis.PyQt.QtCore import ( QUrl, pyqtSignal, QJsonDocument, Qt, QTextCodec ) from qgis.PyQt.QtNetwork import QNetworkReply, QNetworkRequest from qgis.PyQt.QtWidgets import ( QAction, QMessageBox, QDialog, QApplication ) from gn_tools.__about__ import __api_cd_sta__ def get_status(lst,con): sql = """ SELECT t.cd_nom, t.cd_ref, t.regne, t.phylum, t.classe, t.ordre, t.famille, t.group1_inpn, t.group2_inpn, t.group3_inpn, t.nom_vern, t.nom_complet, t.nom_valide, t.lb_nom, --s.* s.cd_sig, s.rq_statut, s.code_statut, s.cd_type_statut, s.label_statut, s.niveau_admin, s.full_citation, s.doc_url FROM taxonomie.taxref t JOIN taxonomie.v_bdc_status s USING (cd_nom) WHERE t.cd_nom IN {cd_nom} ;""".format(cd_nom = tuple(lst)) return pd.read_sql_query(sql,con) def get_type_status(con): sql = """ SELECT * FROM taxonomie.bdc_statut_type ;""" return pd.read_sql_query(sql,con) def get_api_status(api,cd_nom:int): res = requests.api.get('%s/%i'%(api,cd_nom)) if res.status_code == 200: return res.json() else : raise('Error : %i\tcd_nom : %i'%(res.status_code,cd_nom)) def get_taxon_status(lst,api): from datetime import datetime as dt init = dt.now() st = [get_api_status(api,x) for x in lst] # TOO LONG print(dt.now()-init) phylo = { 'cd_ref':[x['cd_ref'] for x in st], 'nom_valide':[x['nom_valide'] if 'nom_valide' in x.keys() else None for x in st], 'nom_vernac':[x['nom_vern'] if 'nom_vern' in x.keys() else None for x in st], 'regne':[x['regne'] if 'regne' in x.keys() else None for x in st], 'group1_inp':[x['group1_inpn'] if 'group1_inpn' in x.keys() else None for x in st], 'group2_inp':[x['group2_inp'] if 'group2_inp' in x.keys() else None for x in st], 'group3_inpn':[x['group3_inpn'] for x in st], 'classe':[x['classe'] if 'classe' in x.keys() else None for x in st], 'ordre':[x['ordre'] if 'ordre' in x.keys() else None for x in st], 'famille':[x['famille'] if 'famille' in x.keys() else None for x in st]} cd_status = { 'AL':[ [val['values'][v]['code_statut'] for val in x['status']['AL']['text'].values() for v in val['values'] ] if 'AL' in x['status'].keys() else None for x in st ], 'BERN':[ [val['values'][v]['code_statut'] for val in x['status']['BERN']['text'].values() for v in val['values'] ] if 'BERN' in x['status'].keys() else None for x in st ], 'BONN':[ [val['values'][v]['code_statut'] for val in x['status']['BONN']['text'].values() for v in val['values'] ] if 'BONN' in x['status'].keys() else None for x in st ], 'DH':[ [val['values'][v]['code_statut'] for val in x['status']['DH']['text'].values() for v in val['values'] ] if 'DH' in x['status'].keys() else None for x in st ], 'DO':[ [val['values'][v]['code_statut'] for val in x['status']['DO']['text'].values() for v in val['values'] ] if 'DO' in x['status'].keys() else None for x in st ], 'LRE':[ [val['values'][v]['code_statut'] for val in x['status']['LRE']['text'].values() for v in val['values'] ] if 'LRE' in x['status'].keys() else None for x in st ], 'LRM':[ [val['values'][v]['code_statut'] for val in x['status']['LRM']['text'].values() for v in val['values'] ] if 'LRM' in x['status'].keys() else None for x in st ], 'LRN':[ [val['values'][v]['code_statut'] for val in x['status']['LRN']['text'].values() for v in val['values'] ] if 'LRN' in x['status'].keys() else None for x in st ], 'LRR':[ [val['values'][v]['code_statut'] for val in x['status']['LRR']['text'].values() for v in val['values'] ] if 'LRR' in x['status'].keys() else None for x in st ], 'PAPNAT':[ [val['values'][v]['code_statut'] for val in x['status']['PAPNAT']['text'].values() for v in val['values'] ] if 'PAPNAT' in x['status'].keys() else None for x in st ], 'PD':[ [val['values'][v]['code_statut'] for val in x['status']['PD']['text'].values() for v in val['values'] ] if 'PD' in x['status'].keys() else None for x in st ], 'PNA':[ [val['values'][v]['code_statut'] for val in x['status']['PNA']['text'].values() for v in val['values'] ] if 'PNA' in x['status'].keys() else None for x in st ], 'PR':[ [val['values'][v]['code_statut'] for val in x['status']['PR']['text'].values() for v in val['values'] ] if 'PR' in x['status'].keys() else None for x in st ], 'REGL':[ [val['values'][v]['code_statut'] for val in x['status']['REGL']['text'].values() for v in val['values'] ] if 'REGL' in x['status'].keys() else None for x in st ], 'REGLII':[ [val['values'][v]['code_statut'] for val in x['status']['REGLII']['text'].values() for v in val['values'] ] if 'REGLII' in x['status'].keys() else None for x in st ], 'REGLLUTTE':[ [val['values'][v]['code_statut'] for val in x['status']['REGLLUTTE']['text'].values() for v in val['values'] ] if 'REGLLUTTE' in x['status'].keys() else None for x in st ], 'REGLSO':[ [val['values'][v]['code_statut'] for val in x['status']['REGLSO']['text'].values() for v in val['values'] ] if 'REGLSO' in x['status'].keys() else None for x in st ], 'SCAP NAT':[ [val['values'][v]['code_statut'] for val in x['status']['SCAP NAT']['text'].values() for v in val['values'] ] if 'SCAP NAT' in x['status'].keys() else None for x in st ], 'SCAP REG':[ [val['values'][v]['code_statut'] for val in x['status']['SCAP REG']['text'].values() for v in val['values'] ] if 'SCAP REG' in x['status'].keys() else None for x in st ], 'SENSNAT':[ [val['values'][v]['code_statut'] for val in x['status']['SENSNAT']['text'].values() for v in val['values'] ] if 'SENSNAT' in x['status'].keys() else None for x in st ], 'ZDET':[ [val['values'][v]['code_statut'] for val in x['status']['ZDET']['text'].values() for v in val['values'] ] if 'ZDET' in x['status'].keys() else None for x in st ], 'exPNA':[ [val['values'][v]['code_statut'] for val in x['status']['exPNA']['text'].values() for v in val['values'] ] if 'exPNA' in x['status'].keys() else None for x in st ] } return pd.DataFrame({**phylo,**cd_status}) def filter_bio_geo(df,zone_bio): idNotBioGeo = [] # Filtre du dommaine biogeographique sur la plaine Rhodanienne et Alpine test_rhod = df.rq_statut.str.contains('rhodanienne : Non déterminante') test_alpi = df.rq_statut.str.contains('Alpine : Non déterminante') if zone_bio == 'rhod': idNotBioGeo = df[test_rhod].index elif zone_bio == 'alpi': idNotBioGeo = df[test_alpi].index elif zone_bio == 'all': idNotBioGeo = df[test_rhod&test_alpi].index if not idNotBioGeo.empty: df.drop(idNotBioGeo, inplace=True) return df def form_territoire(df,terr): is_dep = df.cd_sig.str.contains('INSEED') is_38 = df.cd_sig=='INSEED38' if terr == 'isere': filter_id = df[is_dep&(~is_38)].index df.drop(filter_id,inplace=True) elif terr == 'platiere': lst_stat = df[is_dep&(~is_38)].cd_type_statut.unique() # is_lst = df.cd_type_statut.isin(lst_stat) df.loc[is_dep&(~is_38),'cd_type_statut'] = ( df[is_dep&(~is_38)] .cd_type_statut + '_' + df[is_dep&(~is_38)].cd_sig.str.strip('INSEED') ) df.loc[is_38,'cd_type_statut'] = ( df[is_38] .cd_type_statut + '_38' ) return df dict_dep = { '38':'Isère', '42':'Loire', '07':'Ardèche', '26':'Drôme', } class PivotStatus(QDialog): finished_dl = pyqtSignal() """ Create a filter, specific to GeoNature ZH API and fetch all the features corresponding. """ def __init__( self, parent=None, network_manager=None, layer=None, cd_nom=None ): super().__init__() print(parent.dlg.tab) self.dlg = parent.dlg self.src_name = parent.src_name self.domain_url = parent.domain_url self.network_manager = network_manager self.layer = layer self.cd_nom = cd_nom self.url_api = self.domain_url + __api_cd_sta__ # QApplication.setOverrideCursor(Qt.WaitCursor) self.multi_dep = True self.cookie = (self.network_manager.cookieJar() .cookiesForUrl(QUrl(self.domain_url))[0] ) self._pending_downloads = 0 self.import_status() @property def pending_downloads(self): return self._pending_downloads def import_status(self): idx = self.layer.fields().indexOf(self.cd_nom) unique_cd = self.layer.uniqueValues(idx) print(len(unique_cd)) self.res = {} self.get_status(unique_cd) def get_status(self,cd_nom): self._pending_downloads += 1 url = QUrl( self.url_api + '?fields=status&cd_nom=' + ','.join(map(str,cd_nom)) ) request = QNetworkRequest(url) request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json") request.setHeader(QNetworkRequest.CookieHeader, self.cookie) reply = self.network_manager.get(request) reply.finished.connect(lambda: self.import_finished(reply)) def import_finished(self, reply): self._pending_downloads -= 1 print(reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)) if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200: print(f"code: {reply.error()} message: {reply.errorString()}") else: data_request = reply.readAll().data().decode() data = json.loads(data_request) self.layer_type = self.layer.wkbType().name self.layer_crs = self.layer.crs() bdc = self.v_bdc_status(data) bdc_cd = self._v_bdc_(data,'code') bdc_lab = self._v_bdc_(data,'label') self.add_new_layer(bdc) self.add_new_layer(bdc_cd) self.add_new_layer(bdc_lab) # remak_data = { # "features" :[ # { # "geometry": # json.loads( # QgsGeometry.unaryUnion([ # g.geometry() for g in self.layer.getFeatures() # if g.attribute(name_cd_nom) == i['cd_ref'] # ]).asJson() # ), # "properties" : i , # "type": "Feature" # } # for i in data['items'] # ], # "type": "FeatureCollection", # "geom_type": "Multi"+layer_type # } # fcString = json.dumps(remak_data) # codec = QTextCodec.codecForName("UTF-8") # fields = QgsJsonUtils.stringToFields(fcString, codec) # feats = QgsJsonUtils.stringToFeatureList(fcString, fields, codec) # vl = QgsVectorLayer(remak_data['geom_type'], lay_status_name, "memory") # if vl.sourceCrs() != layer_crs: # print('Changed crs !') # # crs = vl.crs() # # crs.createFromId(self.api['geometry_srid']) # vl.setCrs(layer_crs) # dp = vl.dataProvider() # dp.addAttributes(fields) # vl.updateFields() # dp.addFeatures(feats) # vl.updateExtents() if self.pending_downloads == 0: self.finished_dl.emit() QApplication.restoreOverrideCursor() def v_bdc_status(self,data): return { "features" :[ { "properties" : {**{ ii:i[ii] for ii in i if ii != 'status' }, **ii }, "type": "Feature" } for i in data['items'] for ii in i['status'] ], "type": "FeatureCollection", "geom_type": "No geometry", "layer_name": 'Status - ' + self.layer.name() } def _v_bdc_(self,data,type_status): dtype = 'code_statut' if type_status=='code' else 'label_statut' return { "features" :[ { "geometry": json.loads( QgsGeometry.unaryUnion([ g.geometry() for g in self.layer.getFeatures() if g.attribute(self.cd_nom) == i['cd_ref'] ]).asJson() ), "properties" : { **{ ii:i[ii] for ii in i if ii != 'status' }, **{ ii['cd_type_statut'] + '_' + ii['cd_sig'].strip('INSEED') if self.multi_dep and 'INSEED' in ii['cd_sig'] else ii['cd_type_statut'] : ii[dtype] for ii in i['status'] } }, "type": "Feature" } for i in data['items'] ], "type": "FeatureCollection", "geom_type": self.layer_type if "Multi" in self.layer_type else "Multi" + self.layer_type, "layer_name": 'Status %s - '%type_status + self.layer.name() } def add_new_layer(self,data): geom_type = data['geom_type'] layername = data['layer_name'] fcString = json.dumps(data) codec = QTextCodec.codecForName("UTF-8") fields = QgsJsonUtils.stringToFields(fcString, codec) feats = QgsJsonUtils.stringToFeatureList(fcString, fields, codec) vl = QgsVectorLayer(geom_type, layername, "memory") if vl.sourceCrs() != self.layer_crs: print('Changed crs !') # crs = vl.crs() # crs.createFromId(self.api['geometry_srid']) vl.setCrs(self.layer_crs) dp = vl.dataProvider() dp.addAttributes(fields) vl.updateFields() dp.addFeatures(feats) vl.updateExtents() QgsProject.instance().addMapLayer(vl)