TD 3 - Utiliser et créer des algorithmes processing¶

Tout les algorithmes de la boîte à outils de traitement qui sont mis à dispositions nativement par QGIS ou qui proviennent des extensions installées sont utilisables avec l'API pyQGIS. Dans ce TD nous allons voir comment les utiliser et surtout comment on peut en créer de nouveaux, spécifiques à nos besoins.

1. Utiliser des algorithmes processing¶

Cours

Le module processing est un module Python intégré à QGIS qui permet de lancer des algorithmes de la boîte à outils de traitement. Pour lancer un algorithme on utilise la méthode run() du module processing avec l'identifiant de l'algorithme que l'on souhaite lancer et ses paramètres. L'identifiant est une chaîne de caractère qui a la syntaxe "{provider_id}:{algorithm_id}". Les paramètres sont stockés dans un dictionnaire. La sortie de la commande run() est un dictionnaire contenant les résultats de l'algorithme lancé.

>>> processing.run("native:buffer",
    {'INPUT':'layer',
    'DISTANCE':10,'SEGMENTS':5,
    'END_CAP_STYLE':0,
    'JOIN_STYLE':0,
    'MITER_LIMIT':2,
    'DISSOLVE':False,
    'SEPARATE_DISJOINT':False,
    'OUTPUT':'TEMPORARY_OUTPUT'})

{'OUTPUT': <QgsVectorLayer: 'Mis en tampon' (memory)>}

Pour obtenir l'identifiant et les paramètres de l'algorithme, ouvrez l’algorithme depuis la boîte à outil de traitement puis cliquez sur le bouton 'Aide'. Cela ouvrira la page de la documentation qui correspond à l'algorithme en question avec l'identifiant et les paramètres. Vous pouvez également cliquer sur « Avancé » → « copier en tant que command Python ».

Le paramètre "OUTPUT" est souvent utilisé pour donner la destination des couches résultantes de l'algorithme. On peut choisir la destination selon si l'on souhaite conserver cette couche ou si c'est une couche temporaire (un résultat intermédiaire):

"OUTPUT" : "TEMPORARY_OUTPUT": QGIS décide automatiquement si la couche est conservée uniquement en mémoire (pas de fichier écrit) ou écrit dans le dossier temporaire de votre ordinateur (qui est automatiquement vidé lorsque vous éteignez votre ordinateur) en fonction du type et de la taille de la couche.
"OUTPUT" : "/path/to/file.ext": La couche sera sauvegardé au chemin spécifié.

Créez un projet QGIS et sauvegardez-le à l'emplacement de votre choix sous le nom td3.qgz. Ouvrez-le et chargez les couches « RPG » (fichier rpg.gpkg) et « indices » (fichier indices.gpkg).

1.1 Lancer un algorithme processing¶

Calculez la moyenne de NDVI par polygone du RPG en utilisant l'algorithme « Statistiques de zones » avec processing. Ajoutez la couche de résultats au projet QGIS.

Solution

zonal_stat_outputs = processing.run("native:zonalstatisticsfb", {'INPUT':rpg,'INPUT_RASTER':indices,'RASTER_BAND':1,'COLUMN_PREFIX':'NDVI_','STATISTICS':[2],'OUTPUT':'TEMPORARY_OUTPUT'})
zonal_stat = zonal_stat_outputs["OUTPUT"]
QgsProject.instance().addMapLayer(zonal_stat)

1.2 Enchaîner des algorithmes `processing`¶

En utilisant des algorithmes processing appliquer une rotation de 30° sur les géométries de la couche 'RPG' puis transformez la couche résultante en raster avec une résolution de 5m et une valeur de 1 pour tout les pixels. Ajoutez cette couche au projet.

Tip

Les couches vectorielles résultantes sont majoritairement des QgsVectorLayer mais pour les couches raster ce sont des chemins vers des fichiers (car les couches raster sont souvent plus lourdes). Assurez-vous donc de recharger les résultats dans une QgsRasterLayer si besoin.

Solution

rotation_results = processing.run("native:rotatefeatures", {'INPUT':rpg,'ANGLE':30,'ANCHOR':None,'OUTPUT':'TEMPORARY_OUTPUT'})
rotated = rotation_results["OUTPUT"]
rasterized_results = processing.run("gdal:rasterize", {'INPUT': rotated, 'FIELD':'','BURN':5,'USE_Z':False,'UNITS':1,'WIDTH':5,'HEIGHT':5,'EXTENT':None,'NODATA':0,'CREATION_OPTIONS':None,'DATA_TYPE':5,'INIT':None,'INVERT':False,'EXTRA':'','OUTPUT':'TEMPORARY_OUTPUT'})
raster_path = rasterized_results["OUTPUT"]
raster = QgsRasterLayer(raster_path)
QgsProject.instance().addMapLayer(raster)

2. Créer des algorithmes `processing`¶

Maintenant que nous avons vu comment, avec pyQGIS, nous pouvons travailler à l'échelle d'une couche et de ses entités ainsi qu'avec des algorithmes déjà existants nous allons pouvoir créer nos propres algorithmes.

2.1 Structure d'un algorithme `processing`¶

Ajoutez le modèle fourni « RotationAndRasterize » (fichier modele_rotation_rasterize.model3) dans le modèle builder de QGIS, il s'agit des mêmes opération effectuées dans la partie 1.2. Cliquez sur « Modèle » → « Exporter » → « Exporter comme script Python » puis ouvrez le script obtenu. Inspectez le contenu du script au regard de l'encadré de cours ci-dessous.

Cours

Un algorithme processing est une classe qui hérite de la classe QgsProcessingAlgorithm et doit implémenter les méthodes suivantes:

initAlgorithm(): Définis les paramètres à utiliser dans l'algorithme.
processAlgorithm(): Le coeur algorithmique de l'algorithme.
name(): Le nom de l'algorithme.
display_name_(): Le nom de l'algorithme affiché dans le panneau de la boîte à outils de traitement.
group(): Le nom du groupe dans lequel l'algorithme se trouve dans la boîte à outil de traitement.
groupId(): L'identifiant du groupe.
createInstance(): Là méthode qui crée l'algorithme lorsqu'il est appelé. cette méthode doit rester inchangée.

Les paramètres crées avec la méthode addParameter() dans initParameter() seront accessibles ensuite dans processAlgorithm() via leur identifiants en utilisant le dictionnaire de paramètres donné en entrée de la méthode. Il faut cependant récupérer ces paramètres avec une méthode spéciale qui correspond au type de variable qu'on veut extraire et qui a la syntaxe suivante parameterAs{VariableType}().

def initAlgorithm(self, config: Optional[dict[str, Any]] = None):
    self.addParameter(
        QgsProcessingParameterNumber(
            'angle', # 'angle' est l'identifiant du paramètre
            'Angle', # 'Angle' le nom affiché dans le panneau de l'algorithme
            type=QgsProcessingParameterNumber.Integer,
            minValue=0,
            maxValue=359,
            defaultValue=30))

def processAlgorithm(self, parameters: dict[str, Any], context: QgsProcessingContext, model_feedback: QgsProcessingFeedback) -> dict[str, Any]:
    angle = self.parameterAsInt(parameters, "angle", context) # valeur du paramètre angle en tant qu'entier, p.ex 30 (valeur par défaut).

Selon le type de paramètre il est possible que la syntaxe suivante fonctionne:

def processAlgorithm(self, parameters: dict[str, Any], context: QgsProcessingContext, model_feedback: QgsProcessingFeedback) -> dict[str, Any]:
    angle = parameters["angle"] # on récupère la valeur d'angle directement

Cependant par soucis de clarté et pour éviter des erreurs il est conseillée de tout le temps utiliser les méthodes parameterAs{VariableType}() pour récupérer la valeur des paramètres.

Les méthodes initAlgorithm() et processAlgorithm() possèdent différents arguments qui seront toujours les mêmes:

config: Un dictionnaire de configuration de l'algorithme, il ne sera pas utilisé dans ce cours.
context: Le contexte d’exécution du script (QgsProcessingContext) qui permet à QGIS de retrouver les couches via leur identifiant. Cela permet notamment aux algorithmes processing utilisés dans des algorithmes de récupérer les résultats intermédiaires.
feedback: Une classe QgsProcessingFeedback qui permet de communiquer avec l'utilisateur, il gère par exemple la barre de progression et les logs de l'algorithme affiché dans le panneau d’exécution QGIS.

La sortie de l'algorithme (donc de la méthode processAlgorithm()) est un dictionnaire qui contient le nom des paramètres de sortie, souvent "OUTPUT", qui sont également définis dans la méthode initParameter(). Dans cet exemple il s'agit de QgsProcessingParameterRasterDestination qui permet d'indiquer à QGIS que notre algorithme va retourner un raster.

2.2 Créer son propre algorithme de buffer conditionnel¶

Dans cette question vous allez créer votre propre algorithme processing qui appliquera un buffer aux géométries d'une couche de polygones avec certaines conditions.

2.2.1 Squelette de l'algorithme¶

Dans QGIS allez dans la boîte à outil de traitement et cliquez sur l'icône Python en haut. Sélectionnez « Créer un nouveau script depuis un modèle », enregistrez le à l'emplacement proposé sous le nom « conditional_buffer.py » puis:

Renommez la classe en ConditionalBuffer.
Changez le nom de l'algorithme dans la méthode name() par "conditionalbuffer".
Changez le nom affichée de l'algorithme dans la méthode displayName() par "Conditional buffer".
Changez le groupe de l'algorithme dans la méthode group() par "Cours pyqgis".
Changez l'identifiant du groupe dans la méthode groupId() par "courspyqgis".
Changez la string renvoyée dans shortHelpString() par: "This algorithm apply a conditional buffer on polygon geometries."
Supprimez le contenu de la méthode processAlgorithm() et ajoutez uniquement:
```
return {}
```
Sauvegardez le script puis lancez-le depuis la boîte à outils de traitement.

Vous avez créé un squelette d'algorithme processing prévu pour prendre en entrée une couche vectorielle et retourner une couche vectorielle. Comme processAlgorithm() ne fait rien et retourne un dictionnaire vide {} il ne se passe rien si vous exécutez ce script.

2.2.2 Ajouter des paramètres¶

cours

Un grand nombre de classe sont disponibles pour ajouter des paramètres. Vous trouverez toute ces classes dans le tableau des paramètres.
Par convention et praticité les noms des paramètres sont définis dans des attributs de classes en majuscule. Le tableau donne les noms usuels pour chaque type de paramètre, si vous le pouvez respectez cette convention (mis à part si vous avez plusieurs paramètres du même type, dans ce cas il faudra les différencier).

Dans notre nouvel algorithme on veut appliquer un buffer avec une distance dépendante de l'aire de la géométrie et d'une constante donnée en entrée avec la formule suivante:

\[ D = C + \frac{A_{geometry}}{1000} \]

où:

\(C\) est une constante, comprise entre 0m et 10m inclus, définie en paramètre par l'utilisateur.
\(A_{geometry}\) est l'aire totale de la géométrie.

Ajoutez un paramètre « Distance » à l’algorithme pour que l'utilisateur puisse définir la constante \(C\). Sauvegardez et ouvrez le panneau de l'algorithme pour voir votre paramètre.

Tip

Pensez à importer les classes que vous allez utiliser au début du script. Les classes de paramètres sont toutes dans qgis.core vous n'avez qu'à ajouter à la liste d'imports déjà existants.

Solution

from qgis.core import QgsProcessingParameterDistance, Qgis # à ajouter aux imports existants
# ...
class ConditionalBuffer(QgsProcessingAlgorithm):
    INPUT = "INPUT"
    OUTPUT = "OUTPUT"
    DISTANCE = "DISTANCE" # attribut correspondant au paramètre distance
# ...
def initAlgorithm(self, config: Optional[dict[str, Any]] = None):
# ...
    distance = QgsProcessingParameterDistance(self.DISTANCE, "Distance")
    distance.setDefaultUnit(Qgis.DistanceUnit.Meters)
    self.addParameter(distance)

2.2.3 Implémenter le buffer conditionnel¶

Cours

Pour gérer les entrées et les sorties de type couche vectorielle des algorithmes processing on utilise les deux paramètres suivantes:

QgsProcessingParameterFeatureSource: Une source d'entités géométriques.
QgsProcessingParameterFeatureSink: Une destination pour les entités géométriques crées dans processAlgorithm().

Dans la méthode processAlgorithm() on peut récupérer ces deux paramètres avec:

parameterAsSource(): Retourne une classe QgsProcessingFeatureSource. Cette classe hérite de QgsFeatureSource de même que QgsVectorLayer et possède donc les mêmes méthodes telles que getFeatures qui est essentielle pour itérer sur les entités.
parameterAsSink(): Retourne un tuple (sink, sink_id). La variable sink est une classe QgsFeatureSink qui implémente les méthodes permettant de remplir la couche vectorielle telles que addFeature(). C'est dans cet objet qu'on va stocker les entités issues de notre traitement. Le sink_id est l'identifiant de la couche (équivalent à sink.id()) qui permet d'identifier la couche. C'est sink_id qu'il faut mettre dans le dictionnaire de sortie, cela permettra à QGIS d'identifiant la couche à charger automatiquement dans le projet QGIS. On passe comme argument le type de géométrie, les champs qu'on veut créer dans la couche de sortie et le SCR.

def initAlgorithm(self, config: Optional[dict[str, Any]] = None):
    # ...
    self.addParameter(
        QgsProcessingParameterFeatureSource(
            self.INPUT,
            "Input layer",
            [QgsProcessing.SourceType.TypeVectorAnyGeometry],
        )
    )

    self.addParameter(
        QgsProcessingParameterFeatureSink(self.OUTPUT, "Output layer")
    )
    # ...

def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict[str, Any]:
    # ...
    source = self.parameterAsSource(parameters, self.INPUT, context)
    # ...
    (sink, dest_id) = self.parameterAsSink(
        parameters,
        self.OUTPUT,
        context,
        source.fields(), # même champs que la couche source
        source.wkbType(), # même type de géométrie que la couche source
        source.sourceCrs(), # même SRC que la couche source
    )
    # Ajout des entités dans le sink à la fin de l'algorithme
    features = list(source.getFeatures())

    for feature in enumerate(features):
        sink.addFeature(feature, QgsFeatureSink.Flag.FastInsert)

    return {self.OUTPUT : sink_id} # retourner l'id du sink.

Écrivez le contenu de processAlgorithm() pour appliquer le buffer sur les géométries avec la distance calculée en suivant ces étapes:

Récupérez les paramètres définis dans initAlgorithm() (source, sink et distance).
Itérer sur les entités de source.
Pour chaque entité extrayez l'aire de la géométrie et calculez la distance \(D\) avec la formule.
Appliquez le buffer avec la distance \(D\) sur la géométrie.
Mettez à jour la géométrie de l'entité.
Mettez à jour l'entité dans le sink.
Retournez le dictionnaire contenant le sink_id avec la clef "OUTPUT".

Solution

def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
    ) -> dict[str, Any]:

    source = self.parameterAsSource(parameters, self.INPUT, context)
    (sink, sink_id) = self.parameterAsSink(
        parameters,
        self.OUTPUT,
        context,
        source.fields(),
        source.wkbType(),
        source.sourceCrs(),
    )
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context) # récupération de la distance comme un nombre flottant (double).

    for feature in source.getFeatures():
        geom = feature.geometry()
        area = geom.area()
        buffer_distance = distance + (area/1000)
        buffer_geom = geom.buffer(buffer_distance, 5)
        feature.setGeometry(buffer_geom)
        sink.addFeature(feature)

    return {self.OUTPUT: sink_id}

2.2.4 Conditionner le traitement par la catégorie de culture¶

Pour paramétré encore plus l'utilisation du buffer on peut ajouter la possibilité d'appliquer ce buffer uniquement sur certains types de cultures, définis par un code, au choix de l'utilisateur. Pour cela il faut deux nouveaux paramètres:

Un pour choisir le champ de la couche source qui contient les codes de cultures.
Un pour sélectionner les cultures concernées lors de l’exécution de l'algorithme.

Mettez en choix de cultures les 10 suivantes:

 [
    "TRN",
    "J5M",
    "BTA",
    "J6S",
    "BTH",
    "ORH",
    "PPH",
    "SOG",
    "PTR",
]

Ajoutez ces deux paramètres et modifiez processAlgorithm() pour tenir compte de la sélection de l'utilisateur.

Tip

Les classes de paramètres à utiliser sont QgsProcessingParameterField et QgsProcessingParameterEnum avec l'option allowMultiple=True.
Le paramètre QgsProcessingParameterEnum ne rend pas directement la ou les chaînes de caractères donnée dans les options mais les index des éléments sélectionnés, il faut faire la correspondance dans processAlgorithm(). Pour récupérer une liste d'entiers utilisez parameterAsInts().
Définissez les cultures disponibles au dessus de la classe de l'algorithme comme constante pour éviter de polluer la compréhension de la classe.

Solution

from qgis.core import QgsProcessingParameterEnum, QgsProcessingParameterField
#...
def initAlgorithm(self, config: Optional[dict[str, Any]] = None):
# ...
    self.addParameter(QgsProcessingParameterField(self.FIELD, "Champ de cultures",parentLayerParameterName=self.INPUT))
    self.addParameter(QgsProcessingParameterEnum(self.ENUM, "Cultures", options=CROPS, allowMultiple=True))
# ...
def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
    ) -> dict[str, Any]:
    # ...
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context) # récupération de la distance comme un nombre flottant (double).
    crops_idxs = self.parameterAsInts(parameters, self.ENUM, context) # récupération des index des cultures choisies par l'utilisateur
    crops = [CROPS[i] for i in crops_idxs] # récupération des cultures choisies
    field = self.parameterAsString(parameters, self.FIELD, context) # récupération du nom du champ des codes de cultures

    for feature in source.getFeatures():
        if feature[field] in crops:
            geom = feature.geometry()
            area = geom.area()
            buffer_distance = distance + (area/1000)
            buffer_geom = geom.buffer(buffer_distance, 5)
            feature.setGeometry(buffer_geom)

        sink.addFeature(feature)
# ...

2.2.4 Ajouter un champ dans la couche de sortie¶

Modifier processAlgorithm() pour que la couche en sortie possède un champ booléen « Buffer » qui indique si la géométrie de l'entité a été modifiée.

Tip

Il est nécessaire de créer une nouvelle entité avec les champs de la couche de sortie et de transférer les attributs de l'entité de la couche source, ainsi que la valeur pour le champ « Buffer », à la nouvelle entité. Pour cela vous pouvez utiliser les méthodes attributes() et setAttributes() de la classe QgsFeature.

Solution

from qgis.core import QgsField, QgsFeature # à ajouter aux imports existants
# ...
def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict[str, Any]:

    source = self.parameterAsSource(parameters, self.INPUT, context)
    # creation d'un champ "Buffer"
    buffer_field = QgsField("Buffer", 1)
    fields = source.fields()
    fields.append(buffer_field)
    (sink, sink_id) = self.parameterAsSink(
        parameters,
        self.OUTPUT,
        context,
        fields,
        source.wkbType(),
        source.sourceCrs(),
    )
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
    crops_idxs = self.parameterAsInts(parameters, self.ENUM, context) 
    crops = [CROPS[i] for i in crops_idxs]
    field = self.parameterAsString(parameters, self.FIELD, context)

    for feature in source.getFeatures():
        # clone de l'entité avec ajout du champ Buffer
        new_feature = QgsFeature(fields)
        new_feature.setAttributes(feature.attributes() + [False])
        new_feature.setGeometry(feature.geometry())
        if new_feature[field] in crops:
            geom = new_feature.geometry()
            area = geom.area()
            buffer_distance = distance + (area / 1000)
            buffer_geom = geom.buffer(buffer_distance, 5)
            new_feature.setGeometry(buffer_geom)
            new_feature["Buffer"] = True

        sink.addFeature(new_feature)

    return {self.OUTPUT: sink_id}

2.2.5 Barre de progression¶

Cours

Si l'algorithme lance des traitements qui sont lourd en temps de calcul il est essentiel de fournir à l'utilisateur des informations sur l'état d'avancement de l'algorithme. Pour cela on peut utiliser la classe QgsProcessingFeedback qui permet de gérer:

La barre de progression de l'algorithme.
Les logs dans le panneau d’exécution de l'algorithme.
L'annulation de l'algorithme.

L'instance de cette classe à utiliser dans processAlgorithm est donnée en entrée de la méthode sous le nom de feedback.

La couche « RPG » ne possède pas assez d'entités pour observer correctement les retours utilisateurs car l'algorithme ce termine trop rapidement. Ajoutez dans l'itération sur les entités (la boucle for) un temps d'attente de 0.002s avec time.sleep() afin de ralentir artificiellement le temps d’exécution. Utilisez feedback et la méthode setProgress() pour ajouter une barre de progression qui indiquera l'avancement de l'algorithme.

Tip

La barre de progression va de 0 à 100. Il faut donc calculer le pourcentage d'avancement.

Solution

import time # à ajouter aux imports existants
# ...
def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict[str, Any]:

    source = self.parameterAsSource(parameters, self.INPUT, context)
    buffer_field = QgsField("Buffer", 1)
    fields = source.fields()
    fields.append(buffer_field)
    (sink, sink_id) = self.parameterAsSink(
        parameters,
        self.OUTPUT,
        context,
        fields,
        source.wkbType(),
        source.sourceCrs(),
    )
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
    crop_field = self.parameterAsString(parameters, self.FIELD, context)
    codes_idx = self.parameterAsInts(parameters, self.ENUM, context)
    codes = [CROPS[i] for i in codes_idx]

    total = source.featureCount()

    for i, feature in enumerate(source.getFeatures()):
        code = feature[crop_field]
        new_feature = QgsFeature(fields)
        new_feature.setAttributes(feature.attributes() + [False])
        new_feature.setGeometry(feature.geometry())
        if code in codes:
            geom = new_feature.geometry()
            area = geom.area()
            buffer_distance = distance + (area / 1000)
            buffer_geom = geom.buffer(buffer_distance, 5)
            new_feature.setGeometry(buffer_geom)
            new_feature["Buffer"] = True

        sink.addFeature(new_feature)
        feedback.setProgress((i/total) * 100)
        time.sleep(0.002)

    return {self.OUTPUT: sink_id}

2.2.6 Autoriser l'annulation de l'algorithme¶

Cours

Lorsque l'utilisateur clique sur le bouton « Annuler » du panneau de l'algorithme cela envoie uniquement un signal indiquant que l'utilisateur souhaite annuler l'algorithme mais n'effectue aucune action d'annulation. Certaines opérations sont à risque lorsque l'on fait du traitement de données et une annulation brusque au mauvais moment peut générer des problèmes. C'est donc au programmeur de gérer de la manière qu'il souhaite l'annulation de son algorithme. La class QgsProcessingFeedback contient l'information de l'annulation via sa méthode isCanceled() qui renvoie un booléen.

Pour annuler l'algorithme proprement il suffit de remonter une erreur spécifique de QGIS de la classe QgsProcessingException avec le message souhaité en utilisant l'instruction raise:

if feedback.isCanceled():
    raise QgsProcessingException("Cancel message.")

Ajoutez la logique d'annulation dans processAlgorithm() pour que l'utilisateur puisse annuler le traitement en cours.

Solution

    def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict[str, Any]:

    source = self.parameterAsSource(parameters, self.INPUT, context)
    buffer_field = QgsField("Buffer", 1)
    fields = source.fields()
    fields.append(buffer_field)
    (sink, sink_id) = self.parameterAsSink(
        parameters,
        self.OUTPUT,
        context,
        fields,
        source.wkbType(),
        source.sourceCrs(),
    )
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
    crop_field = self.parameterAsString(parameters, self.FIELD, context)
    codes_idx = self.parameterAsInts(parameters, self.ENUM, context)
    codes = [CROPS[i] for i in codes_idx]

    total = source.featureCount()

    for i, feature in enumerate(source.getFeatures()):
        code = feature[crop_field]
        new_feature = QgsFeature(fields)
        new_feature.setAttributes(feature.attributes() + [False])
        new_feature.setGeometry(feature.geometry())
        if code in codes:
            geom = new_feature.geometry()
            area = geom.area()
            buffer_distance = distance + (area / 1000)
            buffer_geom = geom.buffer(buffer_distance, 5)
            new_feature.setGeometry(buffer_geom)
            new_feature["Buffer"] = True

        sink.addFeature(new_feature)
        feedback.setProgress((i/total) * 100)
        if feedback.isCanceled():
            raise QgsProcessingException("Algorithm is canceled y user.")
        time.sleep(0.002)

    return {self.OUTPUT: sink_id}

2.2.7 Afficher des messages dans le panneau de l'algorithme¶

Utilisez feedback pour informer l'utilisateur avec un message clair:

du nombre d'entités qui vont être traités par l'algorithme (au début de l'algorithme).
du nombre d'entités sur lesquelles un buffer a été appliqué (à la fin de l'algorithme).

Solution

def processAlgorithm(
    self,
    parameters: dict[str, Any],
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict[str, Any]:

    source = self.parameterAsSource(parameters, self.INPUT, context)
    buffer_field = QgsField("Buffer", 1)
    fields = source.fields()
    fields.append(buffer_field)
    (sink, sink_id) = self.parameterAsSink(
        parameters,
        self.OUTPUT,
        context,
        fields,
        source.wkbType(),
        source.sourceCrs(),
    )
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
    crop_field = self.parameterAsString(parameters, self.FIELD, context)
    codes_idx = self.parameterAsInts(parameters, self.ENUM, context)
    codes = [CROPS[i] for i in codes_idx]

    total = source.featureCount()
    feedback.pushInfo(f"L'algorithme va traiter {total} entités.")
    buffered = 0
    for i, feature in enumerate(source.getFeatures()):
        code = feature[crop_field]
        new_feature = QgsFeature(fields)
        new_feature.setAttributes(feature.attributes() + [False])
        new_feature.setGeometry(feature.geometry())
        if code in codes:
            geom = new_feature.geometry()
            area = geom.area()
            buffer_distance = distance + (area / 1000)
            buffer_geom = geom.buffer(buffer_distance, 5)
            new_feature.setGeometry(buffer_geom)
            new_feature["Buffer"] = True
            buffered += 1

        sink.addFeature(new_feature)
        feedback.setProgress((i/total) * 100)
        if feedback.isCanceled():
            raise QgsProcessingException("Algorithm is canceled by user.")
        time.sleep(0.002)

    feedback.pushInfo(f"{buffered} entités ont été modifiées.")

    return {self.OUTPUT: sink_id}

TD 3 - Utiliser et créer des algorithmes processing¶

1. Utiliser des algorithmes processing¶

1.1 Lancer un algorithme processing¶

1.2 Enchaîner des algorithmes processing¶

2. Créer des algorithmes processing¶

2.1 Structure d'un algorithme processing¶

2.2 Créer son propre algorithme de buffer conditionnel¶

2.2.1 Squelette de l'algorithme¶

2.2.2 Ajouter des paramètres¶

2.2.3 Implémenter le buffer conditionnel¶

2.2.4 Conditionner le traitement par la catégorie de culture¶

2.2.4 Ajouter un champ dans la couche de sortie¶

2.2.5 Barre de progression¶

2.2.6 Autoriser l'annulation de l'algorithme¶

2.2.7 Afficher des messages dans le panneau de l'algorithme¶

1.2 Enchaîner des algorithmes `processing`¶

2. Créer des algorithmes `processing`¶

2.1 Structure d'un algorithme `processing`¶