Flask SocketIO를 이용하여 실시간 웹캠 감정데이터 측정하기 2

winterholic·2024년 1월 10일

FaceReview Project

목록 보기
6/9

지난 포스트에서는 본 프로젝트에서 Flask Socketio를 사용하게 된 배경과 기본적인 이론에 대하여 작성하였다.
이번 포스트에서는 Flask Socketio를 사용하여 본 프로젝트에서 진행하려는 감정분석에 대한 테스트 flask와 socketio를 사용하여보는 테스트를 했던 부분에 대하여 작성할 것이다.

일단 AI모델을 이용하여 감정을 분석하기 위해서 tensorflow와 opencv를 이용하였다. 또한 html의 템플릿을 이용하여 라이브서버에서 자신의 웹캠 데이터를 실시간으로 테스트하여 정보를 저장해보는 코드를 작성하였다. 사실 이 테스트에 있어서, 순수한 목적만 생각한다면 socketio를 사용할 필요는 없다. 그러나 socketio에 대한 테스트도 겸하여 진행하였기 때문에 코드가 비효율적인 부분이 많이 존재할 수 있다.

camtest.py

import os
from tensorflow import keras
import cvlib as cv
import shutil
import numpy as np
from PIL import Image
import cv2
import time
import io
from flask import Flask, render_template, Response
from flask_socketio import SocketIO, emit
from datetime import datetime
import base64

app = Flask(__name__)
socketio = SocketIO(app)

# 설정값
frame_interval = 5  # 10초에 1개의 프레임을 전송
frame_data_folder = 'framedata'

# YouTube URL을 Python 코드에서 전달
youtube_url = "https://www.youtube.com/embed/CMHmiau2pQY"

# emotion list
emotion = ["happy", "surprise", "angry", "sad", "neutral"] 

# load model
model = keras.models.load_model('model.h5')


@app.route('/')
def index():
    return render_template('index.html', youtube_url=youtube_url)

def analysis_emotion(string_image):
    # imgdata , time
    imgdata = string_image
    imgdata = base64.b64decode(imgdata) # base64 to image

    # image open
    image = Image.open(io.BytesIO(imgdata))
    image = np.array(image)

     #detect face and crop
    faces, conf = cv.detect_face(image)
    for (x, y, x2, y2) in faces:
        cropped = image[y:y2, x:x2]
    resize = cv2.resize(cropped, (96,96))
    img = cv2.cvtColor(resize, cv2.COLOR_BGR2GRAY)

    # image normalization
    img = img / 255

    img = img.reshape(96, 96, 1)
    img = np.expand_dims(img, axis=0)

    #predict
    pred = model.predict(img, verbose=0)

    return emotion[np.argmax(pred)]

def generate_frames():
    camera = cv2.VideoCapture(0)  # 0은 기본 웹캠을 나타냅니다
    while True:
        success, frame = camera.read()
        if not success:
            break
        ret, buffer = cv2.imencode('.jpg', frame)
        if not ret:
            continue
        if ret:

            # 현재 날짜와 시간을 파일 이름으로 사용하여 문자열 데이터 저장
            current_time = datetime.now().strftime('%Y-%m-%d_%H-%M-%S-%f')

            # 이미지를 저장
            filename2 = f'framedata/{current_time}.jpg'
            cv2.imwrite(filename2, frame)
            
            # 이미지 데이터를 Base64 문자열로 변환
            frame_data = base64.b64encode(buffer.tobytes()).decode('utf-8')
            
            frame_bytes = buffer.tobytes()

            # 현재 날짜와 시간을 파일 이름으로 사용하여 문자열 데이터 저장
            current_time = datetime.now().strftime('%Y-%m-%d_%H-%M-%S-%f')

            this_emotion = analysis_emotion(frame_data)

            name_string = current_time + this_emotion

            filename = os.path.join(frame_data_folder, f'{name_string}.txt')

            with open(filename, 'w') as file:
                file.write(frame_data)

            yield (b'--frame\r\n'
                   b'Content-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n')
            time.sleep(frame_interval)  # 프레임 전송 간격 설정
        else:
            break
    camera.release()

@app.route('/video_feed')
def video_feed():
    return Response(generate_frames(), mimetype='multipart/x-mixed-replace; boundary=frame')

@socketio.on('connect', namespace='/test')
def test_connect():
    print('Web client connected')
    emit('response', {'data': 'Webcam streaming initiated'})

@socketio.on('disconnect', namespace='/test')
def test_disconnect():
    print('Web client disconnected')

@socketio.on('get_frame', namespace='/test')
def get_frame():
    for frame in generate_frames():
        emit('frame', {'data': frame})

if __name__ == '__main__':
    if not os.path.exists(frame_data_folder):
        os.makedirs(frame_data_folder)
    socketio.run(app, debug=True)

templates/index.html

<!DOCTYPE html>
<html>
<head>
    <title>Webcam Streaming</title>
</head>
<body>
    <h1>Real-time Webcam Streaming</h1>
    <div id="content" style="display: flex; justify-content: space-between;">
        <div id="youtube_video" style="flex: 1;">
            <iframe width="900" height="450" src="{{ youtube_url }}" frameborder="0" allowfullscreen></iframe>
        </div>
        <div id="webcam_feed" style="flex: 1;">
            <img id="video_feed" src="{{ url_for('video_feed') }}" width="600" height="450" />
        </div>
    </div>
    <script src="https://ajax.googleapis.com/ajax/libs/jquery/3.5.1/jquery.min.js"></script>
    <script src="{{ url_for('static', filename='js/camera.js') }}"></script>
</body>
</html>

다음 코드를 이용하여 실시간으로 웹캠을 라이브서버에 연결하여 감정분석을 진행할 수 있다.
물론 model = keras.models.load_model('model.h5') 이라는 모델을 불러오는 부분이 있기 때문에 python파일과 같은 위치에 감정분석을 진행하여 주는 딥러닝 모델이 존재하여야 한다.

profile
오늘만 산다.

0개의 댓글