Python OPENCV Threading並行処理で参照するファイルの重複を避ける

Question

PythonでOPENCVで画像処理をしておりますが、
ファイルが多いので高速化検討しようと考えており、並列処理したく思います。

現在は、OPENCVで画像ファイルを並列処理できておりますが、
参照している画像ファイルが時折、重複するときがあります。

そこで、以下コードのどの部分を変更したら画像ファイルの重複を避けれますでしょうか。

WIN10、PsScriter(Spyder)使用しております。

import cv2
import numpy as np
import glob
import os
import pandas as pd
import time
import threading


x_1,y_1,z_1,z_1_1=[],[],[],[]
a,b=[],[]

#保存先
r=r"◎"
j1=["◎"]

for j2 in j1:
    a.clear()
    t1 = time.time()
    dock = r"◎"

    for folder,subfolders,files in os.walk(dock + "\" + str(j2)):
        a.append('{}'.format(folder))
    j=0

    for d in a:
        global q

        def worker1():#--------------------------------------
            global q

            while True:

#            for a in range(5):

                img1 = imread(r_1[q]) # 画像ファイル読み込み
                img1_d_1=r_1[q]
                q=q+1

                if q>d_1-1:
                    break
                img1_d_2=os.path.basename(r_1[q])
                img1_d_3=img1_d_2.replace(".jpg","")

                x_1.append(img1_d_3)

                img_ch0 = cv2.split(img1)[0]
                w_num = cv2.countNonZero(img_ch0)
                
                y_1.append(w_num)

#----------------------------------------------------
        def worker2():
            global q

            while True:
                if q>d_1-1:
                    break
                img2 = imread(r_1[q-1]) # 画像ファイル読み込み
                img2_d_1=r_1[q-1]
                q=q+1
                img2_d_2=os.path.basename(img2_d_1)
                img2_d_3=img2_d_2.replace(".jpg","")
                
                x_1.append(img2_d_3)
                
                img_ch0 = cv2.split(img2)[0]
                w_num = cv2.countNonZero(img_ch0)
                
                y_1.append(w_num)
                
        global q

        r_1 = glob.glob(d + "\*.jpg")
        j=j+1
        print(j)
        print(d)
        
        q=0
        d_1=len(r_1)

        t1 = threading.Thread(target=worker1)
        # t1 スレッドをデーモン化する
        t1.setDaemon(True)
        t2 = threading.Thread(target=worker2)
        x_1=[]
        y_1=[]

        t1.start()
        t2.start()

        print('started')
        t1.join()
        print("--end--")
        t2.join()
        print(set(x_1))    

        print(len(set(x_1)))

        print(y_1)

何卒、宜しくお願いいたします。

Comments

[y_waiwai]

重複したらどういう不具合があるんでしょうか

[goki_gottan]

不具合というより統計を取っておりますので、計算結果がおかしくなるのは避けたいです。
変数であるr_1は200枚程度画像が入っており、今回の評価では200で打ち切りみたいな指示を入れております。
重複は削除の指示を入れて、重複なしで200枚の時点でBreakとしてもよいのですが、他の方法で重複がなければ助かりますし、余分な画像を評価する必要がないので高速化できます。

Answer 1

「並列処理、個々の処理は独立して行える」という条件の場合は concurrent.futures の ThreadPoolExecutor をおつかいください。とても簡単に並列処理が行えます。

threading を使ったやり方だとデータをチャンクに分けたり、スレッドの管理など自前で書く必要が出てきて複雑になります。

例

import time
from concurrent import futures
from pathlib import Path

def func(path):
    pass  # ファイルごとに行う処理を書く


input_dir = Path("hogehoge")


future_list = []
with futures.ThreadPoolExecutor() as executor:
    for path in input_dir.glob("**/*.jpg"):  # input_dir 以下にある jpg ファイル
        # タスクを追加する。
        future = executor.submit(func, str(path))
        # Future オブジェクトを記録する。
        future_list.append(future)

# 結果を受け取る
ret = [x.result() for x in future_list]

参考:

• Python - concurrent.futures を使った並列化の方法について

Comments

[goki_gottan]

ありがとうございます。スレッドごとに処理を書かなくて良い分、非常に魅力的ですが、
こちらは内部的にスレッド何個で処理していることになるのでしょうか。
スレッドの場合やマルチプロセスの場合、数を増やせば、場合によっては処理改善できます。

[tiitoi]

ThreadPoolExecutor の引数でスレッド数を指定します。
数を増やせば、論理コア数までは速度改善が期待できます。

```
ThreadPoolExecutor(max_workers=16)
```

[goki_gottan]

with futures.ThreadPoolExecutor(16) as executor:
このようにしたら良いのですね？試してみます。

[goki_gottan]

非常に高速で速いです。ありがとうございます。この時点で、ベストアンサーにしたく思いますが、当初の別スレッドにしての動作を求めていた、もう一つの理由は、AスレッドとBスレッドで別処理ができる応用も考えておりました。画像の重複はありませんが、その画像らをAスレッドへ、Bスレッドへみたいにすれば、別処理ができます。たとえば、スレッドAはAフォルダへ移動、スレッドBはBスレッドへ移動なのです。
そちらの場合は、ThreadPoolExecutorでできますでしょうか。

[tiitoi]

関数の引数経由で関数に処理内容を指示すればいいのではないでしょうか。
出力先を変えるぐらいであれば、出力先のディレクトリパスを第2引数に渡せばいいと思います。
submit() の第2引数以降に関数に渡す引数をいくつでも指定できるので

executor.submit(func, arg1. arg2. arg3)