Python cv2.VideoWriter 问题

The fist part uses GStreamer syntax, and OpenCV for Python is most likely not built with GStreamer.
The answer is going to be focused on the second part (also because I don't know GStreamer so well).

有几个问题:

• cap = cv2.VideoCapture(stream)应该在while True循环之前.
• out = cv2.VideoWriter(stream, cv2.VideoWriter_fourcc(*'mp4v'), 30, (1920, 1080))应该在while True循环之前.
• cv2.VideoWriter的第一个参数应该是MP4文件名，而不是stream.
• 为了获得有效的输出文件，我们必须在循环之后执行out.release()，但循环可能永远不会结束.

• 建议获取输入视频的帧大小和码率，并相应设置VideoWriter:

   width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
   height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
   fps = int(cap.get(cv2.CAP_PROP_FPS))
  
   video_file_name = 'output.mp4'
  
   out = cv2.VideoWriter(video_file_name, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))  # Open video file for writing
  

• 如果ret是False，建议断开循环:

   ret, frame = cap.read()
  
   if not ret:
       break
  

• One option to end the recording is when user press Esc key.
  Break the loop if cv2.waitKey(1) == 27.
  cv2.waitKey(1) is going to work only after executing cv2.imshow.
  A simple solution is executing cv2.imshow every 30 frames (for example).

   if (frame_counter % 30 == 0):
       cv2.imshow('frame', frame)  # Show frame every 30 frames (for testing)
  
   if cv2.waitKey(1) == 27:  # Press Esc for stop recording (cv2.waitKey is going to work only when cv2.imshow is used).
       break
  

完整的代码示例:

from streamlink import Streamlink
import cv2

def stream_to_url(url, quality='best'):
    session = Streamlink()
    streams = session.streams(url)

    if streams:
        return streams[quality].to_url()
    else:
        raise ValueError('Could not locate your stream.')


url = 'https://www.twitch.tv/noraexplorer'  # Need to login to twitch.tv first (using the browser)...
quality='best'

stream_url = stream_to_url(url, quality)  # Get the video URL
cap = cv2.VideoCapture(stream_url, cv2.CAP_FFMPEG)  # Open video stream for capturing

# Get frame size and rate of the input video
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = int(cap.get(cv2.CAP_PROP_FPS))


video_file_name = 'output.mp4'

out = cv2.VideoWriter(video_file_name, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))  # Open video file for writing


frame_counter = 0
while True:
    ret, frame = cap.read()
    
    if not ret:
        break

    if (frame_counter % 30 == 0):
        cv2.imshow('frame', frame)  # Show frame every 30 frames (for testing)

    out.write(frame)  # Write frame to output.mp4

    if cv2.waitKey(1) == 27:  # Press Esc for stop recording (cv2.waitKey is going to work only when cv2.imshow is used).
        break

    frame_counter += 1

cap.release()
out.release()
cv2.destroyAllWindows()

使用FFplay和subprocess模块测试设置:

from streamlink import Streamlink
import subprocess

def stream_to_url(url, quality='best'):
    session = Streamlink()
    streams = session.streams(url)

    if streams:
        return streams[quality].to_url()
    else:
        raise ValueError('Could not locate your stream.')


#url = 'https://www.twitch.tv/noraexplorer'  # Need to login to twitch.tv first (using the browser)...
url = 'https://www.twitch.tv/valorant'
quality='best'

stream_url = stream_to_url(url, quality)  # Get the video URL

subprocess.run(['ffplay', stream_url])

最新情况:

读取视频使用ffmpeg-python，录制视频使用OpenCV:

In cases where cv2.VideoCapture is not working, we may use FFmpeg CLI as sub-process.
ffmpeg-python module is Python binding for FFmpeg CLI.
Using ffmpeg-python is almost like using subprocess module, it used here mainly for simplifying the usage of FFprobe.

使用FFbe获取视频帧分辨率和帧速率(不使用OpenCV):

p = ffmpeg.probe(stream_url, select_streams='v');
width = p['streams'][0]['width']
height = p['streams'][0]['height']
r_frame_rate = p['streams'][0]['r_frame_rate']  # May return 60000/1001

if '/' in r_frame_rate:
    fps = float(r_frame_rate.split("/")[0]) / float(r_frame_rate.split("/")[1])  # Convert from 60000/1001 to 59.94
elif r_frame_rate != '0':
    fps = float(r_frame_rate)
else:
    fps = 30  # Used as default

获得帧速率可能有点困难……

注:ffprobe CLI应位于执行路径中.

以stdout作为管道启动FFmpeg子进程:

ffmpeg_process = (
    ffmpeg
    .input(stream_url)
    .video
    .output('pipe:', format='rawvideo', pix_fmt='bgr24')
    .run_async(pipe_stdout=True)
)

注:ffmpeg CLI应位于执行路径中.

从管道中读取帧，并将其从字节转换为NumPy数组:

in_bytes = ffmpeg_process.stdout.read(width*height*3)
frame = np.frombuffer(in_bytes, np.uint8).reshape([height, width, 3])

Closing FFmpeg sub-process:
Closing stdout pipe ends FFmpeg (with "broken pipe" error).

ffmpeg_process.stdout.close()
ffmpeg_process.wait()  # Wait for the sub-process to finish

完整的代码示例:

from streamlink import Streamlink
import cv2
import numpy as np
import ffmpeg

def stream_to_url(url, quality='best'):
    session = Streamlink()
    streams = session.streams(url)

    if streams:
        return streams[quality].to_url()
    else:
        raise ValueError('Could not locate your stream.')


#url = 'https://www.twitch.tv/noraexplorer'  # Need to login to twitch.tv first (using the browser)...
url = 'https://www.twitch.tv/valorant'
quality='best'

stream_url = stream_to_url(url, quality)  # Get the video URL

#subprocess.run(['ffplay', stream_url])  # Use FFplay for testing

# Use FFprobe to get video frames resolution and framerate.
################################################################################
p = ffmpeg.probe(stream_url, select_streams='v');
width = p['streams'][0]['width']
height = p['streams'][0]['height']
r_frame_rate = p['streams'][0]['r_frame_rate']  # May return 60000/1001

if '/' in r_frame_rate:
    fps = float(r_frame_rate.split("/")[0]) / float(r_frame_rate.split("/")[1])  # Convert from 60000/1001 to 59.94
elif r_frame_rate != '0':
    fps = float(r_frame_rate)
else:
    fps = 30  # Used as default

#cap = cv2.VideoCapture(stream_url, cv2.CAP_FFMPEG)  # Open video stream for capturing
#width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
#height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
#fps = int(cap.get(cv2.CAP_PROP_FPS))
################################################################################


# Use FFmpeg sub-process instead of using cv2.VideoCapture
################################################################################
ffmpeg_process = (
    ffmpeg
    .input(stream_url, an=None)  # an=None applies -an argument (used for ignoring the input audio - it is not required, just more elegant).
    .video
    .output('pipe:', format='rawvideo', pix_fmt='bgr24')
    .run_async(pipe_stdout=True)
)
################################################################################


video_file_name = 'output.mp4'

out = cv2.VideoWriter(video_file_name, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))  # Open video file for writing


frame_counter = 0
while True:
    #ret, frame = cap.read()    
    in_bytes = ffmpeg_process.stdout.read(width*height*3)  # Read raw video frame from stdout as bytes array.
    
    if len(in_bytes) < width*height*3:  #if not ret:
        break

    frame = np.frombuffer(in_bytes, np.uint8).reshape([height, width, 3])  # Convert bytes array to NumPy array.

    if (frame_counter % 30 == 0):
        cv2.imshow('frame', frame)  # Show frame every 30 frames (for testing)

    out.write(frame)  # Write frame to output.mp4

    if cv2.waitKey(1) == 27:  # Press Esc for stop recording (cv2.waitKey is going to work only when cv2.imshow is used).
        break

    frame_counter += 1

#cap.release()
ffmpeg_process.stdout.close()  # Close stdout pipe (it also closes FFmpeg).
out.release()
cv2.destroyAllWindows()
ffmpeg_process.wait()  # Wait for the sub-process to finish

Note:
In case you care about the quality of the recorded video, using cv2.VideoWriter is not the best choice...