Обновлять прогресс длинного процесса с помощью обратного вызова вместо опроса

Question

в моем скрипте Python я запускаю длинный процесс (drive ()), который инкапсулирован в метод класса:

car.py

import time


class Car(object):

    def __init__(self, sleep_time_in_seconds, miles_to_drive):
        self.sleep_time_in_seconds = sleep_time_in_seconds
        self.miles_to_drive = miles_to_drive

    def drive(self):

        for mile in range(self.miles_to_drive):
            print('driving mile #{}'.format(mile))
            time.sleep(self.sleep_time_in_seconds)

app.py

from car import Car

sleep_time = 2
total_miles = 5

car = Car(sleep_time_in_seconds=sleep_time, miles_to_drive=total_miles)
car.drive()


def print_driven_distance_in_percent(driven_miles):
    print("Driven distance: {}%".format(100 * driven_miles / total_miles))

В основном скрипте app.py Я хотел бы знать, как проходит процесс drive(). Одним из способов решения этой проблемы было бы создание цикла, который опрашивает текущий прогресс из класса Car. Если класс Car будет наследоваться от Thread, опрос, насколько мне кажется, является ожидаемым шаблоном ... Мне просто любопытно, можно ли каким-то образом уведомить основной скрипт из класса Car отекущий прогресс. Я подумал о том, чтобы, возможно, создать класс-обертку, который я смогу передать в качестве аргумента классу Car, и экземпляр машины затем сможет вызвать класс-обертку print_progress. Или есть более питонный способ уведомить сценарий вызывающего абонента по требованию?

Спасибо

РЕДАКТИРОВАТЬ:

Основано на ответе Артема Козырева - вот чего я хотел добиться:

import time
from threading import Thread
from queue import Queue


def ask_queue(q):
    """
    The function to control status of our status display thread
    q - Queue - need to show status of task
    """
    while True:
        x = q.get()  # take element from Queue
        if x == "STOP":
            break
        print("Process completed in {} percents".format(x))
    print("100% finished")


class MyClass:
    """My example class"""
    def __init__(self, name, status_queue):
        self.name = name
        self.status_queue = status_queue

    def my_run(self):
        """
        The function we would like to monitor
        """
        # th = Thread(target=MyClass.ask_queue, args=(self.status_queue,), )  # monitoring thread
        # th.start()  # start monitoring thread
        for i in range(100):  # start doing our main function we would like to monitor
            print("{} {}".format(self.name, i))
            if i % 5 == 0:  # every 5 steps show status of progress
                self.status_queue.put(i)  # send status to Queue
                time.sleep(0.1)
        self.status_queue.put("STOP")  # stop Queue
        # th.join()


if __name__ == "__main__":

    q = Queue()

    th = Thread(target=ask_queue, args=(q,), )  # monitoring thread
    th.start()  # start monitoring thread

    # tests
    x = MyClass("Maria", q)
    x.my_run()

    th.join()

Спасибо всем !!

Answer 1

Спасибо за интересный вопрос, обычно вам не нужно использовать статус как отдельную ветку для дела, вы можете просто напечатать статус в методе, который вы хотите отслеживать, но для целей обучения вы решаете проблему следующим образом,Пожалуйста, следуйте комментариям и не стесняйтесь спрашивать:

import time
from threading import Thread
from queue import Queue


class MyClass:
    """My example class"""
    def __init__(self, name, status_queue):
        self.name = name
        self.status_queue = status_queue

    @staticmethod
    def ask_queue(q):
        """
        The function to control status of our status display thread
        q - Queue - need to show status of task
        """
        while True:
            x = q.get()  # take element from Queue
            if x == "STOP":
                break
            print("Process completed in {} percents".format(x))
        print("100% finished")

    def my_run(self):
        """
        The function we would like to monitor
        """
        th = Thread(target=MyClass.ask_queue, args=(self.status_queue,), )  # monitoring thread
        th.start()  # start monitoring thread
        for i in range(100):  # start doing our main function we would like to monitor
            print("{} {}".format(self.name, i))
            if i % 5 == 0:  # every 5 steps show status of progress
                self.status_queue.put(i)  # send status to Queue
                time.sleep(0.1)
        self.status_queue.put("STOP")  # stop Queue
        th.join()


if __name__ == "__main__":
    # tests
    x = MyClass("Maria", Queue())
    x.my_run()
    print("*" * 200)
    x.my_run()