딥러닝 - OpenCV_004_FastAPI

[4차시] 학습목표

○ 기본적인 FastAPI 사용법 학습하기 ○ 웹 화면에 카메라 영상 출력하기 ○ 웹 출력 영상을 Gray 이미지로 출력하기

1  FastAPI 설치 및 서버 실행하기

• FastAPI : Sebastián Ramírez란 사람이 만든 파이썬 기반 오픈소스 웹 프레임워크
  • 파이썬 3.6 이상
  • 빠른(고성능) API 서버 웹 프레임워크
  • Flask나 Django보다 단순하고 배우기 쉬움
  • GitHub : https://github.com/tiangolo/fastapi
  • 튜토리얼 : https://fastapi.tiangolo.com/ko/
  • 다양한 형태의 Response : https://fastapi.tiangolo.com/ko/advanced/custom-response/

• ASGI(Asynchronous Server Gateway Interface) : 비동기 Python 웹 서버
  • 동기(synchronous) 처리 : 특정 작업이 끝나면 다음 작업을 처리하는 순차처리 방식
    • 파이썬은 기본적으로 동기 방식으로 동작하는 -
  • 비동기(asynchronous) 처리 : 여러 작업을 처리하도록 예약한 뒤 작업이 끝나면 결과를 받는 방식
    • 스레드 하나로 동시 처리를 하기 위한 것

• pyngrok : 방화벽을 넘어서 외부에서 로컬에 접속 가능하게 하는 터널 프로그램

• asyncio : Python 표준 라이브러리와 함께 제공되는 비동기 I/O 프레임워크로 파이썬 3.4에서 추가

• async / await : 비동기(Asynchronous) 처리를 위해 파이썬 3.5에서 추가한 구문
  • async : 비동기 함수를 생성하는 명령
  • await : 비동기 함수를 호출하는 명령

• nest_asyncio : 비동기 방식에서 이벤트가 중첩되는 것을 허용하는 기능을 제공하는 라이브러리

• uvloop : asyncio를 대체하기 위해 제작
  • 동기 함수 내에서 비동기 함수를 호출하려면 asyncio 라이브러리의 이벤트 루프를 이용하는 데 이를 대신하는 기능
  • Cython으로 작성
  • libuv 위에 구축 (libuv : nodejs에서 사용하는 고성능 다중 플랫폼 비동기 I / O 라이브러리)
  • 성능상에서 nodejs, gevent 및 기타 Python 비동기 프레임 워크보다 2배 이상 빠름

• uvicorn : uvloop 및 httptools를 사용하여 만든 빠른 ASGI (Asynchronous Server Gateway Inteface) 서버
  • 다양한 관리 기능으로 fastapi와 함께 사용
  • HTTP / 1.1 및 WebSocket을 지원

1.1  동기 프로그래밍

#실습
import time

def find_users_sync(n) :
    for i in range(1, n+1) :
        print(f"{n}명 중 {i}번째 사용자 조회 중 ...")
        time.sleep(1)    
    print(f"> 총 {n}명 사용자 동기 조회 완료 !!")    
    
find_users_sync(5)

이 코드는 find_users_sync 함수를 정의하고,

n명의 사용자 중에서 1초씩 딜레이를 주면서 각 사용자를 조회하는 동기 방식의 함수입니다.

함수의 인자로 조회할 사용자 수 n을 받고, for문을 이용하여 각 사용자 조회를 수행합니다.

각 사용자 조회마다 time.sleep(1) 함수를 사용하여

1초씩 딜레이를 주어 1초마다 한 명씩 조회할 수 있도록 합니다.

모든 사용자 조회가 완료되면 print 함수를 이용하여 조회 완료 메시지를 출력합니다.

따라서 이 코드는 간단한 동기 방식의 사용자 조회 함수입니다.

 

#코드 리뷰 및 피드백
import time

#사용자 동기 조회 함수
def find_users_sync(n):
for i in range(1, n+1):
print(f"{n}명 중 {i}번째 사용자 조회 중 ...")
time.sleep(1)
print(f"> 총 {n}명 사용자 동기 조회 완료 !!")

#처리 시간을 계산하는 함수
def process_sync():
start = time.time()
find_users_sync(3)
find_users_sync(2)
find_users_sync(1)
end = time.time()
print(f">> 동기 방식으로 처리한 시간 : {end-start}")

process_sync()

time 모듈을 사용하여 시간 측정하는 코드입니다.
find_users_sync() 함수는 매개변수 n으로 전달된 인원 수에 따라 사용자 동기 조회를 진행합니다.
process_sync() 함수에서는 find_users_sync() 함수를 차례로 호출하여 처리 시간을 측정합니다.

 

1.2  비동기 프로그래밍

!pip install nest_asyncio fastapi uvicorn

nest_asyncio, fastapi, uvicorn 패키지를 설치하는 코드입니다.

• nest_asyncio: 비동기식 코드와 같은 이벤트 루프 환경에서 asyncio를 중첩된 방식으로
• 사용할 수 있도록 지원하는 라이브러리입니다.
• fastapi: 빠른 속도로 API를 개발할 수 있도록 도와주는 Python 웹 프레임워크입니다.
• uvicorn: ASGI(Asynchronous Server Gateway Interface) 웹 서버입니다.

따라서 이 코드를 실행하면 위 패키지들이 설치됩니다.

import time
import asyncio

# async : 비동기함수 선언 (코루틴 함수) 
async def find_users_sync(n) :
    for i in range(1, n+1) :
        print(f"{n}명 중 {i}번째 사용자 조회 중 ...")
        await asyncio.sleep(1)    
    print(f"> 총 {n}명 사용자 동기 조회 완료 !!")    
    
find_users_sync(5)

• async 키워드를 사용하여 비동기 함수를 정의하였다.
• await 키워드를 사용하여 asyncio.sleep(1) 함수를 실행한다.
• 이 함수는 async 함수 내부에서만 사용할 수 있는 비동기 함수이다.
• find_users_sync(5)를 호출했지만, 실제로 함수가 실행되지는 않는다.
• 이는 async 함수를 직접 호출할 경우, 비동기적으로 실행되지 않는다.
• asyncio.run(find_users_sync(5)) 형태로 실행해주어야 한다.

import time
import asyncio

# async : 비동기함수 선언 (코루틴 함수) 
async def find_users_sync(n) :
    for i in range(1, n+1) :
        print(f"{n}명 중 {i}번째 사용자 조회 중 ...")
        await asyncio.sleep(1)    
    print(f"> 총 {n}명 사용자 동기 조회 완료 !!")    
    
asyncio.run(find_users_sync(5))

• await asyncio.wait() : 여러 개의 코루틴 함수를 호출
• asyncio.run() : 코루틴 함수를 실행
  
  

# 실습
import nest_asyncio

# 다중 작업 간의 메시지 교환
nest_asyncio.apply()

async def process_async():
    start = time.time()
    
    await asyncio.wait([find_users_sync(3),
                       find_users_sync(2),
                       find_users_sync(1)])
    end = time.time()
    print(f">> 비동기 방식으로 처리한 시간 : {end-start}")

asyncio.run(process_async())

nest_asyncio 패키지를 이용하여 asyncio 라이브러리가 중첩되는 것을 처리함.
import nest_asyncio

#중첩 된 이벤트 루프를 해결하기 위해 필요한 apply() 메소드를 실행
nest_asyncio.apply()

async / await 구문을 이용한 비동기 프로그래밍
async def process_async():
start = time.time()

# wait() 메소드를 이용하여 병렬로 수행할 코루틴들의 집합을 전달받아 비동기 작업을 처리
await asyncio.wait([find_users_sync(3),
                   find_users_sync(2),
                   find_users_sync(1)])
end = time.time()

# 비동기 방식으로 처리한 시간 출력
print(f">> 비동기 방식으로 처리한 시간 : {end-start}")

asyncio.run(process_async())
#asyncio.run() 메소드를 이용하여 이벤트 루프를 생성하고 비동기 함수 실행

1.3  FastAPI 설치하기

!pip install fastapi uvicorn pyngrok nest_asyncio

• FastAPI, uvicorn, pyngrok, nest_asyncio 라이브러리를 설치하는 코드
• FastAPI: API 서버 개발에 사용되는 웹 프레임워크
• uvicorn: ASGI 서버
• pyngrok: 로컬 개발 환경에서 퍼블릭 URL을 생성하는 라이브러리
• nest_asyncio: asyncio 루프를 중첩하여 사용하는 데 필요한 도구

1.4  FastAPI 서버 실행하기

• 참고 : https://blog.naver.com/PostView.naver?blogId=duqrlwjddns1&logNo=222268635237&parentCategoryNo=&categoryNo=&viewDate=&isShowPopularPosts=false&from=postView

• 비동기 처리(asyncio)는 Jupyter Notebook 환경에서 지원하지 않음

• app = FastAPI() : FastAPI 객체를 app 변수에 할당
• nest_asyncio.apply() : 이벤트 루프
• @app.get("/") : GET 메소드로 root url을 접속하는 경우
• async def root() : root() 비동기 함수 (native coroutine)을 만듬
• return {"message": "Hello World"} : message 변수로 해당 값을 JSON 형태로 반환
• uvicorn.run(app, host="127.0.0.1", port=8000) : app 객체에 IP와 Port 설정

#실습
from fastapi import FastAPI
import nest_asyncio
import uvicorn

# 웹서버 초기화
app = FastAPI()
nest_asyncio.apply()

# 라우터 설정(get 방식, root)
@app.get("/")
async def roof() :
    return {"message" : "Hello world !!"}

# 메인함수라면(파이썬을 단독으로 실행했다면)
if __name__ == "__main__" :
    # 서버시작
    uvicorn.run(app, host="172.23.32.1", port = 9000)

• FastAPI, nest_asyncio, uvicorn 모듈을 설치해야 함
• FastAPI 모듈을 app 변수로 초기화
• nest_asyncio 모듈을 이용해 asyncio를 fastAPI와 함께 사용 가능하게 함
• 라우터를 설정해주는 @app.get 데코레이터를 이용해서 라우팅함
• if __name__ == "__main__" 구문을 이용해 해당 파일을 직접 실행한 경우 uvicorn.run() 함수를 이용해 서버를 시작함
• host와 port를 설정해주면 됨

1.4.1  URL를 함수로 연결하기

• @app.get("/hello")
  • URL에 함수명을 설정
  • 브라우저에서 IP와 포트로 서버 접속하고 URL로 함수명을 넘김

from fastapi import FastAPI
import nest_asyncio
import uvicorn

#FastAPI 앱 생성
app = FastAPI()

#nest_asyncio 적용
nest_asyncio.apply()

#라우터 설정 (GET 방식, /hello 경로)
@app.get("/hello")
async def roof() :
return {"message" : "윤효창"}

#메인 함수 (서버 시작)
if name == "main" :
uvicorn.run(app, host="172.23.32.1", port=9000)

• FastAPI 라이브러리를 이용하여, 간단한 웹 어플리케이션을 만드는 코드입니다.
• FastAPI 객체를 생성하고, nest_asyncio 라이브러리를 이용하여 이벤트 루프를 초기화합니다.
• /hello 경로로 GET 요청이 오면, "윤효창" 문자열을 반환합니다.
• if __name__ == "__main__": 을 이용하여, 현재 스크립트가 직접 실행될 때 서버를 구동합니다.
• uvicorn 라이브러리를 사용하여 호스트 주소와 포트를 설정합니다.

1.4.2  URL을 변수로 사용하기

• @app.get("/hello/{value}")
  • URL에 value 값을 설정
  • 브라우저에서 IP와 포트로 서버 접속하고 URL로 100을 넘김
  •  

from fastapi import FastAPI
import nest_asyncio
import uvicorn

app = FastAPI()
nest_asyncio.apply()

# value : 클라이언트에서 넘긴 값을 받는 변수
# {value} : 경로 매개변수로 동적 URL을 생성
@app.get("/hello/{value}")
async def roof(value: str):  # 경로 매개변수에는 데이터 타입 명시 필요
    return {"message": value}

if __name__ == "__main__":
    uvicorn.run(app, host="172.23.32.1", port=9000)

1. FastAPI와 nest_asyncio, uvicorn 라이브러리를 임포트합니다.
2. FastAPI 클래스의 인스턴스를 만듭니다.
3. nest_asyncio.apply()를 호출하여 asyncio 이벤트 루프와 동기 코드의 호환성을 유지합니다.
4. app.get() 데코레이터를 사용하여 라우터를 정의합니다. 이 경로는 클라이언트가 GET 방식으로 호출할 수 있도록 제공됩니다.
5. value는 클라이언트가 전달하는 값으로, 경로 매개 변수에 의해 처리됩니다.
6. 클라이언트가 요청한 값을 반환합니다.
7. uvicorn.run() 함수를 호출하여 서버를 시작합니다. 서버의 호스트 주소와 포트 번호를 지정할 수 있습니다.

위 코드의 주석과 피드백을 참고하여 코드가 어떤 역할을 하는지 이해하시길 바랍니다.

 

1.4.3  html 문서를 반환하기

1.4.3.1  코드 형태로 반환하기

• @app.get("/", response_class=HTMLResponse)
  • HTML 문서를 반환
• 브라우저에서 IP와 포트로 서버 접속

from fastapi import FastAPI
import nest_asyncio
import uvicorn
from fastapi.responses import HTMLResponse

app = FastAPI()
nest_asyncio.apply()

#HTMLResponse를 통해 HTML 파일을 응답하도록 설정
@app.get("/",response_class= HTMLResponse)
async def roof() :
return """
<html><body>
<h1>안녕하세요 !!</h1>
</body></html>
"""

if name == "main" :
# FastAPI를 통해 웹 서버 실행
uvicorn.run(app, host="172.23.32.1", port = 9000)

• 코드 블럭에 대한 주석과 리뷰가 요청되어 코드 블럭에 대한 설명을 작성하였습니다.
• FastAPI에서 HTMLResponse를 사용하여 클라이언트에게 HTML 형식의 응답을 보내도록 구현하였습니다.
• 서버를 시작할 때, uvicorn.run 메소드를 사용하여 주소와 포트번호를 지정하여 서버를 시작합니다.

1.4.3.2  파일 형태로 반환하기

• @app.get("/", response_class=FileResponse)
  • 파일 문서를 반환
• templates 폴더에 저장
• %%writefile 명령 위쪽에는 아무 내용도 있으면 안됨(무조건 첫 줄에 사용) -해당 명령 다음에 있는 내용을 파일에 저장하는 기능

%%writefile ./templates/index.html
<html><body>
    <h2>안녕하세요 !! <br> Fast API입니다 </h2>
</body></html>

실행후

#실습
from fastapi import FastAPI
import nest_asyncio
import uvicorn
from fastapi.responses import FileResponse

app = FastAPI()
nest_asyncio.apply()

@app.get("/",response_class= FileResponse)
async def roof() :
    return "./templates/index.html"

if __name__ == "__main__" :
    
    uvicorn.run(app, host="172.23.32.1", port = 9000)

코드 리뷰와 피드백을 달아드리겠습니다.

• 파일 응답을 보낼 때는 FileResponse를 사용합니다.
• response_class 매개변수를 이용해 응답을 지정할 수 있습니다.
• 파일 경로는 str로 지정합니다.
• if __name__ == "__main__" 구문은 파이썬 파일이 메인으로 실행되었을 때만 실행됩니다.
• host는 로컬 IP 주소이며, 원하는 IP 주소로 수정해서 사용할 수 있습니다.
• port는 서버를 열 포트 번호이며, 1024 이상의 번호를 사용해야 합니다.

따라서 다음과 같이 코드를 수정해 주세요

#실습
from fastapi import FastAPI
import nest_asyncio
import uvicorn
from fastapi.responses import FileResponse

app = FastAPI()
nest_asyncio.apply()

@app.get("/", response_class=FileResponse)
async def roof() :
    return "./templates/index.html"

if __name__ == "__main__" :
    uvicorn.run(app, host="0.0.0.0", port=9000)

또한 host와 port를 각각 0.0.0.0과 9000으로 수정했습니다.

이렇게 하면 외부에서도 접속할 수 있습니다.

 

1.4.4  이미지가 포함된 HTML 문서 반환하기

• static 폴더 : 자원을 담아 놓는 폴더
• static 폴더에 이미지 폴더를 만들고 출력할 이미지를 저장
• jinja2 : HTML 문서에서 파이썬 코드를 사용하도록 지원하는 라이브러리(JSP와 유사)
  
  

!pip install jinja2

해당 코드는 jinja2 라이브러리를 설치하는 코드입니다. jinja2는 파이썬에서

HTML, XML, CSV 등의 템플릿을 작성할 수 있는 라이브러리로,

웹 개발에서 유용하게 쓰이는 라이브러리 중 하나입니다.

%%writefile ./templates/index2.html
<html><body>
    <h2>안녕하세요 !! <br> Fast API입니다 </h2>
    <img src="{{url_for('static',path ='dip.jfif') }}"
</body></html>

해당 코드는 HTML 코드로 구성된 Jinja2 템플릿 파일을 생성하는 코드입니다.

%%writefile ./templates/index2.html은 셀 매직 명령어로,

현재 셀에서 작성한 코드를 ./templates/index2.html 파일로 저장합니다.

이 파일은 HTML 코드를 작성하며, img 태그를 이용해 이미지를 추가했습니다.

src 속성의 값을 {{url_for('static', path='dip.jfif')}}로 지정하여,

Flask의 url_for() 함수를 이용해 이미지 파일의 경로를 지정할 수 있습니다.

해당 경로는 static 폴더에 저장되어 있으며, path 인자를 통해 파일명을 지정합니다.

 

• Jinja2Templates(directory="templates/")
  • Jinja2 템플릿 사용

• app.mount("/static", StaticFiles(directory="static", html=True), name="static")
  • "/static" : 마운트할 경로
  • directory="static" : 정적파일이 들어 있는 폴더명
  • html=True : HTML 문서 여부
  • name="static" : FastAPI에서 내부적으로 사용할 이름

• templates.TemplateResponse("imgdisp2.html", {"request": request})
  • Jinja 템플릿으로 구성된 HTML 문서를 request한 위치로 반환

#실습
from fastapi import FastAPI, Request
import nest_asyncio
import uvicorn
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles

app = FastAPI()
nest_asyncio.apply()

# Jinja 템플릿이 포함된 HTML 문서의 위치
templates = Jinja2Templates(directory ="templates/")

# 이미지의 위치
app.mount("/static",StaticFiles(directory= "static",html=True), name="static")

@app.get("/",response_class= HTMLResponse)
async def root(request : Request) :
    #클라이언트(request)를 설정
    return templates.TemplateResponse("index2.html",{"request":request})

if __name__ == "__main__" :
    
    uvicorn.run(app, host="172.23.32.1", port = 9000)

• FastAPI를 이용하여 웹어플리케이션을 만드는 코드입니다.
• FastAPI는 빠른 처리속도와 API문서 자동화, 타이핑 기능 등의 장점을 가지고 있는 파이썬 프레임워크입니다.
• uvicorn과 같은 높은 생산성과 동시성성 덕분에 FastAPI는 점점 더 인기를 얻고 있습니다.

코드 리뷰

• from fastapi import FastAPI, Request와 같이 import하는 것은 좋습니다. 코드 가독성에 좋습니다.
• 이미지 파일 위치를 추가하기 위해, fastapi.staticfiles.StaticFiles를 이용하여, app.mount 함수를 사용하여 정적파일을 처리해줍니다.
• templates 변수에 Jinja2 템플릿이 포함된 HTML 문서의 위치를 할당합니다.
• root 함수에서 templates.TemplateResponse를 이용하여, 템플릿이 적용된 HTML 문서를 반환합니다.

고칠점

• 코드의 가독성을 높이기 위해 import문은 처음에 한번에 모아서 작성하는 것이 좋습니다.
• 이미지 파일이 있는 폴더 경로를 지정할 때, 상대경로 대신에 절대경로를 사용하면 다른 컴퓨터에서 프로젝트를 실행하는 경우, 경로가 유효하지 않아 오류가 발생할 수 있습니다.
• 템플릿이 포함된 HTML 문서에서 이미지 파일을 표시하려면, img태그에 닫는 괄호 >가 빠져있어서 이미지가 표시되지 않을 수 있습니다.

2  데이터 스트리밍 구현

2.1  구현 방법

• StreamingResponse(test())
  • 스트리밍 함수로 test()를 실행
    
    

#실습

• yield() 함수 배우기

#실습

#실습

2.2  웹 브라우저에 카메라 영상 출력하기

• index.html를 작성하고 templates 폴더에 저장
  • video_feed : 실행할 함수명

• Jinja 템플릿
  • 간단한 문법으로 html에서 {}, {{}} 등의 규칙을 이용하여 python 프로그래밍이 가능
  • 데이터를 담은 변수를 return값으로 html과 함께 넘겨준 뒤, html에서 jinja 템플릿 규칙에 따라 화면에 표현할 수 있음
  • Jinja 템플릿 html에서 주석은 {# 주석으로 처리할 내용 #}으로 표시
  • Jinja2 Tempaltes 변수 넘기기 : {{ 변수명 }}
  • 참고
    • https://velog.io/@jewon119/01.Flask-기초-Jinja-template
    • https://jinja.palletsprojects.com/en/2.11.x/templates/

• url_for() : URL 주소값을 가져온다

#실습

• ret, buffer = cv2.imencode('.jpg', frame) : frame 이미지를 jpg로 인코딩
• frame = buffer.tobytes() : 전송을 위해 인코딩된 이미지를 byte 형식으로 변환

• yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
  • b : byte 형식임을 의미
  • --frame : 프레임을 표시
  • Content-Type: image/jpeg : 문서가 jpg 이미지임을 표시

• StreamingResponse(get_frames(), media_type="multipart/x-mixed-replace; boundary=frame")
  • get_frames() : 호출할 함수명
  • mimetype : 클라이언트에게 전송된 문서의 타입을 알려주기 위한 파라미터 (type/subtype)
  • multipart : 복합문서 타입 (파일, 영상 등)을 의미
  • x-mixed-replace : x (추가적인 확장 형식), mixed (복합문서), repalce (subtype을 다음 메시지로 대체)
  • boundary : 복합문서 내의 각 문서들을 구분하는 분리자 (동영상이므로 frame으로 구분)
• 이미지 전송 부분을 동기방식으로 구현하는 경우

#실습

#실습
import cv2
import nest_asyncio
import asyncio
import uvicorn
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import StreamingResponse

app = FastAPI()
nest_asyncio.apply()
templates = Jinja2Templates(directory="templates/")

cap = cv2.VideoCapture("./images/yadong.mp4")

async def get_frames() :
    while cap.isOpened() :
        ret, frame = cap.read()
        cv2.waitKey(33)
        
        if not ret :
            continue
        else :
            _, 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("/")        
async def root(request : Request) :
    return templates.TemplateResponse("index3.html", {"request" : request})

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

if __name__ == "__main__" :
    uvicorn.run(app, host="172.23.32.1", port=9000)

코드는 비디오 파일을 스트리밍하는 간단한 FastAPI 애플리케이션을 구현합니다.

cv2.VideoCapture()을 사용하여 비디오 파일을 열고,

async def get_frames() 함수를 사용하여 각 프레임을 읽고,

무한 루프에서 각 프레임을 인코딩합니다.

각 프레임은 Multipart로 된 HTTP 응답의 일부로 서비스합니다.

/video_feed 엔드포인트는 / 엔드포인트로 전환되며,

비디오 스트림을 렌더링하려면 html 문서가 필요합니다.

templates 폴더에서 index3.html을 가져오고 @app.get("/")에서 이 문서를 렌더링합니다. response_class=StreamingResponse를 사용하여 video_feed를 스트리밍할 수 있습니다.

2.3  웹 출력 영상을 Binary이미지로 출력하기

%%writefile ./templates/index4.html
<html><body>
    <h3>웹카메라 구현 (원영상, 이진영상)</h3>
    <img src="{{ url_for('video_feed1') }}">   
</body></html>

• <h3>태그 : 웹페이지에 '웹카메라 구현 (원영상, 이진영상)'이라는 제목을 삽입합니다.
• <img>태그 : video_feed1 이라는 이름의 함수가 반환하는 이미지 데이터를 렌더링합니다.
• url_for 함수를 이용하여 해당 함수의 URL 경로를 생성합니다.

#실습
import cv2

#"./images/yadong.mp4" 파일을 비디오 캡쳐 객체로 생성합니다.
cap1 = cv2.VideoCapture("./images/yadong.mp4")

#클라이언트로부터 요청이 올 때마다 비디오의 각 프레임을 처리한 후 전송합니다.
async def get_frames1() :
    while cap1.isOpened() :
        ret, frame = cap1.read()
        cv2.waitKey(33)
         # 프레임을 회색조로 변환합니다.
        frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
		  # 프레임을 이진화 처리합니다.
        frame = cv2.adaptiveThreshold(frame_gray, 255, cv2.ADAPTIVE_THRESH_MEAN_C,
                                     cv2.THRESH_BINARY, 9, 5)
        if not ret :
            continue
        else :
			# JPEG 형식으로 프레임을 인코딩합니다.
            _, 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')        
# 비디오 캡쳐 객체를 해제합니다.
    cap1.release()

웹캠을 연결하는 코드이나,여기서는 비디오 파일을 읽어서 각 프레임을 처리한 후 클라이언트에 전송하는 비디오 스트리밍 서버로 사용됩니다.

import nest_asyncio
import asyncio
import uvicorn
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import StreamingResponse

app = FastAPI()
nest_asyncio.apply()
templates = Jinja2Templates(directory="templates/")

@app.get("/")        
async def root(request : Request) :
    return templates.TemplateResponse("index4.html", {"request" : request})

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

if __name__ == "__main__" :
    uvicorn.run(app, host="218.149.140.29", port=9007)

uvicorn.run 함수를 호출할 때 host 인자로 사용한 IP 주소가 유효한지 확인해야 합니다.

현재 코드에서는 218.149.140.29를 사용하고 있는데,

이는 외부에서 접근할 수 있는 IP 주소인지,

혹은 로컬 네트워크 내에서만 사용 가능한 IP 주소인지에 따라 동작 여부가 결정됩니다.

 

또한, get_frames1() 함수에서 yield 키워드를 사용하고 있으므로,

이 함수를 호출하는 쪽에서는 코루틴을 사용해야 합니다.

현재 코드에서는 StreamingResponse 클래스로 래핑하여 스트리밍 형태로 반환하고 있습니다.

이 경우, yield 키워드를 사용하는 코루틴이 필요합니다.

만약 코루틴이 아닌 일반적인 함수로 프레임을 가져온다면,

StreamingResponse 대신 Response 클래스를 사용해야 합니다.

실습문제
○ 이진 이미지를 출력하는 웹 카메라 만들기
- threshold() 함수 적용
- OTSU 이진화 알고리즘 적용
- 적응형 thresholding 적용

[4차시] 학습요약

○ ret, buffer = cv2.imencode() : 영상 인코딩 함수 ○ yield() : 반복 실행 중에 중간 과정을 반환할 때 사용 ○ FastAPI : 파이썬으로 제작된 웹 프레임워크 (DJango의 축소 버전) ○ app = FastAPI() : FastAPI 객체를 app 변수에 할당 ○ nest_asyncio.apply() : FastAPI와 uvicorn 연결 ○ @app.get("/") : FastAPI에게 어떤 URL이 get방식으로 해당 함수를 실행하는지 알려줌 ○ @app.get("/hello") : 서버 주소에 추가적인 URL을 설정 ○ @app.get("/hello/{value}") : URL에 value 값을 설정 ○ uvicorn.run(app, host="127.0.0.1", port=9600) : 서버 IP, 포트 등을 설정하고 서버를 실행 ○ HTMLResponse() : html 문서를 반환하기 ○ FileResponse() : 파일로 된 문서를 반환하기 ○ TemplateResponse() : 템플릿 형태로 된 문서를 반환하기 ○ StreamingResponse(stream_with_context()) : 데이터 스트리밍을 반환하기

파일다운로드 링크 : https://github.com/yoonhyochang/Deep_learning