에지의 AI는 가장 인기 있는 에지 시나리오 중 하나입니다. 이 시나리오의 구현에는 이미지 분류, 개체 감지, 신체, 얼굴, 제스처 분석과 이미지 조작이 포함됩니다. 이 아키텍처 가이드에서는 Azure IoT Edge를 사용하여 이러한 시나리오를 지원하는 방법을 설명합니다.
AI 모델을 업데이트하여 AI 정확도를 향상시킬 수 있지만 일부 시나리오의 에지 디바이스 네트워크 환경이 좋지 않습니다. 예를 들어 풍력 발전 및 석유 산업에서 장비가 사막이나 바다에 위치할 수 있습니다.
IoT Edge 모듈 쌍은 동적으로 로드되는 AI 모델을 구현하는 데 사용됩니다. IoT Edge 모듈은 Docker를 기반으로 합니다. AI 환경의 IoT Edge 모듈에 대한 이미지의 크기는 일반적으로 1GB 이상이므로 좁은 대역폭 네트워크에서 AI 모델을 증분 업데이트하는 것이 중요합니다. 이 고려 사항은 이 문서의 주요 초점입니다. 이 아이디어는 LiteRT(이전의 TensorFlow Lite) 또는 ONNX(Open Neural Network Exchange) 개체 검색 모델을 로드할 수 있는 IoT Edge AI 모듈을 만드는 것입니다. 모듈을 웹 API로 사용하도록 설정하여 다른 애플리케이션이나 모듈에 도움이 될 수 있습니다.
이 문서에서는 다음과 같은 방법의 솔루션을 제공합니다.
- 에지 디바이스에서 AI 유추를 사용하도록 설정합니다.
- 에지에서 AI 모델을 배포하고 업데이트하는 데 드는 네트워크 비용을 최소화합니다. 이 솔루션은 특히 좁은 대역폭 네트워크 환경에서 사용자 또는 고객의 비용을 절감할 수 있도록 합니다.
- IoT Edge 디바이스의 로컬 스토리지에서 AI 모델 리포지토리를 만들고 관리합니다.
- 에지 디바이스가 AI 모델을 전환할 때 가동 중지 시간이 거의 발생하지 않습니다.
TensorFlow 및 LiteRT 는 Google Inc.의 상표입니다. 이 마크의 사용에 의해 어떠한 보증도 암시되지 않습니다.
아키텍처
이 아키텍처의 Visio 파일을 다운로드합니다.
데이터 흐름
- AI 모델은 Azure Blob Storage 또는 웹 서비스에 업로드됩니다. 모델은 미리 학습된 LiteRT 또는 ONNX 모델 또는 Azure Machine Learning에서 만든 모델일 수 있습니다. IoT Edge 모듈은 이 모델에 액세스하여 나중에 에지 디바이스에 다운로드할 수 있습니다. 더 나은 보안이 필요한 경우 Blob Storage와 에지 디바이스 간에 프라이빗 엔드포인트 연결을 사용하는 것이 좋습니다.
- Azure IoT Hub는 디바이스 모듈 쌍을 AI 모델 정보와 자동으로 동기화합니다. 동기화는 IoT Edge가 오프라인 상태인 경우에도 발생합니다. (경우에 따라 IoT 디바이스는 전원을 절약하거나 네트워크 트래픽을 줄이기 위해 예약된 시간별, 일별 또는 주별로 네트워크에 연결됩니다.)
- 로더 모듈은 API를 통해 모듈 쌍의 업데이트를 모니터링합니다. 업데이트를 감지하면 기계 학습 모델 SAS 토큰을 가져온 다음 AI 모델을 다운로드합니다.
- 자세한 내용은 컨테이너 또는 Blob의 SAS 토큰 만들기를 참조하세요.
- ExpiresOn 속성을 사용하여 리소스의 만료 날짜를 설정할 수 있습니다. 디바이스가 오랫동안 오프라인 상태일 경우 만료 시간을 연장할 수 있습니다.
- 로더 모듈은 IoT Edge 모듈의 공유 로컬 스토리지에 AI 모델을 저장합니다. IoT Edge 배포 JSON 파일에서 공유 로컬 스토리지를 구성해야 합니다.
- 로더 모듈은 LiteRT 또는 ONNX API를 통해 로컬 스토리지에서 AI 모델을 로드합니다.
- 로더 모듈은 POST 요청을 통해 이진 사진을 수신하고 결과를 JSON 파일로 반환하는 웹 API를 시작합니다.
AI 모델을 업데이트하려면 Blob Storage에 새 버전을 업로드하고 증분 업데이트를 위해 디바이스 모듈 쌍을 다시 동기화할 수 있습니다. 전체 IoT Edge 모듈 이미지를 업데이트할 필요가 없습니다.
시나리오 정보
이 솔루션에서는 IoT Edge 모듈을 사용하여 AI 모델을 다운로드한 다음, 기계 학습 유추를 사용하도록 설정합니다. 이 솔루션에서 미리 학습된 LiteRT 또는 ONNX 모델을 사용할 수 있습니다.
LiteRT
.tflite파일은 미리 학습된 AI 모델입니다. TensorFlow.org에서 파일을 다운로드할 수 있습니다. iOS 및 Android와 같은 플랫폼 간 애플리케이션에서 사용할 수 있는 일반적인 AI 모델입니다. LiteRT는 TensorFlow, PyTorch, JAX 및 Keras의 모델을 지원합니다. 메타데이터 및 관련 필드(예:labels.txt)에 대한 자세한 내용은 모델에서 메타데이터 읽기를 참조하세요.개체 감지 모델은 여러 개체 클래스의 존재와 위치를 감지하도록 학습됩니다. 예를 들어 모델은 다양한 과일 조각을 포함하는 이미지와 해당 모델이 나타내는 과일 클래스(예: 사과)를 지정하는 레이블 및 각 개체가 이미지에 표시되는 위치를 지정하는 데이터로 학습될 수 있습니다.
이미지가 모델에 제공되면 감지하는 개체 목록, 각 개체의 경계 상자 위치, 감지의 신뢰도를 나타내는 점수를 출력합니다.
AI 모델을 빌드하거나 사용자 지정 조정하려면 LiteRT Model Maker를 참조하세요.
감지 동물원에서 다양한 대기 시간 및 정밀도 특성을 갖춘 미리 학습된 더 많은 무료 감지 모델을 얻을 수 있습니다. 각 모델은 다음 코드 샘플에 표시된 입력 및 출력 서명을 사용합니다.
ONNX
ONNX는 기계 학습 모델을 표현하기 위한 개방형 표준 형식입니다. 많은 프레임워크 및 도구에서 구현한 파트너 커뮤니티에서 지원됩니다.
- ONNX는 모델을 구축 및 배포하고 다른 작업을 수행하기 위한 도구를 지원합니다. 자세한 내용은 지원되는 ONNX 도구를 참조하세요.
- ONNX 런타임을 사용하여 ONNX 미리 학습된 모델을 실행할 수 있습니다. 미리 학습된 모델에 대한 자세한 내용은 ONNX 모델 동물원을 참조하세요.
- 이 시나리오에서는 개체 감지 및 이미지 구분 모델인 Tiny YOLOv3를 사용할 수 있습니다.
ONNX 커뮤니티는 딥 러닝 모델을 만들고 배포하는 데 도움이 되는 도구를 제공합니다.
학습된 AI 모델 다운로드
학습된 AI 모델을 다운로드하려면 디바이스 쌍을 사용하여 새 모델이 준비될 때 알림을 받는 것이 좋습니다. 디바이스가 오프라인 상태인 경우에도 에지 디바이스가 다시 온라인 상태가 될 때까지 IoT Hub에서 메시지를 캐시할 수 있습니다. 메시지가 자동으로 동기화됩니다.
다음은 디바이스 쌍에 대한 알림을 등록한 다음 AI 모델을 ZIP 파일로 다운로드하는 Python 코드의 예입니다. 또한 다운로드한 파일에 대해 추가 작업을 수행합니다.
이 코드는 다음 작업을 수행합니다.
- 디바이스 쌍 알림을 받습니다. 알림에는 파일 이름, 파일 다운로드 주소, MD5 인증 토큰이 포함됩니다. 파일 이름에 1.0과 같은 버전 정보를 포함할 수 있습니다.
- AI 모델을 ZIP 파일로 로컬 스토리지에 다운로드합니다.
- 필요에 따라 MD5 체크섬을 수행합니다. MD5 확인은 네트워크 전송 중에 변조된 ZIP 파일을 방지하는 데 도움이 됩니다.
- ZIP 파일의 압축을 풀고 로컬에 저장합니다.
- IoT Hub에 알림을 보내거나 또는 라우팅 메시지를 보내 새 AI 모델이 준비되었다는 것을 알립니다.
# define behavior for receiving a twin patch
async def twin_patch_handler(patch):
try:
print( "######## The data in the desired properties patch was: %s" % patch)
if "FileName" in patch:
FileName = patch["FileName"]
if "DownloadUrl" in patch:
DownloadUrl = patch["DownloadUrl"]
if "ContentMD5" in patch:
ContentMD5 = patch["ContentMD5"]
FilePath = "/iotedge/storage/" + FileName
# download AI model
r = requests.get(DownloadUrl)
print ("######## download AI Model Succeeded.")
ffw = open(FilePath, 'wb')
ffw.write(r.content)
ffw.close()
print ("######## AI Model File: " + FilePath)
# MD5 checksum
md5str = content_encoding(FilePath)
if md5str == ContentMD5:
print ( "######## New AI Model MD5 checksum succeeded")
# decompressing the ZIP file
unZipSrc = FilePath
targeDir = "/iotedge/storage/"
filenamenoext = get_filename_and_ext(unZipSrc)[0]
targeDir = targeDir + filenamenoext
unzip_file(unZipSrc,targeDir)
# ONNX
local_model_path = targeDir + "/tiny-yolov3-11.onnx"
local_labelmap_path = targeDir + "/coco_classes.txt"
# LiteRT
# local_model_path = targeDir + "/ssd_mobilenet_v1_1_metadata_1.tflite"
# local_labelmap_path = targeDir + "/labelmap.txt"
# message to module
if client is not None:
print ( "######## Send AI Model Info AS Routing Message")
data = "{\"local_model_path\": \"%s\",\"local_labelmap_path\": \"%s\"}" % (filenamenoext+"/tiny-yolov3-11.onnx", filenamenoext+"/coco_classes.txt")
await client.send_message_to_output(data, "DLModelOutput")
# update the reported properties
reported_properties = {"LatestAIModelFileName": FileName }
print("######## Setting reported LatestAIModelName to {}".format(reported_properties["LatestAIModelFileName"]))
await client.patch_twin_reported_properties(reported_properties)
else:
print ( "######## New AI Model MD5 checksum failed")
except Exception as ex:
print ( "Unexpected error in twin_patch_handler: %s" % ex )
유추
AI 모델을 다운로드한 후 다음 단계는 에지 디바이스에서 모델을 사용하는 것입니다. 모델을 동적으로 로드하고 에지 디바이스에서 개체 감지를 수행할 수 있습니다. 다음 코드 예제에서는 LiteRT AI 모델을 사용하여 에지 디바이스에서 개체를 검색하는 방법을 보여줍니다.
이 코드는 다음 작업을 수행합니다.
- LiteRT AI 모델을 동적으로 로드합니다.
- 이미지 표준화를 수행합니다.
- 개체 감지
- 감지 점수 계산
class InferenceProcedure():
def detect_object(self, imgBytes):
results = []
try:
model_full_path = AI_Model_Path.Get_Model_Path()
if(model_full_path == ""):
raise Exception ("PLEASE SET AI MODEL FIRST")
if '.tflite' in model_full_path:
interpreter = tf.lite.Interpreter(model_path=model_full_path)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
input_shape = input_details[0]['shape']
# bytes to numpy.ndarray
im_arr = np.frombuffer(imgBytes, dtype=np.uint8)
img = cv2.imdecode(im_arr, flags=cv2.IMREAD_COLOR)
im_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
im_rgb = cv2.resize(im_rgb, (input_shape[1], input_shape[2]))
input_data = np.expand_dims(im_rgb, axis=0)
interpreter.set_tensor(input_details[0]['index'], input_data)
interpreter.invoke()
output_data = interpreter.get_tensor(output_details[0]['index'])
detection_boxes = interpreter.get_tensor(output_details[0]['index'])
detection_classes = interpreter.get_tensor(output_details[1]['index'])
detection_scores = interpreter.get_tensor(output_details[2]['index'])
num_boxes = interpreter.get_tensor(output_details[3]['index'])
label_names = [line.rstrip('\n') for line in open(AI_Model_Path.Get_Labelmap_Path())]
label_names = np.array(label_names)
new_label_names = list(filter(lambda x : x != '???', label_names))
for i in range(int(num_boxes[0])):
if detection_scores[0, i] > .5:
class_id = int(detection_classes[0, i])
class_name = new_label_names[class_id]
# top, left, bottom, right
results_json = "{'Class': '%s','Score': '%s','Location': '%s'}" % (class_name, detection_scores[0, i],detection_boxes[0, i])
results.append(results_json)
print(results_json)
except Exception as e:
print ( "detect_object unexpected error %s " % e )
raise
# return results
return json.dumps(results)
다음은 이전 코드의 ONNX 버전입니다. 단계는 대부분 동일합니다. Labelmap 및 모델 출력 매개 변수가 다르기 때문에 감지 점수가 처리되는 방식이 다르다는 것이 유일한 차이점입니다.
class InferenceProcedure():
def letterbox_image(self, image, size):
'''resize image with unchanged aspect ratio using padding'''
iw, ih = image.size
w, h = size
scale = min(w/iw, h/ih)
nw = int(iw*scale)
nh = int(ih*scale)
image = image.resize((nw,nh), Image.BICUBIC)
new_image = Image.new('RGB', size, (128,128,128))
new_image.paste(image, ((w-nw)//2, (h-nh)//2))
return new_image
def preprocess(self, img):
model_image_size = (416, 416)
boxed_image = self.letterbox_image(img, tuple(reversed(model_image_size)))
image_data = np.array(boxed_image, dtype='float32')
image_data /= 255.
image_data = np.transpose(image_data, [2, 0, 1])
image_data = np.expand_dims(image_data, 0)
return image_data
def detect_object(self, imgBytes):
results = []
try:
model_full_path = AI_Model_Path.Get_Model_Path()
if(model_full_path == ""):
raise Exception ("PLEASE SET AI MODEL FIRST")
if '.onnx' in model_full_path:
# input
image_data = self.preprocess(imgBytes)
image_size = np.array([imgBytes.size[1], imgBytes.size[0]], dtype=np.float32).reshape(1, 2)
labels_file = open(AI_Model_Path.Get_Labelmap_Path())
labels = labels_file.read().split("\n")
# Loading ONNX model
print("loading Tiny YOLO...")
start_time = time.time()
sess = rt.InferenceSession(model_full_path)
print("loaded after", time.time() - start_time, "s")
input_name00 = sess.get_inputs()[0].name
input_name01 = sess.get_inputs()[1].name
pred = sess.run(None, {input_name00: image_data,input_name01:image_size})
boxes = pred[0]
scores = pred[1]
indices = pred[2]
results = []
out_boxes, out_scores, out_classes = [], [], []
for idx_ in indices[0]:
out_classes.append(idx_[1])
out_scores.append(scores[tuple(idx_)])
idx_1 = (idx_[0], idx_[2])
out_boxes.append(boxes[idx_1])
results_json = "{'Class': '%s','Score': '%s','Location': '%s'}" % (labels[idx_[1]], scores[tuple(idx_)],boxes[idx_1])
results.append(results_json)
print(results_json)
except Exception as e:
print ( "detect_object unexpected error %s " % e )
raise
# return results
return json.dumps(results)
IoT 에지 디바이스에 이전 코드 및 기능이 통합된 경우 에지 디바이스에는 AI 이미지 개체 감지 기능이 있으며 AI 모델의 동적 업데이트를 지원합니다. 에지 모듈이 웹 API를 통해 다른 애플리케이션 또는 모듈에 AI 기능을 제공하도록 하려면 모듈에서 웹 API를 만들 수 있습니다.
Flask 프레임워크는 API를 빠르게 만드는 데 사용할 수 있는 도구의 한 예입니다. 이미지를 이진 데이터로 수신하고 감지를 위해 AI 모델을 사용한 다음, 결과를 JSON 형식으로 반환할 수 있습니다. 자세한 내용은 Flask: Visual Studio Code의 Flask 자습서를 참조하세요.
참가자
Microsoft에서 이 문서를 유지 관리합니다. 원래 다음 기여자가 작성했습니다.
보안 주체 작성자:
- Bo Wang | 선임 소프트웨어 엔지니어
기타 기여자:
- 프레디 아얄라 | 클라우드 솔루션 설계자
비공개 LinkedIn 프로필을 보려면 LinkedIn에 로그인합니다.
다음 단계
- IoT Hub의 모듈 쌍 이해 및 사용
- IoT Edge에서 모듈을 배포하고 경로를 설정하는 방법 알아보기
- 모듈에 디바이스의 로컬 스토리지에 대한 액세스 권한 부여
- 단일 디바이스 또는 대규모 IoT Edge 자동 배포 이해
- Open Neural Network Exchange
- ONNX 자습서
- IoT 및 에지 디바이스에 기계 학습 모델 배포