Запуск зависимых потоков одновременно в Python - PullRequest
Новая
       4

Запуск зависимых потоков одновременно в Python

0 голосов
/ 13 февраля 2020

У меня есть два класса потоков для извлечения и обнаружения.

Извлечение извлекает кадры из видео и сохраняет его в папке, Detect берет изображения из папки, в которой извлекаются кадры, и обнаруживает объекты.

Но когда я запускаю приведенный ниже код, работает только экстракт: 

global q
q = Queue()

class extract(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        print("T1")
        cam = cv2.VideoCapture(video_name)
        frameNum = 0
        # isCaptured = True
        frameCount = 0
        while True:
            isCapture, frame = cam.read()
            if not isCapture:
                break
            if frameCount % 5 == 0:
                frameNum = frameNum + 1
                fileName = vid + str(frameNum) + '.jpg'
                cv2.imwrite('images/extracted/' + fileName, frame)
                q.put(fileName)
            frameCount += 1
        cam.release()
        cv2.destroyAllWindows()

class detect(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        print("T2")
        #logic to detect objects. 


if __name__ == '__main__':
    thread1 = extract()
    thread1.start()
    thread2 = detect()
    thread2.start()

Это печатает только T1 и без T2. Я подумал, что, вероятно, Detect запускался первым, и очередь была пуста, поэтому ничего не произошло, поэтому я добавил фиктивные записи в очередь, и она работала так, как я хотел.

Но он работал только для фиктивных записей, он не работал для записей, которые функция извлечения добавила в очередь. Посмотрел другие вопросы, и ни один из них, похоже, не решил проблему, поэтому разместил это здесь

Ответы [ 2 ]

0 голосов
/ 13 февраля 2020

Вместо ожидания в жестко заданном времени вы можете использовать Event() и заставить свой поток обнаружения ждать, пока не установится Event(), прежде чем выполнять обнаружение

Если установлено событие, означающее все задачи сделаны. Кроме того, вы также должны следить за очередью, если какие-либо задачи еще не выполнены.

Я написал пример кода, чтобы продемонстрировать, как он работает, вы можете изменить код в соответствии с вашими потребностями.

Здесь extract занимает 5 секунд, чтобы добавить задачу в очередь, а detect проверяет задачу каждую 1 секунду. Поэтому, если извлечение выполняется медленнее, чем когда бы то ни было доступно, оно будет обработано detect. И когда все задачи будут выполнены, обнаружится, что вырвется из л oop. 

import threading
import queue
import time

global q
q = queue.Queue()

class extract(threading.Thread):
    all_tasks_done = threading.Event()
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        counter = 5
        while counter:
            time.sleep(5)
            counter -= 1
            q.put(1)
            print("added a task to queue")
        extract.all_tasks_done.set()



class detect(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while not extract.all_tasks_done.wait(1) or not q.empty():
            print(q.get())
        print("detection done")
        #logic to detect objects. 


if __name__ == '__main__':
    thread1 = extract()
    thread1.start()
    thread2 = detect()
    thread2.start()
    thread1.join()
    thread2.join()
    print("work done")
0 голосов
/ 13 февраля 2020

Вы, вероятно, также хотите, чтобы ваша логика обнаружения c была равна бесконечной l oop. 

class detect(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while True: 
        #detect frame

А если это обнаружение одного кадра. Затем рассмотрите ожидание в обнаружении потока. 

from time import sleep
class detect(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)

        def run(self):
           sleep(120)
           # Detect logic
...