mirror of
https://github.com/qgis/QGIS.git
synced 2025-02-27 00:33:48 -05:00
520 lines
19 KiB
Python
Executable File
520 lines
19 KiB
Python
Executable File
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
***************************************************************************
|
|
GeoAlgorithm.py
|
|
---------------------
|
|
Date : August 2012
|
|
Copyright : (C) 2012 by Victor Olaya
|
|
Email : volayaf 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__ = 'Victor Olaya'
|
|
__date__ = 'August 2012'
|
|
__copyright__ = '(C) 2012, Victor Olaya'
|
|
|
|
# This will get replaced with a git SHA1 when you do a git archive
|
|
|
|
__revision__ = '$Format:%H$'
|
|
|
|
import os.path
|
|
import traceback
|
|
import subprocess
|
|
import copy
|
|
|
|
from qgis.PyQt.QtCore import QCoreApplication
|
|
|
|
from qgis.core import (QgsProcessingFeedback,
|
|
QgsSettings,
|
|
QgsProcessingAlgorithm,
|
|
QgsProject,
|
|
QgsProcessingUtils,
|
|
QgsMessageLog)
|
|
|
|
from builtins import str
|
|
from builtins import object
|
|
from processing.gui.ParametersPanel import ParametersPanel
|
|
from processing.core.ProcessingConfig import ProcessingConfig
|
|
from processing.core.GeoAlgorithmExecutionException import GeoAlgorithmExecutionException
|
|
from processing.core.parameters import ParameterRaster, ParameterVector, ParameterMultipleInput, ParameterTable, Parameter
|
|
from processing.core.outputs import OutputVector, OutputRaster, OutputTable, OutputHTML, Output
|
|
from processing.algs.gdal.GdalUtils import GdalUtils
|
|
from processing.tools import dataobjects
|
|
from processing.algs.help import shortHelp
|
|
|
|
|
|
class GeoAlgorithm(QgsProcessingAlgorithm):
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
# Parameters needed by the algorithm
|
|
self.parameters = list()
|
|
|
|
# Outputs generated by the algorithm
|
|
self.outputs = list()
|
|
|
|
# The crs taken from input layers (if possible), and used when
|
|
# loading output layers
|
|
self.crs = None
|
|
|
|
# If the algorithm is run as part of a model, the parent model
|
|
# can be set in this variable, to allow for customized
|
|
# behavior, in case some operations should be run differently
|
|
# when running as part of a model
|
|
self.model = None
|
|
|
|
self.defineCharacteristics()
|
|
|
|
def getCopy(self):
|
|
"""Returns a new instance of this algorithm, ready to be used
|
|
for being executed.
|
|
"""
|
|
newone = self
|
|
newone.parameters = copy.deepcopy(self.parameters)
|
|
newone.outputs = copy.deepcopy(self.outputs)
|
|
return newone
|
|
|
|
# methods to overwrite when creating a custom geoalgorithm
|
|
|
|
def _formatHelp(self, text):
|
|
return "<h2>%s</h2>%s" % (self.displayName(), "".join(["<p>%s</p>" % s for s in text.split("\n")]))
|
|
|
|
def help(self):
|
|
return False, None
|
|
|
|
def shortHelp(self):
|
|
text = shortHelp.get(self.id(), None)
|
|
if text is not None:
|
|
text = self._formatHelp(text)
|
|
return text
|
|
|
|
def processAlgorithm(self, context, feedback):
|
|
"""Here goes the algorithm itself.
|
|
|
|
There is no return value from this method.
|
|
A GeoAlgorithmExecutionException should be raised in case
|
|
something goes wrong.
|
|
:param context:
|
|
"""
|
|
pass
|
|
|
|
def defineCharacteristics(self):
|
|
"""Here is where the parameters and outputs should be defined.
|
|
"""
|
|
pass
|
|
|
|
def getParametersPanel(self, parent):
|
|
return ParametersPanel(parent, self)
|
|
|
|
def getCustomParametersDialog(self):
|
|
"""If the algorithm has a custom parameters dialog, it should
|
|
be returned here, ready to be executed.
|
|
"""
|
|
return None
|
|
|
|
def getCustomModelerParametersDialog(self, modelAlg, algName=None):
|
|
"""If the algorithm has a custom parameters dialog when called
|
|
from the modeler, it should be returned here, ready to be
|
|
executed.
|
|
"""
|
|
return None
|
|
|
|
def getParameterDescriptions(self):
|
|
"""Returns a dict with param names as keys and detailed
|
|
descriptions of each param as value. These descriptions are
|
|
used as tool tips in the parameters dialog.
|
|
|
|
If a description does not exist, the parameter's
|
|
human-readable name is used.
|
|
"""
|
|
descs = {}
|
|
return descs
|
|
|
|
def checkBeforeOpeningParametersDialog(self):
|
|
"""If there is any check to perform before the parameters
|
|
dialog is opened, it should be done here.
|
|
|
|
This method returns an error message string if there is any
|
|
problem (for instance, an external app not configured yet),
|
|
or None if the parameters dialog can be opened.
|
|
|
|
Note that this check should also be done in the
|
|
processAlgorithm method, since algorithms can be called without
|
|
opening the parameters dialog.
|
|
"""
|
|
return None
|
|
|
|
def checkParameterValuesBeforeExecuting(self):
|
|
"""If there is any check to do before launching the execution
|
|
of the algorithm, it should be done here.
|
|
|
|
If values are not correct, a message should be returned
|
|
explaining the problem.
|
|
|
|
This check is called from the parameters dialog, and also when
|
|
calling from the console.
|
|
"""
|
|
return None
|
|
|
|
def processBeforeAddingToModeler(self, alg, model):
|
|
"""Add here any task that has to be performed before adding an algorithm
|
|
to a model, such as changing the value of a parameter depending on value
|
|
of another one"""
|
|
pass
|
|
|
|
# =========================================================
|
|
|
|
def execute(self, context=None, feedback=None, model=None):
|
|
"""The method to use to call a processing algorithm.
|
|
|
|
Although the body of the algorithm is in processAlgorithm(),
|
|
it should be called using this method, since it performs
|
|
some additional operations.
|
|
|
|
Raises a GeoAlgorithmExecutionException in case anything goes
|
|
wrong.
|
|
"""
|
|
|
|
if feedback is None:
|
|
feedback = QgsProcessingFeedback()
|
|
if context is None:
|
|
context = dataobjects.createContext()
|
|
|
|
self.model = model
|
|
try:
|
|
self.setOutputCRS()
|
|
self.resolveOutputs()
|
|
self.evaluateParameterValues()
|
|
self.runPreExecutionScript(feedback)
|
|
self.processAlgorithm(context, feedback)
|
|
feedback.setProgress(100)
|
|
self.convertUnsupportedFormats(context, feedback)
|
|
self.runPostExecutionScript(feedback)
|
|
except GeoAlgorithmExecutionException as gaee:
|
|
lines = [self.tr('Error while executing algorithm')]
|
|
lines.append(traceback.format_exc())
|
|
QgsMessageLog.logMessage(gaee.msg, self.tr('Processing'), QgsMessageLog.CRITICAL)
|
|
raise GeoAlgorithmExecutionException(gaee.msg, lines, gaee)
|
|
except Exception as e:
|
|
# If something goes wrong and is not caught in the
|
|
# algorithm, we catch it here and wrap it
|
|
lines = [self.tr('Uncaught error while executing algorithm')]
|
|
lines.append(traceback.format_exc())
|
|
QgsMessageLog.logMessage('\n'.join(lines), self.tr('Processing'), QgsMessageLog.CRITICAL)
|
|
raise GeoAlgorithmExecutionException(str(e) + self.tr('\nSee log for more details'), lines, e)
|
|
|
|
def _checkParameterValuesBeforeExecuting(self, context=None):
|
|
if context is None:
|
|
context = dataobjects.createContext()
|
|
for param in self.parameters:
|
|
if isinstance(param, (ParameterRaster, ParameterVector,
|
|
ParameterMultipleInput)):
|
|
if param.value:
|
|
if isinstance(param, ParameterMultipleInput):
|
|
inputlayers = param.value.split(';')
|
|
else:
|
|
inputlayers = [param.value]
|
|
for inputlayer in inputlayers:
|
|
obj = QgsProcessingUtils.mapLayerFromString(inputlayer, context)
|
|
if obj is None:
|
|
return "Wrong parameter value: " + param.value
|
|
return self.checkParameterValuesBeforeExecuting()
|
|
|
|
def runPostExecutionScript(self, feedback):
|
|
scriptFile = ProcessingConfig.getSetting(
|
|
ProcessingConfig.POST_EXECUTION_SCRIPT)
|
|
self.runHookScript(scriptFile, feedback)
|
|
|
|
def runPreExecutionScript(self, feedback):
|
|
scriptFile = ProcessingConfig.getSetting(
|
|
ProcessingConfig.PRE_EXECUTION_SCRIPT)
|
|
self.runHookScript(scriptFile, feedback)
|
|
|
|
def runHookScript(self, filename, feedback):
|
|
if filename is None or not os.path.exists(filename):
|
|
return
|
|
try:
|
|
script = 'import processing\n'
|
|
ns = {}
|
|
ns['feedback'] = feedback
|
|
ns['alg'] = self
|
|
with open(filename) as f:
|
|
lines = f.readlines()
|
|
for line in lines:
|
|
script += line
|
|
exec(script, ns)
|
|
except Exception as e:
|
|
QgsMessageLog.logMessage("Error in hook script: " + str(e), self.tr('Processing'), QgsMessageLog.WARNING)
|
|
# A wrong script should not cause problems, so we swallow
|
|
# all exceptions
|
|
pass
|
|
|
|
def convertUnsupportedFormats(self, context, feedback):
|
|
i = 0
|
|
feedback.setProgressText(self.tr('Converting outputs'))
|
|
for out in self.outputs:
|
|
if isinstance(out, OutputVector):
|
|
if out.compatible is not None:
|
|
layer = QgsProcessingUtils.mapLayerFromString(out.compatible, context)
|
|
if layer is None:
|
|
# For the case of memory layer, if the
|
|
# getCompatible method has been called
|
|
continue
|
|
writer = out.getVectorWriter(layer.fields(), layer.wkbType(), layer.crs(), context)
|
|
features = QgsProcessingUtils.getFeatures(layer, context)
|
|
for feature in features:
|
|
writer.addFeature(feature)
|
|
elif isinstance(out, OutputRaster):
|
|
if out.compatible is not None:
|
|
layer = QgsProcessingUtils.mapLayerFromString(out.compatible, context)
|
|
format = self.getFormatShortNameFromFilename(out.value)
|
|
orgFile = out.compatible
|
|
destFile = out.value
|
|
crsid = layer.crs().authid()
|
|
settings = QgsSettings()
|
|
path = str(settings.value('/GdalTools/gdalPath', ''))
|
|
envval = str(os.getenv('PATH'))
|
|
if not path.lower() in envval.lower().split(os.pathsep):
|
|
envval += '%s%s' % (os.pathsep, path)
|
|
os.putenv('PATH', envval)
|
|
command = 'gdal_translate -of %s -a_srs %s %s %s' % (format, crsid, orgFile, destFile)
|
|
if os.name == 'nt':
|
|
command = command.split(" ")
|
|
else:
|
|
command = [command]
|
|
proc = subprocess.Popen(
|
|
command,
|
|
shell=True,
|
|
stdout=subprocess.PIPE,
|
|
stdin=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=False,
|
|
)
|
|
proc.communicate()
|
|
|
|
elif isinstance(out, OutputTable):
|
|
if out.compatible is not None:
|
|
layer = QgsProcessingUtils.mapLayerFromString(out.compatible, context)
|
|
writer = out.getTableWriter(layer.fields())
|
|
features = QgsProcessingUtils.getFeatures(layer, context)
|
|
for feature in features:
|
|
writer.addRecord(feature)
|
|
feedback.setProgress(100 * i / float(len(self.outputs)))
|
|
|
|
def getFormatShortNameFromFilename(self, filename):
|
|
ext = filename[filename.rfind('.') + 1:]
|
|
supported = GdalUtils.getSupportedRasters()
|
|
for name in list(supported.keys()):
|
|
exts = supported[name]
|
|
if ext in exts:
|
|
return name
|
|
return 'GTiff'
|
|
|
|
def evaluateParameterValues(self):
|
|
for param in self.parameters:
|
|
try:
|
|
param.evaluate(self)
|
|
except ValueError as e:
|
|
traceback.print_exc()
|
|
raise GeoAlgorithmExecutionException(str(e))
|
|
|
|
def resolveOutputs(self):
|
|
"""Sets temporary outputs (output.value = None) with a
|
|
temporary file instead. Resolves expressions as well.
|
|
"""
|
|
try:
|
|
for out in self.outputs:
|
|
out.resolveValue(self)
|
|
except ValueError as e:
|
|
raise GeoAlgorithmExecutionException(str(e))
|
|
|
|
def setOutputCRS(self):
|
|
context = dataobjects.createContext()
|
|
layers = QgsProcessingUtils.compatibleLayers(QgsProject.instance())
|
|
for param in self.parameters:
|
|
if isinstance(param, (ParameterRaster, ParameterVector, ParameterMultipleInput)):
|
|
if param.value:
|
|
if isinstance(param, ParameterMultipleInput):
|
|
inputlayers = param.value.split(';')
|
|
else:
|
|
inputlayers = [param.value]
|
|
for inputlayer in inputlayers:
|
|
for layer in layers:
|
|
if layer.source() == inputlayer:
|
|
self.crs = layer.crs()
|
|
return
|
|
p = QgsProcessingUtils.mapLayerFromString(inputlayer, context)
|
|
if p is not None:
|
|
self.crs = p.crs()
|
|
p = None
|
|
return
|
|
try:
|
|
from qgis.utils import iface
|
|
if iface is not None:
|
|
self.crs = iface.mapCanvas().mapSettings().destinationCrs()
|
|
except:
|
|
pass
|
|
|
|
def resolveDataObjects(self):
|
|
layers = QgsProcessingUtils.compatibleLayers(QgsProject.instance())
|
|
for param in self.parameters:
|
|
if isinstance(param, (ParameterRaster, ParameterVector, ParameterTable,
|
|
ParameterMultipleInput)):
|
|
if param.value:
|
|
if isinstance(param, ParameterMultipleInput):
|
|
inputlayers = param.value.split(';')
|
|
else:
|
|
inputlayers = [param.value]
|
|
for i, inputlayer in enumerate(inputlayers):
|
|
for layer in layers:
|
|
if layer.name() == inputlayer:
|
|
inputlayers[i] = layer.source()
|
|
break
|
|
param.setValue(";".join(inputlayers))
|
|
|
|
def checkInputCRS(self, context=None):
|
|
"""It checks that all input layers use the same CRS. If so,
|
|
returns True. False otherwise.
|
|
"""
|
|
if context is None:
|
|
context = dataobjects.createContext()
|
|
crsList = []
|
|
for param in self.parameters:
|
|
if isinstance(param, (ParameterRaster, ParameterVector, ParameterMultipleInput)):
|
|
if param.value:
|
|
if isinstance(param, ParameterMultipleInput):
|
|
layers = param.value.split(';')
|
|
else:
|
|
layers = [param.value]
|
|
for item in layers:
|
|
crs = QgsProcessingUtils.mapLayerFromString(item, context).crs()
|
|
if crs not in crsList:
|
|
crsList.append(crs)
|
|
return len(crsList) < 2
|
|
|
|
def addOutput(self, output):
|
|
# TODO: check that name does not exist
|
|
if isinstance(output, Output):
|
|
self.outputs.append(output)
|
|
|
|
def addParameter(self, param):
|
|
# TODO: check that name does not exist
|
|
if isinstance(param, Parameter):
|
|
self.parameters.append(param)
|
|
|
|
def setParameterValue(self, paramName, value):
|
|
for param in self.parameters:
|
|
if param.name == paramName:
|
|
return param.setValue(value)
|
|
|
|
def setOutputValue(self, outputName, value):
|
|
for out in self.outputs:
|
|
if out.name == outputName:
|
|
out.setValue(value)
|
|
|
|
def getVisibleOutputsCount(self):
|
|
"""Returns the number of non-hidden outputs.
|
|
"""
|
|
i = 0
|
|
for out in self.outputs:
|
|
if not out.hidden:
|
|
i += 1
|
|
return i
|
|
|
|
def getVisibleParametersCount(self):
|
|
"""Returns the number of non-hidden parameters.
|
|
"""
|
|
i = 0
|
|
for param in self.parameters:
|
|
if not param.hidden:
|
|
i += 1
|
|
return i
|
|
|
|
def getHTMLOutputsCount(self):
|
|
"""Returns the number of HTML outputs.
|
|
"""
|
|
i = 0
|
|
for out in self.outputs:
|
|
if isinstance(out, OutputHTML):
|
|
i += 1
|
|
return i
|
|
|
|
def getOutputValuesAsDictionary(self):
|
|
d = {}
|
|
for out in self.outputs:
|
|
d[out.name] = out.value
|
|
return d
|
|
|
|
def __str__(self):
|
|
s = 'ALGORITHM: ' + self.name() + '\n'
|
|
for param in self.parameters:
|
|
s += '\t' + str(param) + '\n'
|
|
for out in self.outputs:
|
|
if not out.hidden:
|
|
s += '\t' + str(out) + '\n'
|
|
s += '\n'
|
|
return s
|
|
|
|
def removeOutputFromName(self, name):
|
|
for out in self.outputs:
|
|
if out.name == name:
|
|
self.outputs.remove(out)
|
|
|
|
def getOutputFromName(self, name):
|
|
for out in self.outputs:
|
|
if out.name == name:
|
|
return out
|
|
|
|
def getParameterFromName(self, name):
|
|
for param in self.parameters:
|
|
if param.name == name:
|
|
return param
|
|
|
|
def getParameterValue(self, name):
|
|
for param in self.parameters:
|
|
if param.name == name:
|
|
return param.value
|
|
return None
|
|
|
|
def getOutputValue(self, name):
|
|
for out in self.outputs:
|
|
if out.name == name:
|
|
return out.value
|
|
return None
|
|
|
|
def getAsCommand(self):
|
|
"""Returns the command that would run this same algorithm from
|
|
the console.
|
|
|
|
Should return None if the algorithm cannot be run from the
|
|
console.
|
|
"""
|
|
|
|
s = 'processing.run("' + self.id() + '",'
|
|
for param in self.parameters:
|
|
s += param.getValueAsCommandLineParameter() + ','
|
|
for out in self.outputs:
|
|
if not out.hidden:
|
|
s += out.getValueAsCommandLineParameter() + ','
|
|
s = s[:-1] + ')'
|
|
return s
|
|
|
|
def tr(self, string, context=''):
|
|
if context == '':
|
|
context = self.__class__.__name__
|
|
return QCoreApplication.translate(context, string)
|
|
|
|
def trAlgorithm(self, string, context=''):
|
|
if context == '':
|
|
context = self.__class__.__name__
|
|
return string, QCoreApplication.translate(context, string)
|