I have an IP camera that streams video using the RTSP protocol. My system needs to connect to this IP camera and stream the video to a web server. Since most browsers don’t natively support RTSP, I need an intermediary server to handle the RTSP stream. I implemented a FastAPI web server that uses OpenCV to connect to the IP camera and stream the video, and it works; I can see the video in the browser.
Currently, my implementation streams the video as an image element, but I want to stream it as a video element. How can I achieve this with my server?
Additionally, are there any ready-to-use servers that can stream video from an IP camera to an HTML video element that you would recommend?
Reading frame from IP camera implementation:
<code>class VideoStreamService:
def __init__(self):
pass
def get_frame(self, connection_string: str):
cap = cv2.VideoCapture(connection_string)
while True:
# Capture frame-by-frame
success, frame = cap.read() # read the camera frame
if not success:
break
else:
ret, encodded_frame = cv2.imencode('.jpg', frame)
# frame = buffer.tobytes()
yield (b'--framern' b'Content-Type: image/jpegrnrn' + bytearray(encodded_frame) + b'rn')
</code>
<code>class VideoStreamService:
def __init__(self):
pass
def get_frame(self, connection_string: str):
cap = cv2.VideoCapture(connection_string)
while True:
# Capture frame-by-frame
success, frame = cap.read() # read the camera frame
if not success:
break
else:
ret, encodded_frame = cv2.imencode('.jpg', frame)
# frame = buffer.tobytes()
yield (b'--framern' b'Content-Type: image/jpegrnrn' + bytearray(encodded_frame) + b'rn')
</code>
class VideoStreamService:
def __init__(self):
pass
def get_frame(self, connection_string: str):
cap = cv2.VideoCapture(connection_string)
while True:
# Capture frame-by-frame
success, frame = cap.read() # read the camera frame
if not success:
break
else:
ret, encodded_frame = cv2.imencode('.jpg', frame)
# frame = buffer.tobytes()
yield (b'--framern' b'Content-Type: image/jpegrnrn' + bytearray(encodded_frame) + b'rn')
Stream response from the server:
<code>def get_video_stream(
connection_string: str = Query(..., alias="connection_string"),
video_stream_service: VideoStreamService = Depends(Provide[Container.video_stream_service]),) -> StreamingResponse:
try:
encodded_frame = video_stream_service.get_frame(unquote(connection_string))
return StreamingResponse(encodded_frame, media_type="multipart/x-mixed-replace;boundary=frame")
except ValueError as error:
...
</code>
<code>def get_video_stream(
connection_string: str = Query(..., alias="connection_string"),
video_stream_service: VideoStreamService = Depends(Provide[Container.video_stream_service]),) -> StreamingResponse:
try:
encodded_frame = video_stream_service.get_frame(unquote(connection_string))
return StreamingResponse(encodded_frame, media_type="multipart/x-mixed-replace;boundary=frame")
except ValueError as error:
...
</code>
def get_video_stream(
connection_string: str = Query(..., alias="connection_string"),
video_stream_service: VideoStreamService = Depends(Provide[Container.video_stream_service]),) -> StreamingResponse:
try:
encodded_frame = video_stream_service.get_frame(unquote(connection_string))
return StreamingResponse(encodded_frame, media_type="multipart/x-mixed-replace;boundary=frame")
except ValueError as error:
...
I used this SO question as a reference to my implementation