在您的客户端应用程序组件中使用模型 - Amazon Lookout for Vision
在客户端应用程序组件中使用存根启动模型检测异常停止模型列出设备上的模型描述模型获取错误信息

终止支持通知：2025年10月31日， AWS 将停止对亚马逊 Lookout for Vision 的支持。2025 年 10 月 31 日之后，你将无法再访问 Lookout for Vision 主机或 Lookout for Vision 资源。如需更多信息，请访问此博客文章。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

在您的客户端应用程序组件中使用模型

从客户端应用程序组件使用模型时，步骤与使用云端托管的模型类似。

  1. 开始运行模型。

  2. 检测图像中的异常。

  3. 停止模型（如果不再需要）。

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 字段的值为 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)

检测异常

您应使用 DetectAnomalies API 来检测图像中的异常。

DetectAnomalies 操作期望图像位图以 RGB888 压缩格式进行传递。第一个字节代表红色通道，第二个字节代表绿色通道，第三个字节代表蓝色通道。如果您以其他格式（如 BGR）提供图像，则 DetectAnomalies 所做的预测会不正确。

默认情况下，OpenCV 对图像位图使用 BGR 格式。如果要使用 OpenCV 捕获图像以供 DetectAnomalies 分析，则必须先将图像转换为 RGB888 格式，然后才能将其传递给 DetectAnomalies。

在宽度和高度尺寸上，您提供给 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()}")