In addition to not setting is_child_algorithm=True
, you are not retrieving your buffer distance parameter correctly. I have also added a few other improvements and simplifications- you do not need to manually add output features to feature sink, you can just use a QgsProcessingParameterVectorDestination
and set it as the 'OUTPUT'
for your final algorithm. You can also use QgsProcessingMultiStepFeedback
so you get a nicely updating progress bar for the overall algorithm.
Try this modified script:
from PyQt5.QtCore import QCoreApplication, QVariant
from qgis.core import (QgsProcessing,
QgsProcessingException,
QgsProcessingAlgorithm,
QgsProcessingParameterFeatureSource,
QgsProcessingParameterVectorDestination,
QgsProcessingParameterNumber,
QgsProcessingParameterRasterLayer,
QgsProcessingMultiStepFeedback)
import processing
class ExampleProcessingAlgorithm(QgsProcessingAlgorithm):
CENTROIDS = 'CENTROIDS'
INPUT_RASTER_LAND_COVER = 'INPUT_RASTER_LAND_COVER'
VEG_BUFFER_DIST = 'VEG_BUFFER_DIST'
VUN_OUTPUT = 'VUN_OUTPUT'
def tr(self, string):
return QCoreApplication.translate('Processing', string)
def createInstance(self):
return ExampleProcessingAlgorithm()
def name(self):
return 'myscript'
def displayName(self):
return self.tr('My Script')
def group(self):
return self.tr('Example scripts')
def groupId(self):
return 'examplescripts'
def shortHelpString(self):
return self.tr("Example algorithm short description")
def initAlgorithm(self, config=None):
self.addParameter(
QgsProcessingParameterRasterLayer(
self.INPUT_RASTER_LAND_COVER,
self.tr("Input a raster with land cover as distinct bands"),
[QgsProcessing.TypeRaster]
)
)
self.addParameter(
QgsProcessingParameterFeatureSource(
self.CENTROIDS,
self.tr('Building centroids'),
[QgsProcessing.TypeVectorPoint]
)
)
self.addParameter(
QgsProcessingParameterNumber(
self.VEG_BUFFER_DIST,
self.tr('Buffer distance around each property to assess vegetation cover (meters)')
)
)
self.addParameter(
QgsProcessingParameterVectorDestination(
self.VUN_OUTPUT,
self.tr('Vulnerability Output')
)
)
def processAlgorithm(self, parameters, context, model_feedback):
#each source layer is named, it is a two step process
landcover = self.parameterAsRasterLayer(
parameters,
self.INPUT_RASTER_LAND_COVER,
context
)
#if the input is invalid an error is raised
if landcover is None:
raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT_RASTER_LAND_COVER))
#same as above, but with a vector not raster
centroids = self.parameterAsVectorLayer(
parameters,
self.CENTROIDS,
context
)
if centroids is None:
raise QgsProcessingException(self.invalidSourceError(parameters, self.BUILDING_CENTROIDS))
buffer_distance = self.parameterAsInt(parameters,
self.VEG_BUFFER_DIST,
context)
steps = 5
feedback = QgsProcessingMultiStepFeedback(steps, model_feedback)
step = 1
results = {}
outputs = {}
feedback.setCurrentStep(step)
step+=1
outputs['buffer_result'] = processing.run("native:buffer", {
'INPUT': centroids,
'DISTANCE': buffer_distance,
'SEGMENTS': 5,
'END_CAP_STYLE': 0,
'JOIN_STYLE': 0,
'MITER_LIMIT': 2,
'DISSOLVE': False,
'OUTPUT': 'TEMPORARY_OUTPUT'
}, context=context, feedback=feedback, is_child_algorithm=True)
results['buffer_result'] = outputs['buffer_result']['OUTPUT']
feedback.setCurrentStep(step)
step+=1
outputs['zonal_statistics_result'] = processing.run("native:zonalstatisticsfb", {
'INPUT': results['buffer_result'],
'INPUT_RASTER': landcover,
'RASTER_BAND':1,
'COLUMN_PREFIX':'_',
'STATISTICS':[0,1,2],
'OUTPUT':'TEMPORARY_OUTPUT'
},
context=context, feedback=feedback, is_child_algorithm=True)
results['zonal_statistics_result'] = outputs['zonal_statistics_result']['OUTPUT']
feedback.setCurrentStep(step)
step+=1
outputs['join_1'] = processing.run("native:joinattributestable", {
'INPUT':results['buffer_result'],
'FIELD':'OBJECTID',
'INPUT_2': results['zonal_statistics_result'],
'FIELD_2':'OBJECTID',
'FIELDS_TO_COPY':[],
'METHOD':1,
'DISCARD_NONMATCHING':False,
'PREFIX':'',
'OUTPUT':'TEMPORARY_OUTPUT'
}, context=context, feedback=feedback, is_child_algorithm=True)
results['join_1'] = outputs['join_1']['OUTPUT']
feedback.setCurrentStep(step)
step+=1
#drop geometry of the multipolygons, so fields can be attached to point layer
outputs['dropped_geoms'] = processing.run("native:dropgeometries", {
'INPUT': results['join_1'],
'OUTPUT':'TEMPORARY_OUTPUT'
},
context=context, feedback=feedback, is_child_algorithm=True)
results['dropped_geoms'] = outputs['dropped_geoms']['OUTPUT']
feedback.setCurrentStep(step)
step+=1
#attach fields without geometry to point layer
outputs['join_2'] = processing.run("native:joinattributestable", {
'INPUT':centroids,
'FIELD':'OBJECTID',
'INPUT_2': results['dropped_geoms'],
'FIELD_2':'OBJECTID',
'FIELDS_TO_COPY':['_count'],
'METHOD':1,
'DISCARD_NONMATCHING':False,
'PREFIX':'',
'OUTPUT':parameters[self.VUN_OUTPUT]
}, context=context, feedback=feedback, is_child_algorithm=True)
results['join_2'] = outputs['join_2']['OUTPUT']
return results
You can create an intermediate layer with the filtered features, using native:extractbyexpression
(and without a feature sink at any stage):
class myTestAlg(QgsProcessingAlgorithm):
INPUT = 'INPUT'
NRD_INPUT = 'NRD_INPUT'
STRING_1 = 'STRING_1'
STRING_2 = 'STRING_2'
FIELD_1 = 'FIELD_1'
FIELD_2 = 'FIELD_2'
OUTPUT = 'OUTPUT'
OUTPUT_2 = 'OUTPUT_2'
def createInstance(self):
return myTestAlg()
def name(self):
return 'test'
def displayName(self):
return 'Test'
def initAlgorithm(self, config=None):
# mastermap_parameter
self.addParameter(QgsProcessingParameterFeatureSource(
self.INPUT,
self.tr('\'Area\' data from OS Mastermap'),
types=[QgsProcessing.TypeVectorAnyGeometry],
defaultValue='OSMM_AREA')
)
# string1_parameter
self.addParameter(QgsProcessingParameterString(
self.STRING_1,
self.tr('Select features with the following value...'),
'Buildings',
multiLine = False,
optional = False)
)
# field1_parameter
self.addParameter(QgsProcessingParameterField(
self.FIELD_1,
self.tr('...in the following attribute field.'),
'theme',
self.INPUT)
)
# string2_parameter
self.addParameter(QgsProcessingParameterString(
self.STRING_2,
self.tr('Additionally, select features with the following value...'),
'Building',
multiLine = False,
optional = True)
)
#field2_parameter
self.addParameter(QgsProcessingParameterField(
self.FIELD_2,
self.tr('...in the following attribute field.'),
'descriptivegroup',
self.INPUT,
QgsProcessingParameterField.DataType.Any,
allowMultiple = False,
optional = True)
)
self.addParameter(QgsProcessingParameterFeatureSource(
self.NRD_INPUT,
self.tr('Data from National Receptor Dataset'),
[QgsProcessing.TypeVectorAnyGeometry])
)
self.addParameter(QgsProcessingParameterFeatureSink(
self.OUTPUT,
self.tr('Output data'))
)
self.addParameter(QgsProcessingParameterFeatureSink(
self.OUTPUT_2,
self.tr('OS Mastermap buildings without NRD'))
)
def processAlgorithm(self, parameters, context, feedback):
results = {}
outputs = {}
# Extract by expression
field_1 = self.parameterAsString(parameters, 'FIELD_1', context)
field_2 = self.parameterAsString(parameters, 'FIELD_2', context)
alg_params = {
'EXPRESSION': f'attribute( "{field_1}") = \'{parameters["STRING_1"]}\' OR attribute( "{field_2}") = \'{parameters["STRING_2"]}\' ',
'INPUT': parameters[self.INPUT],
'OUTPUT': QgsProcessing.TEMPORARY_OUTPUT
}
mm_building_layer = processing.run('native:extractbyexpression', alg_params, context=context, feedback=feedback, is_child_algorithm=True)['OUTPUT']
outputs['JoinAttributesByLocation'] = processing.run("native:joinattributesbylocation", {
'DISCARD_NONMATCHING' : True,
'INPUT' : mm_building_layer,
'JOIN' : parameters['NRD_INPUT'],
'JOIN_FIELDS' : [],
'METHOD' : 0,
'NON_MATCHING' : parameters['OUTPUT_2'],
'OUTPUT' : parameters['OUTPUT'],
'PREDICATE' : [0],
'PREFIX' : ''
},
is_child_algorithm=True,
context=context,
feedback=feedback)
results[self.OUTPUT] = outputs['JoinAttributesByLocation']['OUTPUT']
results[self.OUTPUT_2] = outputs['JoinAttributesByLocation']['NON_MATCHING']
return results
You could also filter features using getFeatures()
(with an expression) or native:extractbyattribute
(with a single attribute).
Best Answer
Try this: