############################################################################### # # CSW Client # --------------------------------------------------------- # QGIS Catalog Service client. # # Copyright (C) 2010 NextGIS (http://nextgis.org), # Alexander Bruy (alexander.bruy@gmail.com), # Maxim Dubinin (sim@gis-lab.info) # # Copyright (C) 2017 Tom Kralidis (tomkralidis@gmail.com) # # This source is free software; you can redistribute it and/or modify it under # the terms of the GNU General Public License as published by the Free # Software Foundation; either version 2 of the License, or (at your option) # any later version. # # This code is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU General Public License for more # details. # # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. # ############################################################################### import json import os.path from urllib.request import build_opener, install_opener, ProxyHandler from qgis.PyQt.QtCore import Qt from qgis.PyQt.QtWidgets import ( QDialog, QComboBox, QDialogButtonBox, QMessageBox, QTreeWidgetItem, QWidget, ) from qgis.PyQt.QtGui import QColor from qgis.core import ( Qgis, QgsApplication, QgsCoordinateReferenceSystem, QgsCoordinateTransform, QgsGeometry, QgsPointXY, QgsProviderRegistry, QgsSettings, QgsProject, QgsRectangle, QgsSettingsTree, ) from qgis.gui import QgsRubberBand, QgsGui from qgis.utils import OverrideCursor try: from owslib.util import Authentication except ImportError: pass from MetaSearch import link_types from MetaSearch.dialogs.manageconnectionsdialog import ManageConnectionsDialog from MetaSearch.dialogs.newconnectiondialog import NewConnectionDialog from MetaSearch.dialogs.recorddialog import RecordDialog from MetaSearch.dialogs.apidialog import APIRequestResponseDialog from MetaSearch.search_backend import get_catalog_service from MetaSearch.util import ( clean_ows_url, get_connections_from_file, get_ui_class, get_help_url, normalize_text, open_url, render_template, serialize_string, StaticContext, ) BASE_CLASS = get_ui_class("maindialog.ui") class MetaSearchDialog(QDialog, BASE_CLASS): """main dialogue""" def __init__(self, iface): """init window""" QDialog.__init__(self) self.setupUi(self) self.iface = iface self.map = iface.mapCanvas() self.settings = QgsSettings() self.catalog = None self.catalog_url = None self.catalog_username = None self.catalog_password = None self.catalog_type = None self.context = StaticContext() self.leKeywords.setShowSearchIcon(True) self.leKeywords.setPlaceholderText(self.tr("Search keywords")) self.setWindowTitle(self.tr("MetaSearch")) self.rubber_band = QgsRubberBand(self.map, Qgis.GeometryType.Polygon) self.rubber_band.setColor(QColor(255, 0, 0, 75)) self.rubber_band.setWidth(5) # form inputs self.startfrom = 1 self.constraints = [] self.maxrecords = int(self.settings.value("/MetaSearch/returnRecords", 10)) self.timeout = int(self.settings.value("/MetaSearch/timeout", 10)) self.disable_ssl_verification = self.settings.value( "/MetaSearch/disableSSL", False, bool ) self.log_debugging_messages = self.settings.value( "/MetaSearch/logDebugging", False, bool ) # Services tab self.cmbConnectionsServices.activated.connect(self.save_connection) self.cmbConnectionsSearch.activated.connect(self.save_connection) self.btnServerInfo.clicked.connect(self.connection_info) self.btnAddDefault.clicked.connect(self.add_default_connections) self.btnRawAPIResponse.clicked.connect(self.show_api) self.tabWidget.currentChanged.connect(self.populate_connection_list) # server management buttons self.btnNew.clicked.connect(self.add_connection) self.btnEdit.clicked.connect(self.edit_connection) self.btnDelete.clicked.connect(self.delete_connection) self.btnLoad.clicked.connect(self.load_connections) self.btnSave.clicked.connect(save_connections) # Search tab self.treeRecords.itemSelectionChanged.connect(self.record_clicked) self.treeRecords.itemDoubleClicked.connect(self.show_metadata) self.btnSearch.clicked.connect(self.search) self.leKeywords.returnPressed.connect(self.search) # prevent dialog from closing upon pressing enter self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setAutoDefault( False ) # launch help from button self.buttonBox.helpRequested.connect(self.help) self.btnCanvasBbox.setAutoDefault(False) self.btnCanvasBbox.clicked.connect(self.set_bbox_from_map) self.btnGlobalBbox.clicked.connect(self.set_bbox_global) # navigation buttons self.btnFirst.clicked.connect(self.navigate) self.btnPrev.clicked.connect(self.navigate) self.btnNext.clicked.connect(self.navigate) self.btnLast.clicked.connect(self.navigate) self.mActionAddWms.triggered.connect(self.add_to_ows) self.mActionAddWfs.triggered.connect(self.add_to_ows) self.mActionAddWcs.triggered.connect(self.add_to_ows) self.mActionAddAms.triggered.connect(self.add_to_ows) self.mActionAddAfs.triggered.connect(self.add_to_ows) self.mActionAddGisFile.triggered.connect(self.add_gis_file) self.btnViewRawAPIResponse.clicked.connect(self.show_api) self.manageGui() def manageGui(self): """open window""" def _on_timeout_change(value): self.settings.setValue("/MetaSearch/timeout", value) self.timeout = value def _on_records_change(value): self.settings.setValue("/MetaSearch/returnRecords", value) self.maxrecords = value def _on_ssl_state_change(state): self.settings.setValue("/MetaSearch/disableSSL", bool(state)) self.disable_ssl_verification = bool(state) def _on_debugging_state_change(state): self.settings.setValue("/MetaSearch/logDebugging", bool(state)) self.log_debugging_messages = bool(state) self.tabWidget.setCurrentIndex(0) self.populate_connection_list() self.btnRawAPIResponse.setEnabled(False) # load settings self.spnRecords.setValue(self.maxrecords) self.spnRecords.valueChanged.connect(_on_records_change) self.spnTimeout.setValue(self.timeout) self.spnTimeout.valueChanged.connect(_on_timeout_change) self.disableSSLVerification.setChecked(self.disable_ssl_verification) self.disableSSLVerification.stateChanged.connect(_on_ssl_state_change) self.logDebuggingMessages.setChecked(self.log_debugging_messages) self.logDebuggingMessages.stateChanged.connect(_on_debugging_state_change) key = "/MetaSearch/%s" % self.cmbConnectionsSearch.currentText() self.catalog_url = self.settings.value("%s/url" % key) self.catalog_username = self.settings.value("%s/username" % key) self.catalog_password = self.settings.value("%s/password" % key) self.catalog_type = self.settings.value("%s/catalog-type" % key) self.set_bbox_global() self.reset_buttons() # install proxy handler if specified in QGIS settings self.install_proxy() # Services tab def populate_connection_list(self): """populate select box with connections""" self.settings.beginGroup("/MetaSearch/") self.cmbConnectionsServices.clear() self.cmbConnectionsServices.addItems(self.settings.childGroups()) self.cmbConnectionsSearch.clear() self.cmbConnectionsSearch.addItems(self.settings.childGroups()) self.settings.endGroup() self.set_connection_list_position() if self.cmbConnectionsServices.count() == 0: # no connections - disable various buttons state_disabled = False self.btnSave.setEnabled(state_disabled) # and start with connection tab open self.tabWidget.setCurrentIndex(1) # tell the user to add services msg = self.tr( "No services/connections defined. To get " "started with MetaSearch, create a new " "connection by clicking 'New' or click " "'Add default services'." ) self.textMetadata.setHtml("

%s

" % msg) else: # connections - enable various buttons state_disabled = True self.btnServerInfo.setEnabled(state_disabled) self.btnEdit.setEnabled(state_disabled) self.btnDelete.setEnabled(state_disabled) def set_connection_list_position(self): """set the current index to the selected connection""" to_select = self.settings.value("/MetaSearch/selected") conn_count = self.cmbConnectionsServices.count() if conn_count == 0: self.btnDelete.setEnabled(False) self.btnServerInfo.setEnabled(False) self.btnEdit.setEnabled(False) # does to_select exist in cmbConnectionsServices? exists = False for i in range(conn_count): if self.cmbConnectionsServices.itemText(i) == to_select: self.cmbConnectionsServices.setCurrentIndex(i) self.cmbConnectionsSearch.setCurrentIndex(i) exists = True break # If we couldn't find the stored item, but there are some, default # to the last item (this makes some sense when deleting items as it # allows the user to repeatidly click on delete to remove a whole # lot of items) if not exists and conn_count > 0: # If to_select is null, then the selected connection wasn't found # by QgsSettings, which probably means that this is the first time # the user has used CSWClient, so default to the first in the list # of connetions. Otherwise default to the last. if not to_select: current_index = 0 else: current_index = conn_count - 1 self.cmbConnectionsServices.setCurrentIndex(current_index) self.cmbConnectionsSearch.setCurrentIndex(current_index) def save_connection(self): """save connection""" caller = self.sender().objectName() if caller == "cmbConnectionsServices": # servers tab current_text = self.cmbConnectionsServices.currentText() elif caller == "cmbConnectionsSearch": # search tab current_text = self.cmbConnectionsSearch.currentText() self.settings.setValue("/MetaSearch/selected", current_text) key = "/MetaSearch/%s" % current_text if caller == "cmbConnectionsSearch": # bind to service in search tab self.catalog_url = self.settings.value("%s/url" % key) self.catalog_username = self.settings.value("%s/username" % key) self.catalog_password = self.settings.value("%s/password" % key) self.catalog_type = self.settings.value("%s/catalog-type" % key) if caller == "cmbConnectionsServices": # clear server metadata self.textMetadata.clear() self.btnRawAPIResponse.setEnabled(False) def connection_info(self): """show connection info""" current_text = self.cmbConnectionsServices.currentText() key = "/MetaSearch/%s" % current_text self.catalog_url = self.settings.value("%s/url" % key) self.catalog_username = self.settings.value("%s/username" % key) self.catalog_password = self.settings.value("%s/password" % key) self.catalog_type = self.settings.value("%s/catalog-type" % key) # connect to the server if not self._get_catalog(): return if self.catalog: # display service metadata self.btnRawAPIResponse.setEnabled(True) metadata = render_template( "en", self.context, self.catalog.conn, self.catalog.service_info_template, ) style = QgsApplication.reportStyleSheet() self.textMetadata.clear() self.textMetadata.document().setDefaultStyleSheet(style) self.textMetadata.setHtml(metadata) # clear results and disable buttons in Search tab self.clear_results() def add_connection(self): """add new service""" conn_new = NewConnectionDialog() conn_new.setWindowTitle(self.tr("New Catalog Service")) if conn_new.exec() == QDialog.DialogCode.Accepted: # add to service list self.populate_connection_list() self.textMetadata.clear() def edit_connection(self): """modify existing connection""" current_text = self.cmbConnectionsServices.currentText() url = self.settings.value("/MetaSearch/%s/url" % current_text) conn_edit = NewConnectionDialog(current_text) conn_edit.setWindowTitle(self.tr("Edit Catalog Service")) conn_edit.leName.setText(current_text) conn_edit.leURL.setText(url) conn_edit.leUsername.setText( self.settings.value("/MetaSearch/%s/username" % current_text) ) conn_edit.lePassword.setText( self.settings.value("/MetaSearch/%s/password" % current_text) ) conn_edit.cmbCatalogType.setCurrentText( self.settings.value("/MetaSearch/%s/catalog-type" % current_text) ) if conn_edit.exec() == QDialog.DialogCode.Accepted: # update service list self.populate_connection_list() def delete_connection(self): """delete connection""" current_text = self.cmbConnectionsServices.currentText() key = "/MetaSearch/%s" % current_text msg = self.tr("Remove service {0}?").format(current_text) result = QMessageBox.question( self, self.tr("Delete Service"), msg, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No, ) if result == QMessageBox.StandardButton.Yes: # remove service from list self.settings.remove(key) index_to_delete = self.cmbConnectionsServices.currentIndex() self.cmbConnectionsServices.removeItem(index_to_delete) self.cmbConnectionsSearch.removeItem(index_to_delete) self.set_connection_list_position() def load_connections(self): """load services from list""" ManageConnectionsDialog(1).exec() self.populate_connection_list() def add_default_connections(self): """add default connections""" filename = os.path.join( self.context.ppath, "resources", "connections-default.xml" ) doc = get_connections_from_file(self, filename) if doc is None: return self.settings.beginGroup("/MetaSearch/") keys = self.settings.childGroups() self.settings.endGroup() for server in doc.findall("csw"): name = server.attrib.get("name") # check for duplicates if name in keys: msg = self.tr("{0} exists. Overwrite?").format(name) res = QMessageBox.warning( self, self.tr("Loading connections"), msg, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, ) if res != QMessageBox.StandardButton.Yes: continue # no dups detected or overwrite is allowed key = "/MetaSearch/%s" % name self.settings.setValue("%s/url" % key, server.attrib.get("url")) self.settings.setValue( "%s/catalog-type" % key, server.attrib.get("catalog-type", "OGC CSW 2.0.2"), ) self.populate_connection_list() # Settings tab def set_ows_save_title_ask(self): """save ows save strategy as save ows title, ask if duplicate""" self.settings.setValue("/MetaSearch/ows_save_strategy", "title_ask") def set_ows_save_title_no_ask(self): """save ows save strategy as save ows title, do NOT ask if duplicate""" self.settings.setValue("/MetaSearch/ows_save_strategy", "title_no_ask") def set_ows_save_temp_name(self): """save ows save strategy as save with a temporary name""" self.settings.setValue("/MetaSearch/ows_save_strategy", "temp_name") # Search tab def set_bbox_from_map(self): """set bounding box from map extent""" crs = self.map.mapSettings().destinationCrs() crsid = crs.authid() extent = self.map.extent() if crsid != "EPSG:4326": # reproject to EPSG:4326 src = QgsCoordinateReferenceSystem(crsid) dest = QgsCoordinateReferenceSystem("EPSG:4326") xform = QgsCoordinateTransform(src, dest, QgsProject.instance()) minxy = xform.transform(QgsPointXY(extent.xMinimum(), extent.yMinimum())) maxxy = xform.transform(QgsPointXY(extent.xMaximum(), extent.yMaximum())) minx, miny = minxy maxx, maxy = maxxy else: # EPSG:4326 minx = extent.xMinimum() miny = extent.yMinimum() maxx = extent.xMaximum() maxy = extent.yMaximum() self.leNorth.setText(str(maxy)[0:9]) self.leSouth.setText(str(miny)[0:9]) self.leWest.setText(str(minx)[0:9]) self.leEast.setText(str(maxx)[0:9]) def set_bbox_global(self): """set global bounding box""" self.leNorth.setText("90") self.leSouth.setText("-90") self.leWest.setText("-180") self.leEast.setText("180") def search(self): """execute search""" self.catalog = None self.constraints = [] # clear all fields and disable buttons self.clear_results() # set current catalog current_text = self.cmbConnectionsSearch.currentText() key = "/MetaSearch/%s" % current_text self.catalog_url = self.settings.value("%s/url" % key) self.catalog_username = self.settings.value("%s/username" % key) self.catalog_password = self.settings.value("%s/password" % key) self.catalog_type = self.settings.value("%s/catalog-type" % key) # start position and number of records to return self.startfrom = 1 # bbox # CRS is WGS84 with axis order longitude, latitude # defined by 'urn:ogc:def:crs:OGC:1.3:CRS84' minx = self.leWest.text() miny = self.leSouth.text() maxx = self.leEast.text() maxy = self.leNorth.text() bbox = [minx, miny, maxx, maxy] keywords = self.leKeywords.text() # build request if not self._get_catalog(): return # TODO: allow users to select resources types # to find ('service', 'dataset', etc.) try: with OverrideCursor(Qt.CursorShape.WaitCursor): self.catalog.query_records( bbox, keywords, self.maxrecords, self.startfrom ) except Exception as err: QMessageBox.warning( self, self.tr("Search error"), self.tr("Search error: {0}").format(err) ) return if self.catalog.matches == 0: self.lblResults.setText(self.tr("0 results")) return self.display_results() def display_results(self): """display search results""" self.treeRecords.clear() position = self.catalog.returned + self.startfrom - 1 msg = self.tr( "Showing {0} - {1} of %n result(s)", "number of results", self.catalog.matches, ).format(self.startfrom, position) self.lblResults.setText(msg) for rec in self.catalog.records(): item = QTreeWidgetItem(self.treeRecords) if rec["type"]: item.setText(0, normalize_text(rec["type"])) else: item.setText(0, "unknown") if rec["title"]: item.setText(1, normalize_text(rec["title"])) if rec["identifier"]: set_item_data(item, "identifier", rec["identifier"]) self.btnViewRawAPIResponse.setEnabled(True) if self.catalog.matches < self.maxrecords: disabled = False else: disabled = True self.btnFirst.setEnabled(disabled) self.btnPrev.setEnabled(disabled) self.btnNext.setEnabled(disabled) self.btnLast.setEnabled(disabled) self.btnRawAPIResponse.setEnabled(False) def clear_results(self): """clear search results""" self.lblResults.clear() self.treeRecords.clear() self.reset_buttons() def record_clicked(self): """record clicked signal""" # disable only service buttons self.reset_buttons(True, False, False) self.rubber_band.reset() if not self.treeRecords.selectedItems(): return item = self.treeRecords.currentItem() if not item: return identifier = get_item_data(item, "identifier") try: record = next( item for item in self.catalog.records() if item["identifier"] == identifier ) except KeyError: QMessageBox.warning( self, self.tr("Record parsing error"), "Unable to locate record identifier", ) return # if the record has a bbox, show a footprint on the map if record["bbox"] is not None: bx = record["bbox"] rt = QgsRectangle( float(bx["minx"]), float(bx["miny"]), float(bx["maxx"]), float(bx["maxy"]), ) geom = QgsGeometry.fromRect(rt) if geom is not None: src = QgsCoordinateReferenceSystem("EPSG:4326") dst = self.map.mapSettings().destinationCrs() if src.postgisSrid() != dst.postgisSrid(): ctr = QgsCoordinateTransform(src, dst, QgsProject.instance()) try: geom.transform(ctr) except Exception as err: QMessageBox.warning( self, self.tr("Coordinate Transformation Error"), str(err) ) self.rubber_band.setToGeometry(geom, None) # figure out if the data is interactive and can be operated on self.find_services(record, item) def find_services(self, record, item): """scan record for WMS/WMTS|WFS|WCS endpoints""" services = {} for link in record["links"]: link = self.catalog.parse_link(link) if "scheme" in link: link_type = link["scheme"] elif "protocol" in link: link_type = link["protocol"] else: link_type = None if link_type is not None: link_type = link_type.upper() wmswmst_link_types = list(map(str.upper, link_types.WMSWMST_LINK_TYPES)) wfs_link_types = list(map(str.upper, link_types.WFS_LINK_TYPES)) wcs_link_types = list(map(str.upper, link_types.WCS_LINK_TYPES)) ams_link_types = list(map(str.upper, link_types.AMS_LINK_TYPES)) afs_link_types = list(map(str.upper, link_types.AFS_LINK_TYPES)) gis_file_link_types = list(map(str.upper, link_types.GIS_FILE_LINK_TYPES)) # if the link type exists, and it is one of the acceptable # interactive link types, then set all_link_types = ( wmswmst_link_types + wfs_link_types + wcs_link_types + ams_link_types + afs_link_types + gis_file_link_types ) if all([link_type is not None, link_type in all_link_types]): if link_type in wmswmst_link_types: services["wms"] = link["url"] self.mActionAddWms.setEnabled(True) if link_type in wfs_link_types: services["wfs"] = link["url"] self.mActionAddWfs.setEnabled(True) if link_type in wcs_link_types: services["wcs"] = link["url"] self.mActionAddWcs.setEnabled(True) if link_type in ams_link_types: services["ams"] = link["url"] self.mActionAddAms.setEnabled(True) if link_type in afs_link_types: services["afs"] = link["url"] self.mActionAddAfs.setEnabled(True) if link_type in gis_file_link_types: services["gis_file"] = link["url"] services["title"] = record.get("title", "") self.mActionAddGisFile.setEnabled(True) self.tbAddData.setEnabled(True) set_item_data(item, "link", json.dumps(services)) def navigate(self): """manage navigation / paging""" caller = self.sender().objectName() if caller == "btnFirst": self.startfrom = 1 elif caller == "btnLast": self.startfrom = self.catalog.matches - self.maxrecords + 1 elif caller == "btnNext": if self.startfrom > self.catalog.matches - self.maxrecords: msg = self.tr("End of results. Go to start?") res = QMessageBox.information( self, self.tr("Navigation"), msg, (QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel), ) if res == QMessageBox.StandardButton.Ok: self.startfrom = 1 else: return else: self.startfrom += self.maxrecords elif caller == "btnPrev": if self.startfrom == 1: msg = self.tr("Start of results. Go to end?") res = QMessageBox.information( self, self.tr("Navigation"), msg, (QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel), ) if res == QMessageBox.StandardButton.Ok: self.startfrom = self.catalog.matches - self.maxrecords + 1 else: return elif self.startfrom <= self.maxrecords: self.startfrom = 1 else: self.startfrom -= self.maxrecords # bbox # CRS is WGS84 with axis order longitude, latitude # defined by 'urn:ogc:def:crs:OGC:1.3:CRS84' minx = self.leWest.text() miny = self.leSouth.text() maxx = self.leEast.text() maxy = self.leNorth.text() bbox = [minx, miny, maxx, maxy] keywords = self.leKeywords.text() try: with OverrideCursor(Qt.CursorShape.WaitCursor): self.catalog.query_records( bbox, keywords, limit=self.maxrecords, offset=self.startfrom ) except Exception as err: QMessageBox.warning( self, self.tr("Search error"), self.tr("Search error: {0}").format(err) ) return self.display_results() def add_to_ows(self): """add to OWS provider connection list""" conn_name_matches = [] item = self.treeRecords.currentItem() if not item: return item_data = json.loads(get_item_data(item, "link")) caller = self.sender().objectName() if caller == "mActionAddWms": service_type = "OGC:WMS/OGC:WMTS" sname = "WMS" dyn_param = ["wms"] provider_name = "wms" setting_node = ( QgsSettingsTree.node("connections") .childNode("ows") .childNode("connections") ) data_url = item_data["wms"] elif caller == "mActionAddWfs": service_type = "OGC:WFS" sname = "WFS" dyn_param = ["wfs"] provider_name = "WFS" setting_node = ( QgsSettingsTree.node("connections") .childNode("ows") .childNode("connections") ) data_url = item_data["wfs"] elif caller == "mActionAddWcs": service_type = "OGC:WCS" sname = "WCS" dyn_param = ["wcs"] provider_name = "wcs" setting_node = ( QgsSettingsTree.node("connections") .childNode("ows") .childNode("connections") ) data_url = item_data["wcs"] elif caller == "mActionAddAfs": service_type = "ESRI:ArcGIS:FeatureServer" sname = "AFS" dyn_param = [] provider_name = "arcgisfeatureserver" setting_node = QgsSettingsTree.node("connections").childNode( "arcgisfeatureserver" ) data_url = item_data["afs"].split("FeatureServer")[0] + "FeatureServer" keys = setting_node.items(dyn_param) sname = "%s from MetaSearch" % sname for key in keys: if key.startswith(sname): conn_name_matches.append(key) if conn_name_matches: sname = conn_name_matches[-1] # check for duplicates if sname in keys: # duplicate found msg = self.tr("Connection {0} exists. Overwrite?").format(sname) res = QMessageBox.warning( self, self.tr("Saving server"), msg, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No | QMessageBox.StandardButton.Cancel, ) if res == QMessageBox.StandardButton.No: # assign new name with serial sname = serialize_string(sname) elif res == QMessageBox.StandardButton.Cancel: return # no dups detected or overwrite is allowed dyn_param.append(sname) setting_node.childSetting("url").setValue(clean_ows_url(data_url), dyn_param) # open provider window ows_provider = QgsGui.sourceSelectProviderRegistry().createSelectionWidget( provider_name, self, Qt.WindowType.Widget, QgsProviderRegistry.WidgetMode.Embedded, ) # connect dialog signals to iface slots if service_type == "OGC:WMS/OGC:WMTS": ows_provider.addRasterLayer.connect(self.iface.addRasterLayer) conn_cmb = ows_provider.findChild(QWidget, "cmbConnections") connect = "btnConnect_clicked" elif service_type == "OGC:WFS": def addVectorLayer(path, name): self.iface.addVectorLayer(path, name, "WFS") ows_provider.addVectorLayer.connect(addVectorLayer) conn_cmb = ows_provider.findChild(QWidget, "cmbConnections") connect = "connectToServer" elif service_type == "OGC:WCS": ows_provider.addRasterLayer.connect(self.iface.addRasterLayer) conn_cmb = ows_provider.findChild(QWidget, "mConnectionsComboBox") connect = "mConnectButton_clicked" elif service_type == "ESRI:ArcGIS:FeatureServer": def addAfsLayer(path, name): self.iface.addVectorLayer(path, name, "afs") ows_provider.addVectorLayer.connect(addAfsLayer) conn_cmb = ows_provider.findChild(QComboBox) connect = "connectToServer" ows_provider.setModal(False) ows_provider.show() # open provider dialogue against added OWS index = conn_cmb.findText(sname) if index > -1: conn_cmb.setCurrentIndex(index) # only for wfs if service_type == "OGC:WFS": ows_provider.cmbConnections_activated(index) elif service_type == "ESRI:ArcGIS:FeatureServer": ows_provider.cmbConnections_activated(index) getattr(ows_provider, connect)() def add_gis_file(self): """add GIS file from result""" item = self.treeRecords.currentItem() if not item: return item_data = json.loads(get_item_data(item, "link")) gis_file = item_data["gis_file"] title = item_data["title"] layer = self.iface.addVectorLayer(gis_file, title, "ogr") if not layer: self.iface.messageBar().pushWarning(None, "Layer failed to load!") def show_metadata(self): """show record metadata""" if not self.treeRecords.selectedItems(): return item = self.treeRecords.currentItem() if not item: return identifier = get_item_data(item, "identifier") auth = None if self.disable_ssl_verification: try: auth = Authentication(verify=False) except NameError: pass try: with OverrideCursor(Qt.CursorShape.WaitCursor): cat = get_catalog_service( self.catalog_url, # spellok catalog_type=self.catalog_type, timeout=self.timeout, username=self.catalog_username or None, password=self.catalog_password or None, auth=auth, ) record = cat.get_record(identifier) if cat.type == "OGC API - Records": record["url"] = cat.conn.request elif cat.type == "OGC CSW 2.0.2": record.url = cat.conn.request except Exception as err: QMessageBox.warning( self, self.tr("GetRecords error"), self.tr("Error getting response: {0}").format(err), ) return except KeyError as err: QMessageBox.warning( self, self.tr("Record parsing error"), self.tr("Unable to locate record identifier: {0}").format(err), ) return crd = RecordDialog() metadata = render_template( "en", self.context, record, self.catalog.record_info_template ) style = QgsApplication.reportStyleSheet() crd.textMetadata.document().setDefaultStyleSheet(style) crd.textMetadata.setHtml(metadata) crd.exec() def show_api(self): """show API request / response""" crd = APIRequestResponseDialog( self.catalog.request, self.catalog.response, self.catalog.format ) crd.exec() def reset_buttons(self, services=True, api=True, navigation=True): """Convenience function to disable WMS/WMTS|WFS|WCS buttons""" if services: self.tbAddData.setEnabled(False) self.mActionAddWms.setEnabled(False) self.mActionAddWfs.setEnabled(False) self.mActionAddWcs.setEnabled(False) self.mActionAddAms.setEnabled(False) self.mActionAddAfs.setEnabled(False) self.mActionAddGisFile.setEnabled(False) if api: self.btnViewRawAPIResponse.setEnabled(False) if navigation: self.btnFirst.setEnabled(False) self.btnPrev.setEnabled(False) self.btnNext.setEnabled(False) self.btnLast.setEnabled(False) def help(self): """launch help""" open_url(get_help_url()) def reject(self): """back out of dialogue""" QDialog.reject(self) self.rubber_band.reset() def _get_catalog(self): """convenience function to init catalog wrapper""" auth = None if self.disable_ssl_verification: try: auth = Authentication(verify=False) except NameError: pass # connect to the server with OverrideCursor(Qt.CursorShape.WaitCursor): try: self.catalog = get_catalog_service( self.catalog_url, catalog_type=self.catalog_type, timeout=self.timeout, username=self.catalog_username or None, password=self.catalog_password or None, auth=auth, ) return True except Exception as err: msg = self.tr("Error connecting to service: {0}").format(err) QMessageBox.warning(self, self.tr("CSW Connection error"), msg) return False def install_proxy(self): """set proxy if one is set in QGIS network settings""" # initially support HTTP for now if self.settings.value("/proxy/proxyEnabled") == "true": if self.settings.value("/proxy/proxyType") == "HttpProxy": ptype = "http" else: return user = self.settings.value("/proxy/proxyUser") password = self.settings.value("/proxy/proxyPassword") host = self.settings.value("/proxy/proxyHost") port = self.settings.value("/proxy/proxyPort") proxy_up = "" proxy_port = "" if all([user != "", password != ""]): proxy_up = f"{user}:{password}@" if port != "": proxy_port = ":%s" % port conn = f"{ptype}://{proxy_up}{host}{proxy_port}" install_opener(build_opener(ProxyHandler({ptype: conn}))) def save_connections(): """save servers to list""" ManageConnectionsDialog(0).exec() def get_item_data(item, field): """return identifier for a QTreeWidgetItem""" return item.data(_get_field_value(field), 32) def set_item_data(item, field, value): """set identifier for a QTreeWidgetItem""" item.setData(_get_field_value(field), 32, value) def _get_field_value(field): """convenience function to return field value integer""" value = 0 if field == "identifier": value = 0 if field == "link": value = 1 return value