Restore ability to create processing tests from history panel

This commit is contained in:
Nyall Dawson 2017-08-08 23:23:29 +10:00
parent 5177972806
commit 69c991ed87

View File

@ -30,6 +30,7 @@ import os
import re import re
import yaml import yaml
import hashlib import hashlib
import ast
from osgeo import gdal from osgeo import gdal
from osgeo.gdalconst import GA_ReadOnly from osgeo.gdalconst import GA_ReadOnly
@ -37,31 +38,23 @@ from osgeo.gdalconst import GA_ReadOnly
from numpy import nan_to_num from numpy import nan_to_num
from qgis.core import (QgsApplication, from qgis.core import (QgsApplication,
QgsProcessingParameterDefinition) QgsProcessing,
QgsProcessingParameterDefinition,
QgsProcessingParameterBoolean,
QgsProcessingParameterNumber,
QgsProcessingParameterFile,
QgsProcessingParameterString,
QgsProcessingParameterVectorLayer,
QgsProcessingParameterFeatureSource,
QgsProcessingParameterRasterLayer,
QgsProcessingParameterMultipleLayers,
QgsProcessingParameterRasterDestination,
QgsProcessingParameterFeatureSink,
QgsProcessingParameterVectorDestination,
QgsProcessingParameterFileDestination)
from qgis.PyQt.QtCore import QCoreApplication, QMetaObject from qgis.PyQt.QtCore import QCoreApplication, QMetaObject
from qgis.PyQt.QtWidgets import QDialog, QVBoxLayout, QTextEdit, QMessageBox from qgis.PyQt.QtWidgets import QDialog, QVBoxLayout, QTextEdit, QMessageBox
from processing.core.Processing import Processing
from processing.core.outputs import (
OutputNumber,
OutputString,
OutputRaster,
OutputVector,
OutputHTML,
OutputFile
)
from processing.core.parameters import (
ParameterRaster,
ParameterVector,
ParameterMultipleInput,
ParameterFile,
ParameterString,
ParameterNumber,
ParameterBoolean,
ParameterTable
)
def extractSchemaPath(filepath): def extractSchemaPath(filepath):
""" """
@ -132,15 +125,24 @@ def parseParameters(command):
pos = m.end(0) pos = m.end(0)
def splitAlgIdAndParameters(command):
"""
Extracts the algorithm ID and input parameter list from a processing runalg command
"""
exp = re.compile(r"""['"](.*?)['"]\s*,\s*(.*)""")
m = exp.search(command[len('processing.run('):-1])
return m.group(1), ast.literal_eval(m.group(2))
def createTest(text): def createTest(text):
definition = {} definition = {}
tokens = list(parseParameters(text[len('processing.run('):-1])) alg_id, parameters = splitAlgIdAndParameters(text)
cmdname = tokens[0]
alg = QgsApplication.processingRegistry().createAlgorithmById(cmdname)
definition['name'] = 'Test ({})'.format(cmdname) alg = QgsApplication.processingRegistry().createAlgorithmById(alg_id)
definition['algorithm'] = cmdname
definition['name'] = 'Test ({})'.format(alg_id)
definition['algorithm'] = alg_id
params = {} params = {}
results = {} results = {}
@ -151,12 +153,12 @@ def createTest(text):
continue continue
i += 1 i += 1
token = tokens[i] token = parameters[param.name()]
# Handle empty parameters that are optionals # Handle empty parameters that are optionals
if param.flags() & QgsProcessingParameterDefinition.FlagOptional and token is None: if param.flags() & QgsProcessingParameterDefinition.FlagOptional and token is None:
continue continue
if isinstance(param, ParameterVector): if isinstance(param, (QgsProcessingParameterVectorLayer, QgsProcessingParameterFeatureSource)):
schema, filepath = extractSchemaPath(token) schema, filepath = extractSchemaPath(token)
p = { p = {
'type': 'vector', 'type': 'vector',
@ -166,7 +168,7 @@ def createTest(text):
p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]'
params[param.name()] = p params[param.name()] = p
elif isinstance(param, ParameterRaster): elif isinstance(param, QgsProcessingParameterRasterLayer):
schema, filepath = extractSchemaPath(token) schema, filepath = extractSchemaPath(token)
p = { p = {
'type': 'raster', 'type': 'raster',
@ -176,27 +178,18 @@ def createTest(text):
p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]'
params[param.name()] = p params[param.name()] = p
elif isinstance(param, ParameterTable): elif isinstance(param, QgsProcessingParameterMultipleLayers):
schema, filepath = extractSchemaPath(token)
p = {
'type': 'table',
'name': filepath
}
if not schema:
p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]'
params[param.name()] = p
elif isinstance(param, ParameterMultipleInput):
multiparams = token.split(';') multiparams = token.split(';')
newparam = [] newparam = []
# Handle datatype detection # Handle datatype detection
dataType = param.dataType() dataType = param.layerType()
if dataType in ['points', 'lines', 'polygons', 'any vectors']: if dataType in [QgsProcessing.TypeVectorAny, QgsProcessing.TypeVectorPoint, QgsProcessing.TypeVectorLine, QgsProcessing.TypeVectorPolygon, QgsProcessing.TypeTable]:
dataType = 'vector' dataType = 'vector'
else: else:
dataType = 'raster' dataType = 'raster'
schema = None
for mp in multiparams: for mp in multiparams:
schema, filepath = extractSchemaPath(mp) schema, filepath = extractSchemaPath(mp)
newparam.append({ newparam.append({
@ -211,7 +204,7 @@ def createTest(text):
p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]'
params[param.name()] = p params[param.name()] = p
elif isinstance(param, ParameterFile): elif isinstance(param, QgsProcessingParameterFile):
schema, filepath = extractSchemaPath(token) schema, filepath = extractSchemaPath(token)
p = { p = {
'type': 'file', 'type': 'file',
@ -221,12 +214,12 @@ def createTest(text):
p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]'
params[param.name()] = p params[param.name()] = p
elif isinstance(param, ParameterString): elif isinstance(param, QgsProcessingParameterString):
params[param.name()] = token params[param.name()] = token
elif isinstance(param, ParameterBoolean): elif isinstance(param, QgsProcessingParameterBoolean):
params[param.name()] = token params[param.name()] = token
elif isinstance(param, ParameterNumber): elif isinstance(param, QgsProcessingParameterNumber):
if param.isInteger: if param.dataType() == QgsProcessingParameterNumber.Integer:
params[param.name()] = int(token) params[param.name()] = int(token)
else: else:
params[param.name()] = float(token) params[param.name()] = float(token)
@ -235,16 +228,14 @@ def createTest(text):
token = token[1:] token = token[1:]
if token[-1] == '"': if token[-1] == '"':
token = token[:-1] token = token[:-1]
params[param.name] = token params[param.name()] = token
definition['params'] = params definition['params'] = params
for i, out in enumerate([out for out in alg.destinationParameterDefinitions() if not out.flags() & QgsProcessingParameterDefinition.FlagHidden]): for i, out in enumerate([out for out in alg.destinationParameterDefinitions() if not out.flags() & QgsProcessingParameterDefinition.FlagHidden]):
token = tokens[i - len(alg.destinationParameterDefinitions())] token = parameters[out.name()]
if isinstance(out, (OutputNumber, OutputString)): if isinstance(out, QgsProcessingParameterRasterDestination):
results[out.name] = str(out)
elif isinstance(out, OutputRaster):
if token is None: if token is None:
QMessageBox.warning(None, QMessageBox.warning(None,
tr('Error'), tr('Error'),
@ -258,26 +249,26 @@ def createTest(text):
dataArray = nan_to_num(dataset.ReadAsArray(0)) dataArray = nan_to_num(dataset.ReadAsArray(0))
strhash = hashlib.sha224(dataArray.data).hexdigest() strhash = hashlib.sha224(dataArray.data).hexdigest()
results[out.name] = { results[out.name()] = {
'type': 'rasterhash', 'type': 'rasterhash',
'hash': strhash 'hash': strhash
} }
elif isinstance(out, OutputVector): elif isinstance(out, (QgsProcessingParameterVectorDestination, QgsProcessingParameterFeatureSink)):
schema, filepath = extractSchemaPath(token) schema, filepath = extractSchemaPath(token)
results[out.name] = { results[out.name()] = {
'type': 'vector', 'type': 'vector',
'name': filepath 'name': filepath
} }
if not schema: if not schema:
results[out.name]['location'] = '[The expected result data is not in the testdata directory. Please write it to processing/tests/testdata/expected. Prefer gml files.]' results[out.name()]['location'] = '[The expected result data is not in the testdata directory. Please write it to processing/tests/testdata/expected. Prefer gml files.]'
elif isinstance(out, OutputHTML) or isinstance(out, OutputFile): elif isinstance(out, QgsProcessingParameterFileDestination):
schema, filepath = extractSchemaPath(token) schema, filepath = extractSchemaPath(token)
results[out.name] = { results[out.name()] = {
'type': 'file', 'type': 'file',
'name': filepath 'name': filepath
} }
if not schema: if not schema:
results[out.name]['location'] = '[The expected result file is not in the testdata directory. Please redirect the output to processing/tests/testdata/expected.]' results[out.name()]['location'] = '[The expected result file is not in the testdata directory. Please redirect the output to processing/tests/testdata/expected.]'
definition['results'] = results definition['results'] = results
dlg = ShowTestDialog(yaml.dump([definition], default_flow_style=False)) dlg = ShowTestDialog(yaml.dump([definition], default_flow_style=False))