[python] fastapi 실시간 스트리밍 이미지 송출

이번에 새로 진행한 프로젝트의 기능 중 하나는 얼굴 인식으로 하는 로그인 기능이었다

얼굴 인식하는 model은 cvteam에서 받아서 나는 ui에 기능만 구현하면 되는 상황

main.py

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import cv2

app = FastAPI()

def generate_frames():
    cap = cv2.VideoCapture(0)
    while True:
        success, frame = cap.read()
        if not success:
            break
        ret, buffer = cv2.imencode('.jpg', frame)
        frame = buffer.tobytes()
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')

@app.get("/video_feed")
def video_feed():
    return StreamingResponse(
    	generate_frames(),
    	media_type="multipart/x-mixed-replace; boundary=frame")

main.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Live Stream</title>
</head>
<body>
    <h1>Live Video Stream</h1>
    <img id="video-stream" src="/video_feed" alt="Video Stream" />
</body>
</html>

자바스크립트에서 특별한 처리가 필요하지 않고, 이미지 태그의 src 속성을 백엔드 스트리밍 URL로 설정만 하면 잘 작동했다.

여기서 추가로 얼굴 인식 완료되면 등록된 이름을 같이 앞단에 전달해달라는 요청을 받았다.

스트리밍 이미지 송출하고 분리되어서 string도 앞단에 보내려면 media type을 수정해야 한다.

기존 코드

ret, buffer = cv2.imencode('.jpg', frame)
frame = buffer.tobytes()

yield (b'--frame\r\n'
b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')

@app.get("/video_feed")
def video_feed():
    return StreamingResponse(generate_frames(), media_type="multipart/x-mixed-replace; boundary=frame")

변경 코드

media_type = "multipart/x-mixed-replace; boundary=frame" 으로 송출하던 frame을

base24로 encode → 'utf-8'로 decode해서 dict에 담아서 yield하고

추가로 전달해야하는 string도 dict에 담아서 yield 해서 앞단에 전달했다.

그리고 StreamResponse의 media_type을 "text/event-stream"으로 변경해줬다.

ret, buffer = cv2.imencode('.jpg', img)
img = buffer.tobytes()

base64_str = base64.b64encode(img)
base64_string = base64_str.decode('utf-8')
data = { "base64": base64_string }

# yield streaming image
yield f"event: image\ndata: {json.dumps(data)}\n\n"

# yield string
data = {"name": name}
yield f"event: notification\ndata: {json.dumps(data)}\n\n"

@app.get("/video_feed")
def video_feed():
    return StreamingResponse(generate_frames(), media_type="text/event-stream")

그랬더니 Streaming image, String 모두 앞단으로 잘 넘겨주는 것 확인했다.

최종코드

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import cv2

app = FastAPI()

def generate_frames():
    cap = cv2.VideoCapture(0)
    while True:
        success, frame = cap.read()
        if not success:
            break
        ret, buffer = cv2.imencode('.jpg', frame)
        img = buffer.tobytes()
        
        base64_str = base64.b64encode(img)
        base64_string = base64_str.decode('utf-8')
        data = { "base64": base64_string }
        
        # yield streaming image
        yield f"event: image\ndata: {json.dumps(data)}\n\n"
        
        # yield string
        data = {"name": name}
        yield f"event: notification\ndata: {json.dumps(data)}\n\n"

@app.get("/video_feed")
def video_feed():
    return StreamingResponse(generate_frames(), media_type="text/event-stream")

저작자표시

'👩🏻‍💻 > python' 카테고리의 다른 글

[python] Fastapi 애플리케이션 exe 파일로 배포하기 (0)	2024.11.05
[python] socket 동시 연결 (1)	2024.10.22
[python] TypeError: 'type' object is not subscriptable (1)	2024.05.02
[python] Received response with content-encoding: gzip, but failed to decode it. (0)	2024.02.19
[python] 네이버 카페 크롤링 (1)	2024.01.30

Bono-end

[python] fastapi 실시간 스트리밍 이미지 송출

main.py

main.html

기존 코드

변경 코드

최종코드

'👩🏻‍💻 > python' 카테고리의 다른 글

티스토리툴바

[python] fastapi 실시간 스트리밍 이미지 송출

main.py

main.html

기존 코드

변경 코드

최종코드

'👩🏻‍💻 > python' 카테고리의 다른 글

관련글

티스토리툴바