tlaveille/plugin_gn_tools
1
0
Fork 0
forked from CEN38/plugin_gn_tools
Code Pull Requests Activity

init plugin

Browse Source
This commit is contained in:
Colas
2026-03-19 14:06:08 +01:00
commit 5974778b0a
65 changed files with 6606 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gn_tools/processing/__init__.py
+8
View File
@@ -0,0 +1,8 @@
#! python3 # noqa: E265
from .provider import GnToolsProvider # noqa: F401
from .login import GetLogin
from .widget import ParamWidget, LoadingImport
from .draw import RectangleFeatureTool
from .gbif import ImportDataGbif
from .status import PivotStatus
from .filter_synthese import GetFilterFromAPI
gn_tools/processing/draw.py
+104
View File
@@ -0,0 +1,104 @@
# Import PyQt libs
from PyQt5.QtCore import (
pyqtSignal,
Qt,
)
from PyQt5.QtGui import QColor
# Import PyQgis libs
from qgis.core import (
QgsRectangle,
QgsPointXY,
QgsWkbTypes,
QgsCoordinateReferenceSystem,
QgsCoordinateTransform,
)
from qgis.gui import (
QgsMapTool,
QgsMapMouseEvent,
QgsRubberBand,
)
class RectangleFeatureTool(QgsMapTool):
signal = pyqtSignal()
def __init__(self, project, canvas):
super().__init__(canvas)
self.project = project
self.canvas = canvas
# initiate variable to calculate the extent
self.start_point = None
self.end_point = None
self.new_extent = None
# flag to check if the left mouse button was pressed
self.is_left_button_pressed = False
# create a rubber line object
# to display the geometry of the dragged object on the canvas
self.rubber_band = QgsRubberBand(self.canvas, QgsWkbTypes.PolygonGeometry)
self.rubber_band.setColor(QColor(255, 0, 0, 50))
self.rubber_band.setWidth(2)
# Mouse button pressed.
def canvasPressEvent(self, event: QgsMapMouseEvent):
if event.button() == Qt.LeftButton:
self.start_point = self.toMapCoordinates(event.pos())
self.end_point = self.start_point
self.is_left_button_pressed = True
# The cursor moves across the screen.
def canvasMoveEvent(self, event: QgsMapMouseEvent):
# if the left mouse button is pressed
if self.is_left_button_pressed:
self.end_point = self.toMapCoordinates(event.pos())
# draw a rectangle on the canvas
self.showRect(self.start_point, self.end_point)
# The mouse button is released.
def canvasReleaseEvent(self, event: QgsMapMouseEvent):
# if the left mouse button was released
if event.button() == Qt.LeftButton:
if event.type() == 3:
# get the rectangle created from the click point and
# left mouse button release points
self.new_extent = self.rectangle()
self.signal.emit()
# toggle the flag, the left mouse button is no longer held down
self.is_left_button_pressed = False
self.rubber_band.reset() # remove rubber band from canvas
sourceCrs = QgsCoordinateReferenceSystem(self.project.crs())
destCrs = QgsCoordinateReferenceSystem(4326)
tr = QgsCoordinateTransform(sourceCrs, destCrs, self.project)
# Draw the rectangle on the canvas
def showRect(self, startPoint, endPoint):
# Reset the last drawn rectangle
self.rubber_band.reset(QgsWkbTypes.PolygonGeometry)
# Make sure the mouse has moved
if startPoint.x() == endPoint.x() or startPoint.y() == endPoint.y():
return
# Create the rectangle
point1 = QgsPointXY(startPoint.x(), startPoint.y())
point2 = QgsPointXY(startPoint.x(), endPoint.y())
point3 = QgsPointXY(endPoint.x(), endPoint.y())
point4 = QgsPointXY(endPoint.x(), startPoint.y())
self.rubber_band.addPoint(point1, False)
self.rubber_band.addPoint(point2, False)
self.rubber_band.addPoint(point3, False)
self.rubber_band.addPoint(point4, True) # true to update canvas
self.rubber_band.show()
def rectangle(self):
if self.start_point is None or self.end_point is None:
return None
elif self.start_point.x() == self.end_point.x() or self.start_point.y() == self.end_point.y():
return None
return QgsRectangle(self.start_point, self.end_point)
def deactivate(self):
# unload the map tool on the mouse
QgsMapTool.deactivate(self)
self.deactivated.emit()
gn_tools/processing/filter_synthese.py
+227
View File
@@ -0,0 +1,227 @@
# Standard
import json
# PyQt
from qgis.core import QgsAuthMethodConfig, QgsDataSourceUri
from qgis.PyQt.QtCore import QByteArray, QJsonDocument, QObject, QUrl, pyqtSignal, Qt
from qgis.PyQt.QtNetwork import QNetworkReply, QNetworkRequest
from qgis.PyQt.QtWidgets import QAction, QMessageBox, QDialog, QApplication
# Project
from gn_tools.__about__ import (
__orb_tax_url__
)
class GetFilterFromAPI(QDialog):
finished_params = pyqtSignal()
"""
Create a filter, specific to GeoNature ZH API and fetch all the features corresponding.
"""
def __init__(
self,
parent=None,
network_manager=None
):
super().__init__()
print("enfin !")
self.domain_url = parent.domain_url
self.tax_url = self.domain_url if parent.src_name == 'GéoNature' else __orb_tax_url__
if self.tax_url[-1] != '/':
self.tax_url += '/'
self.network_manager = network_manager
self.auth = parent.dlg.mAuthConfigSelect
self.is_connect = parent.dlg.is_connect
self.dlg = parent.dlg_syn
self.parent = parent
self.cookie = (self.network_manager.cookieJar()
.cookiesForUrl(QUrl(self.domain_url))[0]
)
# PyQt5 - Version
# PyQt6 - Version
# QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
self._pending_downloads = 0
# self.dlg.send_to_gn.
self.dlg.send_to_gn.setText(
self.dlg.send_to_gn.text() + '<br><br><a href="{url}">{url}</a>'.format(url=self.domain_url)
)
self.dlg.send_to_gn.setStyleSheet(
"border: 1px solid; border-color:blue; font-weight:bold"
)
self.dlg.send_to_gn.setOpenExternalLinks(True)
self.params = {}
self.get_jdd()
if self.parent.src_name == 'GéoNature':
self.get_phylog()
else:
self.self.dlg.regne.deselectAllOptions()
self.self.dlg.regne.clear()
self.dlg.buttonBox.accepted.connect(self.accept)
self.dlg.buttonBox.rejected.connect(self.reject)
@property
def pending_downloads(self):
return self._pending_downloads
def handle_finished(self, reply):
self._pending_downloads -= 1
# If reply is an error or reply is a redirection, login is not accepted
print(reply.attribute(
QNetworkRequest.HttpStatusCodeAttribute
))
print(reply.error(),' : ',reply.dtype)
self.network_manager.cookieStatus = reply.attribute(
QNetworkRequest.HttpStatusCodeAttribute
)
QApplication.restoreOverrideCursor()
if (
reply.error() != QNetworkReply.NoError
or self.network_manager.cookieStatus != 200
):
# Error code 299 means the logins are invalid
# Redirection not equal to 200 means a redirection happened
if reply.dtype == 'JDD':
self.dlg.jdd_box.setDisabled(True)
self.dlg.jdd_box.clear()
elif reply.dtype in ['PHYLOG','TAX_TREE']:
self.dlg.regne.setDisabled(True)
self.dlg.regne.clear()
if reply.error() == 0:
msg = QMessageBox()
msg.warning(
None,
self.tr("Warning"),
self.tr(
"Houston, we've got a problem ... \n\n L'utilisateur ou le mot de passe est invalide !"
), # noqa: E501
)
else:
print(f"code: {reply.error()} message: {reply.errorString()}")
else:
data_request = reply.readAll().data().decode()
data = json.loads(data_request)
self.params[reply.dtype] = data
self.close()
if self.pending_downloads == 1 :
self.get_taxon_tree()
if self.pending_downloads == 0:
self.set_params_to_widget()
self.finished_params.emit()
QApplication.restoreOverrideCursor()
def get_jdd(self):
self._pending_downloads += 1
QApplication.setOverrideCursor(Qt.WaitCursor)
self.dlg.jdd_box.setEnabled(True)
url_api = self.domain_url + 'api/meta/datasets?'
if self.parent.src_name == 'GéoNature':
url_api += 'orderby=dataset_shortname'
request_type = 'POST'
elif self.parent.src_name == "Biodiv'AuRA":
url_api += 'orderby=dataset_name'
request_type = 'GET'
url = QUrl(url_api)
self.send_request(url,type_request=request_type,type_param='JDD')
def get_phylog(self):
self._pending_downloads += 2
QApplication.setOverrideCursor(Qt.WaitCursor)
self.dlg.regne.setEnabled(True)
if self.parent.src_name == 'GéoNature':
url_api = self.tax_url + 'api/taxhub/api/taxref/regnewithgroupe2'
if self.parent.src_name == "Biodiv'AuRA":
url_api = self.tax_url + 'api/taxref/regnewithgroupe2'
url = QUrl(url_api)
self.send_request(url,type_request='GET',type_param='PHYLOG')
def get_taxon_tree(self):
QApplication.setOverrideCursor(Qt.WaitCursor)
if self.parent.src_name == 'GéoNature':
url_api = self.tax_url + 'api/synthese/taxons_tree'
url = QUrl(url_api)
self.send_request(url,type_request='GET',type_param='TAX_TREE')
else:
self._pending_downloads -= 1
self.set_params_to_widget()
self.finished_params.emit()
QApplication.restoreOverrideCursor()
def send_request(self,url,type_request,type_param):
print(url)
request = QNetworkRequest(url)
request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
request.setHeader(QNetworkRequest.CookieHeader, self.cookie)
if type_request == 'GET':
reply = self.network_manager.get(request)
elif type_request == 'POST':
doc = QJsonDocument({})
reply = self.network_manager.post(request,doc.toJson())
reply.dtype = type_param
reply.finished.connect(lambda: self.handle_finished(reply))
def set_params_to_widget(self):
if 'JDD' in self.params.keys():
jdd_items = [i['dataset_shortname'] for i in self.params['JDD']]
self.dlg.jdd_box.addItems(jdd_items)
if 'PHYLOG' in self.params.keys():
rgn_items = [i for i in self.params['PHYLOG'].keys()]
self.dlg.regne.addItems(rgn_items)
# self.dlg.jdd_box.setCheckedItems(items)
print('END loading params ...')
def accept(self):
params = {
'id_dataset':[ i['id_dataset']
for i in self.params['JDD']
if i['dataset_shortname'] in self.dlg.jdd_box.checkedItems()
] if 'JDD' in self.params.keys() else [],
'cd_ref_parent': [ i['cd_ref']
for i in self.params['TAX_TREE']
if i['nom_regne'] in self.dlg.regne.checkedItems()
] if 'PHYLOG' in self.params.keys() else [],
'date_min' : None if self.dlg.date_min.isNull()
else self.dlg.date_min.date().toString('yyyy-MM-dd'),
'date_max' : None if self.dlg.date_max.isNull()
else self.dlg.date_max.date().toString('yyyy-MM-dd'),
}
self.parent.syn_params = {
i : params[i] for i in params if params[i]
}
def reject(self):
self.clear_widgets()
QApplication.restoreOverrideCursor()
print("Rejected")
def clear_widgets(self):
self.dlg.jdd_box.deselectAllOptions()
self.dlg.regne.deselectAllOptions()
self.dlg.date_min.clear()
self.dlg.date_max.clear()
gn_tools/processing/gbif.py
+205
View File
@@ -0,0 +1,205 @@
# Import basic libs
import json
import time
# Import PyQt libs
from PyQt5.QtCore import QObject, pyqtSignal, QUrl, Qt, QVariant
from PyQt5.QtGui import QColor
from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply
from PyQt5.QtWidgets import QDialog, QPushButton, QApplication, QVBoxLayout, QHBoxLayout, QButtonGroup, QCheckBox, QLabel
# Import PyQgis libs
from qgis.utils import iface
from qgis.core import QgsProject, QgsFeature, QgsDataSourceUri, QgsRectangle, QgsNetworkAccessManager, QgsVectorLayer, QgsPointXY, QgsWkbTypes, QgsMapLayerProxyModel, QgsGeometry, QgsCoordinateReferenceSystem, QgsCoordinateTransform, QgsField
from qgis.gui import QgsMapTool, QgsMapMouseEvent, QgsRubberBand, QgsMapLayerComboBox
class GetTaxrefId(QObject):
finished_dl = pyqtSignal()
"""Get multiples informations from a getcapabilities request.
List all layers available, get the maximal extent of all the Wfs' data."""
def __init__(self, network_manager=None, project=None, layer=None, field_name=None, field_rank=None):
super().__init__()
self.network_manager = network_manager
self.project = project
self.layer = layer
self.field_name = field_name
self.field_rank = field_rank
self._pending_downloads = 0
self.ids = []
self.names = []
self.rank = []
for obs in self.layer.getFeatures():
self.ids.append(obs.id())
self.names.append(obs[self.field_name])
self.rank.append(obs[self.field_rank])
self._pending_downloads = len(self.names)
self._iterate_names = 0
self.download(self.ids[self._iterate_names], self.names[self._iterate_names], self.rank[self._iterate_names])
@property
def pending_downloads(self):
return self._pending_downloads
@property
def iterate_names(self):
return self._iterate_names
def download(self, feature_id, nom, rank):
if rank.lower() == 'stateofmatter':
self._iterate_names += 1
self.download(self.ids[self._iterate_names], self.names[self._iterate_names], self.rank[self._iterate_names])
else:
if rank.lower() == 'complex' or rank.lower() == 'hybrid':
rank = 'species'
elif rank.lower() == 'epifamily' or rank.lower() == 'section':
rank = 'genus'
elif rank.lower() == 'subtribe':
rank = 'subfamily'
url = "https://api.checklistbank.org/nameusage/search?content=SCIENTIFIC_NAME&datasetKey=2008&facet=datasetKey&facet=rank&facet=issue&facet=status&facet=nomStatus&facet=nameType&facet=field&facet=authorship&facet=authorshipYear&facet=extinct&facet=environment&facet=origin&limit=50&offset=0&q={nom}&rank={rank}&type=PREFIX".format(nom=nom.split("(")[0], rank=rank.lower())
print(url)
request = QNetworkRequest(QUrl(url))
request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
reply = self.network_manager.get(request)
reply.finished.connect(lambda: self.handle_finished(reply, feature_id))
self._iterate_names += 1
def handle_finished(self, reply, feature_id):
self._pending_downloads -= 1
if reply.error() != QNetworkReply.NoError:
print(f"code: {reply.error()} message: {reply.errorString()}")
if reply.error() == 403:
print("Service down")
else:
data_request = reply.readAll().data().decode()
if data_request != "":
res = json.loads(data_request)
if "result" in res:
taxref_id = res["result"][0]["id"]
taxref_name = res["result"][0]["usage"]["label"]
self.layer.startEditing()
self.layer.changeAttributeValue(feature_id, self.layer.fields().indexFromName("cd_nom"), taxref_id)
self.layer.changeAttributeValue(feature_id, self.layer.fields().indexFromName("nom_scien"), taxref_name)
self.layer.commitChanges()
self.layer.triggerRepaint()
if self.pending_downloads == 0:
self.project.addMapLayer(self.layer)
self.finished_dl.emit()
else:
self.download(self.ids[self._iterate_names], self.names[self._iterate_names], self.rank[self._iterate_names])
class ImportDataGbif(QObject):
finished_dl = pyqtSignal()
"""Get multiples informations from a getcapabilities request.
List all layers available, get the maximal extent of all the Wfs' data."""
def __init__(self, network_manager=None, project=None, geom=None, ):
super().__init__()
self._pending_downloads = 0
self._pending_pages = 1
self.network_manager = network_manager
self.project = project
self.layer = self.new_layer()
self.geom = geom
self.new_features = []
self.max_obs = 0
self.total_pages = 0
self.download()
@property
def pending_downloads(self):
return self._pending_downloads
@property
def pending_pages(self):
return self._pending_pages
def new_layer(self):
geom_type = "Point"
new_layer = QgsVectorLayer(geom_type + "?crs=epsg:4326", "GBIF", "memory")
new_layer.startEditing()
new_layer.addAttribute(QgsField("id", QVariant.String, "string", 254))
new_layer.addAttribute(QgsField("cd_nom", QVariant.Int, "integer", 10))
new_layer.addAttribute(QgsField("niveau", QVariant.String, "string", 254))
new_layer.addAttribute(QgsField("nom", QVariant.String, "string", 254))
new_layer.addAttribute(QgsField("nom_scien", QVariant.String, "string", 100000))
new_layer.addAttribute(QgsField("observat", QVariant.String, "string", 254))
new_layer.addAttribute(QgsField("date", QVariant.String, "string", 254))
new_layer.addAttribute(QgsField("projet", QVariant.String, "string", 254))
new_layer.addAttribute(QgsField("effectif", QVariant.Int, "integer", 10))
new_layer.addAttribute(QgsField("url", QVariant.String, "string", 254))
new_layer.commitChanges()
new_layer.triggerRepaint()
return new_layer
def download(self):
url = "https://api.gbif.org/v1/occurrence/search?advanced=false&geometry={polygon}&offset={offset}&limit=1000".format(
offset=self.pending_pages, polygon=self.geom.asWkt())
url = QUrl(url)
self.nb_request = 1
print(url)
request = QNetworkRequest(url)
request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
reply = self.network_manager.get(request)
reply.finished.connect(lambda: self.handle_finished(reply))
self._pending_downloads += 1
def handle_finished(self, reply):
self._pending_downloads -= 1
if reply.error() != QNetworkReply.NoError:
print(f"code: {reply.error()} message: {reply.errorString()}")
if reply.error() == 403:
print("Service down")
else:
data_request = reply.readAll().data().decode()
if self.pending_downloads == 0:
res = json.loads(data_request)
self.max_obs = res["count"]
self.total_pages = int(self.max_obs/20) + 1
for obs in res["results"]:
new_geom = QgsGeometry.fromPointXY(QgsPointXY(obs["decimalLongitude"], obs["decimalLatitude"]))
new_feature = QgsFeature(self.layer.fields())
new_feature.setGeometry(new_geom)
new_feature.setAttribute(0, str(obs["key"]))
new_feature.setAttribute(1, 0)
new_feature.setAttribute(2, obs["taxonRank"])
if obs["taxonRank"].lower() in ['kingdom', 'phylum', 'order', 'family', 'genus', 'species']:
new_feature.setAttribute(3, obs[obs["taxonRank"].lower()])
else:
new_feature.setAttribute(3, obs["scientificName"])
new_feature.setAttribute(4, "")
new_feature.setAttribute(5, "")
new_feature.setAttribute(6, str(obs["eventDate"]))
# new_feature.setAttribute(7, obs["_datasetKey"]["title"])
new_feature.setAttribute(7, obs["datasetName"] if "datasetName" in obs.keys() else None )
new_feature.setAttribute(8, 1)
new_feature.setAttribute(9, "https://www.gbif.org/fr/occurrence/" + str(obs["key"]))
self.new_features.append(new_feature)
if self.nb_request == 10:
time.sleep(10)
if self.pending_pages < self.total_pages:
self._pending_pages += 300
self.nb_request += 1
self.download()
else:
# Add the new feature to the temporary layer
self.layer.startEditing()
self.layer.dataProvider().addFeatures(self.new_features)
self.layer.updateExtents()
self.layer.commitChanges()
self.layer.triggerRepaint()
GetTaxrefId(self.network_manager, self.project, self.layer, "nom", "niveau")
self.finished_dl.emit()
gn_tools/processing/login.py
+173
View File
@@ -0,0 +1,173 @@
# Standard
import json
# PyQt
from qgis.core import QgsAuthMethodConfig, QgsDataSourceUri
from qgis.PyQt.QtCore import QByteArray, QJsonDocument, QObject, QUrl, pyqtSignal
from qgis.PyQt.QtNetwork import QNetworkReply, QNetworkRequest
from qgis.PyQt.QtWidgets import QAction, QMessageBox, QDialog
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QApplication
# from PyQt6.QtCore import Qt
# from PyQt6.QtWidgets import QApplication
# Project
from gn_tools.__about__ import __login_url__
class GetLogin(QDialog):
finished_login = pyqtSignal()
"""
Create a filter, specific to GeoNature ZH API and fetch all the features corresponding.
"""
def __init__(
self,
parent=None,
network_manager=None,
params=None
):
super().__init__()
print("enfin !")
self.domain_url = parent.domain_url
self.network_manager = network_manager
self.auth = parent.dlg.mAuthConfigSelect
self.is_connect = parent.dlg.is_connect
self.params = params
print(self.auth.configId())
# self.zhs = None
# self.json_filter = {}
# self.total_zh = 0
# PyQt5 - Version
QApplication.setOverrideCursor(Qt.WaitCursor)
# PyQt6 - Version
# QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
self._pending_downloads = 0
self.check_password()
@property
def pending_downloads(self):
return self._pending_downloads
def handle_finished(self, reply):
self._pending_downloads -= 1
QApplication.restoreOverrideCursor()
# If reply is an error or reply is a redirection, login is not accepted
print(reply.attribute(
QNetworkRequest.HttpStatusCodeAttribute
))
print('Login : ',reply.error())
self.network_manager.cookieStatus = reply.attribute(
QNetworkRequest.HttpStatusCodeAttribute
)
if (
reply.error() != QNetworkReply.NoError
or self.network_manager.cookieStatus != 200
):
# Error code 299 means the logins are invalid
# Redirection not equal to 200 means a redirection happened
if reply.error() == 0:
self.is_connect.setText("Aie !")
msg = QMessageBox()
msg.warning(
None,
self.tr("Warning"),
self.tr(
"Houston, we've got a problem ... \n\n L'utilisateur ou le mot de passe est invalide !"
), # noqa: E501
)
elif reply.error() == 3:
msg = QMessageBox()
msg.warning(
None,
self.tr("Warning"),
self.tr(
"GéoNature's URL not found, add it to the plugin settings"
), # noqa: E501
)
self.close()
self.parent.iface.showOptionsDialog(
currentPage="mOptionsPage{}".format(__title__)
)
self.parent.create_project_from_gn()
else:
print(f"code: {reply.error()} message: {reply.errorString()}")
else:
self.cookies = self.network_manager.cookieJar()
self.is_connect.setText("Good Job !")
# If the authenfication configuration already exists,
# modify it with the new values
# self.manager_config.setConfigMap(
# {
# "password": self.usr,
# "username": self.pwd,
# }
# )
# self.manager.storeAuthenticationConfig(
# self.manager_config, True
# )
# delete sensitive informations... even if it's in the database now...
self.network_manager.accepted = True
# Add the cookie created with the login to the request
self.network_manager.setCookieJar(self.cookies)
# self.network_manager.setCookiesFromUrl(self.cookies,self.domain_url)
self.close()
if self.pending_downloads == 0:
self.finished_login.emit()
return self.network_manager
def get_param_connect(self):
if self.params:
self.usr = self.params["usr"]
self.pwd = self.params["pwd"]
return True
elif self.auth.configId():
authCfg = self.auth.configId()
quri = QgsDataSourceUri()
quri.setParam("authcfg", authCfg)
quri.setParam("url", self.domain_url)
self.usr = quri.uri().split("user='")[1].split("' ",1)[0]
self.pwd = quri.uri().split("password='")[1].split("' ",1)[0]
return True
else :
return False
def check_password(self):
# Test if the user wrote something
if self.auth.configId() or self.params:
is_connect = self.get_param_connect()
url = QUrl(self.domain_url + __login_url__)
print(self.usr)
data_json = {
"login": (
self.usr
),
"password": (
self.pwd
),
}
print(url)
request = QNetworkRequest(url)
request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
document = QJsonDocument(data_json)
print(document.toJson())
reply = self.network_manager.post(request, document.toJson())
reply.finished.connect(lambda: self.handle_finished(reply))
self._pending_downloads += 1
else:
msg = QMessageBox()
msg.warning(
None,
self.tr("Warning"),
self.tr(
"GéoNature's Auth not found, add it to the plugin settings"
), # noqa: E501
)
gn_tools/processing/provider.py
+88
View File
@@ -0,0 +1,88 @@
#! python3 # noqa: E265
"""
Processing provider module.
"""
# PyQGIS
from qgis.core import QgsProcessingProvider
from qgis.PyQt.QtCore import QCoreApplication
from qgis.PyQt.QtGui import QIcon
# project
from gn_tools.__about__ import (
__icon_path__,
__title__,
__version__,
)
# ############################################################################
# ########## Classes ###############
# ##################################
class GnToolsProvider(QgsProcessingProvider):
"""Processing provider class."""
def loadAlgorithms(self):
"""Loads all algorithms belonging to this provider."""
pass
def id(self) -> str:
"""Unique provider id, used for identifying it. This string should be unique, \
short, character only string, eg "qgis" or "gdal". \
This string should not be localised.
:return: provider ID
:rtype: str
"""
return "gn_tools"
def name(self) -> str:
"""Returns the provider name, which is used to describe the provider
within the GUI. This string should be short (e.g. "Lastools") and localised.
:return: provider name
:rtype: str
"""
return __title__
def longName(self) -> str:
"""Longer version of the provider name, which can include
extra details such as version numbers. E.g. "Lastools LIDAR tools". This string should be localised. The default
implementation returns the same string as name().
:return: provider long name
:rtype: str
"""
return self.tr("{} - Tools".format(__title__))
def icon(self) -> QIcon:
"""QIcon used for your provider inside the Processing toolbox menu.
:return: provider icon
:rtype: QIcon
"""
return QIcon(str(__icon_path__))
def tr(self, message: str) -> str:
"""Get the translation for a string using Qt translation API.
:param message: String for translation.
:type message: str, QString
:returns: Translated version of message.
:rtype: str
"""
# noinspection PyTypeChecker,PyArgumentList,PyCallByClass
return QCoreApplication.translate(self.__class__.__name__, message)
def versionInfo(self) -> str:
"""Version information for the provider, or an empty string if this is not \
applicable (e.g. for inbuilt Processing providers). For plugin based providers, \
this should return the plugin’s version identifier.
:return: version
:rtype: str
"""
return __version__
gn_tools/processing/status.py
+469
View File
@@ -0,0 +1,469 @@
#!/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)
gn_tools/processing/widget.py
+441
View File
@@ -0,0 +1,441 @@
# Standard
import json
# PyQt
from qgis.core import (
QgsAuthMethodConfig,
QgsDataSourceUri,QgsVectorLayer,
QgsJsonUtils,
QgsProject,
QgsCoordinateReferenceSystem,
QgsCoordinateTransform
)
from qgis.PyQt.QtCore import (
QByteArray,
QJsonDocument,
QObject,
QUrl,
pyqtSignal,
QTextCodec,
Qt
)
from qgis.PyQt.QtNetwork import QNetworkReply, QNetworkRequest
from qgis.PyQt.QtWidgets import QAction, QMessageBox, QDialog, QApplication
# from PyQt5.QtCore import Qt
# from PyQt5.QtWidgets import QApplication
# from PyQt6.QtCore import Qt
# from PyQt6.QtWidgets import QApplication
# Project
from gn_tools.__about__ import (
__api_sn_web__,
__api_sn_exp__,
__api_lst_ex__,
__exept_export__,
)
# from gn_tools.processing import ImportDataGbif
from .gbif import ImportDataGbif
dict_exp = []
class ParamWidget(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
):
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.zhs = None
# self.json_filter = {}
# self.total_zh = 0
self._pending_downloads = 0
if self.src_name == 'GBIF':
self.dlg.source_import.addItems(['Synthèse GBIF'])
self.dlg.source_import.currentTextChanged.connect(
self.print_change
)
else :
QApplication.setOverrideCursor(Qt.WaitCursor)
self.cookie = (network_manager.cookieJar()
.cookiesForUrl(QUrl(self.domain_url))[0]
)
self.send_tab_import()
@property
def pending_downloads(self):
return self._pending_downloads
def send_tab_import(self):
if self.src_name == 'GBIF':
self.dlg.source_import.addItems(['Synthèse GBIF'])
self.dlg.source_import.currentTextChanged.connect(
self.print_change
)
else :
self.dlg.source_import.clear()
request = self.get_api(api=__api_lst_ex__)
reply = self.network_manager.get(request)
reply.finished.connect(lambda: self.import_finished(reply))
# reply_exp
# dict_exp = {i for i in exp_all if exp_all[i].geometry_srid and 'sinp' not in exp_all[i].label.lower}
# lst_exp = [i.label for i in dict_exp]
# # self.dlg.source_import.setText(self.tr("Choose source data"))
# self.dlg.source_import.addItems(lst_exp)
# self.dlg.source_import.currentTextChanged.connect(
# self.print_change
# )
def import_finished(self, reply):
self._pending_downloads -= 1
QApplication.restoreOverrideCursor()
print(reply.attribute(QNetworkRequest.HttpStatusCodeAttribute))
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200:
print(f"code: {reply.error()} message: {reply.errorString()}")
else:
self.lst_exp = ['Synthèse taxons']
if self.src_name == 'GéoNature':
data_request = reply.readAll().data().decode()
data = json.loads(data_request)
items = data['items']
self.dlg.dict_exp = [i for i in items if i['geometry_srid'] ]
if len(__exept_export__) > 0:
for j in __exept_export__ :
self.dlg.dict_exp = [i for i in self.dlg.dict_exp if j not in i['label'].lower()]
self.lst_exp += [i['label'] for i in self.dlg.dict_exp]
self.dlg.source_import.addItems(self.lst_exp)
self.dlg.source_import.currentTextChanged.connect(
self.print_change
)
if self.pending_downloads == 0:
self.finished_dl.emit()
def print_change(self):
print(self.dlg.source_import.currentIndex())
def get_api(self,api=None,apiFilter=None):
self._pending_downloads += 1
# Function launched two times, one to check zh number the other to dl them
url_api = self.domain_url
if api:
url_api += str(api)
url = QUrl(url_api)
print(url)
# Define request properties (cookie, header)
request = QNetworkRequest(url)
request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
request.setHeader(QNetworkRequest.CookieHeader, self.cookie)
if apiFilter:
doc = QJsonDocument(self.json_filter)
else:
doc = QJsonDocument({})
return request
# data_request = reply.readAll().data().decode()
# print('tom')
# print(data_request)
# return json.loads(data_request)
# print(data)
class LoadingImport(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
):
super().__init__()
print(parent.dlg.tab)
self.dlg = parent.dlg
self.syn_params = parent.syn_params
self.source = self.dlg.source_import.currentText()
self.src_name = parent.src_name
self.domain_url = parent.domain_url
self.network_manager = network_manager
self.cookie = (network_manager.cookieJar()
.cookiesForUrl(QUrl(self.domain_url))[0]
) if self.src_name != 'GBIF' else None
self.rectangle = parent.rectangle
self.syn_params = parent.syn_params
self.project = parent.project
# self.zhs = None
# self.json_filter = {}
# self.total_zh = 0
QApplication.setOverrideCursor(Qt.WaitCursor)
self._pending_downloads = 0
self.run()
@property
def pending_downloads(self):
return self._pending_downloads
def run(self,items=None):
self._pending_downloads += 1
self.api = None
if self.source == 'Synthèse taxons':
url_api = self.domain_url + __api_sn_web__
self.transform_bouds(4326)
url_api += '?limit=800000'
if self.src_name == 'GéoNature':
geo_param = {
"geoIntersection": {
"type": "Feature",
"properties": {},
"geometry": json.loads(self.rectangle.asJson())
}
} if self.rectangle else {}
elif self.src_name == "Biodiv'AuRA":
geo_param = {
"geoIntersection": self.rectangle.asWkt(),
"with_areas":False
} if self.rectangle else {}
json_param = {
"modif_since_validation": False,
**geo_param,
**self.syn_params
}
doc = QJsonDocument(
json_param
)
print(doc.toJson())
elif self.source == 'Synthèse taxons export':
url_api = self.domain_url + __api_sn_exp__
url_api += '?export_format=geojson'
if self.src_name == 'GéoNature':
lst_id_synth = [
i['properties']['id_synthese']
for i in items['features']
]
elif self.src_name == "Biodiv'AuRA":
lst_id_synth = [
j['id']
for i in items['features']
for j in i['properties']['observations']
]
doc = QJsonDocument(lst_id_synth)
elif self.source == 'Synthèse GBIF':
ImportDataGbif(
self.network_manager, self.project, self.rectangle
)
# self.close()
return QApplication.restoreOverrideCursor()
# self.cookie = None
# url_api = self.domain_url
# url_api += '?advanced=false&geometry={polygon}&offset={offset}'.format(
# offset=1, polygon=self.rectangle.asWkt())
elif self.source in [i['label'] for i in self.dlg.dict_exp]:
self.api = [i for i in self.dlg.dict_exp if i['label'] == self.source][0]
url_api = self.domain_url + 'api/exports/api/%i'%self.api['id']
self.transform_bouds(self.api['geometry_srid'])
url_api += '?limit=0'
if self.rectangle :
url_api += "&geometry=" + str(self.rectangle.asWkt())
url = QUrl(url_api)
print(url)
# print('crs rect : ',self.rectangle.sourceCrs())
# Define request properties (cookie, header)
request = QNetworkRequest(url)
request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
if self.cookie:
request.setHeader(QNetworkRequest.CookieHeader, self.cookie)
if 'Synthèse taxons' in self.source :
reply = self.network_manager.post(request,doc.toJson())
else:
reply = self.network_manager.get(request)
reply.finished.connect(lambda: self.loading_finished(reply))
def loading_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)
items = data['items'] if 'items' in data.keys() else data
if self.source == 'Synthèse taxons':
self.source = 'Synthèse taxons export'
self.run(items)
return
items_point,items_poly,items_line = self.filter_gn_geom(items)
if items_point['features']:
self.addMapLayer(items_point,items_point['geom_type'])
if items_poly['features']:
self.addMapLayer(items_poly,items_poly['geom_type'])
if items_line['features']:
self.addMapLayer(items_line,items_line['geom_type'])
if self.pending_downloads == 0:
self.finished_dl.emit()
QApplication.restoreOverrideCursor()
print("END")
def filter_gn_geom(self,items):
nb_items = len(items['features'])
print('total_count : ',nb_items)
items_poly = {"features":[],"type": "FeatureCollection","geom_type": "MultiPolygon"}
items_line = {"features":[],"type": "FeatureCollection","geom_type": "MultiLine"}
items_point = {
"features":[i for i in items['features'] if 'point' in i['geometry']['type'].lower()],
"type": "FeatureCollection",
"geom_type": "Point"
}
len_data = len(items_point['features'])
if nb_items> len_data:
items_poly = {
"features":[i for i in items['features'] if 'polygon' in i['geometry']['type'].lower()],
"type": "FeatureCollection",
"geom_type": "MultiPolygon"
}
len_data += len(items_poly['features'])
if nb_items> len_data:
items_line = {
"features":[i for i in items['features'] if 'line' in i['geometry']['type'].lower()],
"type": "FeatureCollection",
"geom_type": "MultiLineString"
}
len_data += len(items_line['features'])
return items_point,items_poly,items_line
# def filter_biodiv_geom(self,items):
# items_point = {
# "features":[ {
# 'geometry' : i['geometry'],
# 'properties': j
# }
# for i in items['features'] if 'point' in i['geometry']['type'].lower()
# for j in i['properties']['observations']
# ],
# "type": "FeatureCollection",
# "geom_type": "Point"
# }
# items_poly = {
# "features":[ {
# 'geometry' : i['geometry'],
# 'properties': j
# }
# for i in items['features'] if 'polygon' in i['geometry']['type'].lower()
# for j in i['properties']['observations']
# ],
# "type": "FeatureCollection",
# "geom_type": "MultiPolygon"
# }
# items_line = {
# "features":[ {
# 'geometry' : i['geometry'],
# 'properties': j
# }
# for i in items['features'] if 'line' in i['geometry']['type'].lower()
# for j in i['properties']['observations']
# ],
# "type": "FeatureCollection",
# "geom_type": "MultiLineString"
# }
# return items_point,items_poly,items_line
def addMapLayer(self,data,geom_type):
mon_layer = ' - '.join([self.src_name[:6], self.source])
fcString = json.dumps(data)
codec = QTextCodec.codecForName("UTF-8")
fields = QgsJsonUtils.stringToFields(fcString, codec)
feats = QgsJsonUtils.stringToFeatureList(fcString, fields, codec)
if self.api:
crs_api = QgsCoordinateReferenceSystem(self.api['geometry_srid'])
else :
crs_api = QgsCoordinateReferenceSystem(4326)
vl = QgsVectorLayer(geom_type, mon_layer, "memory")
if vl.sourceCrs() != crs_api:
print('Changed crs !')
# crs = vl.crs()
# crs.createFromId(self.api['geometry_srid'])
vl.setCrs(crs_api)
dp = vl.dataProvider()
dp.addAttributes(fields)
vl.updateFields()
dp.addFeatures(feats)
vl.updateExtents()
if dp.errors():
msg = QMessageBox()
msg.warning(
None,
self.tr("Warning"),
self.tr(
"Houston, we've got a problem ... \n\n %s !"%str(dp.errors())
), # noqa: E501
)
QgsProject.instance().addMapLayer(vl)
def transform_bouds(self,crs):
sourceCrs = QgsCoordinateReferenceSystem(4326)
destCrs = QgsCoordinateReferenceSystem(crs)
if sourceCrs != destCrs:
tr = QgsCoordinateTransform(sourceCrs, destCrs, self.project)
self.rectangle.transform(tr)