La inteligencia artificial en el perímetro es uno de los escenarios perimetrales más populares. Las implementaciones de este escenario incluyen la clasificación de imágenes, la detección de objetos, el cuerpo, la cara y el análisis de gestos, y la manipulación de imágenes. En esta guía de arquitectura se describe cómo usar Azure IoT Edge para admitir estos escenarios.
Puede mejorar la precisión de la inteligencia artificial mediante la actualización del modelo de IA, pero en algunos escenarios el entorno de red del dispositivo perimetral no es recomendable. Por ejemplo, en las industrias de energía eólica y petróleo, el equipo podría encontrarse en el desierto o en el océano.
Los módulos gemelos de IoT Edge se usan para implementar el modelo de inteligencia artificial cargado dinámicamente. Los módulos de IoT Edge se basan en Docker. Normalmente, una imagen de un módulo de IoT Edge en un entorno de inteligencia artificial tiene un tamaño de al menos 1 GB, por lo que la actualización incremental del modelo de IA es importante en una red con un ancho de banda limitado. Esa consideración es el enfoque principal de este artículo. La idea es crear un módulo de inteligencia artificial de IoT Edge AI que pueda cargar modelos de detección de objetos LiteRT (antes conocido como TensorFlow Lite) u Open Neural Network Exchange (ONNX). También puede habilitar el módulo como una API web para que pueda usarlo para beneficiar a otras aplicaciones o módulos.
La solución descrita en este artículo puede ayudarle de estas maneras:
- Habilite la inferencia de IA en dispositivos perimetrales.
- Minimice el costo de red de la implementación y actualización de modelos de inteligencia artificial en el perímetro. La solución puede ahorrar dinero para usted o sus clientes, especialmente en un entorno de una red con un ancho de banda limitado.
- Cree y administre un repositorio de modelos de IA en el almacenamiento local de un dispositivo de IoT Edge.
- Logre un tiempo de inactividad prácticamente inexistente cuando el dispositivo perimetral cambia los modelos de IA.
TensorFlow y LiteRT son una marca comercial de Google Inc. El uso de esta marca no implica ninguna aprobación.
Arquitectura
Descargue un archivo Visio de esta arquitectura.
Flujo de datos
- El modelo de IA se carga en Azure Blob Storage o en un servicio web. El modelo puede ser un modelo LiteRT u ONNX previamente entrenado o un modelo creado en Azure Machine Learning. El módulo de IoT Edge puede acceder a este modelo y descargarlo en el dispositivo perimetral más adelante. Si necesita una mejor seguridad, considere la posibilidad de usar conexiones de punto de conexión privado entre Blob Storage y el dispositivo perimetral.
- Azure IoT Hub sincroniza automáticamente los módulos gemelos de dispositivo con la información del modelo de IA. La sincronización se produce incluso si IoT Edge ha estado sin conexión. (En algunos casos, los dispositivos IoT están conectados a redes a horas programadas, diarias o semanales, para ahorrar energía o reducir el tráfico de red).
- El módulo del cargador supervisa las actualizaciones de los módulos gemelos mediante la API. Cuando detecta una actualización, obtiene el token de SAS del modelo de Machine Learning y, a continuación, descarga el modelo de IA.
- Para obtener más información, consulte Creación de un token de SAS para un contenedor o blob.
- Puede usar la propiedad ExpiresOn para establecer la fecha de expiración de los recursos. Si el dispositivo estará sin conexión durante mucho tiempo, puede ampliar la fecha de expiración.
- El módulo del cargador guarda el modelo de IA en el almacenamiento local compartido del módulo de IoT Edge. Debe configurar el almacenamiento local compartido en el archivo JSON de implementación de IoT Edge.
- El módulo del cargador carga el modelo de IA desde el almacenamiento local mediante LiteRT o la API de ONNX.
- El módulo del cargador inicia una API web que recibe la foto binaria a través de la solicitud POST y devuelve los resultados en un archivo JSON.
Para actualizar el modelo de IA, puede cargar la nueva versión en Blob Storage y volver a sincronizar los módulos gemelos del dispositivo para una actualización incremental. No es necesario actualizar toda la imagen del módulo de IoT Edge.
Detalles del escenario
En esta solución, se usa un módulo de IoT Edge para descargar un modelo de IA y, a continuación, habilitar la inferencia de aprendizaje automático. Puede usar los modelos LiteRT u ONNX previamente entrenados en esta solución.
LiteRT
Un archivo
.tflitees un modelo de IA previamente entrenado. Puede descargar uno de TensorFlow.org. Es un modelo de inteligencia artificial genérico que puede usar en aplicaciones multiplataforma como iOS y Android. LiteRT es compatible con modelos de TensorFlow, PyTorch, JAX y Keras. Para obtener más información sobre los metadatos y los campos asociados (por ejemplo,labels.txt), consulte Lectura de los metadatos de los modelos.Un modelo de detección de objetos se entrena para detectar la presencia y la ubicación de varias clases de objetos. Por ejemplo, un modelo podría entrenarse con imágenes que contienen varias frutas, junto con una etiqueta que especifica la clase de fruta que representan (por ejemplo, manzana) y datos que especifican dónde aparece cada objeto en la imagen.
Cuando se proporciona una imagen al modelo, genera una lista de los objetos que detecta, la ubicación de un rectángulo de selección para cada objeto y una puntuación que indica la confianza de la detección.
Si quiere crear o ajustar un modelo de IA personalizado, consulte LiteRT Model Maker.
Puede obtener más modelos de detección previamente entrenados gratuitos, con varias características de latencia y precisión, en Detection Zoo. Cada modelo usa las firmas de entrada y salida que se muestran en los ejemplos de código siguientes.
ONNX
ONNX es un formato estándar abierto para representar modelos de aprendizaje automático. Es compatible con una comunidad de asociados que lo han implementado en muchos marcos y herramientas.
- ONNX admite herramientas para crear e implementar modelos y para realizar otras tareas. Para obtener más información, consulte Herramientas de ONNX compatibles.
- Puede usar ONNX Runtime para ejecutar modelos previamente entrenados de ONNX. Para obtener información sobre los modelos previamente entrenados, consulte ONNX Model Zoo.
- En este escenario, puede usar un modelo de segmentación de imágenes y detección de objetos: Tiny YOLOv3.
La comunidad de ONNX proporciona herramientas para ayudarle a crear e implementar el modelo de aprendizaje profundo.
Descarga de modelos de IA entrenados
Para descargar modelos de inteligencia artificial entrenados, se recomienda usar dispositivos gemelos para recibir notificaciones cuando un nuevo modelo esté listo. Incluso si el dispositivo está sin conexión, el mensaje se puede almacenar en caché en IoT Hub hasta que el dispositivo perimetral vuelva a estar en línea. El mensaje se sincronizará automáticamente.
A continuación se muestra un ejemplo de código de Python que registra notificaciones para los dispositivos gemelos y, a continuación, descarga el modelo de IA en un archivo ZIP. También realiza operaciones adicionales en el archivo descargado.
El código realiza estas tareas:
- Reciba la notificación de dispositivos gemelos. La notificación incluye el nombre de archivo, la dirección de descarga de archivos y el token de autenticación MD5. (En el nombre de archivo, puede incluir información de versión, como 1.0)
- Descargue el modelo de IA como un archivo ZIP en el almacenamiento local.
- Opcionalmente, realice la suma de comprobación MD5. La verificación MD5 ayuda a evitar archivos ZIP que se han alterado durante la transmisión de red.
- Descomprima el archivo ZIP y guárdelo localmente.
- Envíe una notificación a IoT Hub o un mensaje de enrutamiento para informar de que el nuevo modelo de IA está listo.
# 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 )
Inferencia
Una vez descargado el modelo de IA, el siguiente paso consiste en usar el modelo en el dispositivo perimetral. Puede cargar dinámicamente el modelo y realizar la detección de objetos en dispositivos perimetrales. En el ejemplo de código siguiente se muestra cómo usar el modelo de IA LiteRT para detectar objetos en dispositivos perimetrales.
El código realiza estas tareas:
- Cargue dinámicamente el modelo de IA de LiteRT.
- Realice la estandarización de la imagen.
- Detección de objetos.
- Procese las puntuaciones de detección.
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)
A continuación se muestra la versión de ONNX del código anterior. Los pasos son principalmente los mismos. La única diferencia es cómo se controla la puntuación de detección, ya que Labelmap y los parámetros de salida del modelo son diferentes.
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)
Si el dispositivo de IoT Edge incorpora el código y las características anteriores, el dispositivo perimetral tiene detección de objetos de imagen de IA y admite la actualización dinámica de modelos de IA. Si quiere que el módulo perimetral proporcione funcionalidad de inteligencia artificial a otras aplicaciones o módulos mediante una API web, puede crear una API web en el módulo.
El marco de Flask es un ejemplo de una herramienta que puede usar para crear rápidamente una API. Puede recibir imágenes como datos binarios, usar un modelo de IA para la detección y, a continuación, devolver los resultados en un formato JSON. Para obtener más información, consulte Flask: Tutorial de Flask en Visual Studio Code.
Colaboradores
Microsoft mantiene este artículo. Originalmente lo escribieron los siguientes colaboradores.
Autor principal:
- Bo Wang | Ingeniero de software sénior
Otro colaborador:
- Freddy Ayala | Arquitecto de soluciones en la nube
Para ver los perfiles no públicos de LinkedIn, inicie sesión en LinkedIn.
Pasos siguientes
- Descripción y uso de módulos gemelos en IoT Hub
- Obtenga información sobre cómo implementar módulos y establecer rutas en IoT Edge
- Acceso de los módulos al almacenamiento local de un dispositivo
- Descripción de las implementaciones automáticas de IoT Edge en un único dispositivo o a escala
- Open Neural Network Exchange
- Tutoriales de ONNX
- Implementación de un modelo de Machine Learning en dispositivos IoT y perimetrales
Recursos relacionados
- arquitecturas de IoT de
- Elección de una solución de IoT en Azure
- Diseño de arquitectura de IA