실시간 카메라 - 백엔드 서버 연결

목업으로 처리하고 있던 유동인구 정보를 실제 카메라에 잡히는 인원으로 추정할 수 있도록 하는 과정

 

인구를 감지하는 방법중 제시한건 2개

 

1 그냥 그 때 그 때 화면에 나오는 사람 수를 기준으로 추정하는 방법과

2 특정 사람을 인식해서 몇 명 정도인지 대략 측정하는 방법 (중복으로 카운트 되지 않도록 함)

 

2번은 최대 가능한 인원이 3개정도고, 가격이 너무 비싼 문제 등 여러가지 하자가 많았기 때문에 1을 선택했다.

 

1에서도 4개로 나눌 수 있었고 간단하게 정리했다

	연결 방법
웹캠 구매	라즈베리파이 사용 -> 직접 웹캠을 연결
맥북 등 노트북 카메라 사용	그냥 터미널에서 돌리면 됨
아이폰 연결	와이파이 스트림 사용 (맥북이나 라즈베리파이 아무거나 상관 x)
안드로이드 연결	라즈베리파이 (USB 웹캠 모드로 하고 파이랑 연결)

 

그중 아이폰 연결은 와이파이 스트림 (동일 와이파이를 사용하는 경우에 한함) 으로 연결하는 방식이었고 

앱을 깔아서 진행했다. 사용한 앱은 SIMPLE IP CAM 

들어가면 FPS  나 해상도 ? 등을 조정할 수 있었고 

http://x.x.x.x:8080/stream.mjpg

조정 이후 Start server 누르면 이런 큐알 화면으로 이동

굳이 큐알을 사용할 필요는 없고 저 밑에잇는 URL 로 사용하면 됨 

지금 나는 보이다시피 데이터로 연결되어있어서 0.0.0.0 으로 뜨지만 와이파이 연결하면 그 ip 번호로 뜨고

 

비디오 화면 코드에 해당 url 값을 넣어주면 된다.

# webcam, android, macbook
cap = cv2.VideoCapture(0)
# iphone
cap = cv2.VedioCapture("http://x.x.x.x:8080/stream.mjpg")

 

코드는 처음에 

"""
사람 수 검출 테스트 스크립트 (터미널 5초 주기 최대치 출력 버전)
- 화면 창: 실시간 박스 표시 (기존과 동일)
- 터미널: 5초마다 그 구간 동안 감지된 인원수의 최대값을 출력
- 종료: 영상 창 활성화 상태에서 'q' 키
"""
import cv2
import time
import requests

# 로컬 테스트용
# BACKEND_URL = "http://localhost:8000/api/crowd/live"

# 배포용 
BACKEND_URL = "http://x.x.x.x:4327/api/crowd/live"

# 경로는 백엔드 확정 후 교체
last_sent = 0


hog = cv2.HOGDescriptor()
hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())

# 1. 아이폰 카메라 (Simple IP Camera) 사용하여 연결
# cap = cv2.VideoCapture("http://x.x.x.x:8080/stream.mjpg")

# 2. 노트북 카메라, 웹캠, 안드로이드 USB-C 카메라 사용하여 연결
# 추가: 웹캠이랑 안드로이드 USB-C 사용하는 경우 라즈베리파이 필요 
cap = cv2.VideoCapture(0)

if not cap.isOpened():
    print("카메라를 열 수 없습니다.")
    print("- 맥북이면: 시스템 설정 > 개인정보 보호 및 보안 > 카메라에서 터미널 권한 확인")
    print("- 다른 프로그램이 카메라를 이미 쓰고 있는지도 확인해보세요 (Zoom, 카메라 앱 등)")
    exit()

print("카메라 열림. 영상 창에서 'q' 키를 누르면 종료됩니다.")
print("터미널에는 5초마다 그 구간의 최대 인원수가 출력됩니다.\n")

window_start = time.time()
window_max = 0

while True:
    ret, frame = cap.read()
    if not ret:
        # 아이폰으로 테스트 시 break 대신 countinue 로 바꿔주기 
        print("프레임을 읽지 못했습니다. 카메라 연결을 확인하세요.")
        break

    small = cv2.resize(frame, (640, 480))

    boxes, weights = hog.detectMultiScale(
        small, winStride=(8, 8), padding=(8, 8), scale=1.05
    )

    for (x, y, w, h) in boxes:
        cv2.rectangle(small, (x, y), (x + w, y + h), (0, 255, 0), 2)

    count = len(boxes)
    window_max = max(window_max, count)

    cv2.putText(
        small, f"People: {count}", (10, 30),
        cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2
    )

    cv2.imshow("Person Detection Test", small)

    now = time.time()
    if now - window_start >= 5:
        print(f"[{time.strftime('%H:%M:%S')}] 최근 5초간 최대 인원수: {window_max}")
        try:
            requests.post(BACKEND_URL, json={"count": window_max}, timeout=1)
        except requests.RequestException:
            pass  # 백엔드 없거나 꺼져 있어도 검출은 계속
        window_start = now
        window_max = 0

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

이렇게 짰음

 

이미지가 들어오는 경로를 cap 으로 받아주고 (카메라를 킴)

cap 이 잘 열려있다면 프레임 마다 사람이 몇 명 있는지 세어서 박스로 표시

 

hog 가 사람 검출기로 사용된다 

hog = cv2.HOGDescriptor()
hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())

 

해당 화면을 resizing 해준다 크기를 줄이면 계산이 빨라지기 때문에 매 프레임마다 사람을 인식해야하는 경우 유용함!

boxes, weights 를 찾고 사람을 인식하면 사각형 목록으로 받고

참고로 winStride, scale 이 얼마나 ... 자세히 어떤 크기까지 인식할지를 정하는 세부 설정으로 사용된다. 

 

boxes 를 for문으로 돌면서 초록색 네모를 그리고 

나는 매 프레임마다 서버로 보내지 않고 5초 내에 인식된 최대 사람수를 보내기 때문에 해당 값을 count 와 window_max 를 사용하여 저장하고 보낸다 

now = time.time()
if now - window_start >= 5:

5초를 세기 위해 now 값을 저장해주고 현재 시각 - 타이머 시작시간 이 5초 이상이면 보내고 아니면 pass ... 

    print(f"[{time.strftime('%H:%M:%S')}] 최근 5초간 최대 인원수: {window_max}")
    try:
        requests.post(BACKEND_URL, json={"count": window_max}, timeout=1)
    except requests.RequestException:
        pass

5초가 지나면 로그를 찍고

post 를 사용해서 앞서 저장한 backend 서버로 몇 명이 인식되었는지 보낸다.

이 과정에서 생긴 오류는 RequestException 으로 예외처리 해줌 

requests 는 기본으로 제공되는 라이브러리가 아니기 때문에 pip3 install requests 로 다운받아주기!

 

반복문 종료 이후 (q 또는 ctrl C) release 해줌 

 

근데 솔직히..>!!!! 성능이 너무 쓰레기였다 

혼자 개발하고 있는.. 인턴이다보니 사람들을 부를수 없어서 핸드폰으로 여러 사람을 캡처해서 보여줘도 인원이 안 늘거나

가방 같은것도 사람으로 인식해서 갑자기 2가 되는 경우가 생겼다 

 

성능개선 타임. . . 

옛날에 대학교 때 yolo 로 자전거 인식 프로그램 만들었던 기억이 나서 yolo 를 도입하기로 했다 

 

"""
사람 수 검출 테스트 스크립트 (YOLO 버전)
화면 창: 실시간 박스 표시
터미널: 5초마다 그 구간 최대 인원수 출력
종료: 영상 창에서 'q'
"""
import cv2
import time
import requests
from ultralytics import YOLO

# 로컬 테스트용
# BACKEND_URL = "http://localhost:8000/api/crowd/live"
# 배포 서버용
BACKEND_URL = "http://xxxx.xxxxx.co.rk/api/crowd/live"

모델 로드 (최초 실행 시 가중치 자동 다운로드, yolo11n.pt는 ~5MB)
model = YOLO("yolo11n.pt")   # 정확도 더 필요하면 yolo11s.pt / yolo11m.pt
CONF = 0.4                   # 신뢰도 임계값 (낮추면 더 많이 잡지만 오검출↑)
DEVICE = "cpu"               # 애플 실리콘이면 "mps"로 바꿔 GPU 가속 시도 가능

# 아이폰 카메라 (Simple IP Camera) 사용하여 연결
# cap = cv2.VideoCapture("http://x.x.x.x:8080/stream.mjpg")
# 노트북 카메라, 웹캠, 안드로이드 USB-C 카메라 사용하여 연결
# 추가: 웹캠이랑 안드로이드 USB-C 사용하는 경우 라즈베리파이 필요
cap = cv2.VideoCapture(0)
if not cap.isOpened():
    print("카메라를 열 수 없습니다.")
    exit()

print("카메라 열림. 'q'로 종료.\n")

window_start = time.time()
window_max = 0

while True:
    ret, frame = cap.read()
    if not ret:
        print("프레임을 읽지 못했습니다.")
        break

    small = cv2.resize(frame, (640, 480))

    # classes=[0] → COCO 'person' 클래스만 검출
    results = model(small, classes=[0], conf=CONF, device=DEVICE, verbose=False)
    boxes = results[0].boxes

    count = len(boxes)
    for box in boxes:
        x1, y1, x2, y2 = map(int, box.xyxy[0])
        score = float(box.conf[0])
        cv2.rectangle(small, (x1, y1), (x2, y2), (0, 255, 0), 2)
        cv2.putText(small, f"{score:.2f}", (x1, y1 - 6),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1)

    window_max = max(window_max, count)

    cv2.putText(small, f"People: {count}", (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
    cv2.imshow("Person Detection Test", small)

    now = time.time()
    if now - window_start >= 5:
        print(f"[{time.strftime('%H:%M:%S')}] 최근 5초간 최대 인원수: {window_max}")
try:
            requests.post(BACKEND_URL, json={"count": window_max}, timeout=1)
        except requests.RequestException:
            pass
        window_start = now
        window_max = 0

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

yolo 를 사용하기 위해 ultralytics 를 pip3 install 해주고 import 해줌 

model = YOLO("yolo11n.pt")   # 학습된 모델 파일 불러오기
CONF = 0.4                   # 신뢰도 임계값
DEVICE = "cpu"               # 계산을 CPU로 할지 GPU로 할지

hog 를 사용하던 검출기 대신 model 을 yolo11 로 불러왔다 

results = model(small, classes=[0], conf=CONF, device=DEVICE, verbose=False)
boxes = results[0].boxes

yolo 는 진짜 여러 클래스들을 구분할 수 잇는데 여기선 [0] 을 사용해서 사람만 검출할 수 잇도록 지정해주고 

conf 값 (몇퍼센트 이상으로 사람이라고 판단되면 count 할지)

device 는 뭘 쓸지 등등을 매개변수로 넘겨서 처리 

 

이후는 박스를 그리고 사람을 세어 추가한다. 

 

처음 다운로드랑 세팅에는 시간이 약간 걸렸는데 다 받고 나니까 성능이 진짜!!! 최고 . .. . 

뒷모습만 겨우인식하던 hog 랑 다르게 사람 수를 거의 정확하게 세는 모습을 확인할 수 있었다 

	HOG (기존)	YOLO (신규)
import	cv2, time, requests	+ ultralytics
검출기 생성	HOGDescriptor 2줄	YOLO("yolo11n.pt") + CONF/DEVICE
검출 실행	detectMultiScale(...)	model(small, classes=[0], ...)
좌표 형식	(x, y, w, h)	(x1, y1, x2, y2)

post 를 사용해서 보내준 값들은 백엔드 코드에서

@PostMapping("/live") 
public Map<String, Object> live (@RequestBody LiveRequest body) {
	return crowdService.sampleAndStore(body.peopleCount(), body.level());
}

로 받고 서비스 코드에서 확인한다.

 

시뮬레이션을 위해 만들어둔 수동 사람 조정 요청이 1순위

2순위가 캠으로 들어오는 정보

3순위가 목업 센서 (그냥 만들어지는 가짜 정보들)

 

body 값으로 들어온 사람 수, level (혼잡도 레벨: 한산, 보통, 붐빔,, 등)을 사용해서 캐시와 DB 에 각각 저장

변화 정도에 따라 알림을 주거나 오토파일럿을 실행하는 동기처리를 수행