mirror of
https://github.com/qgis/QGIS.git
synced 2025-10-25 00:05:24 -04:00
852 lines
27 KiB
Python
852 lines
27 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
***************************************************************************
|
|
utils.py
|
|
---------------------
|
|
Date : November 2009
|
|
Copyright : (C) 2009 by Martin Dobias
|
|
Email : wonder dot sk at gmail dot com
|
|
***************************************************************************
|
|
* *
|
|
* This program 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. *
|
|
* *
|
|
***************************************************************************
|
|
"""
|
|
|
|
__author__ = 'Martin Dobias'
|
|
__date__ = 'November 2009'
|
|
__copyright__ = '(C) 2009, Martin Dobias'
|
|
|
|
"""
|
|
QGIS utilities module
|
|
|
|
"""
|
|
|
|
from qgis.PyQt.QtCore import QCoreApplication, QLocale, QThread, qDebug, QUrl
|
|
from qgis.PyQt.QtGui import QDesktopServices
|
|
from qgis.PyQt.QtWidgets import QPushButton, QApplication
|
|
from qgis.core import Qgis, QgsMessageLog, qgsfunction, QgsMessageOutput
|
|
from qgis.gui import QgsMessageBar
|
|
|
|
import os
|
|
import sys
|
|
import traceback
|
|
import glob
|
|
import os.path
|
|
import configparser
|
|
import warnings
|
|
import codecs
|
|
import time
|
|
import functools
|
|
|
|
import builtins
|
|
builtins.__dict__['unicode'] = str
|
|
builtins.__dict__['basestring'] = str
|
|
builtins.__dict__['long'] = int
|
|
builtins.__dict__['Set'] = set
|
|
|
|
# ######################
|
|
# ERROR HANDLING
|
|
|
|
warnings.simplefilter('default')
|
|
warnings.filterwarnings("ignore", "the sets module is deprecated")
|
|
|
|
|
|
def showWarning(message, category, filename, lineno, file=None, line=None):
|
|
stk = ""
|
|
for s in traceback.format_stack()[:-2]:
|
|
if hasattr(s, 'decode'):
|
|
stk += s.decode(sys.getfilesystemencoding())
|
|
else:
|
|
stk += s
|
|
if hasattr(filename, 'decode'):
|
|
decoded_filename = filename.decode(sys.getfilesystemencoding())
|
|
else:
|
|
decoded_filename = filename
|
|
QgsMessageLog.logMessage(
|
|
u"warning:{}\ntraceback:{}".format(warnings.formatwarning(message, category, decoded_filename, lineno), stk),
|
|
QCoreApplication.translate("Python", "Python warning")
|
|
)
|
|
|
|
|
|
def showException(type, value, tb, msg, messagebar=False, level=Qgis.Warning):
|
|
if msg is None:
|
|
msg = QCoreApplication.translate('Python', 'An error has occurred while executing Python code:')
|
|
|
|
logmessage = ''
|
|
for s in traceback.format_exception(type, value, tb):
|
|
logmessage += s.decode('utf-8', 'replace') if hasattr(s, 'decode') else s
|
|
|
|
title = QCoreApplication.translate('Python', 'Python error')
|
|
QgsMessageLog.logMessage(logmessage, title, level)
|
|
|
|
try:
|
|
blockingdialog = QApplication.instance().activeModalWidget()
|
|
window = QApplication.instance().activeWindow()
|
|
except:
|
|
blockingdialog = QApplication.activeModalWidget()
|
|
window = QApplication.activeWindow()
|
|
|
|
# Still show the normal blocking dialog in this case for now.
|
|
if blockingdialog or not window or not messagebar or not iface:
|
|
open_stack_dialog(type, value, tb, msg)
|
|
return
|
|
|
|
bar = iface.messageBar() if iface else None
|
|
|
|
# If it's not the main window see if we can find a message bar to report the error in
|
|
if not window.objectName() == "QgisApp":
|
|
widgets = window.findChildren(QgsMessageBar)
|
|
if widgets:
|
|
# Grab the first message bar for now
|
|
bar = widgets[0]
|
|
|
|
item = bar.currentItem()
|
|
if item and item.property("Error") == msg:
|
|
# Return of we already have a message with the same error message
|
|
return
|
|
|
|
widget = bar.createMessage(title, msg + " " + QCoreApplication.translate("Python", "See message log (Python Error) for more details."))
|
|
widget.setProperty("Error", msg)
|
|
stackbutton = QPushButton(QCoreApplication.translate("Python", "Stack trace"), pressed=functools.partial(open_stack_dialog, type, value, tb, msg))
|
|
button = QPushButton(QCoreApplication.translate("Python", "View message log"), pressed=show_message_log)
|
|
widget.layout().addWidget(stackbutton)
|
|
widget.layout().addWidget(button)
|
|
bar.pushWidget(widget, Qgis.Warning)
|
|
|
|
|
|
def show_message_log(pop_error=True):
|
|
if pop_error:
|
|
iface.messageBar().popWidget()
|
|
|
|
iface.openMessageLog()
|
|
|
|
|
|
def open_stack_dialog(type, value, tb, msg, pop_error=True):
|
|
if pop_error and iface is not None:
|
|
iface.messageBar().popWidget()
|
|
|
|
if msg is None:
|
|
msg = QCoreApplication.translate('Python', 'An error has occurred while executing Python code:')
|
|
|
|
# TODO Move this to a template HTML file
|
|
txt = u'''<font color="red"><b>{msg}</b></font>
|
|
<br>
|
|
<h3>{main_error}</h3>
|
|
<pre>
|
|
{error}
|
|
</pre>
|
|
<br>
|
|
<b>{version_label}</b> {num}
|
|
<br>
|
|
<b>{qgis_label}</b> {qversion} {qgisrelease}, {devversion}
|
|
<br>
|
|
<h4>{pypath_label}</h4>
|
|
<ul>
|
|
{pypath}
|
|
</ul>'''
|
|
|
|
error = ''
|
|
lst = traceback.format_exception(type, value, tb)
|
|
for s in lst:
|
|
error += s.decode('utf-8', 'replace') if hasattr(s, 'decode') else s
|
|
error = error.replace('\n', '<br>')
|
|
|
|
main_error = lst[-1].decode('utf-8', 'replace') if hasattr(lst[-1], 'decode') else lst[-1]
|
|
|
|
version_label = QCoreApplication.translate('Python', 'Python version:')
|
|
qgis_label = QCoreApplication.translate('Python', 'QGIS version:')
|
|
pypath_label = QCoreApplication.translate('Python', 'Python Path:')
|
|
txt = txt.format(msg=msg,
|
|
main_error=main_error,
|
|
error=error,
|
|
version_label=version_label,
|
|
num=sys.version,
|
|
qgis_label=qgis_label,
|
|
qversion=Qgis.QGIS_VERSION,
|
|
qgisrelease=Qgis.QGIS_RELEASE_NAME,
|
|
devversion=Qgis.QGIS_DEV_VERSION,
|
|
pypath_label=pypath_label,
|
|
pypath=u"".join(u"<li>{}</li>".format(path) for path in sys.path))
|
|
|
|
txt = txt.replace(' ', ' ') # preserve whitespaces for nicer output
|
|
|
|
dlg = QgsMessageOutput.createMessageOutput()
|
|
dlg.setTitle(msg)
|
|
dlg.setMessage(txt, QgsMessageOutput.MessageHtml)
|
|
dlg.showMessage()
|
|
|
|
|
|
def qgis_excepthook(type, value, tb):
|
|
# detect if running in the main thread
|
|
in_main_thread = QCoreApplication.instance() is None or QThread.currentThread() == QCoreApplication.instance().thread()
|
|
|
|
# only use messagebar if running in main thread - otherwise it will crash!
|
|
showException(type, value, tb, None, messagebar=in_main_thread)
|
|
|
|
|
|
def installErrorHook():
|
|
"""
|
|
Installs the QGIS application error/warning hook. This causes Python exceptions
|
|
to be intercepted by the QGIS application and shown in the main window message bar
|
|
and in custom dialogs.
|
|
|
|
Generally you shouldn't call this method - it's automatically called by
|
|
the QGIS app on startup, and has no use in standalone applications and scripts.
|
|
"""
|
|
sys.excepthook = qgis_excepthook
|
|
warnings.showwarning = showWarning
|
|
|
|
|
|
def uninstallErrorHook():
|
|
sys.excepthook = sys.__excepthook__
|
|
|
|
|
|
# initialize 'iface' object
|
|
iface = None
|
|
|
|
|
|
def initInterface(pointer):
|
|
from qgis.gui import QgisInterface
|
|
from sip import wrapinstance
|
|
|
|
global iface
|
|
iface = wrapinstance(pointer, QgisInterface)
|
|
|
|
|
|
#######################
|
|
# PLUGINS
|
|
|
|
# list of plugin paths. it gets filled in by the QGIS python library
|
|
plugin_paths = []
|
|
|
|
# dictionary of plugins
|
|
plugins = {}
|
|
|
|
plugin_times = {}
|
|
|
|
# list of active (started) plugins
|
|
active_plugins = []
|
|
|
|
# list of plugins in plugin directory and home plugin directory
|
|
available_plugins = []
|
|
|
|
# dictionary of plugins providing metadata in a text file (metadata.txt)
|
|
# key = plugin package name, value = config parser instance
|
|
plugins_metadata_parser = {}
|
|
|
|
|
|
def findPlugins(path):
|
|
""" for internal use: return list of plugins in given path """
|
|
for plugin in glob.glob(path + "/*"):
|
|
if not os.path.isdir(plugin):
|
|
continue
|
|
if not os.path.exists(os.path.join(plugin, '__init__.py')):
|
|
continue
|
|
|
|
metadataFile = os.path.join(plugin, 'metadata.txt')
|
|
if not os.path.exists(metadataFile):
|
|
continue
|
|
|
|
cp = configparser.ConfigParser()
|
|
|
|
try:
|
|
with codecs.open(metadataFile, "r", "utf8") as f:
|
|
cp.read_file(f)
|
|
except:
|
|
cp = None
|
|
|
|
pluginName = os.path.basename(plugin)
|
|
yield (pluginName, cp)
|
|
|
|
|
|
def metadataParser():
|
|
"""Used by other modules to access the local parser object"""
|
|
return plugins_metadata_parser
|
|
|
|
|
|
def updateAvailablePlugins():
|
|
""" Go through the plugin_paths list and find out what plugins are available. """
|
|
# merge the lists
|
|
plugins = []
|
|
metadata_parser = {}
|
|
for pluginpath in plugin_paths:
|
|
for pluginName, parser in findPlugins(pluginpath):
|
|
if parser is None:
|
|
continue
|
|
if pluginName not in plugins:
|
|
plugins.append(pluginName)
|
|
metadata_parser[pluginName] = parser
|
|
|
|
global available_plugins
|
|
available_plugins = plugins
|
|
global plugins_metadata_parser
|
|
plugins_metadata_parser = metadata_parser
|
|
|
|
|
|
def pluginMetadata(packageName, fct):
|
|
""" fetch metadata from a plugin - use values from metadata.txt """
|
|
try:
|
|
return plugins_metadata_parser[packageName].get('general', fct)
|
|
except Exception:
|
|
return "__error__"
|
|
|
|
|
|
def loadPlugin(packageName):
|
|
""" load plugin's package """
|
|
|
|
try:
|
|
__import__(packageName)
|
|
return True
|
|
except:
|
|
pass # continue...
|
|
|
|
# snake in the grass, we know it's there
|
|
sys.path_importer_cache.clear()
|
|
|
|
# retry
|
|
try:
|
|
__import__(packageName)
|
|
return True
|
|
except:
|
|
msg = QCoreApplication.translate("Python", "Couldn't load plugin '{0}'").format(packageName)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True, level=Qgis.Critical)
|
|
return False
|
|
|
|
|
|
def _startPlugin(packageName):
|
|
""" initializes a plugin, but does not load GUI """
|
|
global plugins, active_plugins, iface, plugin_times
|
|
|
|
if packageName in active_plugins:
|
|
return False
|
|
|
|
if packageName not in sys.modules:
|
|
return False
|
|
|
|
package = sys.modules[packageName]
|
|
|
|
# create an instance of the plugin
|
|
try:
|
|
plugins[packageName] = package.classFactory(iface)
|
|
except:
|
|
_unloadPluginModules(packageName)
|
|
errMsg = QCoreApplication.translate("Python", "Couldn't load plugin '{0}'").format(packageName)
|
|
msg = QCoreApplication.translate("Python", "{0} due to an error when calling its classFactory() method").format(errMsg)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True, level=Qgis.Critical)
|
|
return False
|
|
return True
|
|
|
|
|
|
def _addToActivePlugins(packageName, duration):
|
|
""" Adds a plugin to the list of active plugins """
|
|
active_plugins.append(packageName)
|
|
plugin_times[packageName] = "{0:02f}s".format(duration)
|
|
|
|
|
|
def startPlugin(packageName):
|
|
""" initialize the plugin """
|
|
global plugins, active_plugins, iface, plugin_times
|
|
start = time.process_time()
|
|
if not _startPlugin(packageName):
|
|
return False
|
|
|
|
# initGui
|
|
try:
|
|
plugins[packageName].initGui()
|
|
except:
|
|
del plugins[packageName]
|
|
_unloadPluginModules(packageName)
|
|
errMsg = QCoreApplication.translate("Python", "Couldn't load plugin '{0}'").format(packageName)
|
|
msg = QCoreApplication.translate("Python", "{0} due to an error when calling its initGui() method").format(errMsg)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True, level=Qgis.Critical)
|
|
return False
|
|
|
|
end = time.process_time()
|
|
_addToActivePlugins(packageName, end - start)
|
|
return True
|
|
|
|
|
|
def startProcessingPlugin(packageName):
|
|
""" initialize only the Processing components of a plugin """
|
|
global plugins, active_plugins, iface, plugin_times
|
|
start = time.process_time()
|
|
if not _startPlugin(packageName):
|
|
return False
|
|
|
|
errMsg = QCoreApplication.translate("Python", "Couldn't load plugin '{0}'").format(packageName)
|
|
if not hasattr(plugins[packageName], 'initProcessing'):
|
|
del plugins[packageName]
|
|
_unloadPluginModules(packageName)
|
|
msg = QCoreApplication.translate("Python", "{0} - plugin has no initProcessing() method").format(errMsg)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True, level=Qgis.Critical)
|
|
return False
|
|
|
|
# initProcessing
|
|
try:
|
|
plugins[packageName].initProcessing()
|
|
except:
|
|
del plugins[packageName]
|
|
_unloadPluginModules(packageName)
|
|
msg = QCoreApplication.translate("Python", "{0} due to an error when calling its initProcessing() method").format(errMsg)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True)
|
|
return False
|
|
|
|
end = time.process_time()
|
|
_addToActivePlugins(packageName, end - start)
|
|
|
|
return True
|
|
|
|
|
|
def canUninstallPlugin(packageName):
|
|
""" confirm that the plugin can be uninstalled """
|
|
global plugins, active_plugins
|
|
|
|
if packageName not in plugins:
|
|
return False
|
|
if packageName not in active_plugins:
|
|
return False
|
|
|
|
try:
|
|
metadata = plugins[packageName]
|
|
if "canBeUninstalled" not in dir(metadata):
|
|
return True
|
|
return bool(metadata.canBeUninstalled())
|
|
except:
|
|
msg = "Error calling " + packageName + ".canBeUninstalled"
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True)
|
|
return True
|
|
|
|
|
|
def unloadPlugin(packageName):
|
|
""" unload and delete plugin! """
|
|
global plugins, active_plugins
|
|
|
|
if packageName not in plugins:
|
|
return False
|
|
if packageName not in active_plugins:
|
|
return False
|
|
|
|
try:
|
|
plugins[packageName].unload()
|
|
del plugins[packageName]
|
|
active_plugins.remove(packageName)
|
|
_unloadPluginModules(packageName)
|
|
return True
|
|
except Exception as e:
|
|
msg = QCoreApplication.translate("Python", "Error while unloading plugin {0}").format(packageName)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg, messagebar=True)
|
|
return False
|
|
|
|
|
|
def _unloadPluginModules(packageName):
|
|
""" unload plugin package with all its modules (files) """
|
|
global _plugin_modules
|
|
mods = _plugin_modules[packageName]
|
|
|
|
for mod in mods:
|
|
if mod not in sys.modules:
|
|
continue
|
|
|
|
# if it looks like a Qt resource file, try to do a cleanup
|
|
# otherwise we might experience a segfault next time the plugin is loaded
|
|
# because Qt will try to access invalid plugin resource data
|
|
try:
|
|
if hasattr(sys.modules[mod], 'qCleanupResources'):
|
|
sys.modules[mod].qCleanupResources()
|
|
except:
|
|
# Print stack trace for debug
|
|
qDebug("qCleanupResources error:\n%s" % traceback.format_exc())
|
|
|
|
# try removing path
|
|
if hasattr(sys.modules[mod], '__path__'):
|
|
for path in sys.modules[mod].__path__:
|
|
try:
|
|
sys.path.remove(path)
|
|
except ValueError:
|
|
# Discard if path is not there
|
|
pass
|
|
|
|
# try to remove the module from python
|
|
try:
|
|
del sys.modules[mod]
|
|
except:
|
|
qDebug("Error when removing module:\n%s" % traceback.format_exc())
|
|
# remove the plugin entry
|
|
del _plugin_modules[packageName]
|
|
|
|
|
|
def isPluginLoaded(packageName):
|
|
""" find out whether a plugin is active (i.e. has been started) """
|
|
global plugins, active_plugins
|
|
|
|
if packageName not in plugins:
|
|
return False
|
|
return (packageName in active_plugins)
|
|
|
|
|
|
def reloadPlugin(packageName):
|
|
""" unload and start again a plugin """
|
|
global active_plugins
|
|
if packageName not in active_plugins:
|
|
return # it's not active
|
|
|
|
unloadPlugin(packageName)
|
|
loadPlugin(packageName)
|
|
startPlugin(packageName)
|
|
|
|
|
|
def showPluginHelp(packageName: str = None, filename: str = "index", section: str = ""):
|
|
"""Open help in the user's html browser. The help file should be named index-ll_CC.html or index-ll.html or index.html.
|
|
|
|
:param str packageName: name of package folder, if None it's using the current file package. Defaults to None. Optional.
|
|
:param str filename: name of file to open. It can be a path like 'doc/index' for example. Defaults to 'index'.
|
|
:param str section: URL path to open. Defaults to empty string.
|
|
"""
|
|
try:
|
|
source = ""
|
|
if packageName is None:
|
|
import inspect
|
|
|
|
source = inspect.currentframe().f_back.f_code.co_filename
|
|
else:
|
|
source = sys.modules[packageName].__file__
|
|
except:
|
|
return
|
|
path = os.path.dirname(source)
|
|
locale = str(QLocale().name())
|
|
helpfile = os.path.join(path, filename + "-" + locale + ".html")
|
|
if not os.path.exists(helpfile):
|
|
helpfile = os.path.join(path, filename + "-" + locale.split("_")[0] + ".html")
|
|
if not os.path.exists(helpfile):
|
|
helpfile = os.path.join(path, filename + "-en.html")
|
|
if not os.path.exists(helpfile):
|
|
helpfile = os.path.join(path, filename + "-en_US.html")
|
|
if not os.path.exists(helpfile):
|
|
helpfile = os.path.join(path, filename + ".html")
|
|
if os.path.exists(helpfile):
|
|
url = "file://" + helpfile
|
|
if section != "":
|
|
url = url + "#" + section
|
|
QDesktopServices.openUrl(QUrl(url))
|
|
|
|
|
|
def pluginDirectory(packageName):
|
|
""" return directory where the plugin resides. Plugin must be loaded already """
|
|
return os.path.dirname(sys.modules[packageName].__file__)
|
|
|
|
|
|
def reloadProjectMacros():
|
|
# unload old macros
|
|
unloadProjectMacros()
|
|
|
|
from qgis.core import QgsProject
|
|
|
|
code, ok = QgsProject.instance().readEntry("Macros", "/pythonCode")
|
|
if not ok or not code or code == '':
|
|
return
|
|
|
|
# create a new empty python module
|
|
import importlib
|
|
mod = importlib.util.module_from_spec(importlib.machinery.ModuleSpec("proj_macros_mod", None))
|
|
|
|
# set the module code and store it sys.modules
|
|
exec(str(code), mod.__dict__)
|
|
sys.modules["proj_macros_mod"] = mod
|
|
|
|
# load new macros
|
|
openProjectMacro()
|
|
|
|
|
|
def unloadProjectMacros():
|
|
if "proj_macros_mod" not in sys.modules:
|
|
return
|
|
# unload old macros
|
|
closeProjectMacro()
|
|
# destroy the reference to the module
|
|
del sys.modules["proj_macros_mod"]
|
|
|
|
|
|
def openProjectMacro():
|
|
if "proj_macros_mod" not in sys.modules:
|
|
return
|
|
mod = sys.modules["proj_macros_mod"]
|
|
if hasattr(mod, 'openProject'):
|
|
mod.openProject()
|
|
|
|
|
|
def saveProjectMacro():
|
|
if "proj_macros_mod" not in sys.modules:
|
|
return
|
|
mod = sys.modules["proj_macros_mod"]
|
|
if hasattr(mod, 'saveProject'):
|
|
mod.saveProject()
|
|
|
|
|
|
def closeProjectMacro():
|
|
if "proj_macros_mod" not in sys.modules:
|
|
return
|
|
mod = sys.modules["proj_macros_mod"]
|
|
if hasattr(mod, 'closeProject'):
|
|
mod.closeProject()
|
|
|
|
|
|
#######################
|
|
# SERVER PLUGINS
|
|
#
|
|
# TODO: move into server_utils.py ?
|
|
|
|
# list of plugin paths. it gets filled in by the QGIS python library
|
|
server_plugin_paths = []
|
|
|
|
# dictionary of plugins
|
|
server_plugins = {}
|
|
|
|
# list of active (started) plugins
|
|
server_active_plugins = []
|
|
|
|
|
|
# initialize 'serverIface' object
|
|
serverIface = None
|
|
|
|
|
|
def initServerInterface(pointer):
|
|
from qgis.server import QgsServerInterface
|
|
from sip import wrapinstance
|
|
sys.excepthook = sys.__excepthook__
|
|
global serverIface
|
|
serverIface = wrapinstance(pointer, QgsServerInterface)
|
|
|
|
|
|
def startServerPlugin(packageName):
|
|
""" initialize the plugin """
|
|
global server_plugins, server_active_plugins, serverIface
|
|
|
|
if packageName in server_active_plugins:
|
|
return False
|
|
if packageName not in sys.modules:
|
|
return False
|
|
|
|
package = sys.modules[packageName]
|
|
|
|
errMsg = QCoreApplication.translate("Python", "Couldn't load server plugin {0}").format(packageName)
|
|
|
|
# create an instance of the plugin
|
|
try:
|
|
server_plugins[packageName] = package.serverClassFactory(serverIface)
|
|
except:
|
|
_unloadPluginModules(packageName)
|
|
msg = QCoreApplication.translate("Python",
|
|
"{0} due to an error when calling its serverClassFactory() method").format(errMsg)
|
|
showException(sys.exc_info()[0], sys.exc_info()[1], sys.exc_info()[2], msg)
|
|
return False
|
|
|
|
# add to active plugins
|
|
server_active_plugins.append(packageName)
|
|
return True
|
|
|
|
|
|
def spatialite_connect(*args, **kwargs):
|
|
"""returns a dbapi2.Connection to a SpatiaLite db
|
|
using the "mod_spatialite" extension (python3)"""
|
|
import sqlite3
|
|
import re
|
|
|
|
def fcnRegexp(pattern, string):
|
|
result = re.search(pattern, string)
|
|
return True if result else False
|
|
|
|
con = sqlite3.dbapi2.connect(*args, **kwargs)
|
|
con.enable_load_extension(True)
|
|
cur = con.cursor()
|
|
libs = [
|
|
# SpatiaLite >= 4.2 and Sqlite >= 3.7.17, should work on all platforms
|
|
("mod_spatialite", "sqlite3_modspatialite_init"),
|
|
# SpatiaLite >= 4.2 and Sqlite < 3.7.17 (Travis)
|
|
("mod_spatialite.so", "sqlite3_modspatialite_init"),
|
|
# SpatiaLite < 4.2 (linux)
|
|
("libspatialite.so", "sqlite3_extension_init")
|
|
]
|
|
found = False
|
|
for lib, entry_point in libs:
|
|
try:
|
|
cur.execute("select load_extension('{}', '{}')".format(lib, entry_point))
|
|
except sqlite3.OperationalError:
|
|
continue
|
|
else:
|
|
found = True
|
|
break
|
|
if not found:
|
|
raise RuntimeError("Cannot find any suitable spatialite module")
|
|
if any(['.gpkg' in arg for arg in args]):
|
|
try:
|
|
cur.execute("SELECT EnableGpkgAmphibiousMode()")
|
|
except (sqlite3.Error, sqlite3.DatabaseError, sqlite3.NotSupportedError):
|
|
QgsMessageLog.logMessage(u"warning:{}".format("Could not enable geopackage amphibious mode"),
|
|
QCoreApplication.translate("Python", "Python warning"))
|
|
|
|
cur.close()
|
|
con.enable_load_extension(False)
|
|
con.create_function("regexp", 2, fcnRegexp)
|
|
return con
|
|
|
|
|
|
class OverrideCursor():
|
|
"""
|
|
Executes a code block with a different cursor set and makes sure the cursor
|
|
is restored even if exceptions are raised or an intermediate ``return``
|
|
statement is hit.
|
|
|
|
Example:
|
|
```
|
|
with OverrideCursor(Qt.WaitCursor):
|
|
do_a_slow(operation)
|
|
```
|
|
"""
|
|
|
|
def __init__(self, cursor):
|
|
self.cursor = cursor
|
|
|
|
def __enter__(self):
|
|
QApplication.setOverrideCursor(self.cursor)
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
QApplication.restoreOverrideCursor()
|
|
|
|
|
|
#######################
|
|
# IMPORT wrapper
|
|
|
|
if os.name == 'nt' and sys.version_info < (3, 8):
|
|
import ctypes
|
|
from ctypes import windll, wintypes
|
|
|
|
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
|
|
|
|
_hasAddDllDirectory = hasattr(kernel32, 'AddDllDirectory')
|
|
if _hasAddDllDirectory:
|
|
_import_path = os.environ['PATH']
|
|
_import_paths = {}
|
|
|
|
def _errcheck_zero(result, func, args):
|
|
if not result:
|
|
raise ctypes.WinError(ctypes.get_last_error())
|
|
return args
|
|
|
|
DLL_DIRECTORY_COOKIE = wintypes.LPVOID
|
|
|
|
_AddDllDirectory = kernel32.AddDllDirectory
|
|
_AddDllDirectory.errcheck = _errcheck_zero
|
|
_AddDllDirectory.restype = DLL_DIRECTORY_COOKIE
|
|
_AddDllDirectory.argtypes = (wintypes.LPCWSTR,)
|
|
|
|
_RemoveDllDirectory = kernel32.RemoveDllDirectory
|
|
_RemoveDllDirectory.errcheck = _errcheck_zero
|
|
_RemoveDllDirectory.argtypes = (DLL_DIRECTORY_COOKIE,)
|
|
|
|
_uses_builtins = True
|
|
try:
|
|
import builtins
|
|
_builtin_import = builtins.__import__
|
|
except AttributeError:
|
|
_uses_builtins = False
|
|
import __builtin__
|
|
_builtin_import = __builtin__.__import__
|
|
|
|
_plugin_modules = {}
|
|
|
|
|
|
def _import(name, globals={}, locals={}, fromlist=[], level=None):
|
|
"""
|
|
Wrapper around builtin import that keeps track of loaded plugin modules and blocks
|
|
certain unsafe imports
|
|
"""
|
|
if level is None:
|
|
level = 0
|
|
|
|
if 'PyQt4' in name:
|
|
msg = 'PyQt4 classes cannot be imported in QGIS 3.x.\n' \
|
|
'Use {} or the version independent {} import instead.'.format(name.replace('PyQt4', 'PyQt5'), name.replace('PyQt4', 'qgis.PyQt'))
|
|
raise ImportError(msg)
|
|
|
|
if os.name == 'nt' and sys.version_info < (3, 8):
|
|
global _hasAddDllDirectory
|
|
if _hasAddDllDirectory:
|
|
global _import_path
|
|
global _import_paths
|
|
|
|
old_path = _import_path
|
|
new_path = os.environ['PATH']
|
|
if old_path != new_path:
|
|
global _AddDllDirectory
|
|
global _RemoveDllDirectory
|
|
|
|
for p in set(new_path.split(';')) - set(old_path.split(';')):
|
|
if p is not None and p not in _import_path and os.path.isdir(p):
|
|
_import_paths[p] = _AddDllDirectory(p)
|
|
|
|
for p in set(old_path.split(';')) - set(new_path.split(';')):
|
|
if p in _import_paths:
|
|
_RemoveDllDirectory(_import_paths.pop(p))
|
|
|
|
_import_path = new_path
|
|
|
|
mod = _builtin_import(name, globals, locals, fromlist, level)
|
|
|
|
if mod and getattr(mod, '__file__', None):
|
|
module_name = mod.__name__ if fromlist else name
|
|
package_name = module_name.split('.')[0]
|
|
# check whether the module belongs to one of our plugins
|
|
if package_name in available_plugins:
|
|
if package_name not in _plugin_modules:
|
|
_plugin_modules[package_name] = set()
|
|
_plugin_modules[package_name].add(module_name)
|
|
# check the fromlist for additional modules (from X import Y,Z)
|
|
if fromlist:
|
|
for fromitem in fromlist:
|
|
frmod = module_name + "." + fromitem
|
|
if frmod in sys.modules:
|
|
_plugin_modules[package_name].add(frmod)
|
|
|
|
return mod
|
|
|
|
|
|
if not os.environ.get('QGIS_NO_OVERRIDE_IMPORT'):
|
|
if _uses_builtins:
|
|
builtins.__import__ = _import
|
|
else:
|
|
__builtin__.__import__ = _import
|
|
|
|
|
|
def run_script_from_file(filepath):
|
|
"""
|
|
Runs a Python script from a given file. Supports loading processing scripts.
|
|
:param filepath: The .py file to load.
|
|
"""
|
|
import sys
|
|
import inspect
|
|
from qgis.processing import alg
|
|
try:
|
|
from qgis.core import QgsApplication, QgsProcessingAlgorithm, QgsProcessingFeatureBasedAlgorithm
|
|
from qgis.processing import execAlgorithmDialog
|
|
_locals = {}
|
|
exec(open(filepath.replace("\\\\", "/").encode(sys.getfilesystemencoding())).read(), _locals)
|
|
alginstance = None
|
|
try:
|
|
alginstance = alg.instances.pop().createInstance()
|
|
except IndexError:
|
|
for name, attr in _locals.items():
|
|
if inspect.isclass(attr) and issubclass(attr, (QgsProcessingAlgorithm, QgsProcessingFeatureBasedAlgorithm)) and attr.__name__ not in ("QgsProcessingAlgorithm", "QgsProcessingFeatureBasedAlgorithm"):
|
|
alginstance = attr()
|
|
break
|
|
if alginstance:
|
|
alginstance.setProvider(QgsApplication.processingRegistry().providerById("script"))
|
|
alginstance.initAlgorithm()
|
|
execAlgorithmDialog(alginstance)
|
|
except ImportError:
|
|
pass
|