Correct way to specify Inputs and Outputs for QGIS processing scripts that use custom widgets?

Question

Processing scripts do not allow parameter inputs to be changed by other parameter inputs. This can be overcome by defining a custom widget which does have the desired interactions, then using a 'widget_wrapper' to add the custom widget to the processing parameter inputs. The source for this can be found in the link: QGIS processing parameter using values from layer feature in processing plugin and How to chose a Vector layer, chose a field, then chose values using parameterAsEnums

def initAlgorithm(self, config=None):
    file_param = QgsProcessingParameterMatrix(self.INPUT_PARAMS, 'Input Parameters')
    file_param.setMetadata({'widget_wrapper': {'class': CustomParametersWidget}})
    self.addParameter(file_param)

The perameters are prased by using 'parameterAsMatrix' input_params_list = self.parameterAsMatrix(parameters, self.INPUT_PARAMS, context)

However I have not found any way to define the parameters the way they would normaly be defined in a processing algorithim, eg:

class MergeEMFandGPS(QgsProcessingAlgorithm):
    INPUT_PARAMS = 'INPUT_PARAMS'
    WORKDIR = 'WORKDIR'
    FILEEMF = 'FILEEMF'
    FILEGPS = 'FILEGPS'
    OUTPUT = 'OUTPUT'

Then it is tied to the parameter in the initAlortihim then prased in the process processAlgorithm

def initAlgorithm(self, config=None):
    self.addParameter(
        QgsProcessingParameterFeatureSource(
            self.INPUT,
            self.tr('Input layer'),
            [QgsProcessing.TypeVectorAnyGeometry]
        )
    )

def processAlgorithm(self, parameters, context, feedback):
    source = self.parameterAsSource(
        parameters,
        self.INPUT,
        context
     )

Is there away to tie the inputs from the parameterMatrix to the individual parameter constants? Is it necessary for other functionality?

The full code I am using is below:

from processing.gui.wrappers import WidgetWrapper # for custom widgets

from qgis.PyQt.QtWidgets import (QLabel,
                                QWidget,
                                QGridLayout)
                                
from qgis.gui import QgsFileWidget

from qgis.PyQt.QtCore import (QCoreApplication,
                                QVariant,
                                Qt)

from qgis.core import (QgsProcessing,
                       QgsFeatureSink,
                       QgsProcessingException,
                       QgsProcessingAlgorithm,
                       QgsProcessingParameterFeatureSource,
                       QgsProcessingParameterFeatureSink,
                       QgsProcessingParameterFile,
                       QgsProcessingParameterFileDestination,
                       QgsProcessingParameterDefinition,
                       QgsProcessingParameterMatrix)

from qgis import processing

import os

# Widget Wrapper class
class CustomParametersWidget(WidgetWrapper):
    '''
    Custom widgets so file inputs and outputs can be populated
    using the selected woring direcetory
    '''
    
    def createWidget(self):
        self.cpw = CustomFileSelectWidget()
        return self.cpw
        
    def value(self):
        # This method gets the parameter values and returns them in a list...
        # which will be retrieved and parsed in the processAlgorithm() method
        self.Wdir = self.cpw.getWdir()
        self.Femf = self.cpw.getFemf()
        self.Fgps = self.cpw.getFgps()
        self.Fgmf = self.cpw.getFgmf()
        return [self.Wdir, self.Femf, self.Fgps, self.Fgmf]

# Custom Widget class
class CustomFileSelectWidget(QWidget):
    
    def __init__(self):
        super(CustomFileSelectWidget, self).__init__()
        self.Wdir_lbl = QLabel('Select Working Directory', self)
        self.Wdir_sel = QgsFileWidget(self)
        self.Wdir_sel.setDialogTitle('Select Working Directory')
        self.Wdir_sel.setStorageMode(QgsFileWidget.StorageMode.GetDirectory)
        
        self.Femf_lbl = QLabel('Select EMF File', self)
        self.Femf_sel = QgsFileWidget(self)
        self.Femf_sel.setDialogTitle('Select EMF File')
        self.Femf_sel.setStorageMode(QgsFileWidget.StorageMode.GetFile)
        self.Femf_sel.setFilter('emfhistory*.csv')
        self.Femf_sel.setFileWidgetButtonVisible(False)
        
        self.Fgps_lbl = QLabel('Select GPS File', self)
        self.Fgps_sel = QgsFileWidget(self)
        self.Fgps_sel.setStorageMode(QgsFileWidget.StorageMode.GetFile)
        self.Fgps_sel.setFilter('*.csv')
        self.Fgps_sel.setFileWidgetButtonVisible(False)
        
        self.Fgmf_lbl = QLabel('Select Output File', self)
        self.Fgmf_sel = QgsFileWidget(self)
        self.Fgmf_sel.setStorageMode(QgsFileWidget.StorageMode.GetFile)
        self.Fgmf_sel.setFilter('*.csv')
        self.Fgmf_sel.setFileWidgetButtonVisible(False)
        
        # Define the layout
        self.layout = QGridLayout(self)
        self.layout.addWidget(self.Wdir_lbl, 0, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Wdir_sel, 0, 1, 1, 2)
        self.layout.addWidget(self.Femf_lbl, 1, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Femf_sel, 1, 1, 1, 2)
        self.layout.addWidget(self.Fgps_lbl, 2, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Fgps_sel, 2, 1, 1, 2)
        self.layout.addWidget(self.Fgmf_lbl, 3, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Fgmf_sel, 3, 1, 1, 2)
        
        # whenever the working directory is changed, repopulate the input files
        self.Wdir_sel.fileChanged.connect(self.populateInputFiles)
        # update the output files when the GPS or EMF file is changed.
        self.Femf_sel.fileChanged.connect(self.updateOutputFile, Qt.QueuedConnection)
        self.Fgps_sel.fileChanged.connect(self.updateOutputFile, Qt.QueuedConnection)
    
    # populates the input and outut files from working directory
    def populateInputFiles(self):
        folderPath = self.Wdir_sel.filePath()
        for file in os.listdir(folderPath):
            if file.endswith(".csv"):
                if 'emfhistory' in file:
                    self.Femf_sel.setFilePath(os.path.join(folderPath, file))
                    self.Femf_sel.setFileWidgetButtonVisible(True)
                    print("EMF :", file)
                elif 'gmf' not in file:
                    print("GPS :", file)
                    self.Fgps_sel.setFilePath(os.path.join(folderPath, file))
                    self.Fgps_sel.setFileWidgetButtonVisible(True)

    def updateOutputFile(self):
        self.Fgmf_sel.setFileWidgetButtonVisible(True)
        if self.Fgps_sel.filePath() == '':
            print('Select GPS file')
        if self.Femf_sel.filePath() == '':
            print('Select EMF file')
        if self.Fgps_sel.filePath() != '' and self.Femf_sel.filePath() != '':
            folderPath = self.Wdir_sel.filePath()
            fileNameStart = 'gmf_'+ os.path.basename(self.Fgps_sel.filePath())[:10] + '_'
            fileNameEnd = os.path.basename(self.Femf_sel.filePath()).split('_')[3]
            fileName = fileNameStart + fileNameEnd
            self.Fgmf_sel.setFilePath(os.path.join(folderPath, fileName))
            print ('update File Name : ', fileName)
        
    def getWdir(self):
        return self.Wdir_sel.filePath()
        
    def getFemf(self):
        return self.Femf_sel.filePath()
        
    def getFgps(self):
        return self.Fgps_sel.filePath()
        
    def getFgmf(self):
        return self.Fgmf_sel.filePath()

class MergeEMFandGPS(QgsProcessingAlgorithm):
    """
    All Processing algorithms should extend the QgsProcessingAlgorithm
    class.
    """

    # Constants used to refer to parameters and outputs. They will be
    # used when calling the algorithm from another algorithm, or when
    # calling from the QGIS console.
    
    INPUT = 'INPUT'
    INPUT_PARAMS = 'INPUT_PARAMS'
    WORKDIR = 'WORKDIR'
    FILEEMF = 'FILEEMF'
    FILEGPS = 'FILEGPS'
    OUTPUT = 'OUTPUT'

    def createInstance(self):
        return MergeEMFandGPS()

    def initAlgorithm(self, config=None):
        """
        Here we define the inputs and output of the algorithm, along
        with some other properties.
        """
        
        file_param = QgsProcessingParameterMatrix(self.INPUT_PARAMS, 'Input Parameters')
        file_param.setMetadata({'widget_wrapper': {'class': CustomParametersWidget}})
        self.addParameter(file_param)

    def processAlgorithm(self, parameters, context, feedback):
        """
        Here is where the processing itself takes place.
        """
        
        # Retrieve the list of parameters returned by the custom widget wrapper
        input_params_list = self.parameterAsMatrix(parameters, self.INPUT_PARAMS, context)
        
        WorkDIR = input_params_list[0]
        FileEMF = input_params_list[1]
        FileGPS = input_params_list[2]
        FileOUT = input_params_list[3]

        #do something with the FileEMF and FileGPS files and create a new file FileOUT
        return {'OUTPUT': FileOUT}

Answer 1

To be honest, I'm not sure what problem you are having (I don't really see any problem here) or why you think your approach is not correct, but the correct way to specify inputs and outputs is simply to add the parameters you are going to use in the algorithm.

These constants you refer to here:

INPUT = 'INPUT'
INPUT_PARAMS = 'INPUT_PARAMS'
WORKDIR = 'WORKDIR'
FILEEMF = 'FILEEMF'
FILEGPS = 'FILEGPS'
OUTPUT = 'OUTPUT'

are just string literals which are used as keys in the input parameters dictionary. There is nothing special about them. The only thing which is necessary is to ‘tie’ the parameters which are added to the algorithm to the method which retrieves and parses them in the processAlgorithm() method. In your example you are only using one input parameter, a QgsProcessingParameterMatrix, which will be retrieved as a list and which should contain 4 strings. These strings should also be valid file paths.

Your code example is also missing a couple of required overridden methods. If you incorporate these and add the algorithm to the processing toolbox:

def name(self):
    # Mandatory
    return "mergeemfandgps"
     
def displayName(self):
    # Mandatory
    return "Merge EMF and GPS"

def group(self):
    return "Examples"

def groupId(self):
    return "examples"

def shortHelpString(self):
    return "Example script with custom widgets"

You can call this algorithm from the Python console like this and it works perfectly fine:

params = {'INPUT_PARAMS':['/home/ben/Test/CSVs',
                            '/home/ben/Test/CSVs/emfhistory_test.csv',
                            '/home/ben/Test/CSVs/Darwin_area_records.csv',
                            '/home/ben/Test/CSVs/gmf_Darwin_are_test.csv']}
                            
results = processing.run("script:mergeemfandgps", params)

print(results['OUTPUT'])

And you call it from another algorithm in the same way as long as you set is_child_algorithm=True.

Of course, if you pass an incorrect list item inside the input parameter list you will get an error (like any other algorithm). To deal with this, I would suggest that you could override the checkParameterValues() and run a series of checks, returning False and a helpful error message which will be shown if any of the input list items cannot be parsed into a valid file path. If this method returns False, the algorithm will not proceed to run. E.g.

def checkParameterValues(self, parameters, context):
    """
    Check the input parameter values and return a helpful message.
    This method is called before processAlgorithm and if it returns False,
    the algorithm will not proceed.
    """
    input_params_list = self.parameterAsMatrix(parameters, self.INPUT_PARAMS, context)
    input_params_count = len(input_params_list)
    if not isinstance(input_params_list, list):
        return False, 'Incorrect value for INPUT_PARAMS- expected a list'
    if input_params_count != 4:
        if input_params_count < 4:
            return False, f'Not enough parameters- 4 expected, but received {input_params_count}'
        return False, f'Too many parameters- 4 expected, but received {input_params_count}'
    if not [isinstance(input_params_list[0], str) and isinstance(input_params_list[1], str) and isinstance(input_params_list[2], str) and isinstance(input_params_list[3], str)][0]:
        return False, 'Input parameters must be strings'
    if not [os.path.exists(input_params_list[0]) and os.path.exists(input_params_list[1]) and os.path.exists(input_params_list[2]) and os.path.exists(input_params_list[3])][0]:
        return False, 'One or more input parameters is not a valid file path'
    # After adding our overridden implementation we should also check
    # and return the base class implementation
    return super().checkParameterValues(parameters, context)

Complete example using your script:

from processing.gui.wrappers import WidgetWrapper # for custom widgets

from qgis.PyQt.QtWidgets import (QLabel,
                                QWidget,
                                QGridLayout)
                                
from qgis.gui import QgsFileWidget

from qgis.PyQt.QtCore import (QCoreApplication,
                                QVariant,
                                Qt)

from qgis.core import (QgsProcessing,
                       QgsFeatureSink,
                       QgsProcessingException,
                       QgsProcessingAlgorithm,
                       QgsProcessingParameterFeatureSource,
                       QgsProcessingParameterFeatureSink,
                       QgsProcessingParameterFile,
                       QgsProcessingParameterFileDestination,
                       QgsProcessingParameterDefinition,
                       QgsProcessingParameterMatrix)

from qgis import processing

import os

# Widget Wrapper class
class CustomParametersWidget(WidgetWrapper):
    '''
    Custom widgets so file inputs and outputs can be populated
    using the selected woring direcetory
    '''
    
    def createWidget(self):
        self.cpw = CustomFileSelectWidget()
        return self.cpw
        
    def value(self):
        # This method gets the parameter values and returns them in a list...
        # which will be retrieved and parsed in the processAlgorithm() method
        self.Wdir = self.cpw.getWdir()
        self.Femf = self.cpw.getFemf()
        self.Fgps = self.cpw.getFgps()
        self.Fgmf = self.cpw.getFgmf()
        return [self.Wdir, self.Femf, self.Fgps, self.Fgmf]

# Custom Widget class
class CustomFileSelectWidget(QWidget):
    
    def __init__(self):
        super(CustomFileSelectWidget, self).__init__()
        self.Wdir_lbl = QLabel('Select Working Directory', self)
        self.Wdir_sel = QgsFileWidget(self)
        self.Wdir_sel.setDialogTitle('Select Working Directory')
        self.Wdir_sel.setStorageMode(QgsFileWidget.StorageMode.GetDirectory)
        
        self.Femf_lbl = QLabel('Select EMF File', self)
        self.Femf_sel = QgsFileWidget(self)
        self.Femf_sel.setDialogTitle('Select EMF File')
        self.Femf_sel.setStorageMode(QgsFileWidget.StorageMode.GetFile)
        self.Femf_sel.setFilter('emfhistory*.csv')
        self.Femf_sel.setFileWidgetButtonVisible(False)
        
        self.Fgps_lbl = QLabel('Select GPS File', self)
        self.Fgps_sel = QgsFileWidget(self)
        self.Fgps_sel.setStorageMode(QgsFileWidget.StorageMode.GetFile)
        self.Fgps_sel.setFilter('*.csv')
        self.Fgps_sel.setFileWidgetButtonVisible(False)
        
        self.Fgmf_lbl = QLabel('Select Output File', self)
        self.Fgmf_sel = QgsFileWidget(self)
        self.Fgmf_sel.setStorageMode(QgsFileWidget.StorageMode.GetFile)
        self.Fgmf_sel.setFilter('*.csv')
        self.Fgmf_sel.setFileWidgetButtonVisible(False)
        
        # Define the layout
        self.layout = QGridLayout(self)
        self.layout.addWidget(self.Wdir_lbl, 0, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Wdir_sel, 0, 1, 1, 2)
        self.layout.addWidget(self.Femf_lbl, 1, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Femf_sel, 1, 1, 1, 2)
        self.layout.addWidget(self.Fgps_lbl, 2, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Fgps_sel, 2, 1, 1, 2)
        self.layout.addWidget(self.Fgmf_lbl, 3, 0, 1, 1, Qt.AlignRight)
        self.layout.addWidget(self.Fgmf_sel, 3, 1, 1, 2)
        
        # whenever the working directory is changed, repopulate the input files
        self.Wdir_sel.fileChanged.connect(self.populateInputFiles)
        # update the output files when the GPS or EMF file is changed.
        self.Femf_sel.fileChanged.connect(self.updateOutputFile, Qt.QueuedConnection)
        self.Fgps_sel.fileChanged.connect(self.updateOutputFile, Qt.QueuedConnection)
    
    # populates the input and outut files from working directory
    def populateInputFiles(self):
        folderPath = self.Wdir_sel.filePath()
        for file in os.listdir(folderPath):
            if file.endswith(".csv"):
                if 'emfhistory' in file:
                    self.Femf_sel.setFilePath(os.path.join(folderPath, file))
                    self.Femf_sel.setFileWidgetButtonVisible(True)
                    print("EMF :", file)
                elif 'gmf' not in file:
                    print("GPS :", file)
                    self.Fgps_sel.setFilePath(os.path.join(folderPath, file))
                    self.Fgps_sel.setFileWidgetButtonVisible(True)

    def updateOutputFile(self):
        self.Fgmf_sel.setFileWidgetButtonVisible(True)
        if self.Fgps_sel.filePath() == '':
            print('Select GPS file')
        if self.Femf_sel.filePath() == '':
            print('Select EMF file')
        if self.Fgps_sel.filePath() != '' and self.Femf_sel.filePath() != '':
            folderPath = self.Wdir_sel.filePath()
            fileNameStart = 'gmf_'+ os.path.basename(self.Fgps_sel.filePath())[:10] + '_'
            fileNameEnd = os.path.basename(self.Femf_sel.filePath()).split('_')[3]
            fileName = fileNameStart + fileNameEnd
            self.Fgmf_sel.setFilePath(os.path.join(folderPath, fileName))
            print ('update File Name : ', fileName)
        
    def getWdir(self):
        return self.Wdir_sel.filePath()
        
    def getFemf(self):
        return self.Femf_sel.filePath()
        
    def getFgps(self):
        return self.Fgps_sel.filePath()
        
    def getFgmf(self):
        return self.Fgmf_sel.filePath()

class MergeEMFandGPS(QgsProcessingAlgorithm):
    """
    All Processing algorithms should extend the QgsProcessingAlgorithm
    class.
    """

    # Constants used to refer to parameters and outputs. They will be
    # used when calling the algorithm from another algorithm, or when
    # calling from the QGIS console.
    
    INPUT_PARAMS = 'INPUT_PARAMS'
    OUTPUT = 'OUTPUT'
    
    def __init__(self):
        super().__init__()
 
    def name(self):
        return "mergeemfandgps"
         
    def displayName(self):
        return "Merge EMF and GPS"
 
    def group(self):
        return "Examples"
 
    def groupId(self):
        return "examples"
 
    def shortHelpString(self):
        return "Example script with custom widgets"
 
    def helpUrl(self):
        return "https://qgis.org"

    def createInstance(self):
        return MergeEMFandGPS()
        
    def checkParameterValues(self, parameters, context):
        """
        Check the input parameter values and return a helpful message.
        This method is called before processAlgorithm and if it returns False,
        the algorithm will not proceed.
        """
        input_params_list = self.parameterAsMatrix(parameters, self.INPUT_PARAMS, context)
        input_params_count = len(input_params_list)
        if not isinstance(input_params_list, list):
            return False, 'Incorrect value for INPUT_PARAMS- expected a list'
        if input_params_count != 4:
            if input_params_count < 4:
                return False, f'Not enough parameters- 4 expected, but received {input_params_count}'
            return False, f'Too many parameters- 4 expected, but received {input_params_count}'
        if not [isinstance(input_params_list[0], str) and isinstance(input_params_list[1], str) and isinstance(input_params_list[2], str) and isinstance(input_params_list[3], str)][0]:
            return False, 'Input parameters must be strings'
        if not [os.path.exists(input_params_list[0]) and os.path.exists(input_params_list[1]) and os.path.exists(input_params_list[2]) and os.path.exists(input_params_list[3])][0]:
            return False, 'One or more input parameters is not a valid file path'
        # After adding our overridden implementation we should also check
        # and return the base class implementation
        return super().checkParameterValues(parameters, context)

    def initAlgorithm(self, config=None):
        """
        Here we define the inputs and output of the algorithm, along
        with some other properties.
        """
        
        file_param = QgsProcessingParameterMatrix(self.INPUT_PARAMS, 'Input Parameters')
        file_param.setMetadata({'widget_wrapper': {'class': CustomParametersWidget}})
        self.addParameter(file_param)

    def processAlgorithm(self, parameters, context, feedback):
        """
        Here is where the processing itself takes place.
        """
        
        # Retrieve the list of parameters returned by the custom widget wrapper
        input_params_list = self.parameterAsMatrix(parameters, self.INPUT_PARAMS, context)
        
        WorkDIR = input_params_list[0]
        FileEMF = input_params_list[1]
        FileGPS = input_params_list[2]
        FileOUT = input_params_list[3]

        #do something with the FileEMF and FileGPS files and create a new file FileOUT
        return {'OUTPUT': FileOUT}