1

I am trying to implement threading to some pyqgis code.

I have been trying to use QgsTask.fromFunction.

I have implemented code from this article but it isn't working. I used the first of the three methods - basically verbatim, with minor changes to the structure to make it work in my plugin and to try to pin down the source of the error.

When I call testThreading I get the error

Exception: <[my plugin] object at 0x00000267FA2E39B0>

Why?

def test_run(task, wait_time):

    from time import sleep
    import random

    QgsMessageLog.logMessage('Started task {}'.format(task.description()),level=Qgis.Info)
    wait_time = wait_time / 100
    total = 0
    iterations = 0
    for i in range(101):
        sleep(wait_time)

        # use task.setProgress to report progress
        task.setProgress(i)

        total += random.randint(0, 100)
        iterations += 1

        # check task.isCanceled() to handle cancellation
        if task.isCanceled():
            test_stopped(task)
            return None

        # raise exceptions to abort task
        if random.randint(0, 500) == 42:
            raise Exception('bad value!')

    return {
      'total': total, 'iterations': iterations, 'task': task.description()
    }


  def test_stopped(task):
    QgsMessageLog.logMessage('This does not happen',level=Qgis.Info)


  def test_completed(exception, result=None):

    if exception is None:
        QgsMessageLog.logMessage('This does not happen',level=Qgis.Info)
    else:
        QgsMessageLog.logMessage("Exception: {}".format(exception),level=Qgis.Info)
        raise exception


  def testThreading(self):
    task1 = QgsTask.fromFunction(
      'waste cpu 1', self.test_run, on_finished=self.test_completed, wait_time=4)
    task2 = QgsTask.fromFunction(
      'waste cpu 2', self.test_run, on_finished=self.test_completed, wait_time=3)
    QgsApplication.taskManager().addTask(task1)
    QgsApplication.taskManager().addTask(task2)
    QgsMessageLog.logMessage('Getting here',level=Qgis.Info)
0

It could be that you are missing a few self, try to change def test_run(task, wait_time): to def test_run(self, task, wait_time): and def test_completed(exception, result=None): to def test_completed(self, exception, result=None):

I made a similar post trying to get clarity on the function isCanceled() here: What happens when you press cancel on a QgsTask?

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.