Procesamiento de datos no estructurados desde la nube hasta el borde
Los datos no estructurados ahora representan alrededor del 80% de los datos que manejamos, desde texto e imágenes hasta videos, audio y formatos más complejos. A medida que crece su volumen, también aumenta la necesidad de mejores formas de procesarlos y analizarlos.
En un webinar reciente, Tim Spann, Principal Developer Advocate en Zilliz, compartió sus perspectivas sobre la gestión de datos no estructurados, particularmente desde entornos en la nube hasta dispositivos edge. También destacó cómo las bases de datos vectoriales como Milvus son clave para dar sentido a estos datos.
En esta publicación, recapitularemos los puntos principales de Tim, analizando los desafíos al manejar datos no estructurados y cómo las bases de datos vectoriales lideran el camino. También veremos cómo los dispositivos edge impulsados por estas bases de datos están promoviendo nuevos avances en IA. Si te interesan más detalles, recomendamos ver la repetición completa de la charla de Tim en YouTube.
Comprender los datos no estructurados
Los datos no estructurados se refieren a información que carece de un formato o esquema predefinido. A diferencia de los datos estructurados, que están organizados ordenadamente en bases de datos con campos y formatos específicos, los datos no estructurados son sin procesar y no están organizados. A menudo incluyen:
Texto: Correos electrónicos, documentos de Word, publicaciones en redes sociales
Imágenes: Fotos, capturas de pantalla, diagramas
Videos: Reuniones grabadas, material de vigilancia, contenido multimedia
Audio: Grabaciones de voz, archivos de música, podcasts
Con el aumento del contenido digital, los datos no estructurados han crecido exponencialmente, ofreciendo tanto oportunidades como desafíos para las organizaciones. Para aprovechar esta abundancia de información, las empresas pueden:
Mejorar la experiencia del cliente analizando las interacciones y los comentarios de los clientes
Mejorar la eficiencia operativa mediante información obtenida de datos de sensores y registros del sistema
Impulsar la innovación aprovechando contenido multimedia para el desarrollo de productos
Las bases de datos tradicionales, diseñadas para manejar datos estructurados, tienen dificultades con la irregularidad de los datos no estructurados, lo que dificulta buscar, consultar o extraer información significativa. Aquí es donde las bases de datos vectoriales se vuelven esenciales, ofreciendo una solución para gestionar y recuperar información no estructurada de manera eficiente.
¿Por qué usar una base de datos vectorial para el procesamiento de datos no estructurados?
Una base de datos vectorial es una base de datos especializada diseñada para almacenar, indexar y consultar representaciones vectoriales de datos (también conocidas como embeddings vectoriales), a menudo derivadas de datos no estructurados como texto, imágenes y audio. Estas bases de datos permiten búsquedas de similitud eficientes en un espacio de alta dimensión, lo que las hace ideales para la búsqueda semántica, el procesamiento del lenguaje natural (NLP), los sistemas de recomendación, las búsquedas de imágenes y la generación aumentada por recuperación (RAG).
Las bases de datos vectoriales como Milvus y Zilliz Cloud ofrecen una variedad de funcionalidades más allá de las capacidades de búsqueda de alto rendimiento que las hacen cruciales para gestionar datos no estructurados.
Más allá de la búsqueda de alto rendimiento
Aunque las bases de datos vectoriales suelen asociarse con capacidades rápidas de búsqueda por similitud, su utilidad va mucho más allá. Proporcionan otras funciones esenciales que garantizan la gestión y utilización eficaces de datos no estructurados:
Operaciones CRUD: Las bases de datos vectoriales, al igual que las bases de datos tradicionales, te permiten Crear, Leer, Actualizar y Eliminar (CRUD) datos. Esto garantiza que, pese a trabajar con tipos de datos complejos, las operaciones básicas sigan siendo intuitivas y accesibles.
Actualidad de los datos: Una de las principales fortalezas de las bases de datos vectoriales es garantizar que tus datos se mantengan actualizados. En casos de uso como los sistemas de recomendación, mantener los datos al día es esencial para generar información precisa.
Persistencia: A diferencia de las estructuras de datos en memoria, las bases de datos vectoriales ofrecen almacenamiento persistente, lo que significa que tus datos se almacenan de forma segura y permanecen accesibles incluso después de reinicios del sistema.
Disponibilidad: Las bases de datos vectoriales mantienen los datos fácilmente accesibles para consultas y recuperación en tiempo real. Esto garantiza que, incluso a medida que tus datos crecen, puedas seguir realizando búsquedas rápidas y eficientes, lo que permite tomar decisiones impulsadas por IA sin demora.
Escalabilidad: A medida que los datos aumentan en volumen, las bases de datos vectoriales escalan de manera eficiente para adaptarse a ese crecimiento sin degradación del rendimiento. Esto es esencial al gestionar miles de millones de puntos de datos no estructurados.
Gestión completa de datos
Más allá de las funciones anteriores, las bases de datos vectoriales proporcionan herramientas sólidas para gestionar tus datos no estructurados, incluida la ingesta de datos, la indexación y las consultas, lo que permite una recuperación y un análisis eficientes incluso de los conjuntos de datos más grandes. Funciones como copia de seguridad y migración garantizan que tus datos permanezcan seguros y recuperables, lo que convierte a las bases de datos vectoriales en una opción fiable para gestionar información crítica.
Facilidad operativa
Las bases de datos vectoriales también están diseñadas para simplificar la implementación y la gestión:
Implementación en la nube o local: Muchas bases de datos vectoriales diseñadas específicamente son versátiles y pueden implementarse fácilmente en nubes públicas o en infraestructura local, lo que garantiza flexibilidad según las necesidades de la organización.
Observabilidad: Muchas bases de datos vectoriales ofrecen herramientas de monitoreo para hacer seguimiento del estado y el rendimiento de la base de datos, lo que permite a los usuarios optimizar sus sistemas de forma proactiva.
Multi-tenancy: Las bases de datos vectoriales como Zilliz Cloud admiten múltiples usuarios o aplicaciones, garantizando un acceso seguro y un manejo aislado de los datos para cada inquilino. Esto las hace ideales para implementaciones a gran escala donde varios equipos necesitan entornos de datos aislados.
El papel de Milvus en la gestión de datos no estructurados
Milvus es una base de datos vectorial de código abierto diseñada para almacenar, indexar y recuperar de manera eficiente vectores de alta dimensionalidad a escala de miles de millones, lo que la hace ideal para casos de uso relacionados con IA y machine learning, como sistemas de recomendación, reconocimiento de imágenes o generación aumentada por recuperación (RAG).
Milvus ofrece múltiples opciones de implementación para satisfacer diversas necesidades de los clientes.
Milvus de código abierto: De código abierto, autogestionado, y puede alojarse en cualquier máquina con soporte de la comunidad. Milvus también ofrece múltiples opciones de implementación para diversas necesidades y entornos, incluidas Milvus Lite, Milvus Standalone y Milvus Distributed. Consulta la documentación de Milvus para obtener más detalles.
Zilliz Cloud: versión totalmente gestionada de Milvus, rediseñada para la nube y disponible en las principales nubes públicas como AWS, GCP y Azure.
Zilliz BYOC: Milvus listo para empresas para VPC privadas y que puede implementarse en tu nube privada virtual.
Fig 1- Formas de implementar Milvus .png
Fig 1: Formas de implementar Milvus
Características clave de Milvus
Milvus está diseñado para un alto rendimiento y escalabilidad, ofreciendo características como:
Figura- Características clave de Milvus .png
Figura: Características clave de Milvus
Multi-tenancy: Milvus permite que múltiples usuarios o aplicaciones trabajen en el mismo sistema sin interferir entre sí, proporcionando acceso seguro a datos aislados.
Cómputo acelerado por hardware: Optimizado para aprovechar hardware como GPUs, lo que lo hace más rápido y eficiente para tareas intensivas en recursos, como la inferencia de modelos de IA.
Compatibilidad con lenguajes y API: Milvus es muy versátil, compatible con Python, Java, Golang, NodeJS, etc., lo que lo hace accesible para una amplia gama de desarrolladores.
Arquitectura escalable y elástica: Milvus está diseñado para gestionar necesidades crecientes de datos. Escala automáticamente para satisfacer la demanda, garantizando que el rendimiento siga siendo óptimo a medida que crecen los datos.
Compatibilidad con diversos índices: Admite múltiples tipos de índices, incluidos HNSW, PQ, Binary y DiskANN, lo que permite flexibilidad en la forma en que se almacenan y buscan los datos.
Consistencia ajustable: Milvus te permite ajustar los niveles de consistencia de tus datos, lo que te permite equilibrar el rendimiento y la precisión de los datos según las necesidades de la aplicación.
Para obtener información más detallada, consulta la documentación de Milvus.
Tecnologías que impulsan Milvus para diversos casos de uso
Milvus está impulsado por un conjunto de tecnologías que se adaptan a diferentes tipos de aplicaciones, garantizando rendimiento, flexibilidad y escalabilidad en diversos entornos:
Figura- Tecnologías de Milvus para diversos casos de uso .png
Figura: Tecnologías de Milvus para diversos casos de uso
Tipos de cómputo: Milvus está optimizado para diferentes entornos de hardware, incluidos AVX512 (un conjunto de instrucciones de CPU para computación de alta velocidad), Neon para SIMD (Single Instruction, Multiple Data) y aceleración por GPU. Esto garantiza que Milvus pueda utilizar hardware de alto rendimiento para ofrecer procesamiento rápido y escalabilidad rentable.
Tipos de búsqueda: Milvus admite una amplia variedad de métodos de búsqueda, como top-K ANN (Approximate Nearest Neighbors) para encontrar los puntos de datos más similares, range ANN, búsquedas dispersas y densas y búsquedas filtradas. Estos nos permiten adaptar la funcionalidad de búsqueda a las necesidades específicas de nuestras aplicaciones, ya sea identificar imágenes, videos o texto similares.
Multi-tenancy: Milvus admite gestión de colecciones y particiones, lo que habilita multi-tenancy. Esta característica permite que diferentes equipos o aplicaciones compartan la misma base de datos sin interferir con los datos de los demás, garantizando un manejo de datos seguro y aislado.
Tipos de índices: Milvus proporciona una amplia gama de 15 tipos de indexación, incluidos HNSW (Hierarchical Navigable Small Worlds) para búsquedas de alta velocidad, PQ (Product Quantization) para almacenamiento compacto y DiskANN para gestionar conjuntos de datos masivos. Puedes elegir el tipo de indexación que mejor equilibre rendimiento, precisión y costo, según tu caso de uso.
Estas tecnologías hacen de Milvus una buena opción para una amplia gama de aplicaciones, desde implementaciones en la nube a gran escala hasta dispositivos edge con recursos limitados.
El cambio hacia la computación edge
A medida que crece la demanda de información en tiempo real a partir de datos no estructurados, muchas organizaciones descubren que depender únicamente de entornos en la nube no es suficiente. Esto ha impulsado un cambio hacia la computación edge, donde el procesamiento ocurre más cerca de donde se generan los datos, ofreciendo ventajas claras para aplicaciones de IA sensibles al tiempo.
Computación edge se refiere a la práctica de procesar datos más cerca de donde se generan, a menudo en dispositivos edge como sensores, cámaras y dispositivos IoT. En lugar de enviar datos sin procesar de vuelta a centros de datos centralizados para su procesamiento, la computación edge permite realizar cálculos en los propios dispositivos o en ubicaciones cercanas.
Con la creciente disponibilidad de dispositivos edge potentes, se ha vuelto posible implementar modelos de aprendizaje automático e IA directamente en estos dispositivos. Este enfoque permite la toma de decisiones en tiempo real en aplicaciones como vehículos autónomos, ciudades inteligentes y automatización industrial.
¿Por qué procesar datos en el edge?
Procesar datos no estructurados en el edge tiene varias ventajas:
Baja latencia: Dado que los datos se procesan cerca de su fuente, hay un retraso mínimo en la obtención de resultados.
Ancho de banda reducido: Al procesar datos localmente, reduces la necesidad de transmitir grandes volúmenes de datos sin procesar de vuelta a la nube.
Privacidad y seguridad: Los datos sensibles pueden permanecer en el dispositivo edge, minimizando el riesgo de exposición durante la transmisión.
IA en el edge y bases de datos vectoriales
Aunque la computación edge acerca el procesamiento de datos a la fuente en dispositivos como sensores, cámaras y dispositivos pequeños como la Raspberry Pi, es la base de datos vectorial la que realmente permite a estos dispositivos manejar el creciente flujo de datos no estructurados en tiempo real.
Por ejemplo, en un entorno industrial, los dispositivos edge monitorean el rendimiento de la maquinaria y detectan posibles problemas usando datos de sensores. Una base de datos vectorial como Milvus permite al dispositivo edge comparar rápidamente los datos entrantes con patrones históricos, identificando anomalías que indican necesidades de mantenimiento. Sin la base de datos vectorial, el dispositivo edge tendría que enviar todos los datos sin procesar a un servidor en la nube para su procesamiento (introduciendo retrasos y costos más altos) o arriesgarse a perder información crítica.
Milvus Lite: potenciando los dispositivos edge con capacidades de IA
Milvus Lite es la versión ligera de Milvus, diseñada específicamente para entornos con recursos limitados como los dispositivos edge. Proporciona todas las capacidades esenciales de una base de datos vectorial, pero está optimizada para su implementación en hardware más pequeño y menos potente. Esto la convierte en una solución ideal para potenciar dispositivos edge y permitirles manejar tareas complejas de IA, incluso en entornos con recursos limitados.
Con Milvus Lite ejecutándose en un dispositivo edge, el dispositivo se transforma en un procesador de datos impulsado por IA. Esta combinación permite que el dispositivo edge realice reconocimiento de imágenes localizado, búsquedas de similitud de video o incluso procesamiento del lenguaje natural. En un entorno minorista, por ejemplo, un sistema de pago inteligente impulsado por Milvus Lite podría reconocer productos al instante y compararlos con una base de datos vectorial almacenada de artículos, acelerando las transacciones sin necesidad de sistemas basados en la nube.
Veamos un ejemplo práctico en el que Milvus potencia un dispositivo edge que se ejecuta en una Raspberry Pi.
Creación de un sistema de estimación de pose en tiempo real con Raspberry Pi y Milvus
Raspberry Pi es una serie de computadoras pequeñas, asequibles y de placa única desarrolladas por la Raspberry Pi Foundation. En este ejemplo, usaremos Raspberry Pi como dispositivo edge y Milvus Lite para gestionar datos no estructurados.
Este sistema de estimación de pose capturará transmisiones de video, las procesará para estimar poses humanas y almacenará los vectores de características extraídos en Milvus para una recuperación eficiente por similitud. Además, integraremos Slack para notificaciones, alertándonos cuando se detecten poses específicas.
Puedes consultar el código fuente de este ejemplo en el notebook de Tim en GitHub.
Requisitos previos
Raspberry Pi (con GStreamer y Python instalados)
Configuración de Milvus (ya sea localmente o en un servidor remoto)
Cuenta de Slack para notificaciones
Importación de las bibliotecas necesarias y configuración inicial
Comienza importando las bibliotecas necesarias y estableciendo algunas configuraciones iniciales:
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import os
import argparse
import multiprocessing
import numpy as np
import setproctitle
import cv2
import time
from datetime import datetime
import uuid
import glob
import torch
from torchvision import transforms
from PIL import Image
import timm
from sklearn.preprocessing import normalize
from timm.data import resolve_data_config
from timm.data.transforms_factory import create_transform
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from pymilvus import connections
from pymilvus import utility
from pymilvus import FieldSchema, CollectionSchema, DataType, Collection
from pymilvus import MilvusClient
from pymilvus import MilvusClient
import hailo
from hailo_rpi_common import (
get_default_parser,
QUEUE,
get_caps_from_pad,
get_numpy_from_buffer,
GStreamerApp,
app_callback_class,
)
DIMENSION = 512
MILVUS_URL = "https://in05-7bd87b945683c8d.serverless.gcp-us-west1.cloud.zilliz.com"
COLLECTION_NAME = "rpipose"
TOKEN = os.environ["ZILLIZ_TOKEN"]
PATH = "/opt/demo/images"
time_list = [ 0, 5, 10, 20, 30, 40, 50, 59 ]
El código anterior importa varias bibliotecas para procesamiento de video (GStreamer), procesamiento de imágenes (OpenCV), aprendizaje profundo (PyTorch, timm), operaciones de base de datos (pymilvus) y comunicación (Slack SDK). También importa la biblioteca Hailo para trabajar con el acelerador de IA, diseñado para habilitar aplicaciones de aprendizaje profundo en dispositivos edge. Configuramos variables globales para la configuración de la base de datos Milvus y definimos una lista de tiempos para acciones periódicas. MILVUSURL Y TOKEN nos autenticarán con la nube de Zilliz.
Configuración de la base de datos Milvus
A continuación, configuramos nuestra conexión a la base de datos vectorial Milvus. Usaremos Milvus para almacenar las características extraídas del modelo de estimación de pose.
# Connect to Milvus
# Milvus Lite
# milvus_client = MilvusClient(uri="pipose.db")
# Cloud Server
milvus_client = MilvusClient( uri=MILVUS_URL, token=TOKEN )
# Create Milvus collection which includes the id, filepath of the image, and image embedding
fields = [
FieldSchema(name='id', dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name='label', dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name='lefteye', dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name='righteye', dtype=DataType.VARCHAR, max_length=200),
FieldSchema(name='confidence', dtype=DataType.FLOAT),
FieldSchema(name='vector', dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
schema = CollectionSchema(fields=fields)
milvus_client.create_collection(COLLECTION_NAME, DIMENSION, schema=schema, metric_type="COSINE", auto_id=True)
index_params = milvus_client.prepare_index_params()
index_params.add_index(field_name = "vector", metric_type="COSINE")
milvus_client.create_index(COLLECTION_NAME, index_params)
Aquí estamos configurando nuestra base de datos Milvus. Creamos una conexión de cliente, definimos el esquema para nuestra colección (incluidos campos para ID, etiqueta, posiciones de los ojos, confianza y el vector de características), creamos la colección y configuramos un índice para búsquedas de similitud eficientes.
Configuración de Slack para notificaciones
A continuación, configura una conexión a Slack para enviar notificaciones:
slack_token = os.environ["SLACK_BOT_TOKEN"]
client = WebClient(token=slack_token)
Esta configuración nos permitirá enviar mensajes y archivos a Slack usando un token de bot almacenado como variable de entorno.
Implementación del extractor de características
Nuestro sistema de estimación de poses requiere un extractor de características para convertir las poses detectadas en un formato adecuado para su almacenamiento en Milvus. Definamos una clase para esto.
class FeatureExtractor:
def __init__(self, modelname):
# Load the pre-trained model
self.model = timm.create_model(
modelname, pretrained=True, num_classes=0, global_pool="avg"
)
self.model.eval()
# Get the input size required by the model
self.input_size = self.model.default_cfg["input_size"]
config = resolve_data_config({}, model=modelname)
# Get the preprocessing function provided by TIMM for the model
self.preprocess = create_transform(**config)
def __call__(self, imagepath):
# Preprocess the input image
input_image = Image.open(imagepath).convert("RGB") # Convert to RGB if needed
input_image = self.preprocess(input_image)
# Convert the image to a PyTorch tensor and add a batch dimension
input_tensor = input_image.unsqueeze(0)
# Perform inference
with torch.no_grad():
output = self.model(input_tensor)
# Extract the feature vector
feature_vector = output.squeeze().numpy()
return normalize(feature_vector.reshape(1, -1), norm="l2").flatten()
extractor = FeatureExtractor("resnet34")
Esta clase FeatureExtractor convertirá nuestras imágenes en vectores de características. Estamos usando un modelo ResNet34 preentrenado para esta tarea. El método __call__ abre una imagen, la preprocesa, la ejecuta a través del modelo y devuelve un vector de características normalizado.
Definición de una clase de callback de usuario
Continúa y define una clase simple de callback de usuario.
class user_app_callback_class(app_callback_class):
def __init__(self):
super().__init__()
Esta clase hereda de app_callback_class y no añade ninguna funcionalidad nueva. Es un marcador de posición que permite una futura personalización del comportamiento del callback.
Creación de una función de callback de la aplicación
La función app_callback realiza la mayor parte del procesamiento. Esta función procesa fotogramas de video, estima poses y almacena los resultados en Milvus.
def app_callback(pad, info, user_data):
# Get the GstBuffer from the probe info
lefteye = ""
righteye = ""
buffer = info.get_buffer()
# Comprobar si el buffer es válido
if buffer is None:
return Gst.PadProbeReturn.OK
# Usar user_data para contar el número de fotogramas
user_data.increment()
string_to_print = f"Conteo de fotogramas: {user_data.get_count()}\n"
# Obtener las caps del pad
format, width, height = get_caps_from_pad(pad)
# Si user_data.use_frame está establecido en True, podemos obtener el fotograma de video del buffer
frame = None
if user_data.use_frame and format is not None and width is not None and height is not None:
# Obtener fotograma de video
frame = get_numpy_from_buffer(buffer, format, width, height)
# Obtener las detecciones del buffer
roi = hailo.get_roi_from_buffer(buffer)
detections = roi.get_objects_typed(hailo.HAILO_DETECTION)
# Analizar las detecciones
for detection in detections:
label = detection.get_label()
bbox = detection.get_bbox()
confidence = detection.get_confidence()
if label == "person":
string_to_print += (f"Detección: {label} {confidence:.2f}\n")
# Puntos de referencia de estimación de pose de la detección (si están disponibles)
landmarks = detection.get_objects_typed(hailo.HAILO_LANDMARKS)
if len(landmarks) != 0:
points = landmarks[0].get_points()
left_eye = points[1] # asumiendo que 1 es el índice del ojo izquierdo
right_eye = points[2] # asumiendo que 2 es el índice del ojo derecho
# Los puntos de referencia están normalizados al cuadro delimitador; también necesitamos convertirlos al tamaño del fotograma
left_eye_x = int((left_eye.x() * bbox.width() + bbox.xmin()) * width)
left_eye_y = int((left_eye.y() * bbox.height() + bbox.ymin()) * height)
right_eye_x = int((right_eye.x() * bbox.width() + bbox.xmin()) * width)
right_eye_y = int((right_eye.y() * bbox.height() + bbox.ymin()) * height)
string_to_print += (f" Ojo izquierdo: x: {left_eye_x:.2f} y: {left_eye_y:.2f} Ojo derecho: x: {right_eye_x:.2f} y: {right_eye_y:.2f}\n")
if user_data.use_frame:
# Añadir marcadores al fotograma para mostrar los puntos de referencia de los ojos
cv2.circle(frame, (left_eye_x, left_eye_y), 5, (0, 255, 0), -1)
cv2.circle(frame, (right_eye_x, right_eye_y), 5, (0, 255, 0), -1)
if user_data.use_frame:
# Convertir el fotograma a BGR
framesave = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
user_data.set_frame(framesave)
time_now = datetime.now()
current_time = int(time_now.strftime("%S"))
if current_time in time_list and len(label) > 4:
# Guardar imagen
strfilename = PATH + "/personpose.jpg"
cv2.imwrite(strfilename, framesave)
lefteye = (f"x: {left_eye_x:.2f} y: {left_eye_y:.2f}")
righteye = (f"x: {right_eye_x:.2f} y: {right_eye_y:.2f}")
# Slack
try:
response = client.chat_postMessage(
channel="C06NE1FU6SE",
text=(f"Detección: {label} {confidence:.2f}")
)
except SlackApiError as e:
# Recibirás un SlackApiError si "ok" es False
assert e.response["error"]
try:
response = client.chat_postMessage(
channel="C06NE1FU6SE",
text=(f" Ojo izquierdo: x: {left_eye_x:.2f} y: {left_eye_y:.2f} Ojo derecho: x: {right_eye_x:.2f} y: {right_eye_y:.2f}\n")
)
except SlackApiError as e:
# Recibirás un SlackApiError si "ok" es False
assert e.response["error"]
try:
response = client.files_upload_v2(
channel="C06NE1FU6SE",
file=strfilename,
title=label,
initial_comment="Live Camera image ",
)
except SlackApiError as e:
assert e.response["error"]
# Milvus insert
try:
imageembedding = extractor(strfilename)
milvus_client.insert( COLLECTION_NAME, {"vector": imageembedding, "lefteye": lefteye, "righteye": righteye, "label": label, "confidence": confidence})
except Exception as e:
print("An error:", e)
print(string_to_print)
return Gst.PadProbeReturn.OK
Esta función de callback procesa fotogramas de video para detectar personas, centrándose en extraer detalles como el cuadro delimitador, la confianza y los puntos de referencia de los ojos. Marca las posiciones de los ojos detectadas en el fotograma y guarda la imagen del fotograma si hay una persona en él. Además, envía notificaciones a Slack con los detalles de la detección y la imagen marcada. Luego, la función extrae un vector de incrustación de la imagen guardada. Actualiza Milvus con estos datos, incluida la incrustación (vector), las coordenadas de los ojos (lefteye y righteye), la etiqueta del objeto (label) y la confianza de detección (confidence).
Creación de la función de utilidad para los puntos clave de COCO
Crea una función de utilidad para obtener los puntos clave de COCO. Estos son puntos específicos del cuerpo humano que se utilizan para la estimación de la pose.
def get_keypoints():
"""Get the COCO keypoints and their left/right flip correspondence map."""
keypoints = {
'nose': 1,
'left_eye': 2,
'right_eye': 3,
'left_ear': 4,
'right_ear': 5,
'left_shoulder': 6,
'right_shoulder': 7,
'left_elbow': 8,
'right_elbow': 9,
'left_wrist': 10,
'right_wrist': 11,
'left_hip': 12,
'right_hip': 13,
'left_knee': 14,
'right_knee': 15,
'left_ankle': 16,
'right_ankle': 17,
}
return keypoints
Esta función devuelve un diccionario que asigna partes del cuerpo a sus índices de puntos clave correspondientes en el formato del conjunto de datos COCO.
Creación de la canalización de GStreamer
Continúa y crea la clase principal de la aplicación que configurará una canalización de GStreamer. Esta es una secuencia de elementos que procesan datos multimedia en un orden específico. Procesará transmisiones de video en vivo.
# This class inherits from the hailo_rpi_common.GStreamerApp class
class GStreamerPoseEstimationApp(GStreamerApp):
def __init__(self, args, user_data):
# Call the parent class constructor
super().__init__(args, user_data)
# Additional initialization code can be added here
# Set Hailo parameters these parameters should be set based on the model used
self.batch_size = 2
self.network_width = 640
self.network_height = 640
self.network_format = "RGB"
self.default_postprocess_so = os.path.join(self.postprocess_dir, 'libyolov8pose_post.so')
self.post_function_name = "filter"
self.hef_path = os.path.join(self.current_path, '../resources/yolov8s_pose_h8l_pi.hef')
self.app_callback = app_callback
# Set the process title
setproctitle.setproctitle("Hailo Pose Estimation with Milvus")
self.create_pipeline()
def get_pipeline_string(self):
if (self.source_type == "rpi"):
source_element = f"libcamerasrc name=src_0 auto-focus-mode=2 ! "
source_element += f"video/x-raw, format={self.network_format}, width=1536, height=864 ! "
source_element += QUEUE("queue_src_scale")
source_element += f"videoscale ! "
source_element += f"video/x-raw, format={self.network_format}, width={self.network_width}, height={self.network_height}, framerate=30/1 ! "
elif (self.source_type == "usb"):
source_element = f"v4l2src device={self.video_source} name=src_0 ! "
source_element += f"video/x-raw, width=640, height=480, framerate=30/1 ! "
else:
source_element = f"filesrc location={self.video_source} name=src_0 ! "
source_element += QUEUE("queue_dec264")
source_element += f" qtdemux ! h264parse ! avdec_h264 max-threads=2 ! "
source_element += f" video/x-raw,format=I420 ! "
source_element += QUEUE("queue_scale")
source_element += f"videoscale n-threads=2 ! "
source_element += QUEUE("queue_src_convert")
source_element += f"videoconvert n-threads=3 name=src_convert qos=false ! "
source_element += f"video/x-raw, format={self.network_format}, width={self.network_width}, height={self.network_height}, pixel-aspect-ratio=1/1 ! "
pipeline_string = "hailomuxer name=hmux "
pipeline_string += source_element
pipeline_string += "tee name=t ! "
pipeline_string += QUEUE("bypass_queue", max_size_buffers=20) + "hmux.sink_0 "
pipeline_string += "t. ! " + QUEUE("queue_hailonet")
pipeline_string += "videoconvert n-threads=3 ! "
pipeline_string += f"hailonet hef-path={self.hef_path} batch-size={self.batch_size} force-writable=true ! "
pipeline_string += QUEUE("queue_hailofilter")
pipeline_string += f"hailofilter function-name={self.post_function_name} so-path={self.default_postprocess_so} qos=false ! "
pipeline_string += QUEUE("queue_hmuc") + " hmux.sink_1 "
pipeline_string += "hmux. ! " + QUEUE("queue_hailo_python")
pipeline_string += QUEUE("queue_user_callback")
pipeline_string += f"identity name=identity_callback ! "
pipeline_string += QUEUE("queue_hailooverlay")
pipeline_string += f"hailooverlay ! "
pipeline_string += QUEUE("queue_videoconvert")
pipeline_string += f"videoconvert n-threads=3 qos=false ! "
pipeline_string += QUEUE("queue_hailo_display")
pipeline_string += f"fpsdisplaysink video-sink={self.video_sink} name=hailo_display sync={self.sync} text-overlay={self.options_menu.show_fps} signal-fps-measurements=true "
print(pipeline_string)
return pipeline_string
Esta clase configura nuestra pipeline de GStreamer. Ejecuta un modelo de estimación de pose en una Raspberry Pi usando un acelerador de IA Hailo, que es un procesador de IA en el borde diseñado para ejecutar tareas de aprendizaje profundo e inferencia de IA en dispositivos de borde. Inicializa parámetros como el tamaño de lote, las dimensiones de entrada, el formato de color y las rutas de archivo para el modelo de red neuronal y el posprocesamiento. Según la fuente de entrada (por ejemplo, cámara de Raspberry Pi, cámara USB o archivo de video), construye una pipeline de GStreamer para capturar video, procesarlo a través del modelo Hailo, aplicar posprocesamiento y mostrar la salida. La configuración de la pipeline se especifica en el método get_pipeline_string, que ensambla los elementos de GStreamer necesarios para esta tarea.
Ejecución del programa
El paso final es ejecutar el programa.
if __name__ == "__main__":
# Create an instance of the user app callback class
user_data = user_app_callback_class()
parser = get_default_parser()
args = parser.parse_args()
app = GStreamerPoseEstimationApp(args, user_data)
app.run()
Aquí es donde conectamos todo. Creamos una instancia de nuestra user_app_callback_class, analizamos cualquier argumento de línea de comandos, creamos nuestra GStreamerPoseEstimationApp y la ejecutamos.
Aquí tienes un resultado de muestra de lo que puedes esperar cuando ejecutes el programa.
Fig 2- Output of a pose estimation program in Slack with eye landmarks marked with green dots.png
Fig 2: Salida de un programa de estimación de pose en Slack con puntos de referencia de los ojos marcados con puntos verdes
Esta es la actualización en Slack cuando se detecta una persona. Tiene una puntuación de confianza del 81%, las coordenadas de los ojos izquierdo y derecho, y la imagen con los puntos de referencia de los ojos marcados (círculos verdes en las posiciones del ojo izquierdo y derecho).
Casos de uso de combinar IA y bases de datos vectoriales
Combinar IA en el edge y bases de datos vectoriales da lugar a numerosos casos de uso. A continuación se muestran algunos ejemplos.
Robótica
La IA en el edge y las bases de datos vectoriales mejoran las capacidades de los robots autónomos, permitiéndoles procesar datos de sensores localmente y tomar decisiones al instante. Esta tecnología mejora significativamente la eficiencia y la adaptabilidad en entornos dinámicos.
Ejemplo: En la automatización de almacenes, los robots móviles autónomos (AMR) utilizan cámaras integradas y bases de datos vectoriales locales para el reconocimiento de objetos en tiempo real. Mientras navegan por el almacén, estos robots identifican y categorizan productos al instante comparando datos visuales con vectores almacenados localmente. Esto les permite adaptarse a las condiciones cambiantes del almacén y optimizar sus rutas sin depender de infraestructura en la nube.
Ciudades inteligentes
La IA en el edge y las bases de datos vectoriales desempeñan un papel crucial en la optimización de las operaciones urbanas, desde la gestión del tráfico hasta la seguridad pública. Estas tecnologías permiten un procesamiento rápido de datos y la toma de decisiones en el edge, mejorando la eficiencia de la ciudad y la calidad de vida de los residentes.
Ejemplo: Los sistemas inteligentes de gestión del tráfico aprovechan la IA en el edge para procesar localmente transmisiones de video de cámaras en intersecciones. El sistema convierte los patrones de tráfico en representaciones vectoriales y los compara con datos históricos almacenados en una base de datos vectorial local. Esto permite ajustes en tiempo real de los semáforos para reducir la congestión, garantizando tiempos de respuesta rápidos sin necesidad de transmisión constante de datos a servidores centrales.
Automatización industrial
La IA en el edge y las bases de datos vectoriales impulsan el mantenimiento predictivo y la supervisión de equipos en entornos de fabricación e industriales. Este enfoque permite el análisis en tiempo real del estado de las máquinas, evitando tiempos de inactividad y optimizando los procesos de producción.
Ejemplo: Las plantas de fabricación inteligentes despliegan sensores en maquinaria crítica para recopilar continuamente datos de vibración. Los dispositivos edge convierten estos datos en representaciones vectoriales y los comparan con patrones de vibración conocidos como "saludables" y "defectuosos" almacenados en una base de datos vectorial local. Este análisis en tiempo real permite la detección inmediata de posibles fallos de equipos, facilitando el mantenimiento proactivo y minimizando interrupciones costosas.
Atención médica
La IA en el edge y las bases de datos vectoriales transforman la atención al paciente al permitir la monitorización continua de la salud sin comprometer la privacidad. Estas tecnologías permiten un análisis sofisticado de datos de salud directamente en dispositivos portátiles o sistemas locales.
Ejemplo: Los monitores de ECG portátiles utilizan IA en el edge para procesar localmente los datos del ritmo cardíaco. El dispositivo convierte los patrones de ECG en vectores y los compara con una base de datos de ritmos normales y anormales almacenada a bordo. Este enfoque permite la detección inmediata de posibles problemas cardíacos mientras mantiene los datos de salud sensibles en el dispositivo, mejorando tanto la capacidad de respuesta como la privacidad del paciente.
Conclusión
A medida que los datos no estructurados crecen en escala e importancia, las soluciones de procesamiento eficientes se vuelven cada vez más cruciales. Tim hizo un buen trabajo al destacar el potencial transformador de las bases de datos vectoriales como Milvus, que ofrecen capacidades avanzadas tanto para implementaciones en la nube como en el borde. Ya sea para aplicaciones de IA a gran escala o para dispositivos de borde con recursos limitados, Milvus y su contraparte ligera, Milvus Lite, permiten a las organizaciones aprovechar los datos no estructurados para la toma de decisiones en tiempo real, la innovación y la eficiencia operativa. Con la flexibilidad de escalar desde la nube hasta el borde, estas tecnologías están a la vanguardia de las soluciones modernas impulsadas por IA.
Recursos adicionales
Proyectos de IA de Tim en GitHub:
Procesamiento de datos no estructurados con un Raspberry Pi AI Kit
Modelos de IA con mejor rendimiento para tus apps de GenAI | Zilliz
Sigue leyendo
3 Easiest Ways to Use Claude Code on Your Mobile Phone
Run Claude Code from your phone with Remote Control, Happy Coder, or SSH + Tailscale. Comparison table, setup steps, and tools for typing, memory, and parallel tasks.
Expanding Our Global Reach: Zilliz Cloud Launches in Azure Central India
Zilliz Cloud expands to Azure Central India. This new region helps customers meet compliance, reduce latency, and optimize cloud costs when building AI applications.
Vector Databases vs. Hierarchical Databases
Use a vector database for AI-powered similarity search; use a hierarchical database for organizing data in parent-child relationships with efficient top-down access patterns.