미니 프로젝트 - OLED 캡처 프로그램

개발 목적

현재까지 학습한 MQTT 통신과 OpenCV 기반 이미지 처리 기술을 바탕으로 하나의 응용 프로그램을 제작하는 것이 목적이다.
OLED 실습을 통해 흑백 이미지가 선명하게 출력된다는 점을 확인했으며,
이를 바탕으로 최종적으로 캡처한 화면을 OLED에 출력하는 것을 목표로 한다.

 

시스템 구현

 

1. MQTT

스마트폰과의 MQTT 통신을 통해 원격 제어 기능을 구현하였다.
버튼 패널을 이용해 토글 방식으로 각 기능을 제어할 수 있다.

2. OpenCV

OLED에 출력할 이미지를 전처리하기 위한 목적으로 OpenCV를 활용하였다.
이미지를 HSV 색 공간으로 변환 후, 피부색 계열을 필터링하여 추출하고,
추출된 영역은 흰색, 나머지는 검은색으로 처리하여 선명한 흑백 이미지로 변환한다.

3. YOLO

YOLO 객체 탐지 모델을 이용하여 특정 물체를 인식하고,
탐지된 물체에 따라 OLED에 미리 정의된 이미지를 출력하도록 제어한다.

4. OLED

• OpenCV를 이용하여 웹캠 화면을 실시간으로 캡처하여 출력한다.
• YOLO 탐지 결과에 따라 특정 상황에 맞는 이미지(아이콘 등)를 OLED에 표시한다.

기술 요약

• 사용 보드: Rasberry Pi
• 입력 장치: 웹캠, 모바일 애플리케이션
• 출력 장치: OLED
• 주요 구현 기술: Yolo 객체 탐지, OpenCV 이미지 처리, MQTT 통신
• 사용 언어: Python

 

시스템 구성도

 

 

플로우 차트

 

문제점 해결

MQTT 통신 확인 -> OpenCV 이미지 처리 확인 -> 이미지 프레임 통신 확인 -> OLED 출력 확인
위 단계를 차례대로 진행하며 합병하는 방식으로 구현했다. 

 

문제점

1. 이미지 처리 문제

HSV 색 공간 변환을 통해 피부색을 추출하고, 마스크 처리로 흑백 이미지를 생성하는 과정에서 색상 범위를 적절히 지정하지 못해 정확한 추출이 어려웠다.
범위를 점차 넓혀가며 테스트를 반복한 결과, 적절한 HSV 값 범위를 찾아 문제를 해결할 수 있었다.

2. 이미지 전송 문제

웹캠은 Windows에서 실행되고, OLED는 라즈베리 파이에 연결되어 있기 때문에

이미지 캡처 후 전송 방식에 대한 문제가 발생했다.
이미지를 그대로 전송하면 데이터 손상 가능성이 높고, 전송 시간도 길어지기 때문이다.

이를 해결하기 위해 이미지 데이터를 바이트 배열로 변환하여 전송하는 방식을 선택했다.
이미지는 픽셀 값의 집합이므로, tobytes() 함수를 사용하면 간단히 바이트 배열로 전환할 수 있다.
Base64 인코딩 방식도 고려했지만, 속도와 효율성 면에서 바이트 배열 전송 방식이 더 적절하다고 판단했다.

 

3. 카메라 속도 문제

웹캠을 통해 정보를 받은 것을 토대로 모든 기능이 구현되었었는데, 속도가 매우 느려서 확인이 잘 안될 정도였다.

스레드를 이용하여 카메라를 실행했는데, 이 과정에서 "pass"를 사용한 것이 문제였다.

try:
	while True:
    		pass
except KeyboardInterrupt:
	#오류

위 코드는 CPU를 100%로 사용하게 된다.

계속 루프를 돌아버리기 때문이다. 

Python은 기본적으로 싱글 스레드 성능에 제약이 있고, 이 코드가 계속해서 CPU를 독점하면
다른 쓰레드가 제대로 스케줄링 되지 못하여 카메라 처리 속도나 화면 표시가 매우 느려지거나 끊기게 된다.

 

해결 방법은 time.sleep(0.1)을 사용하는 것이다.

매 0.1초마다 잠시 쉬면서 CPU 점유율을 낮춰준다.

이 코드는 스레드 간 리소스 분배를 원할하게 해주어 정상 작동이 되게 한다.

 

코드

#Rasberry Pi
import paho.mqtt.client as mqtt
import cv2
import re
import threading
import numpy as np
from ultralytics import YOLO
import time
model = YOLO(r'best.pt')
#MQTT
BROKER = "test.mosquitto.org"
PORT = 1883
TOPIC = "karist"

received_message = ""

#capture
count=0
capture_frame=[]

id = ""
def capture_image():
  global capture_frame
  capture_frame = np.array(capture_frame)
  global count
  file_name = f"./t/img_test{count}.jpg"
  cv2.imwrite(file_name,capture_frame)
  count+=1


def on_message(client, userdata, msg):
  global id

  msg_text = msg.payload.decode()
  print(f"Received message: {msg_text}")
  if(msg.topic ==  TOPIC+"/Click"and msg_text=="Click"):
    capture_image()
    _,buffer = cv2.imencode('.jpg',capture_frame)
    client.publish(TOPIC+"/Image",buffer.tobytes())
  if(msg.topic == TOPIC+"/Detect"):
    print("DETECT")
    client.publish(TOPIC+"/Detect_Image",str(id))


def camera_thread():
    global id
    global received_message
    global capture_frame
    cap = cv2.VideoCapture(0)
    if not cap.isOpened():
        print("Failed to open camera")
        return

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        results = model(frame)
        for i in results:          #box                
            for box in i.boxes:    #'xyxy':box좌표, 'cls':클래스id,'conf':신뢰도 
              class_id = int(box.cls[0]) #클래스ID(라벨링한)0,
              id = str(class_id)
              confidence = box.conf[0]   #확률
              #print(f"Class ID: {class_id}, Confidence: {confidence:.2f}")
        annotated_frame = results[0].plot() 
       
        cv2.imshow('Project',annotated_frame)
        capture_frame=frame.copy()
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break

    cap.release()
    cv2.destroyAllWindows()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_message = on_message
client.connect(BROKER, PORT, 60)

client.loop_start()
client.subscribe(TOPIC+"/Click")
client.subscribe(TOPIC+"/Detect")
thread = threading.Thread(target=camera_thread)
thread.daemon = True
thread.start()

try:
    while True:
        time.sleep(0.1)
except KeyboardInterrupt:
    print("Program terminated")
    client.loop_stop()
    client.disconnect()

 

#windows
import paho.mqtt.client as mqtt
import cv2
import re
import threading
from gpiozero import Button
import numpy as np
import board
import busio
from adafruit_ssd1306 import SSD1306_I2C
from PIL import Image, ImageDraw
#MQTT
BROKER = "test.mosquitto.org"
PORT = 1883
TOPIC = "karist"

# OLED
i2c = busio.I2C(board.SCL, board.SDA)
display = SSD1306_I2C(128, 64, i2c)

display.fill(0)
display.show()
image = Image.new('1', (display.width, display.height))
draw = ImageDraw.Draw(image)
button = Button(26,bounce_time=0.05)
button2 = Button(21,bounce_time=0.05)
def image_show(img):
	image = Image.new('1',(display.width,display.height))
	draw = ImageDraw.Draw(image)
	name = Image.open(img+".png").convert('1')
	
	image.paste(name,(46,16))
	display.image(image)
	display.show()

def on_message():
    MESSAGE = "Click"
    client.publish(TOPIC+"/Click",MESSAGE)
    print("send")

def detect_image():
	MESSAGE = "Detect"
	client.publish(TOPIC+"/Detect",MESSAGE)
	print("send2")

def oled(client,userdata,msg):
	if msg.topic == TOPIC+"/Image":
		print("image:")
		nparr = np.frombuffer(msg.payload,np.uint8)
		img = cv2.imdecode(nparr,cv2.IMREAD_COLOR)
		hsv = cv2.cvtColor(img,cv2.COLOR_BGR2HSV)
			
		lower_skin = np.array(([0,20,70]),dtype=np.uint8)
		upper_skin = np.array(([20,255,255]),dtype=np.uint8)
		
		mask = cv2.inRange(hsv,lower_skin,upper_skin)
		mask_image = Image.fromarray(mask)
		mask_image = mask_image.convert('1')
		mask_image = mask_image.resize((display.width,display.height))
				
		display.image(image)
		display.show()
		image.paste(mask_image,(0,0))
	elif msg.topic == TOPIC+"/Detect_Image":
		print(1)
		data = msg.payload.decode()
		
		print(data)
		if data == "0":
			image_show("verygood")
		elif data == "1":
			image_show("verybad")
			
		else:
			pass


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
button.when_pressed = on_message
button2.when_pressed = detect_image
client.on_message = oled
client.connect(BROKER, PORT, 60)
client.subscribe(TOPIC+"/Image")
client.subscribe(TOPIC+"/Detect_Image")
client.loop_start()

try:
    while True:
        pass 
except KeyboardInterrupt:
    print("exit")
    client.loop_stop()
    client.disconnect()

테스트

 

이미지 캡처

객체 탐지

HSV 색 공간 변환 시 여러 값을 넣으며 테스트를 진행하였다. 
경계값 분석을 통하여 최대한 감지가 잘 되는 범위를 구할 수 있었고 해당 결과를 얻을 수 있었다.

단위 테스트를 진행하며 병합했고, 마지막 통합 테스트를 통해 기능 테스트를 진행했다.

 

향후 발전 가능성

현재 배운 내용으로는 완전한 색 공간 변환, 윤곽선 추출이 이루어지지 않은 것 같다.

기법을 조금 더 조사하여 적용을 시킨다면, 손실되는 데이터 없이 완전한 흑백 사진을

캡처할 수 있을 것이라고 예상한다.

 

구현 결과

사진 캡처

 

객체 탐지