클라이언트 애플리케이션 구성 요소에서 모델 사용 - Amazon Lookout for Vision
클라이언트 애플리케이션 구성 요소의 스텁 사용모델 시작이상 감지모델 중지디바이스의 모델 목록모델 설명오류 정보 가져오기

지원 종료 알림: 2025 AWS 년 10월 31일에 는 Amazon Lookout for Vision에 대한 지원을 중단할 예정입니다. 2025년 10월 31일 이후에는 Lookout for Vision 콘솔 또는 Lookout for Vision 리소스에 더 이상 액세스할 수 없습니다. 자세한 내용은 이 블로그 게시물을 참조하세요.

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

클라이언트 애플리케이션 구성 요소에서 모델 사용

클라이언트 애플리케이션 구성 요소의 모델을 사용하는 단계는 클라우드에 호스팅된 모델을 사용하는 단계와 비슷합니다.

  1. 모델 실행을 시작합니다.

  2. 이미지에서 이상을 감지합니다.

  3. 더 이상 필요하지 않은 경우 모델을 중지합니다.

Amazon Lookout for Vision Edge 에이전트는 모델을 시작하고, 이미지에서 이상을 감지하고, 모델을 중지할 수 있는 API를 제공합니다. API를 사용하여 디바이스의 모델을 나열하고 배포된 모델에 대한 정보를 얻을 수도 있습니다. 자세한 내용은 Amazon Lookout for Vision Edge Agent API 참조 섹션을 참조하세요.

gRPC 상태 코드를 확인하여 오류 정보를 얻을 수 있습니다. 자세한 내용은 오류 정보 가져오기 섹션을 참조하세요.

gRPC가 지원하는 모든 언어를 사용하여 코드를 작성할 수 있습니다. 예제 Python 코드를 제공합니다.

클라이언트 애플리케이션 구성 요소의 스텁 사용

다음 코드를 사용하여 Lookout for Vision 엣지 에이전트를 통해 모델에 대한 액세스를 설정합니다.

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 필드 값이 Running이면 모델이 실행 중인 것입니다.

예제 코드

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)

이상 감지

DetectAnomaliesAPI를 사용하여 이미지의 이상을 감지합니다.

이 DetectAnomalies 작업에서는 이미지 비트맵이 RGB888 압축 형식으로 전달될 것으로 예상합니다. 첫 번째 바이트는 빨간색 채널을, 두 번째 바이트는 녹색 채널을, 세 번째 바이트는 파란색 채널을 나타냅니다. 이미지를 다른 형식 (예: 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()}")