【GCP】Google Cloud Vision APIのOCRチュートリアルで遊んでみました ~GCFを用いたOCRドキュメントを理解していく~
はじめに
こんにちは、がんがんです。前回の備忘録にてCloud Runについて試してみました。
gangannikki.hatenadiary.jp
次のステップとして、Cloud Vision APIを触っております。将来的にはPDFの解析 → DBに保存ということを実行していきたいと考えております。
ドキュメントを見てみるとCloud Functionsを用いたOCRのチュートリアルがありました。今回はドキュメントを通じながらVision APIやCloud Functionsの使い方を学んでいきたいと思います。
目次
今回やっていくこと
今回の手順は以下の通りです(本家より引用)。今回使用する言語はPythonを選択します。
OCR チュートリアル アプリケーションのデータの流れでは、次の手順が行われます。
1. 任意の言語のテキストを含む画像が Cloud Storage にアップロードされます。
2. Cloud 関数がトリガーされ、Vision API を使用してテキストを抽出し、ソース言語を検出します。
3. Pub/Sub トピックにメッセージが発行されることで、テキストが翻訳のためにキューに配置されます。翻訳は、ソース言語とは異なるターゲット言語ごとにキューに配置されます。
4. ターゲット言語がソース言語と一致する場合、翻訳キューがスキップされ、テキストは結果キュー（別の Pub/Sub トピック）に送信されます。
5. Cloud 関数が、Translation API を使用して翻訳キューのテキストを翻訳します。翻訳結果は結果キューに送信されます。
6. 別の Cloud 関数が、翻訳されたテキストを結果キューから Cloud Storage に保存します。
7. 結果は、翻訳ごとに txt ファイルとして Cloud Storage に保存されます。
大雑把にまとめると以下の通りです。
1. 画像をCloud Storageにアップロード 2. Vision APIでテキスト検出し、Translation APIで言語翻訳 3. 翻訳結果をCloud Storageに保存
下準備
まずはプロジェクトやCloud Storageのバケットを準備していきます。コードはGitHubのコードをcloneして理解していきます。
$ git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
次に、Cloud Storageのバケットを準備します。gsutilはStorageで使用するコマンドです。
# 画像をアップロードするバケット $ gsutil mb gs://YOUR_IMAGE_BUCKET_NAME # 翻訳テキストを保存するバケット $ gsutil mb gs://YOUR_TEXT_BUCKET_NAME
最後に、config.jsonを設定します。TO_LANGが翻訳言語を選択してる部分です。
| python-docs-samples/functions/ocr/app/config.json |
{
"RESULT_TOPIC": "YOUR_RESULT_TOPIC_NAME",
"RESULT_BUCKET": "YOUR_TEXT_BUCKET_NAME",
"TRANSLATE_TOPIC": "YOUR_TRANSLATE_TOPIC_NAME",
"TRANSLATE": true,
"TO_LANG": ["en", "fr", "es", "ja", "ru"]
}
コードの詳細と手順
ドキュメント、コードを見ながら1つずつ内容を理解していきます。
GitHubのコード
綺麗な全体コードはこちらを参照ください。
github.com
画像の処理部分
この部分では、画像からテキストを抽出する部分を処理しています。Vision APIとTranslation APIに関する参考ドキュメントをメモとして置いておきます。
画像内のテキストを検出する | Cloud Vision API | Google Cloud
言語の検出 | Cloud Translation | Google Cloud
# [START functions_ocr_process] def process_image(file, context): """Cloud Function triggered by Cloud Storage when a file is changed. Args: file (dict): Metadata of the changed file, provided by the triggering Cloud Storage event. context (google.cloud.functions.Context): Metadata of triggering event. Returns: None; the output is written to stdout and Stackdriver Logging """ bucket = validate_message(file, 'bucket') name = validate_message(file, 'name') detect_text(bucket, name) print('File {} processed.'.format(file['name'])) # [END functions_ocr_process] # [START functions_ocr_detect] def detect_text(bucket, filename): print('Looking for text in image {}'.format(filename)) futures = [] # Vision APIを用いてテキスト検出 text_detection_response = vision_client.text_detection({ 'source': {'image_uri': 'gs://{}/{}'.format(bucket, filename)} }) annotations = text_detection_response.text_annotations if len(annotations) > 0: text = annotations[0].description else: text = '' print('Extracted text {} from image ({} chars).'.format(text, len(text))) # Translation APIを用いて言語の特定 detect_language_response = translate_client.detect_language(text) src_lang = detect_language_response['language'] print('Detected language {} for text {}.'.format(src_lang, text)) # Submit a message to the bus for each target language for target_lang in config.get('TO_LANG', []): topic_name = config['TRANSLATE_TOPIC'] if src_lang == target_lang or src_lang == 'und': topic_name = config['RESULT_TOPIC'] message = { 'text': text, 'filename': filename, 'lang': target_lang, 'src_lang': src_lang } message_data = json.dumps(message).encode('utf-8') topic_path = publisher.topic_path(project_id, topic_name) future = publisher.publish(topic_path, data=message_data) futures.append(future) for future in futures: future.result() # [END functions_ocr_detect] # [START message_validatation_helper] def validate_message(message, param): var = message.get(param) if not var: raise ValueError('{} is not provided. Make sure you have \ property {} in the request'.format(param, param)) return var # [END message_validatation_helper]
テキスト翻訳部分
本関数ではテキストの翻訳を行っていきます。
テキストの翻訳 | Cloud Translation | Google Cloud
# [START functions_ocr_translate] def translate_text(event, context): if event.get('data'): message_data = base64.b64decode(event['data']).decode('utf-8') message = json.loads(message_data) else: raise ValueError('Data sector is missing in the Pub/Sub message.') text = validate_message(message, 'text') filename = validate_message(message, 'filename') target_lang = validate_message(message, 'lang') src_lang = validate_message(message, 'src_lang') print('Translating text into {}.'.format(target_lang)) # テキスト翻訳 translated_text = translate_client.translate(text, target_language=target_lang, source_language=src_lang) topic_name = config['RESULT_TOPIC'] message = { 'text': translated_text['translatedText'], 'filename': filename, 'lang': target_lang, } message_data = json.dumps(message).encode('utf-8') topic_path = publisher.topic_path(project_id, topic_name) future = publisher.publish(topic_path, data=message_data) future.result() # [END functions_ocr_translate]
翻訳テキストの保存部分
翻訳テキストをバケットに保存する処理を行っている部分です。
この関数はtranslate_textの部分と比較的似ています。
Cloud Functionsへのデプロイ
完成したコードをfunctionsにデプロイしていきます。各種トリガーについては
gcloud functions deployのドキュメントを参照ください。
# 画像の処理用関数をデプロイ $ gcloud functions deploy ocr-extract --runtime python37 --trigger-bucket YOUR_IMAGE_BUCKET_NAME --entry-point process_image # テキスト翻訳関数をデプロイ $ gcloud functions deploy ocr-translate --runtime python37 --trigger-topic YOUR_TRANSLATE_TOPIC_NAME --entry-point translate_text # 翻訳テキストの保存関数をデプロイ $ gcloud functions deploy ocr-save --runtime python37 --trigger-topic YOUR_RESULT_TOPIC_NAME --entry-point save_result
実験と実験結果
画像をバケットにアップロードする実験を行っていきます。
$ gsutil cp PATH_TO_IMAGE gs://YOUR_IMAGE_BUCKET_NAME
特に問題なければYOUR_TEXT_BUCKET_NAMEにテキストファイルが保存されていると思います。
もし保存されてなければLoggingやError Reportingから問題を検証していきます。
実際に解析したいPDFのスクショ画像で実験してみましたが、思ったよりもきちんと出力できていました。
これは思ったよりも期待が出来そうな予感です。
