Approche progressive du multiprocessing en Python

Question

J'ai été occupé à écrire mon premier code de multiprocessing et il fonctionne, yay. Cependant, j'aimerais maintenant avoir un retour sur la progression et je ne suis pas sûr de la meilleure approche à adopter.

Ce que mon code (voir ci-dessous) fait en bref :

• Un répertoire cible est analysé à la recherche de fichiers mp4.
• Chaque fichier est analysé par un processus distinct, le processus enregistre un résultat (une image).

Ce que je cherche pourrait être :

1. Simple

• Chaque fois qu'un processus termine un fichier, il envoie un message "terminé".
• Le code principal compte le nombre de fichiers terminés.

2. Fantaisie

   Core 0 processing file 20 of 317 ||||||_ 60% completed Core 1 processing file 21 of 317 ||||||||| 90% completed ... Core 7 processing file 18 of 317 ||__ 20% completed

J'ai lu toutes sortes d'informations sur les files d'attente, les pools, le tqdm et je ne suis pas sûr de la voie à suivre. Quelqu'un pourrait-il m'indiquer une approche qui fonctionnerait dans ce cas ?

Merci d'avance !

EDIT : J'ai modifié mon code qui lance les processus comme suggéré par gsb22.

Mon code :

# file operations
import os
import glob
# Multiprocessing
from multiprocessing import Process
# Motion detection
import cv2

# >>> Enter directory to scan as target directory
targetDirectory = "E:\Projects\Programming\Python\OpenCV\\videofiles"

def get_videofiles(target_directory):

    # Find all video files in directory and subdirectories and put them in a list
    videofiles = glob.glob(target_directory + '/**/*.mp4', recursive=True)
    # Return the list
    return videofiles

def process_file(videofile):

    '''
    What happens inside this function:
    - The video is processed and analysed using openCV
    - The result (an image) is saved to the results folder
    - Once this function receives the videofile it completes
      without the need to return anything to the main program
    '''

    # The processing code is more complex than this code below, this is just a test
    cap = cv2.VideoCapture(videofile)

    for i in range(10):
        succes, frame = cap.read()

        # cv2.imwrite('{}/_Results/{}_result{}.jpg'.format(targetDirectory, os.path.basename(videofile), i), frame)

        if succes:
            try:
                cv2.imwrite('{}/_Results/{}_result_{}.jpg'.format(targetDirectory, os.path.basename(videofile), i), frame)
            except:
                print('something went wrong')

if __name__ == "__main__":

    # Create directory to save results if it doesn't exist
    if not os.path.exists(targetDirectory + '/_Results'):
        os.makedirs(targetDirectory + '/_Results')

    # Get a list of all video files in the target directory
    all_files = get_videofiles(targetDirectory)

    print(f'{len(all_files)} video files found')

    # Create list of jobs (processes)
    jobs = []

    # Create and start processes
    for file in all_files:
        proc = Process(target=process_file, args=(file,))
        jobs.append(proc)

    for job in jobs:
        job.start()

    for job in jobs:
        job.join()

    # TODO: Print some form of progress feedback

    print('Finished :)')

Answer 1

J'ai lu toutes sortes d'informations sur les files d'attente, les pools, le tqdm et je ne suis pas sûr de la voie à suivre. Quelqu'un pourrait-il m'indiquer une approche qui fonctionnerait dans ce cas ?

Voici un moyen très simple d'obtenir une indication de progression à un coût minimal :

from multiprocessing.pool import Pool
from random import randint
from time import sleep

from tqdm import tqdm

def process(fn) -> bool:
    sleep(randint(1, 3))
    return randint(0, 100) < 70

files = [f"file-{i}.mp4" for i in range(20)]

success = []
failed = []
NPROC = 5
pool = Pool(NPROC)

for status, fn in tqdm(zip(pool.imap(process, files), files), total=len(files)):
    if status:
        success.append(fn)
    else:
        failed.append(fn)

print(f"{len(success)} succeeded and {len(failed)} failed")

Quelques commentaires :

• tqdm est une bibliothèque tierce qui implémente très bien les barres de progression. Il en existe d'autres. pip install tqdm .
• nous utilisons un pool (il n'y a presque jamais de raison de gérer soi-même les processus pour des choses simples comme celle-ci) de NPROC processus. Nous laissons le pool gérer l'itération de notre fonction de processus sur les données d'entrée.
• nous signalons l'état en faisant en sorte que la fonction renvoie un booléen (dans cet exemple, nous choisissons au hasard, en privilégiant le succès). Nous ne renvoyons pas le nom du fichier, bien que nous pourrions le faire, car il devrait être sérialisé et envoyé par le sous-processus, ce qui constitue une surcharge inutile.
• nous utilisons Pool.imap qui renvoie un itérateur qui garde le même ordre que l'itérable que nous lui passons. Nous pouvons donc utiliser zip pour itérer files directement. Puisque nous utilisons un itérateur de taille inconnue, tqdm a besoin qu'on lui dise quelle est sa durée. (Nous aurions pu utiliser pool.map mais il n'y a pas besoin d'engager la ram - bien que pour un bool cela ne fasse probablement aucune différence).

J'ai délibérément écrit ceci comme une sorte de recette. Vous pouvez faire beaucoup de choses avec le multitraitement en utilisant simplement les paradigmes de haut niveau, et Pool.[i]map est l'un des plus utiles.

Références

https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.Pool https://tqdm.github.io/