FastAPI를 사용하여 웹캠 스트리밍을 제공하고, 흉기 감지를 감지하여 서버에 알림을 전송.
1. 필요한 라이브러리
from fastapi import FastAPI
import cv2
import uvicorn
from starlette.responses import StreamingResponse
import httpx # 비동기 HTTP 요청을 위한 라이브러리
import asyncio # 비동기 프로그래밍을 지원하는 라이브러리
import nest_asyncio
from fastapi.middleware.cors import CORSMiddleware
2. FastAPI 애플리케이션 생성 및 CORS 설정
app = FastAPI()
# CORS 설정 추가
origins = ["http://localhost:3000", "http://localhost:5000"]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
- FastAPI 애플리케이션 인스턴스를 생성.
- CORS 설정: 특정 도메인에서의 요청을 허용합니다.
- 클라이언트(http://localhost:3000)와 서버 (http://localhost:5000)에서의 요청을 허용.
3. nest_asyncio 적용
nest_asyncio.apply()
- 현재 이벤트 루프에 nest_asyncio를 적용하여 중첩된 이벤트 루프를 허용.
- 주로 Jupyter Notebook과 같은 환경에서 필요할 수 있음.
4. 웹캠 스트리밍 기능
def generate_frames():
cap = cv2.VideoCapture(0)
if not cap.isOpened():
print("웹캠을 열 수 없습니다.")
return
while True:
success, frame = cap.read()
if not success:
break
# 프레임 해상도를 50%로 줄임
frame = cv2.resize(frame, (0, 0), fx=0.5, fy=0.5)
# JPEG 포맷으로 인코딩
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')
- generate_frames: 웹캠에서 비디오 프레임을 캡처하고 스트리밍을 위한 형식으로 인코딩하는 제너레이터 함수.
- cv2.VideoCapture(0): 기본 웹캠을 열기. 실패 시 오류 메시지를 출력.
- 무한 루프를 통해 지속적으로 프레임을 캡처하고, 프레임의 해상도를 줄인 후 JPEG 형식으로 인코딩하여 반환.
- yield를 사용하여 스트리밍 응답을 생성.
5. 웹캠 비디오 피드 엔드포인트
@app.get('/video_feed')
async def video_feed():
return StreamingResponse(generate_frames(),
media_type='multipart/x-mixed-replace; boundary=frame')
- /video_feed 엔드포인트는 클라이언트가 웹캠 비디오 스트림을 요청할 때 호출.
- StreamingResponse를 사용하여 generate_frames 함수의 출력을 스트리밍 응답으로 반환.
6. 흉기 감지 알림을 Express 서버에 POST 요청
@app.post('/detect_weapon')
async def detect_weapon():
url = "http://localhost:8082/alert" # Express 서버의 websocket 알림 수신 경로
data = {"detected": True}
async with httpx.AsyncClient() as client:
try:
response = await client.post(url, json=data)
response.raise_for_status() # HTTP 오류 발생 시 예외 발생
except httpx.HTTPStatusError as exc:
return {"error": f"흉기 감지 알림 전송 실패: {exc.response.status_code}"}
except Exception as exc:
return {"error": f"알림 전송 중 오류 발생: {str(exc)}"}
return {"message": "Weapon detection alert sent to Express server"}
- /detect_weapon 엔드포인트는 흉기가 감지되었을 때 Express 서버에 POST 요청을 보내는 기능을 구현.
- httpx.AsyncClient를 사용하여 비동기적으로 POST 요청을 보내며, 요청 본문에 {"detected": True} 데이터를 포함.
- 요청이 성공하면 Express 서버에 알림을 보냈다는 메시지를 반환하고, 오류가 발생할 경우 적절한 에러 메시지를 반환.
7. FastAPI 애플리케이션 실행
@app.get('/')
async def index():
return {"message": "웹캠 스트리밍 서버가 실행 중입니다. /video_feed 에 접속하세요."}
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
- 이 FastAPI 애플리케이션은 웹캠에서 비디오를 스트리밍하며, 흉기가 감지되면 Express 서버에 알림을 POST 요청으로 전송.
- CORS 설정을 통해 특정 도메인에서의 요청을 허용하고, 비동기 프로그래밍을 통해 효율적인 처리를 지원
실행 명령어
python .\main.py
or
uvicorn main:app --reload --host 127.0.0.1 --port 8000
개발자