""" *************************************************************************** algfactory.py --------------------- Date : November 2018 Copyright : (C) 2018 by Nathan Woodrow Email : woodrow dot nathan 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__ = "Nathan Woodrow" __date__ = "November 2018" __copyright__ = "(C) 2018, Nathan Woodrow" from collections import OrderedDict from functools import partial from qgis.PyQt.QtCore import QCoreApplication from qgis.PyQt.QtGui import QIcon from qgis.core import ( QgsProcessingParameterDefinition, QgsProcessingAlgorithm, QgsProcessingParameterString, QgsProcessingParameterAuthConfig, QgsProcessingParameterNumber, QgsProcessingParameterDistance, QgsProcessingParameterDuration, QgsProcessingParameterFeatureSource, QgsProcessingParameterFeatureSink, QgsProcessingParameterFileDestination, QgsProcessingParameterFolderDestination, QgsProcessingParameterRasterDestination, QgsProcessingParameterVectorDestination, QgsProcessingParameterPointCloudDestination, QgsProcessingParameterBand, QgsProcessingParameterBoolean, QgsProcessingParameterCrs, QgsProcessingParameterEnum, QgsProcessingParameterExpression, QgsProcessingParameterExtent, QgsProcessingParameterField, QgsProcessingParameterFile, QgsProcessingParameterMapLayer, QgsProcessingParameterMatrix, QgsProcessingParameterMultipleLayers, QgsProcessingParameterPoint, QgsProcessingParameterGeometry, QgsProcessingParameterRange, QgsProcessingParameterRasterLayer, QgsProcessingParameterVectorLayer, QgsProcessingParameterMeshLayer, QgsProcessingParameterColor, QgsProcessingParameterScale, QgsProcessingParameterLayout, QgsProcessingParameterLayoutItem, QgsProcessingParameterDateTime, QgsProcessingParameterMapTheme, QgsProcessingParameterProviderConnection, QgsProcessingParameterDatabaseSchema, QgsProcessingParameterDatabaseTable, QgsProcessingParameterCoordinateOperation, QgsProcessingParameterPointCloudLayer, QgsProcessingParameterAnnotationLayer, QgsProcessingOutputString, QgsProcessingOutputBoolean, QgsProcessingOutputFile, QgsProcessingOutputFolder, QgsProcessingOutputHtml, QgsProcessingOutputLayerDefinition, QgsProcessingOutputMapLayer, QgsProcessingOutputMultipleLayers, QgsProcessingOutputNumber, QgsProcessingOutputRasterLayer, QgsProcessingOutputVectorLayer, QgsProcessingOutputPointCloudLayer, QgsMessageLog, QgsApplication, ) def _log(*args, **kw): """ Log messages to the QgsMessageLog viewer """ QgsMessageLog.logMessage(" ".join(map(str, args)), "Factory") def _make_output(**args): """ Create a processing output class type. :param args: 'cls' The class object type. 'name' the name of the output 'description' The description used on the output :return: """ cls = args["cls"] del args["cls"] newargs = { "name": args["name"], "description": args["description"], } return cls(**newargs) class ProcessingAlgFactoryException(Exception): """ Exception raised when using @alg on a function. """ def __init__(self, message): super().__init__(message) class AlgWrapper(QgsProcessingAlgorithm): """ Wrapper object used to create new processing algorithms from @alg. """ def __init__( self, name=None, display=None, group=None, group_id=None, inputs=None, outputs=None, func=None, help=None, icon=None, ): super().__init__() self._inputs = OrderedDict(inputs or {}) self._outputs = OrderedDict(outputs or {}) self._icon = icon self._name = name self._group = group self._group_id = group_id self._display = display self._func = func self._help = help def _get_parent_id(self, parent): """ Return the id of the parent object. """ if isinstance(parent, str): return parent else: raise NotImplementedError() # Wrapper logic def define( self, name, label, group, group_label, help=None, icon=QgsApplication.iconPath("processingScript.svg"), ): self._name = name self._display = label self._group = group_label self._group_id = group self._help = help self._icon = icon def end(self): """ Finalize the wrapper logic and check for any invalid config. """ if not self.has_outputs: raise ProcessingAlgFactoryException( "No outputs defined for '{}' alg" "At least one must be defined. Use @alg.output" ) def add_output(self, type, **kwargs): parm = self._create_param(type, output=True, **kwargs) self._outputs[parm.name()] = parm def add_help(self, helpstring, *args, **kwargs): self._help = helpstring def add_input(self, type, **kwargs): parm = self._create_param(type, **kwargs) self._inputs[parm.name()] = parm @property def inputs(self): return self._inputs @property def outputs(self): return self._outputs def _create_param(self, type, output=False, **kwargs): name = kwargs["name"] if name in self._inputs or name in self._outputs: raise ProcessingAlgFactoryException(f"{name} already defined") parent = kwargs.get("parent") if parent: parentname = self._get_parent_id(parent) if parentname == name: raise ProcessingAlgFactoryException( "{} can't depend on itself. " "We know QGIS is smart but it's not that smart".format(name) ) if parentname not in self._inputs and parentname not in self._outputs: raise ProcessingAlgFactoryException( f"Can't find parent named {parentname}" ) kwargs["description"] = kwargs.pop("label", "") kwargs["defaultValue"] = kwargs.pop("default", None) advanced = kwargs.pop("advanced", False) help_str = kwargs.pop("help", "") try: if output: try: make_func = output_type_mapping[type] except KeyError: raise ProcessingAlgFactoryException( f"{type} is a invalid output type" ) else: try: make_func = input_type_mapping[type] except KeyError: raise ProcessingAlgFactoryException( f"{type} is a invalid input type" ) parm = make_func(**kwargs) if advanced: parm.setFlags( parm.flags() | QgsProcessingParameterDefinition.Flag.FlagAdvanced ) if not output: parm.setHelp(help_str) return parm except KeyError as ex: raise NotImplementedError(f"{str(type)} not supported") def set_func(self, func): self._func = func # Only take the help from the function if it hasn't already been set. if self._func and not self._help: self._help = self._func.__doc__.strip() @property def has_outputs(self): """ True if this alg wrapper has outputs defined. """ dests = [p for p in self._inputs.values() if p.isDestination()] return bool(self._outputs) or bool(dests) @property def has_inputs(self): """ True if this alg wrapper has outputs defined. """ return bool(self._inputs) def _get_parameter_value(self, parm, parameters, name, context): """ Extract the real value from the parameter. """ if isinstance( parm, (QgsProcessingParameterString, QgsProcessingParameterAuthConfig) ): value = self.parameterAsString(parameters, name, context) return value elif isinstance(parm, QgsProcessingParameterNumber): if parm.dataType() == QgsProcessingParameterNumber.Type.Integer: value = self.parameterAsInt(parameters, name, context) return value if parm.dataType() == QgsProcessingParameterNumber.Type.Double: value = self.parameterAsDouble(parameters, name, context) return value # Overloads def name(self): return self._name def displayName(self): return self._display def group(self): return self._group def groupId(self): return self._group_id def processAlgorithm(self, parameters, context, feedback): values = {} for parm in self._inputs.values(): name = parm.name() values[name] = self._get_parameter_value(parm, parameters, name, context) output = self._func(self, parameters, context, feedback, values) if output is None: return {} return output def createInstance(self): return AlgWrapper( self._name, self._display, self._group, self._group_id, inputs=self._inputs, outputs=self._outputs, func=self._func, help=self._help, icon=self._icon, ) def initAlgorithm(self, configuration=None, p_str=None, Any=None, *args, **kwargs): for parm in self._inputs.values(): self.addParameter(parm.clone()) for parm in self._outputs.values(): clsname = parm.__class__.__name__ klass = globals()[clsname] clone = klass(parm.name(), parm.description()) self.addOutput(clone) def shortHelpString(self): return self._help def icon(self): return QIcon(self._icon) class ProcessingAlgFactory: STRING = ("STRING",) INT = ("INT",) NUMBER = ("NUMBER",) DISTANCE = ("DISTANCE",) SINK = "SINK" SOURCE = "SOURCE" FILE = ("FILE",) FOLDER = ("FOLDER",) HTML = ("HTML",) LAYERDEF = ("LAYERDEF",) MAPLAYER = ("MAPLAYER",) MULTILAYER = ("MULTILAYER",) RASTER_LAYER = ("RASTER_LAYER",) VECTOR_LAYER = ("VECTOR_LAYER",) MESH_LAYER = ("MESH_LAYER",) POINT_CLOUD_LAYER = ("POINT_CLOUD_LAYER",) FILE_DEST = ("FILE_DEST",) FOLDER_DEST = ("FOLDER_DEST",) RASTER_LAYER_DEST = ("RASTER_LAYER_DEST",) VECTOR_LAYER_DEST = ("VECTOR_LAYER_DEST",) POINTCLOUD_LAYER_DEST = ("POINTCLOUD_LAYER_DEST",) BAND = ("BAND",) BOOL = ("BOOL",) CRS = ("CRS",) ENUM = ("ENUM",) EXPRESSION = ("EXPRESSION",) EXTENT = ("EXTENT",) FIELD = ("FIELD",) MATRIX = ("MATRIX",) POINT = ("POINT",) GEOMETRY = ("GEOMETRY",) RANGE = ("RANGE",) AUTH_CFG = "AUTH_CFG" SCALE = "SCALE" COLOR = "COLOR" LAYOUT = "LAYOUT" LAYOUT_ITEM = "LAYOUT_ITEM" DATETIME = "DATETIME" DATE = "DATE" TIME = "TIME" MAP_THEME = "MAP_THEME" PROVIDER_CONNECTION = "PROVIDER_CONNECTION" DATABASE_SCHEMA = "DATABASE_SCHEMA" DATABASE_TABLE = "DATABASE_TABLE" COORDINATE_OPERATION = "COORDINATE_OPERATION" POINTCLOUD_LAYER = "POINTCLOUD_LAYER" ANNOTATION_LAYER = "ANNOTATION_LAYER" def __init__(self): self._current = None self.instances = [] def tr(self, string): """ Returns a translatable string with the self.tr() function. """ return QCoreApplication.translate("Processing", string) @property def current(self): return self._current @property def current_defined(self): return self._current is not None def __call__(self, *args, **kwargs): return self._define(*args, **kwargs) def _initnew(self): self._current = AlgWrapper() def _pop(self): self.instances.append(self.current) self._current = None def _define(self, *args, **kwargs): self._initnew() self.current.define(*args, **kwargs) def dec(f): self.current.set_func(f) self.current.end() self._pop() return f return dec def output(self, type, *args, **kwargs): """ Define a output parameter for this algorithm using @alg.output. Apart from `type` this method will take all arguments and pass them though to the correct `QgsProcessingOutputDefinition ` type. Types: str: QgsProcessingOutputString int: QgsProcessingOutputNumber float: QgsProcessingOutputNumber alg.NUMBER: QgsProcessingOutputNumber alg.DISTANCE: QgsProcessingOutputNumber alg.INT: QgsProcessingOutputNumber alg.STRING: QgsProcessingOutputString alg.FILE: QgsProcessingOutputFile alg.FOLDER: QgsProcessingOutputFolder alg.HTML: QgsProcessingOutputHtml alg.LAYERDEF: QgsProcessingOutputLayerDefinition alg.MAPLAYER: QgsProcessingOutputMapLayer alg.MULTILAYER: QgsProcessingOutputMultipleLayers alg.RASTER_LAYER: QgsProcessingOutputRasterLayer alg.VECTOR_LAYER: QgsProcessingOutputVectorLayer alg.POINTCLOUD_LAYER: QgsProcessingOutputPointCloudLayer alg.BOOL: QgsProcessingOutputBoolean :param type: The type of the input. This should be a type define on `alg` like alg.STRING, alg.DISTANCE :keyword label: The label of the output. Will convert into `description` arg. :keyword parent: The string ID of the parent parameter. Parent parameter must be defined before its here. """ def dec(f): return f self.current.add_output(type, *args, **kwargs) return dec def help(self, helpstring, *args, **kwargs): """ Define the help for the algorithm using @alg.help. This method will be used instead of the doc string on the function as the help in the processing dialogs. :param helpstring: The help text. Use QCoreApplication.translate() for translation support. """ def dec(f): return f self.current.add_help(helpstring, *args, **kwargs) return dec def input(self, type, *args, **kwargs): """ Define a input parameter for this algorithm using @alg.input. Apart from `type` this method will take all arguments and pass them though to the correct `QgsProcessingParameterDefinition` type. Types: str: QgsProcessingParameterString int: QgsProcessingParameterNumber float: QgsProcessingParameterNumber bool: QgsProcessingParameterBoolean alg.NUMBER: QgsProcessingParameterNumber alg.INT: QgsProcessingParameterNumber alg.STRING: QgsProcessingParameterString alg.DISTANCE: QgsProcessingParameterDistance alg.SINK: QgsProcessingParameterFeatureSink alg.SOURCE: QgsProcessingParameterFeatureSource alg.FILE_DEST: QgsProcessingParameterFileDestination alg.FOLDER_DEST: QgsProcessingParameterFolderDestination alg.RASTER_LAYER: QgsProcessingParameterRasterLayer alg.RASTER_LAYER_DEST: QgsProcessingParameterRasterDestination alg.VECTOR_LAYER_DEST: QgsProcessingParameterVectorDestination alg.POINTCLOUD_LAYER_DEST: QgsProcessingParameterPointCloudDestination alg.BAND: QgsProcessingParameterBand alg.BOOL: QgsProcessingParameterBoolean alg.CRS: QgsProcessingParameterCrs alg.ENUM: QgsProcessingParameterEnum alg.EXPRESSION: QgsProcessingParameterExpression alg.EXTENT: QgsProcessingParameterExtent alg.FIELD: QgsProcessingParameterField alg.FILE: QgsProcessingParameterFile alg.MAPLAYER: QgsProcessingParameterMapLayer alg.MATRIX: QgsProcessingParameterMatrix alg.MULTILAYER: QgsProcessingParameterMultipleLayers alg.POINT: QgsProcessingParameterPoint alg.GEOMETRY: QgsProcessingParameterGeometry alg.RANGE: QgsProcessingParameterRange alg.VECTOR_LAYER: QgsProcessingParameterVectorLayer alg.AUTH_CFG: QgsProcessingParameterAuthConfig alg.MESH_LAYER: QgsProcessingParameterMeshLayer alg.SCALE: QgsProcessingParameterScale alg.LAYOUT: QgsProcessingParameterLayout alg.LAYOUT_ITEM: QgsProcessingParameterLayoutItem alg.COLOR: QgsProcessingParameterColor alg.DATETIME: QgsProcessingParameterDateTime(type=QgsProcessingParameterDateTime.Type.DateTime) alg.DATE: QgsProcessingParameterDateTime(type=QgsProcessingParameterDateTime.Type.Date) alg.TIME: QgsProcessingParameterDateTime(type=QgsProcessingParameterDateTime.Type.Time) alg.MAP_THEME: QgsProcessingParameterMapTheme alg.PROVIDER_CONNECTION: QgsProcessingParameterProviderConnection alg.DATABASE_SCHEMA: QgsProcessingParameterDatabaseSchema alg.DATABASE_TABLE: QgsProcessingParameterDatabaseTable alg.COORDINATE_OPERATION: QgsProcessingParameterCoordinateOperation alg.POINTCLOUD_LAYER: QgsProcessingParameterPointCloudLayer alg.ANNOTATION_LAYER: QgsProcessingParameterAnnotationLayer :param type: The type of the input. This should be a type define on `alg` like alg.STRING, alg.DISTANCE :keyword label: The label of the output. Translates into `description` arg. :keyword parent: The string ID of the parent parameter. Parent parameter must be defined before its here. :keyword default: The default value set for that parameter. Translates into `defaultValue` arg. """ def dec(f): return f self.current.add_input(type, *args, **kwargs) return dec input_type_mapping = { str: QgsProcessingParameterString, int: partial( QgsProcessingParameterNumber, type=QgsProcessingParameterNumber.Type.Integer ), float: partial( QgsProcessingParameterNumber, type=QgsProcessingParameterNumber.Type.Double ), bool: QgsProcessingParameterBoolean, ProcessingAlgFactory.NUMBER: partial( QgsProcessingParameterNumber, type=QgsProcessingParameterNumber.Type.Double ), ProcessingAlgFactory.INT: partial( QgsProcessingParameterNumber, type=QgsProcessingParameterNumber.Type.Integer ), ProcessingAlgFactory.STRING: QgsProcessingParameterString, ProcessingAlgFactory.DISTANCE: QgsProcessingParameterDistance, ProcessingAlgFactory.SINK: QgsProcessingParameterFeatureSink, ProcessingAlgFactory.SOURCE: QgsProcessingParameterFeatureSource, ProcessingAlgFactory.FILE_DEST: QgsProcessingParameterFileDestination, ProcessingAlgFactory.FOLDER_DEST: QgsProcessingParameterFolderDestination, ProcessingAlgFactory.RASTER_LAYER: QgsProcessingParameterRasterLayer, ProcessingAlgFactory.RASTER_LAYER_DEST: QgsProcessingParameterRasterDestination, ProcessingAlgFactory.VECTOR_LAYER_DEST: QgsProcessingParameterVectorDestination, ProcessingAlgFactory.POINTCLOUD_LAYER_DEST: QgsProcessingParameterPointCloudDestination, ProcessingAlgFactory.BAND: QgsProcessingParameterBand, ProcessingAlgFactory.BOOL: QgsProcessingParameterBoolean, ProcessingAlgFactory.CRS: QgsProcessingParameterCrs, ProcessingAlgFactory.ENUM: QgsProcessingParameterEnum, ProcessingAlgFactory.EXPRESSION: QgsProcessingParameterExpression, ProcessingAlgFactory.EXTENT: QgsProcessingParameterExtent, ProcessingAlgFactory.FIELD: QgsProcessingParameterField, ProcessingAlgFactory.FILE: QgsProcessingParameterFile, ProcessingAlgFactory.MAPLAYER: QgsProcessingParameterMapLayer, ProcessingAlgFactory.MATRIX: QgsProcessingParameterMatrix, ProcessingAlgFactory.MULTILAYER: QgsProcessingParameterMultipleLayers, ProcessingAlgFactory.POINT: QgsProcessingParameterPoint, ProcessingAlgFactory.GEOMETRY: QgsProcessingParameterGeometry, ProcessingAlgFactory.RANGE: QgsProcessingParameterRange, ProcessingAlgFactory.VECTOR_LAYER: QgsProcessingParameterVectorLayer, ProcessingAlgFactory.AUTH_CFG: QgsProcessingParameterAuthConfig, ProcessingAlgFactory.MESH_LAYER: QgsProcessingParameterMeshLayer, ProcessingAlgFactory.SCALE: QgsProcessingParameterScale, ProcessingAlgFactory.LAYOUT: QgsProcessingParameterLayout, ProcessingAlgFactory.LAYOUT_ITEM: QgsProcessingParameterLayoutItem, ProcessingAlgFactory.COLOR: QgsProcessingParameterColor, ProcessingAlgFactory.DATETIME: partial( QgsProcessingParameterDateTime, type=QgsProcessingParameterDateTime.Type.DateTime, ), ProcessingAlgFactory.DATE: partial( QgsProcessingParameterDateTime, type=QgsProcessingParameterDateTime.Type.Date ), ProcessingAlgFactory.TIME: partial( QgsProcessingParameterDateTime, type=QgsProcessingParameterDateTime.Type.Time ), ProcessingAlgFactory.MAP_THEME: QgsProcessingParameterMapTheme, ProcessingAlgFactory.PROVIDER_CONNECTION: QgsProcessingParameterProviderConnection, ProcessingAlgFactory.DATABASE_SCHEMA: QgsProcessingParameterDatabaseSchema, ProcessingAlgFactory.DATABASE_TABLE: QgsProcessingParameterDatabaseTable, ProcessingAlgFactory.COORDINATE_OPERATION: QgsProcessingParameterCoordinateOperation, ProcessingAlgFactory.POINTCLOUD_LAYER: QgsProcessingParameterPointCloudLayer, ProcessingAlgFactory.ANNOTATION_LAYER: QgsProcessingParameterAnnotationLayer, } output_type_mapping = { str: partial(_make_output, cls=QgsProcessingOutputString), int: partial(_make_output, cls=QgsProcessingOutputNumber), float: partial(_make_output, cls=QgsProcessingOutputNumber), ProcessingAlgFactory.NUMBER: partial(_make_output, cls=QgsProcessingOutputNumber), ProcessingAlgFactory.DISTANCE: partial(_make_output, cls=QgsProcessingOutputNumber), ProcessingAlgFactory.INT: partial(_make_output, cls=QgsProcessingOutputNumber), ProcessingAlgFactory.STRING: partial(_make_output, cls=QgsProcessingOutputString), ProcessingAlgFactory.FILE: partial(_make_output, cls=QgsProcessingOutputFile), ProcessingAlgFactory.FOLDER: partial(_make_output, cls=QgsProcessingOutputFolder), ProcessingAlgFactory.HTML: partial(_make_output, cls=QgsProcessingOutputHtml), ProcessingAlgFactory.LAYERDEF: partial( _make_output, cls=QgsProcessingOutputLayerDefinition ), ProcessingAlgFactory.MAPLAYER: partial( _make_output, cls=QgsProcessingOutputMapLayer ), ProcessingAlgFactory.MULTILAYER: partial( _make_output, cls=QgsProcessingOutputMultipleLayers ), ProcessingAlgFactory.RASTER_LAYER: partial( _make_output, cls=QgsProcessingOutputRasterLayer ), ProcessingAlgFactory.VECTOR_LAYER: partial( _make_output, cls=QgsProcessingOutputVectorLayer ), ProcessingAlgFactory.POINTCLOUD_LAYER: partial( _make_output, cls=QgsProcessingOutputPointCloudLayer ), ProcessingAlgFactory.BOOL: partial(_make_output, cls=QgsProcessingOutputBoolean), }