클라우드에서 엣지까지의 비정형 데이터 처리
비정형 데이터는 이제 우리가 처리하는 데이터의 약 80%를 차지하며, 텍스트와 이미지부터 동영상, 오디오, 더 복잡한 형식에 이르기까지 다양합니다. 그 양이 증가함에 따라 이를 처리하고 분석하는 더 나은 방법에 대한 필요성도 커지고 있습니다.
최근 웨비나에서 Zilliz의 Principal Developer Advocate인 Tim Spann은 특히 클라우드 환경부터 엣지 디바이스에 이르기까지 비정형 데이터를 관리하는 방법에 대한 인사이트를 공유했습니다. 또한 Milvus와 같은 벡터 데이터베이스가 이 데이터를 이해하는 데 핵심적이라고 강조했습니다.
이 글에서는 Tim의 주요 내용을 요약하며, 비정형 데이터를 처리할 때의 과제와 벡터 데이터베이스가 어떻게 그 길을 이끌고 있는지 논의합니다. 또한 이러한 데이터베이스로 구동되는 엣지 디바이스가 AI의 새로운 발전을 어떻게 이끌고 있는지도 살펴보겠습니다. 더 자세한 내용이 궁금하시다면 YouTube에서 Tim의 발표 전체 다시 보기를 시청하시길 권장합니다.
비정형 데이터 이해하기
비정형 데이터는 미리 정의된 형식이나 스키마가 없는 정보를 의미합니다. 특정 필드와 형식으로 데이터베이스에 깔끔하게 정리된 정형 데이터와 달리, 비정형 데이터는 원시적이고 정리되지 않은 상태입니다. 여기에는 흔히 다음이 포함됩니다:
텍스트: 이메일, 워드 문서, 소셜 미디어 게시물
이미지: 사진, 스크린샷, 다이어그램
동영상: 녹화된 회의, 감시 영상, 멀티미디어 콘텐츠
오디오: 음성 녹음, 음악 파일, 팟캐스트
디지털 콘텐츠가 급증하면서 비정형 데이터는 기하급수적으로 증가했으며, 조직에 기회와 과제를 동시에 제공하고 있습니다. 이러한 풍부한 정보를 활용하기 위해 기업은 다음을 수행할 수 있습니다:
고객 상호작용과 피드백을 분석하여 고객 경험 향상
센서 데이터와 시스템 로그에서 얻은 인사이트를 통해 운영 효율성 개선
제품 개발에 멀티미디어 콘텐츠를 활용하여 혁신 촉진
정형 데이터를 처리하도록 설계된 기존 데이터베이스는 비정형 데이터의 불규칙성을 다루는 데 어려움을 겪어 검색, 쿼리 또는 의미 있는 인사이트 추출을 어렵게 만듭니다. 바로 이 지점에서 벡터 데이터베이스가 필수적이 되며, 비정형 정보를 효율적으로 관리하고 검색하기 위한 솔루션을 제공합니다.
비정형 데이터 처리에 벡터 데이터베이스를 사용하는 이유는 무엇인가요?
벡터 데이터베이스는 데이터의 벡터 표현(일명 벡터 임베딩)을 저장, 인덱싱, 쿼리하도록 설계된 특수 데이터베이스로, 주로 텍스트, 이미지, 오디오와 같은 비정형 데이터에서 파생됩니다. 이러한 데이터베이스는 고차원 공간에서 효율적인 유사도 검색을 가능하게 하여 시맨틱 검색, 자연어 처리(NLP), 추천 시스템, 이미지 검색, 검색 증강 생성(RAG)에 이상적입니다.
Milvus 및 Zilliz Cloud와 같은 벡터 데이터베이스는 고성능 검색 기능을 넘어 비정형 데이터를 관리하는 데 핵심적인 다양한 기능을 제공합니다.
고성능 검색을 넘어
벡터 데이터베이스는 빠른 유사도 검색 기능과 자주 연관되지만, 그 활용성은 훨씬 더 넓습니다. 벡터 데이터베이스는 비정형 데이터의 효과적인 관리와 활용을 보장하는 다른 필수 기능도 제공합니다:
CRUD 작업: 벡터 데이터베이스는 전통적인 데이터베이스처럼 데이터를 Create, Read, Update, Delete(CRUD)할 수 있게 해줍니다. 이를 통해 복잡한 데이터 유형을 다루더라도 기본 작업은 직관적이고 접근 가능하게 유지됩니다.
데이터 최신성: 벡터 데이터베이스의 핵심 강점 중 하나는 데이터가 최신 상태로 유지되도록 보장하는 것입니다. 추천 시스템과 같은 사용 사례에서는 정확한 인사이트를 생성하기 위해 데이터를 최신 상태로 유지하는 것이 필수적입니다.
영속성: 인메모리 데이터 구조와 달리, 벡터 데이터베이스는 영구 저장소를 제공하므로 시스템 재부팅 후에도 데이터가 안전하게 저장되고 접근 가능합니다.
가용성: 벡터 데이터베이스는 실시간 쿼리와 검색을 위해 데이터를 즉시 접근 가능한 상태로 유지합니다. 이를 통해 데이터가 증가하더라도 빠르고 효율적인 검색을 계속 수행할 수 있어, 지연 없이 AI 기반 의사결정을 가능하게 합니다.
확장성: 데이터의 양이 증가함에 따라, 벡터 데이터베이스는 성능 저하 없이 그 증가를 수용하도록 효율적으로 확장됩니다. 이는 수십억 개의 비정형 데이터 포인트를 관리할 때 필수적입니다.
완전한 데이터 관리
위 기능들을 넘어, 벡터 데이터베이스는 데이터 수집, 인덱싱, 쿼리를 포함해 비정형 데이터를 관리하기 위한 강력한 도구를 제공하여, 가장 큰 규모의 데이터셋도 효율적으로 검색하고 분석할 수 있게 합니다. 백업 및 마이그레이션과 같은 기능은 데이터가 안전하게 유지되고 복구 가능하도록 보장하여, 벡터 데이터베이스를 중요한 정보를 관리하기 위한 신뢰할 수 있는 선택지로 만듭니다.
운영 편의성
벡터 데이터베이스는 배포와 관리를 단순화하도록 설계되기도 했습니다:
클라우드 또는 온프레미스 배포: 많은 목적 특화 벡터 데이터베이스는 다재다능하며 퍼블릭 클라우드 또는 온프레미스 인프라에 쉽게 배포할 수 있어, 조직의 요구에 기반한 유연성을 보장합니다.
관측 가능성: 많은 벡터 데이터베이스는 데이터베이스 상태와 성능을 추적하기 위한 모니터링 도구를 제공하여, 사용자가 시스템을 사전에 최적화할 수 있게 합니다.
멀티테넌시: Zilliz Cloud와 같은 벡터 데이터베이스는 여러 사용자 또는 애플리케이션을 지원하여, 각 테넌트에 대해 안전한 접근과 격리된 데이터 처리를 보장합니다. 이는 여러 팀이 격리된 데이터 환경을 필요로 하는 대규모 배포에 이상적입니다.
비정형 데이터 관리에서 Milvus의 역할
Milvus는 10억 규모의 고차원 벡터를 효율적으로 저장, 인덱싱, 검색하도록 설계된 오픈소스 벡터 데이터베이스로, 추천 시스템, 이미지 인식 또는 검색 증강 생성(RAG)과 같이 AI 및 머신러닝이 관련된 사용 사례에 이상적입니다.
Milvus는 다양한 고객 요구를 충족하기 위해 여러 배포 옵션을 제공합니다.
오픈소스 Milvus: 오픈소스이며, 자체 관리형이고, 커뮤니티 지원을 통해 어떤 머신에서도 호스팅할 수 있습니다. Milvus는 또한 Milvus Lite, Milvus Standalone, Milvus Distributed를 포함해 다양한 요구와 환경을 위한 여러 배포 옵션을 제공합니다. 자세한 내용은 Milvus 문서를 참조하세요.
Zilliz Cloud: 클라우드용으로 재설계되어 AWS, GCP, Azure와 같은 주요 퍼블릭 클라우드에서 사용 가능한 Milvus의 완전 관리형 버전입니다.
Zilliz BYOC: Private VPC를 위한 엔터프라이즈급 Milvus이며, 사용자의 가상 프라이빗 클라우드에 배포할 수 있습니다.
Fig 1- Ways of deploying Milvus .png
그림 1: Milvus 배포 방법
Milvus의 주요 기능
Milvus는 고성능과 확장성을 위해 설계되었으며, 다음과 같은 기능을 제공합니다:
Figure- Milvus key features .png
그림: Milvus 주요 기능
멀티 테넌시: Milvus는 여러 사용자 또는 애플리케이션이 서로 간섭하지 않고 동일한 시스템에서 작업할 수 있도록 하며, 격리된 데이터에 대한 안전한 액세스를 제공합니다.
하드웨어 가속 컴퓨팅: GPU와 같은 하드웨어를 활용하도록 최적화되어 AI 모델 추론과 같은 리소스 집약적 작업을 더 빠르고 효율적으로 수행할 수 있습니다.
언어 및 API 지원: Milvus는 Python, Java, Golang, NodeJS 등을 지원하는 매우 다재다능한 제품으로, 다양한 개발자가 쉽게 사용할 수 있습니다.
확장 가능하고 탄력적인 아키텍처: Milvus는 증가하는 데이터 요구 사항을 처리하도록 설계되었습니다. 수요에 맞춰 자동으로 확장되어 데이터가 증가해도 성능이 최적으로 유지되도록 보장합니다.
다양한 인덱스 지원: HNSW, PQ, Binary, DiskANN을 포함한 여러 유형의 인덱스를 지원하여 데이터 저장 및 검색 방식에 유연성을 제공합니다.
조정 가능한 일관성: Milvus는 데이터의 일관성 수준을 조정할 수 있게 해 주어, 애플리케이션의 요구 사항에 따라 성능과 데이터 정확성 사이의 균형을 맞출 수 있습니다.
자세한 내용은 Milvus documentation을 참조하세요.
다양한 사용 사례를 위한 Milvus의 기반 기술
Milvus는 다양한 유형의 애플리케이션에 대응하는 일련의 기술로 구동되어, 여러 환경에서 성능, 유연성, 확장성을 보장합니다:
Figure- Milvus technologies for various use cases .png
그림: 다양한 사용 사례를 위한 Milvus 기술
컴퓨팅 유형: Milvus는 고속 컴퓨팅을 위한 CPU 명령어 세트인 AVX512, SIMD(Single Instruction, Multiple Data)를 위한 Neon, GPU acceleration을 포함한 다양한 하드웨어 환경에 최적화되어 있습니다. 이를 통해 Milvus는 고성능 하드웨어를 사용하여 빠른 처리와 비용 효율적인 확장성을 제공할 수 있습니다.
검색 유형: Milvus는 가장 유사한 데이터 포인트를 찾기 위한 top-K ANN(Approximate Nearest Neighbors), range ANN, sparse & dense searches, 필터링 검색과 같은 다양한 검색 방법을 지원합니다. 이를 통해 유사한 이미지, 비디오 또는 텍스트를 식별하는 등 애플리케이션의 특정 요구에 맞게 검색 기능을 조정할 수 있습니다.
멀티 테넌시: Milvus는 collection and partition management를 지원하여 멀티 테넌시를 가능하게 합니다. 이 기능을 통해 서로 다른 팀이나 애플리케이션이 서로의 데이터에 간섭하지 않고 동일한 데이터베이스를 공유할 수 있어, 안전하고 격리된 데이터 처리를 보장합니다.
인덱스 유형: Milvus는 고속 검색을 위한 HNSW (Hierarchical Navigable Small Worlds), 컴팩트한 저장을 위한 PQ (Product Quantization), 대규모 데이터셋 처리를 위한 DiskANN을 포함해 15가지의 광범위한 인덱싱 유형을 제공합니다. 사용 사례에 따라 성능, 정확도, 비용의 균형을 가장 잘 맞추는 인덱싱 유형을 선택할 수 있습니다.
이러한 기술 덕분에 Milvus는 대규모 클라우드 배포부터 리소스가 제한된 엣지 디바이스까지 다양한 애플리케이션에 적합한 선택입니다.
엣지 컴퓨팅으로의 전환
비정형 데이터에서 실시간 인사이트에 대한 수요가 증가함에 따라, 많은 조직은 클라우드 환경에만 의존하는 것으로는 충분하지 않다는 사실을 깨닫고 있습니다. 이로 인해 데이터가 생성되는 위치에 더 가까운 곳에서 처리가 이루어지는 엣지 컴퓨팅으로의 전환이 일어나고 있으며, 이는 시간에 민감한 AI 애플리케이션에 뚜렷한 이점을 제공합니다.
엣지 컴퓨팅은 데이터가 생성되는 위치에 더 가까운 곳에서 데이터를 처리하는 관행을 의미하며, 주로 센서, 카메라, IoT 디바이스와 같은 엣지 디바이스에서 이루어집니다. 원시 데이터를 처리를 위해 중앙 집중식 데이터 센터로 다시 보내는 대신, 엣지 컴퓨팅은 디바이스 자체 또는 인근 위치에서 계산을 수행할 수 있게 합니다.
강력한 엣지 디바이스의 가용성이 증가하면서, 머신 러닝 및 AI 모델을 이러한 디바이스에 직접 배포하는 것이 가능해졌습니다. 이 접근 방식은 자율주행차, 스마트 시티, 산업 자동화와 같은 애플리케이션에서 실시간 의사결정을 가능하게 합니다.
왜 엣지에서 데이터를 처리해야 할까요?
엣지에서 비정형 데이터를 처리하면 여러 가지 장점이 있습니다:
낮은 지연 시간: 데이터가 소스 근처에서 처리되므로 결과를 얻는 데 지연이 최소화됩니다.
대역폭 감소: 데이터를 로컬에서 처리함으로써 대량의 원시 데이터를 클라우드로 다시 전송할 필요를 줄일 수 있습니다.
개인정보 보호 및 보안: 민감한 데이터가 엣지 디바이스에 남아 있을 수 있어, 전송 중 노출 위험을 최소화합니다.
엣지 AI와 벡터 데이터베이스
엣지 컴퓨팅은 센서, 카메라, Raspberry Pi와 같은 소형 디바이스 등에서 데이터 처리를 소스에 더 가깝게 가져오지만, 이러한 디바이스가 증가하는 비정형 데이터의 홍수를 실시간으로 처리할 수 있도록 진정으로 힘을 실어주는 것은 벡터 데이터베이스입니다.
예를 들어, 산업 환경에서 엣지 디바이스는 센서 데이터를 사용해 기계 성능을 모니터링하고 잠재적인 문제를 감지합니다. Milvus와 같은 벡터 데이터베이스는 엣지 디바이스가 들어오는 데이터를 과거 패턴과 빠르게 비교하여 유지보수가 필요함을 나타내는 이상 징후를 식별할 수 있도록 지원합니다. 벡터 데이터베이스가 없다면, 엣지 디바이스는 모든 원시 데이터를 처리를 위해 클라우드 서버로 보내야 하거나(지연 및 더 높은 비용 초래), 중요한 인사이트를 놓칠 위험이 있습니다.
Milvus Lite: AI 기능으로 엣지 디바이스 강화
Milvus Lite는 Milvus의 경량 버전으로, 엣지 디바이스와 같이 리소스가 제한된 환경을 위해 특별히 설계되었습니다. 벡터 데이터베이스의 모든 필수 기능을 제공하지만, 더 작고 성능이 낮은 하드웨어에 배포하도록 최적화되어 있습니다. 따라서 제한된 리소스의 환경에서도 엣지 디바이스가 복잡한 AI 작업을 처리할 수 있도록 지원하는 이상적인 솔루션입니다.
엣지 디바이스에서 Milvus Lite를 실행하면, 해당 디바이스는 AI 기반 데이터 프로세서로 변환됩니다. 이 조합을 통해 엣지 디바이스는 로컬 이미지 인식, 비디오 유사도 검색, 또는 자연어 처리까지 수행할 수 있습니다. 예를 들어 리테일 환경에서 Milvus Lite로 구동되는 스마트 체크아웃 시스템은 제품을 즉시 인식하고 저장된 상품 벡터 데이터베이스와 매칭하여, 클라우드 기반 시스템 없이도 거래 속도를 높일 수 있습니다.
Raspberry Pi에서 실행되는 엣지 디바이스에 Milvus가 어떻게 활용되는지 실용적인 예제를 살펴보겠습니다.
Raspberry Pi와 Milvus로 실시간 자세 추정 시스템 구축하기
Raspberry Pi 는 Raspberry Pi Foundation에서 개발한 작고 저렴한 싱글 보드 컴퓨터 시리즈입니다. 이 예제에서는 Raspberry Pi를 엣지 디바이스로 사용하고, 비정형 데이터를 처리하기 위해 Milvus Lite를 사용합니다.
이 자세 추정 시스템은 비디오 스트림을 캡처하고, 이를 처리하여 사람의 자세를 추정하며, 추출된 특징 벡터를 효율적인 유사도 검색을 위해 Milvus에 저장합니다. 또한 Slack을 통합하여 특정 자세가 감지될 때 알림을 받도록 합니다.
이 예제의 소스 코드는 GitHub의 Tim’s notebook에서 확인할 수 있습니다.
사전 요구 사항
Raspberry Pi (GStreamer와 Python이 설치되어 있어야 함)
알림을 위한 Slack 계정
필요한 라이브러리 가져오기 및 초기 설정
필요한 라이브러리를 가져오고 몇 가지 초기 설정을 구성하는 것으로 시작합니다:
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 ]
위 코드는 비디오 처리(GStreamer), 이미지 처리(OpenCV), 딥러닝(PyTorch, timm), 데이터베이스 작업(pymilvus), 통신(Slack SDK)을 위한 다양한 라이브러리를 가져옵니다. 또한 엣지 디바이스에서 딥러닝 애플리케이션을 지원하도록 설계된 AI 가속기와 작업하기 위해 Hailo 라이브러리를 가져옵니다. Milvus 데이터베이스 구성을 위한 전역 변수를 설정하고 주기적 작업을 위한 시간 목록을 정의합니다. MILVUSURL 및 TOKEN은 Zilliz cloud에서 인증하는 데 사용됩니다.
Milvus 데이터베이스 설정
다음으로 Milvus 벡터 데이터베이스에 대한 연결을 설정합니다. Milvus를 사용하여 자세 추정 모델에서 추출한 특징을 저장합니다.
# Connect to Milvus
# Milvus Lite
# milvus_client = MilvusClient(uri="pipose.db")
# 클라우드 서버
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)
여기서는 Milvus 데이터베이스를 설정하고 있습니다. 클라이언트 연결을 생성하고, 컬렉션의 스키마(ID, 라벨, 눈 위치, 신뢰도, 특징 벡터 필드 포함)를 정의한 뒤, 컬렉션을 생성하고 효율적인 유사도 검색을 위한 인덱스를 설정합니다.
알림을 위한 Slack 설정
다음으로, 알림을 보내기 위해 Slack 연결을 설정합니다:
slack_token = os.environ["SLACK_BOT_TOKEN"]
client = WebClient(token=slack_token)
이 설정을 통해 환경 변수로 저장된 봇 토큰을 사용하여 Slack에 메시지와 파일을 보낼 수 있습니다.
특징 추출기 구현
우리의 포즈 추정 시스템은 감지된 포즈를 Milvus에 저장하기 적합한 형식으로 변환하기 위한 특징 추출기가 필요합니다. 이를 위한 클래스를 정의해 보겠습니다.
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")
이 FeatureExtractor 클래스는 이미지를 특징 벡터로 변환합니다. 이 작업에는 사전 학습된 ResNet34 모델을 사용합니다. __call__ 메서드는 이미지를 열고, 전처리한 뒤, 모델에 통과시키고, 정규화된 특징 벡터를 반환합니다.
사용자 콜백 클래스 정의
계속해서 간단한 사용자 콜백 클래스를 정의합니다.
class user_app_callback_class(app_callback_class):
def __init__(self):
super().__init__()
이 클래스는 app_callback_class를 상속하며 새로운 기능을 추가하지 않습니다. 콜백 동작의 향후 사용자 지정을 가능하게 하는 플레이스홀더입니다.
애플리케이션 콜백 함수 생성
app_callback 함수가 대부분의 처리를 수행합니다. 이 함수는 비디오 프레임을 처리하고, 포즈를 추정하며, 결과를 Milvus에 저장합니다.
def app_callback(pad, info, user_data):
# Get the GstBuffer from the probe info
lefteye = ""
righteye = ""
buffer = info.get_buffer()
# 버퍼가 유효한지 확인
if buffer is None:
return Gst.PadProbeReturn.OK
# user_data를 사용하여 프레임 수 계산
user_data.increment()
string_to_print = f"프레임 수: {user_data.get_count()}\n"
# 패드에서 caps 가져오기
format, width, height = get_caps_from_pad(pad)
# user_data.use_frame이 True로 설정되어 있으면 버퍼에서 비디오 프레임을 가져올 수 있습니다
frame = None
if user_data.use_frame and format is not None and width is not None and height is not None:
# 비디오 프레임 가져오기
frame = get_numpy_from_buffer(buffer, format, width, height)
# 버퍼에서 감지 결과 가져오기
roi = hailo.get_roi_from_buffer(buffer)
detections = roi.get_objects_typed(hailo.HAILO_DETECTION)
# 감지 결과 파싱
for detection in detections:
label = detection.get_label()
bbox = detection.get_bbox()
confidence = detection.get_confidence()
if label == "person":
string_to_print += (f"감지: {label} {confidence:.2f}\n")
# detection에서 포즈 추정 랜드마크(사용 가능한 경우)
landmarks = detection.get_objects_typed(hailo.HAILO_LANDMARKS)
if len(landmarks) != 0:
points = landmarks[0].get_points()
left_eye = points[1] # 1이 왼쪽 눈의 인덱스라고 가정
right_eye = points[2] # 2가 오른쪽 눈의 인덱스라고 가정
# 랜드마크는 경계 상자에 정규화되어 있으므로 프레임 크기로도 변환해야 합니다
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" 왼쪽 눈: x: {left_eye_x:.2f} y: {left_eye_y:.2f} 오른쪽 눈: x: {right_eye_x:.2f} y: {right_eye_y:.2f}\n")
if user_data.use_frame:
# 눈 랜드마크를 표시하기 위해 프레임에 마커 추가
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:
# 프레임을 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:
# 이미지 저장
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"감지: {label} {confidence:.2f}")
)
except SlackApiError as e:
# "ok"가 False이면 SlackApiError가 발생합니다
assert e.response["error"]
try:
response = client.chat_postMessage(
channel="C06NE1FU6SE",
text=(f" 왼쪽 눈: x: {left_eye_x:.2f} y: {left_eye_y:.2f} 오른쪽 눈: x: {right_eye_x:.2f} y: {right_eye_y:.2f}\n")
)
except SlackApiError as e:
# "ok"가 False이면 SlackApiError가 발생합니다
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
이 콜백 함수는 비디오 프레임을 처리하여 사람을 감지하며, 바운딩 박스, 신뢰도, 눈 랜드마크와 같은 세부 정보를 추출하는 데 중점을 둡니다. 프레임에서 감지된 눈 위치를 표시하고, 프레임에 사람이 있으면 프레임 이미지를 저장합니다. 또한 감지 세부 정보와 표시된 이미지를 Slack으로 알림으로 보냅니다. 그런 다음 함수는 저장된 이미지에서 임베딩 벡터를 추출합니다. 이 데이터로 Milvus를 업데이트하며, 여기에는 임베딩(vector), 눈 좌표(lefteye 및 righteye), 객체 레이블(label), 감지 신뢰도(confidence)가 포함됩니다.
COCO 키포인트용 유틸리티 함수 만들기
COCO 키포인트를 가져오는 유틸리티 함수를 만드세요. 이는 포즈 추정에 사용되는 인체의 특정 지점입니다.
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
이 함수는 신체 부위를 COCO 데이터셋 형식의 해당 키포인트 인덱스에 매핑하는 딕셔너리를 반환합니다.
GStreamer 파이프라인 만들기
계속해서 GStreamer 파이프라인을 설정할 메인 애플리케이션 클래스를 만드세요. 이는 특정 순서로 멀티미디어 데이터를 처리하는 요소들의 시퀀스입니다. 라이브 비디오 스트림을 처리합니다.
# 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
이 클래스는 우리의 GStreamer 파이프라인을 설정합니다. 엣지 디바이스에서 딥러닝 및 AI 추론 작업을 실행하도록 설계된 엣지 AI 프로세서인 Hailo AI 가속기를 사용하여 Raspberry Pi에서 포즈 추정 모델을 실행합니다. 신경망 모델 및 후처리를 위한 배치 크기, 입력 크기, 색상 형식, 파일 경로와 같은 매개변수를 초기화합니다. 입력 소스(예: Raspberry Pi 카메라, USB 카메라 또는 비디오 파일)에 따라 비디오를 캡처하고, Hailo 모델을 통해 처리하며, 후처리를 적용하고, 출력을 표시하는 GStreamer 파이프라인을 구성합니다. 파이프라인 구성은 이 작업에 필요한 GStreamer 요소를 조립하는 get_pipeline_string 메서드에 지정되어 있습니다.
프로그램 실행
마지막 단계는 프로그램을 실행하는 것입니다.
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()
여기가 모든 것을 연결하는 부분입니다. user_app_callback_class의 인스턴스를 생성하고, 명령줄 인수를 파싱하며, GStreamerPoseEstimationApp을 생성한 다음 실행합니다.
프로그램을 실행했을 때 기대할 수 있는 샘플 결과는 다음과 같습니다.
Fig 2- Slack에서 초록색 점으로 눈 랜드마크가 표시된 포즈 추정 프로그램의 출력.png
그림 2: Slack에서 초록색 점으로 눈 랜드마크가 표시된 포즈 추정 프로그램의 출력
사람이 감지되었을 때 Slack에 업데이트되는 내용입니다. 81%의 신뢰도 점수, 왼쪽 및 오른쪽 눈의 좌표, 그리고 표시된 눈 랜드마크(왼쪽 및 오른쪽 눈 위치의 초록색 원)가 있는 이미지가 포함되어 있습니다.
AI와 벡터 데이터베이스 결합의 사용 사례
엣지 AI와 벡터 데이터베이스를 결합하면 다양한 사용 사례가 생깁니다. 아래는 몇 가지 예시입니다.
로보틱스
엣지 AI와 벡터 데이터베이스는 자율 로봇의 역량을 향상시켜 센서 데이터를 로컬에서 처리하고 즉각적인 결정을 내릴 수 있게 합니다. 이 기술은 동적인 환경에서 효율성과 적응성을 크게 향상시킵니다.
예시: 창고 자동화에서 자율 이동 로봇(AMR)은 온보드 카메라와 로컬 벡터 데이터베이스를 사용하여 실시간 객체 인식을 수행합니다. 창고를 이동하면서 이러한 로봇은 시각 데이터를 로컬에 저장된 벡터와 비교해 제품을 즉시 식별하고 분류합니다. 이를 통해 클라우드 인프라에 의존하지 않고도 변화하는 창고 조건에 적응하고 경로를 최적화할 수 있습니다.
스마트 시티
엣지 AI와 벡터 데이터베이스는 교통 관리부터 공공 안전에 이르기까지 도시 운영을 최적화하는 데 중요한 역할을 합니다. 이러한 기술은 엣지에서 빠른 데이터 처리와 의사결정을 가능하게 하여 도시 효율성과 주민의 삶의 질을 향상시킵니다.
예시: 지능형 교통 관리 시스템은 엣지 AI를 활용하여 교차로 카메라의 비디오 피드를 로컬에서 처리합니다. 시스템은 교통 패턴을 벡터 표현으로 변환하고 이를 로컬 벡터 데이터베이스에 저장된 과거 데이터와 비교합니다. 이를 통해 혼잡을 줄이기 위한 실시간 신호등 조정이 가능해지며, 중앙 서버로 지속적으로 데이터를 전송할 필요 없이 빠른 응답 시간을 보장합니다.
산업 자동화
엣지 AI와 벡터 데이터베이스는 제조 및 산업 환경에서 예측 유지보수와 장비 모니터링을 촉진합니다. 이 접근 방식은 기계 상태에 대한 실시간 분석을 가능하게 하여 가동 중단을 방지하고 생산 프로세스를 최적화합니다.
예시: 스마트 제조 공장은 핵심 기계에 센서를 배치하여 진동 데이터를 지속적으로 수집합니다. 엣지 디바이스는 이 데이터를 벡터 표현으로 변환하고, 로컬 벡터 데이터베이스에 저장된 알려진 "정상" 및 "결함" 진동 패턴과 비교합니다. 이러한 실시간 분석은 잠재적인 장비 고장을 즉시 감지할 수 있게 하여, 선제적 유지보수를 촉진하고 비용이 많이 드는 중단을 최소화합니다.
헬스케어
엣지 AI와 벡터 데이터베이스는 개인정보 보호를 저해하지 않으면서 지속적인 건강 모니터링을 가능하게 하여 환자 치료를 혁신합니다. 이러한 기술은 웨어러블 디바이스나 로컬 시스템에서 직접 건강 데이터의 정교한 분석을 가능하게 합니다.
예시: 웨어러블 ECG 모니터는 엣지 AI를 활용하여 심장 리듬 데이터를 로컬에서 처리합니다. 디바이스는 ECG 패턴을 벡터로 변환하고 온보드에 저장된 정상 및 비정상 리듬 데이터베이스와 비교합니다. 이 접근 방식은 민감한 건강 데이터를 디바이스에 유지하면서 잠재적인 심장 문제를 즉시 감지할 수 있게 하여, 대응성과 환자 개인정보 보호를 모두 향상시킵니다.
결론
비정형 데이터의 규모와 중요성이 커짐에 따라, 효율적인 처리 솔루션은 점점 더 중요해지고 있습니다. Tim은 Milvus와 같은 벡터 데이터베이스의 혁신적 잠재력을 잘 강조했으며, 클라우드와 엣지 배포 모두를 위한 고급 기능을 제공합니다. 대규모 AI 애플리케이션이든 리소스가 제한된 엣지 디바이스이든, Milvus와 그 경량 대응 제품인 Milvus Lite는 조직이 실시간 의사결정, 혁신, 운영 효율성을 위해 비정형 데이터를 활용할 수 있도록 지원합니다. 클라우드에서 엣지까지 확장할 수 있는 유연성을 갖춘 이러한 기술은 현대 AI 기반 솔루션의 최전선에 있습니다.
추가 리소스
계속 읽기
Introducing Loon: A New Storage Engine for Vector Data That Never Stops Changing
Loon is a new storage engine for Milvus 3.0 and Zilliz Vector Lakebase, built to manage evolving vector datasets with ColumnGroups, row ID alignment, and Manifests.
My Wife Wanted Dior. I Spent $600 on Claude Code to Vibe-Code a 2M-Line Database Instead.
Write tests, not code reviews. How a test-first workflow with 6 parallel Claude Code sessions turns a 2M-line C++ codebase into a daily shipping pipeline.
Zilliz Cloud Delivers Better Performance and Lower Costs with Arm Neoverse-based AWS Graviton
Zilliz Cloud adopts Arm-based AWS Graviton3 CPUs to cut costs, speed up AI vector search, and power billion-scale RAG and semantic search workloads.