RxPYを使ってOpenCVするとき、asyncio使えば別スレッドからでもimshowできる #Python

RxPY+OpenCVでimshowをするとき、asyncioを使えばsubscribeがメインスレッドじゃなくてもimshowすることができたのでメモとして残しておく。

＜2019年10月9日追記＞
タイトルに語弊がありました。imshow自体は別スレッドで使っているわけではないです。

サンプルコード(失敗例)

下のコードはRxPYで画像処理を適用し、最後に画面に表示するためのプログラムです。
一見、問題ないように思いますが、このコードは最後の「show」関数がメインスレッドではない、別のスレッドで実行されるため、cv2.imshow関数でエラー終了してしまいます。

import cv2
import rx
from rx import Observable

def show(image):
    """
    動画を画面に表示する
    """
    cv2.imshow('res', image)
    cv2.waitKey(1)

def imgproc(image):
    """
    画像処理するプログラム (とりあえずここではグレースケールにするだけ)
    """
    return cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

video = cv2.VideoCapture(0)

# 1. 15FPSでインターバル
# 2. VideoCaptureからデータを取り出す
# 3. ちゃんと画像が取り出せた時だけ処理する
# 4. 画像だけを次のストリームに渡す
# 5. 画像処理を適用する
# 6. 結果を画面に表示する
Observable.interval(1 / 15 * 1000) \
    .map(lambda x: video.read()) \
    .filter(lambda data: data[0]) \
    .map(lambda data: data[1]) \
    .map(imgproc) \
    .subscribe(show)

# プログラムが停止しないように入力待ちする
a = input()

asyncioを使う

asyncioはPythonで非同期するためのモジュールで、いわゆる「ノンブロッキング処理」を実現できます。
このasyncioに、「run_coroutine_threadsafe」という、別スレッドで実行されているイベントループに処理を渡すことができる関数があります。

これを使って、メインスレッドでイベントループを作成し、run_coroutine_threadsafeを使ってimshowを呼び出してやれば上記のようなPGでもimshowを実行することができる。

サンプル

import cv2
import rx
from rx import Observable
import asyncio

def imgproc(image):
    """
    画像処理するプログラム (とりあえずここではグレースケールにするだけ)
    """
    return cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

video = cv2.VideoCapture(0)
# イベントループを作成
loop = asyncio.get_event_loop()

async def show_coroutine(image):
    """
    asyncioで実行するために、GUI表示部分をコルーチン化
    """
    cv2.imshow('res', image)
    return cv2.waitKey(1) & 0xff

def show(image):
    """
    run_coroutine_threadsafeでコルーチン
    """
    future = asyncio.run_coroutine_threadsafe(show_coroutine(image), loop)
    key = future.result()
    if key == ord('q'): # qが押されたらループを停止する
        loop.stop()

Observable.interval(1 / 15 * 1000) \
    .map(lambda x: video.read()) \
    .filter(lambda data: data[0]) \
    .map(lambda data: data[1]) \
    .map(imgproc) \
    .subscribe(show)

# イベントループを開始する
loop.run_forever()
video.release()
loop.close()

ちなみに、RxPYにはAsyncIOSchedulerというスケジューラもあるので、
もしかするともっと上手く書くこともできるかもしれないです。