サポート終了通知: 2025 年 10 月 31 日、 AWS は Amazon Lookout for Vision のサポートを終了します。2025 年 10 月 31 日以降、Lookout for Vision コンソールまたは Lookout for Vision リソースにアクセスできなくなります。詳細については、このブログ記事 を参照してください。

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

クライアントアプリケーションコンポーネントでのモデルの使用

クライアントアプリケーションコンポーネントからのモデルを使用するステップは、クラウドでホストされているモデルを使用するステップと似ています。

Amazon Lookout for Vision Edge Agent には、モデルを開始し、画像内の異常を検出し、モデルを停止するための API が用意されています。API を使用してデバイス上のモデルをリストし、デプロイされたモデルに関する情報を取得することもできます。詳細については「Amazon Lookout for Vision Edge Agent API リファレンス」を参照してください。

gRPC ステータスコードをチェックすることでエラー情報を取得できます。詳細については「エラー情報の取得」を参照してください。

コードを記述するには、gRPC がサポートする任意の言語を使用できます。Python コードの例を提供します。

クライアントアプリケーションコンポーネントでのスタブの使用

次のコードを使用して、Lookout for Vision Edge Agent を通じてモデルへのアクセスをセットアップします。

import grpc
from edge_agent_pb2_grpc import EdgeAgentStub
import edge_agent_pb2 as pb2

# Creating stub.
 with grpc.insecure_channel("unix:///tmp/aws.iot.lookoutvision.EdgeAgent.sock") as channel:
    stub = EdgeAgentStub(channel)
    # Add additional code that works with Edge Agent in this block to prevent resources leakage

モデルの開始

StartModel API を呼び出してモデルを開始します。モデルの開始までに時間がかかる場合があります。DescribeModel を呼び出すことで、現在のステータスをチェックできます。status フィールドの値が「実行中」の場合、モデルは実行中です。

コードの例

component_name はモデルコンポーネントの名前に置き換えてください。

import time

import grpc
from edge_agent_pb2_grpc import EdgeAgentStub
import edge_agent_pb2 as pb2


model_component_name = "component_name"


def start_model_if_needed(stub, model_name):
    # Starting model if needed.
    while True:
        model_description_response = stub.DescribeModel(pb2.DescribeModelRequest(model_component=model_name))
        print(f"DescribeModel() returned {model_description_response}")
        if model_description_response.model_description.status == pb2.RUNNING:
            print("Model is already running.")
            break
        elif model_description_response.model_description.status == pb2.STOPPED:
            print("Starting the model.")
            stub.StartModel(pb2.StartModelRequest(model_component=model_name))
            continue
        elif model_description_response.model_description.status == pb2.FAILED:
            raise Exception(f"model {model_name} failed to start")
        print(f"Waiting for model to start.")
        if model_description_response.model_description.status != pb2.STARTING:
            break
        time.sleep(1.0)


# Creating stub.
with grpc.insecure_channel("unix:///tmp/aws.iot.lookoutvision.EdgeAgent.sock") as channel:
    stub = EdgeAgentStub(channel)
    start_model_if_needed(stub, model_component_name)

異常の検出

画像内の異常を検出するには DetectAnomalies API を使用します。

DetectAnomalies オペレーションでは、画像ビットマップが RGB888 圧縮形式で渡されることを想定しています。最初のバイトは赤チャンネル、2 番目のバイトは緑チャンネル、3 番目のバイトは青チャンネルを表します。BGR などの別の形式で画像を提供した場合、DetectAnomalies による予測は正しくありません。

デフォルトでは、OpenCV は画像ビットマップに BGR 形式を使用します。OpenCV を使用して DetectAnomalies により分析用の画像をキャプチャする場合は 、画像を DetectAnomalies に渡す前に画像を RGB888 形式に変換する必要があります。

DetectAnomalies に提供する画像は、モデルのトレーニングに使用した画像と同じ幅と高さである必要があります。

画像バイトによる異常の検出

画像を画像バイトとして提供することで、画像内の異常を検出できます。次の例では、ローカルファイルシステムに保存されている画像から画像バイトが取得されます。

sample.jpg を、分析する画像ファイルの名前に置き換えてください。component_name を、モデルコンポーネントの名前に置き換えてください。

import time

from PIL import Image
import grpc
from edge_agent_pb2_grpc import EdgeAgentStub
import edge_agent_pb2 as pb2


model_component_name = "component_name"

....
# Detecting anomalies.
def detect_anomalies(stub, model_name, image_path):
    image = Image.open(image_path)
    image = image.convert("RGB")
    detect_anomalies_response = stub.DetectAnomalies(
        pb2.DetectAnomaliesRequest(
            model_component=model_name,
            bitmap=pb2.Bitmap(
                width=image.size[0],
                height=image.size[1],
                byte_data=bytes(image.tobytes())
            ) 
        )
    )
    print(f"Image is anomalous - {detect_anomalies_response.detect_anomaly_result.is_anomalous}")
    return detect_anomalies_response.detect_anomaly_result


# Creating stub.
with grpc.insecure_channel("unix:///tmp/aws.iot.lookoutvision.EdgeAgent.sock") as channel:
    stub = EdgeAgentStub(channel)
    start_model_if_needed(stub, model_component_name)
    detect_anomalies(stub, model_component_name, "sample.jpg")

共有メモリーセグメントを使った異常検知

画像を POSIX 共有メモリセグメントの画像バイトとして提供することで、画像内の異常を検出できます。最高のパフォーマンスを得るために、DetectAnomalies リクエストに共有メモリを使用することをお勧めします。詳細については「DetectAnomalies」を参照してください。

モデルの停止

モデルを使用しなくなった場合は、StopModel API を使用してモデルの実行を停止します。

stop_model_response = stub.StopModel(
    pb2.StopModelRequest(
         model_component=model_component_name   
    )
)
print(f"New status of the model is {stop_model_response.status}") 

デバイス上のモデルの一覧表示

ListModels API を使用して、デバイスにデプロイされているモデルを一覧表示できます。

models_list_response = stub.ListModels(
    pb2.ListModelsRequest()
)
for model in models_list_response.models:
    print(f"Model Details {model}")

モデルの記述

DescribeModel API を呼び出すと、デバイスにデプロイされたモデルに関する情報を取得できます。DescribeModel を使用すると、モデルの現在のステータスを取得するのに便利です。たとえば、DetectAnomalies を呼び出す前にモデルが実行中かどうかを知る必要があります。サンプルコードについては、「モデルの開始」を参照してください。

エラー情報の取得

gRPC ステータスコードは API 結果の報告に使用されます。

次の例に示すように、RpcError 例外をキャッチすることでエラー情報を取得できます。エラーステータスコードの詳細については、API のリファレンストピックを参照してください。

# Error handling.
try:
    stub.DetectAnomalies(detect_anomalies_request)
except grpc.RpcError as e:
    print(f"Error code: {e.code()}, Status: {e.details()}")