이게 뭐야?

환경

참고로 한 자료

코드

#! /usr/bin/python
# -*- coding: utf-8 -*-

# Google Cloud Vision API：LABEL_DETECTION

import sys
import base64
import cv2

from requests import Request, Session
import json
import time
import threading

# GCPのAPIキー
api_key = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'

# 検出する顔の数の最大数 (増やすとレスポンスが遅くなる）
max_results = 5

# END POINTS
DISCOVERY_URL = 'https://vision.googleapis.com/v1/images:annotate?key='

# cv画像と画像ファイルへのPathと検出最大数が引数
def googleAPI(img, max_results):
    # 通信不良等を考慮してTry/expectしておく
    try:
        # カメラ画像をJPG画像へ変換
        retval, image = cv2.imencode('.jpg', img)

        # Headerやpayload
        str_headers = {'Content-Type': 'application/json'}
        batch_request = {'requests': [{'image': {'content': base64.b64encode(image)}, 'features': [{'type': 'LABEL_DETECTION', 'maxResults': max_results, }]}]}

        # セッション作ってリクエストSend
        obj_session = Session()
        obj_request = Request("POST", DISCOVERY_URL + api_key, data=json.dumps(batch_request), headers=str_headers)
        obj_prepped = obj_session.prepare_request(obj_request)
        obj_response = obj_session.send(obj_prepped, verify=True, timeout=180)

        # Responseからjsonを抽出
        response_json = json.loads(obj_response.text)

        # return
        return response_json

    except:
        return img, ""

class googleApiThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.return_value = None   # RETURN VALUE
        self.frame = None
        self.flagStop = True # TRUE = LOOPし続ける
        self.processTime = 0.0

    def run(self):
        # ちゃんとしたカメラ画像が格納されるまで少しWait
        # カメラ画像が格納されないうちにgoogleAPIを呼び出すと、cv2.imencodeがExceptionとなる。。
        time.sleep(1)

        while(self.flagStop):

            # 処理時間を測定
            timeStart = time.time()

            # Google Cloud Vision APIの呼び出し
            self.return_value = googleAPI(self.frame, max_results)

            # debug
            # print(self.return_value)

            timeEnd = time.time()
            self.processTime = timeEnd - timeStart

    def set_frame(self, frame):
        self.frame = frame

    def set_stopFlag(self):
        self.flagStop = False

    def get_value(self):
        return self.return_value

    def get_processTime(self):
        return self.processTime

if __name__ == '__main__':
    # カメラ映像の取り込みスタート
    cap = cv2.VideoCapture(0)

    # 別Threadの起動
    threadGoogleApi = googleApiThread()
    threadGoogleApi.start()

    while(True):
        ret, frame = cap.read()

        # 別スレッドの認識処理の画像を更新
        threadGoogleApi.set_frame(frame)

        # 別スレッドの最新の処理結果を受け取る（1秒に1回ぐらいしか更新されないけど）
        response_json = threadGoogleApi.get_value()
        processTime = threadGoogleApi.get_processTime()

        # 'labelAnnotations'があれば何かラベルを検出した
        if response_json is not None:
            if 'labelAnnotations' in response_json['responses'][0]:
                labels = response_json['responses'][0]['labelAnnotations']

                count = 0
                for label in labels:
                    # ラベルを表示（画像に書き込み）
                    cv2.putText(frame, str(label['description']), (0, 80 + 30 * count), cv2.FONT_HERSHEY_PLAIN, 2, (255, 255, 255), 2, cv2.LINE_AA)
                    count += 1

        # 処理速度を表示（画像に書き込み）
        cv2.putText(frame, str('%03.1f' % processTime) + " sec", (0, 30), cv2.FONT_HERSHEY_PLAIN, 2, (255, 255, 255), 2, cv2.LINE_AA)

        cv2.imshow("camera image", frame)
        if cv2.waitKey(1) == 27:    # ESCキーで終了
            break

    # 終了処理
    threadGoogleApi.set_stopFlag()
    cap.release()

실행 결과