프로젝트 개요
본 프로젝트는 저조도 환경에서 특정 인물(유재석)이 감지될 때 자동으로 LED를 켜는 지능형 시스템을 구축하는 것을 목표로 한다.
AWS 클라우드 서비스와 Raspberry Pi를 활용한 엣지 컴퓨팅을 결합하여 실시간 이미지 처리, 얼굴 인식, 데이터 수집 및 머신러닝 기반 의사결정을 구현했다.
Architecture
데이터 흐름 및 처리 과정
- 센서 데이터 수집 및 전송
- Raspberry Pi의 MCP3208 ADC를 통해 아날로그 조도 센서 값을 디지털로 변환
- 변환된 조도 값을 AWS IoT Core를 통해 DynamoDB에 저장하여 시계열 데이터 구축
- 영상 수집 및 전송
- Raspberry Pi 카메라에서 캡처된 영상을 Kinesis Video Streams로 전송
- Kinesis Video Streams의 자동 이미지 캡처 기능을 통해 S3에 주기적으로 이미지 저장
- 얼굴 인식 처리
- AWS Rekognition의 Face Collection 기능을 활용하여 사전 등록된 유재석 얼굴과 비교
- 95% 이상의 신뢰도를 기준으로 일치 여부 판단
- 데이터 병합
- Lambda 함수에서 Rekognition 분석 결과와 DynamoDB의 조도 센서 데이터를 시간 기반으로 정확히 매칭
- 이미지 생성 시간을 기준으로 전후 30초 내의 가장 가까운 조도 데이터를 선택
- 병합된 결과를 CSV 파일로 생성하여 S3에 저장함으로써 학습 데이터 구축
- 모델 학습
- AWS Glue ETL을 통해 개별 CSV 파일들을 하나의 통합 데이터셋으로 병합
- 파생 변수 생성: 조도 50 미만을 저조도로 정의(is_low_light), 저조도와 유재석 감지 여부 조합(is_target_in_low_light)
- SageMaker Canvas에서 AutoML 기반 모델 학습 진행
- SageMaker 엔드포인트로 배포하여 실시간 추론 서비스 구축
- 추론
- Raspberry Pi에서 SageMaker 엔드포인트를 호출하여 실시간 상황 판단
- 예측 결과에 따라 LED 제어
핵심 기술 구성 요소
- SageMaker Canvas
- 정형 데이터에 특화되어 있음
- 그래서 비디오 데이터 직접 처리 불가능
- 코드 없이 머신러닝 모델 구축 가능
- 자동화된 특성 중요도 분석
- 다양한 모델 비교 및 평가 기능
- AWS Glue
- 스키마 자동 감지 및 테이블 생성
- Spark 기반 데이터 처리 파이프라인
- Lambda
- 서버리스 컴퓨팅 서비스로 이벤트 기반 처리 로직 구현
- S3 객체 생성 이벤트에 반응하여 자동으로 데이터 처리 수행
- Rekognition과 DynamoDB 연동을 통한 데이터 병합 및 전처리
- 확장성 있는 이미지 처리 및 분석 파이프라인 구축
- Dynamo DB
- 완전 관리형 NoSQL 데이터베이스로 시계열 조도 데이터 저장에 활용
- TTL(Time To Live) 기능을 통한 자동 데이터 수명 주기 관리
- 파티션 키(timestamp) 기반 효율적인 시간 범위 쿼리 지원
- Rekognition - 전 블로그 내용 참고
- Kinesis Video Streams - 전 블로그 내용 참고
- IoT Core - 전 블로그 내용 참고
실제 구현을 시작해보자.
1. 조도 센서 데이터 수집 및 저장
- 조도 센서 설정
- MCP3208 ADC를 통해 아날로그 조도 센서 값을 디지털로 변환
- 0-1 범위의 원시값을 0-100 스케일로 정규화하여 가독성 향상
- 조도값 50을 기준으로 저조도 여부 판단
라즈베리 파이에서 직접 전압값을 읽을 수 없다. 그래서 MCP3208과 전압을 읽어 통신으로 데이터를 보내주는 소자를 이용하여 전압을 읽어야 함. MCP3208 IC는 SPI 통신을 사용했다.
- Raspberry Pi SPI 통신 설정
- SPI 인터페이스 활성화:
- [Preferences] → [Raspberry Pi Configuration] → [Interfaces] 탭에서 SPI를 Enabled로 설정
- 회로 구성
- SPI 인터페이스 활성화:
| 라즈베리파이 | MCP3208 | 센서 |
| 3.3V | 16번 VDD | |
| 3.3V | 15번 VREF | |
| GND | 14번 AGND | |
| GPIO11 | 13번 CLK | |
| GPIO9 | 12번 DOUT | |
| GPIO10 | 11번 DIN | |
| GPIO8 | 10번 CS | |
| GND | 9번 DGND | |
| 1번핀 | CDS 조도센서 & 10K옴(갈빨검검갈) | |
| GPIO18 | LED 긴 다리 |
- IoT Core 설정
- 사물(Thing) 생성
- 사물 이름 지정 및 디바이스 인증서 구성
- 인증서에 연결할 정책 생성( 필요한 리소스 접근 권한 부여 )
- 인증서에 생성한 정책 연결
- MQTT 주제 구성으로 메시지 라우팅 설정
- DynamoDB 테이블 설계
- 테이블 생성
- 파티션 키: timestamp (ISO 8601 형식)
- 테이블 생성
- Raspberry pi에서 조도센서 데이터 수집 코드 작성
import time
import json
import boto3
from decimal import Decimal # Decimal 타입 추가
from datetime import datetime
from gpiozero import MCP3208
from awscrt import mqtt
from awsiot import mqtt_connection_builder
# AWS IoT Core 설정
AWS_IOT_ENDPOINT = "[Endpoint ARN]"
CLIENT_ID = "light-sensor-client"
TOPIC = "sensor/light-value"
CERT_PATH = "[Cert Path]/device.pem.crt"
KEY_PATH = "[Key Path]/private.pem.key"
CA_PATH = "[Ca Path]/AmazonRootCA1.pem"
# DynamoDB 설정
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('LightSensorData')
# MQTT 연결 설정
mqtt_connection = mqtt_connection_builder.mtls_from_path(
endpoint=AWS_IOT_ENDPOINT,
port=8883,
cert_filepath=CERT_PATH,
pri_key_filepath=KEY_PATH,
ca_filepath=CA_PATH,
client_id=CLIENT_ID,
clean_session=False,
keep_alive_secs=30
)
# 연결 시도
print("🚀 AWS IoT Core 연결 시도...")
connect_future = mqtt_connection.connect()
connect_future.result()
print("✅ 연결 성공!")
# MCP3208 조도 센서 초기화 (채널 0)
cds = MCP3208(channel=0)
try:
while True:
# 현재 타임스탬프 생성 (ISO 8601 형식, UTC)
timestamp = datetime.utcnow().isoformat()
# 조도 값 측정 (0~1 범위 → 0~100 스케일 변환)
light_value = round(cds.value * 100, 2)
print(f"💡 {timestamp} | 조도 값: {light_value:.2f}")
# JSON 데이터 생성
payload = {
"timestamp": timestamp,
"light-value": light_value
}
# MQTT 메시지 전송
mqtt_connection.publish(
topic=TOPIC,
payload=json.dumps(payload),
qos=mqtt.QoS.AT_LEAST_ONCE
)
print(f"📤 MQTT 전송 완료: {payload}")
# DynamoDB에 데이터 저장
try:
# DynamoDB 항목 구성 - float를 Decimal로 변환
item = {
'timestamp': timestamp,
'lightValue': Decimal(str(light_value)), # float를 문자열로 변환 후 Decimal로 변환
'ttl': int(time.time()) + 86400 # 24시간 후 데이터 만료 (TTL)
}
# DynamoDB에 항목 추가
table.put_item(Item=item)
print(f"📊 DynamoDB 저장 완료: {timestamp}")
except Exception as e:
print(f"❌ DynamoDB 저장 실패: {e}")
# 1초 대기 후 반복
time.sleep(1)
except KeyboardInterrupt:
print("\n⏹ 프로그램 종료... MQTT 연결 해제")
disconnect_future = mqtt_connection.disconnect()
disconnect_future.result()
print("🔌 연결 종료 완료!")
2. 영상 데이터 수집 및 얼굴인식 시스템 구축
- Kinesis Video Streams 설정
- 새 비디오 스트림 생성
- 스트림 이름: test-vide-stream
- Raspberry Pi 카메라 연결
- 이미지 캡쳐 후 S3 버킷으로 자동 저장 json 설정
- SamplingInterval: 3000ms마다 이미지 저장 (3초)
- Format: JPEG 이미지로 저장
- ImageSelectorType: Producer Timestamp 기준
- WidthPixels, HeightPixels: 640x480 해상도 설정
- 이미지 캡쳐 후 S3 버킷으로 자동 저장 json 설정
{
"StreamName": "test-vide-stream",
"ImageGenerationConfiguration": {
"Status": "ENABLED",
"DestinationConfig": {
"DestinationRegion": "us-east-1",
"Uri": "s3://jr-detection/kinesis-video-images"
},
"SamplingInterval": 3000,
"ImageSelectorType": "PRODUCER_TIMESTAMP",
"Format": "JPEG",
"FormatConfig": {
"JPEGQuality": "80"
},
"WidthPixels": 640,
"HeightPixels": 480
}
}- update-image-generation-input.json 파일의 설정이 Kinesis Video Stream에 반영되도록 명령어 수행
aws kinesisvideo update-image-generation-configuration --cli-input-json file://update-image-generation-input.json- GStreamer 애플리케이션 실행
cd amazon-kinesis-video-streams-producer-sdk-cpp/build
cmake .. -DBUILD_GSTREAMER_PLUGIN=ON -DBUILD_DEPENDENCIES=FALSE
make
# GStreamer 플러그인 경로 설정
export GST_PLUGIN_PATH=/amazon-kinesis-video-streams-producer-sdk-cpp/build
# AWS 리전 설정
export AWS_DEFAULT_REGION=us-east-1
# AWS 액세스 키 설정
export AWS_ACCESS_KEY_ID=[AWS ACCESS KEY ID]
# AWS 시크릿 액세스 키 설정
export AWS_SECRET_ACCESS_KEY=[AWS SECRET ACCESS KEY]
#GStreamer sample 애플리케이션 실행
./kvs_gstreamer_sample test-vide-stream- Rekognition Face Collection
- s3 폴더 만들기
- Rekognition Collection 생성
# Rekognition Collection 생성
aws rekognition create-collection \
--collection-id "face-detect" \
--profile default
# 생성된 Collection 확인
aws rekognition list-collections --profile default- Rekognition Collection에 S3에 업로드한 얼굴(유재석) 이미지 등록하기
# S3의 이미지를 Collection에 등록
aws rekognition index-faces \
--image '{"S3Object":{"Bucket":"[Your S3 Bucket Name]","Name":"Yoo3.jpg"}}' \
--collection-id "face-detect" \
--max-faces 1 \
--quality-filter "AUTO" \
--detection-attributes "ALL" \
--external-image-id "Yoo3.jpg" \
--profile default
다음 part2 글에서는 Lambda 함수와 Glue를 사용하여 데이터 처리를 구현하고자 한다.