如何使用 openCV python 阅读 Youtube 直播？

Question

我想从 youtube 读取实时流来执行一些基本的 CV 操作，可能我们必须以某种方式剥离 youtube URL 以将其转换为 openCV 可以读取的格式，例如？：

cap = cv2.VideoCapture('https://www.youtube.com/watch?v=_9OBhtLA9Ig')

有人做过吗？

Answer 1

我相信你现在已经知道答案了，但我会为其他搜索相同主题的人回答。您可以使用Pafy（可能与youtube_dl 一起使用）来做到这一点。

import pafy
import cv2

url = "https://www.youtube.com/watch?v=_9OBhtLA9Ig"
video = pafy.new(url)
best = video.getbest(preftype="mp4")

capture = cv2.VideoCapture(best.url)
while True:
    grabbed, frame = capture.read()
    # ...

应该就是这样。

Answer 2

在 100-120 帧之后，@lee hannigan 的回答让我在 youtube 上进行了直播。

我与 Pafy 一起制定了一种方法，只需抓取 x 个帧并将它们拼接在一起。然而，这最终将这些块拼接在一起很糟糕，并给出了不稳定的结果。 Pafy 可能不是为直播而设计的，我找不到将帧无缝拼接在一起的方法。

最终起作用的方法如下，根据guttentag_liu 在this post 上的回答稍作修改。它需要更多的包，而且很长，但是可以。因为文件是活动的，所以它需要分块，因此保存到临时文件中。您可能可以对每个块进行 openCV 工作，然后最后保存到文件中，而不是重新打开。

# pip install urllib
# pip install m3u8
# pip install streamlink
from datetime import datetime, timedelta, timezone
import urllib
import m3u8
import streamlink
import cv2 #openCV

def get_stream(url):

    """
    Get upload chunk url
    input: youtube URL
    output: m3u8 object segment
    """
    #Try this line tries number of times, if it doesn't work, 
    # then show the exception on the last attempt
    # Credit, theherk, https://stackoverflow.com/questions/2083987/how-to-retry-after-exception
    tries = 10
    for i in range(tries):
    try:
        streams = streamlink.streams(url)
    except:
        if i < tries - 1: # i is zero indexed
            print(f"Attempt {i+1} of {tries}")
            time.sleep(0.1) #Wait half a second, avoid overload
            continue
        else:
            raise
    break

    stream_url = streams["best"] #Alternate, use '360p'

    m3u8_obj = m3u8.load(stream_url.args['url'])
    return m3u8_obj.segments[0] #Parsed stream


def dl_stream(url, filename, chunks):
    """
    Download each chunk to file
    input: url, filename, and number of chunks (int)
    output: saves file at filename location
    returns none.
    """
    pre_time_stamp = datetime(1, 1, 1, 0, 0, tzinfo=timezone.utc)

    
    #Repeat for each chunk
    #Needs to be in chunks because 
    #  1) it's live
    #  2) it won't let you leave the stream open forever
    i=1
    while i <= chunks:
       
        #Open stream
        stream_segment = get_stream(url)
    
        #Get current time on video
        cur_time_stamp = stream_segment.program_date_time
        #Only get next time step, wait if it's not new yet
        if cur_time_stamp <= pre_time_stamp:
            #Don't increment counter until we have a new chunk
            print("NO   pre: ",pre_time_stamp, "curr:",cur_time_stamp)
            time.sleep(0.5) #Wait half a sec
            pass
        else:
            print("YES: pre: ",pre_time_stamp, "curr:",cur_time_stamp)
            print(f'#{i} at time {cur_time_stamp}')
            #Open file for writing stream
            file = open(filename, 'ab+') #ab+ means keep adding to file
            #Write stream to file
            with urllib.request.urlopen(stream_segment.uri) as response:
                html = response.read()
                file.write(html)
            
            #Update time stamp
            pre_time_stamp = cur_time_stamp
            time.sleep(stream_segment.duration) #Wait duration time - 1

            i += 1 #only increment if we got a new chunk

    return None

def openCVProcessing(saved_video_file):
    '''View saved video with openCV
    Add your other steps here'''
    capture = cv2.VideoCapture(saved_video_file)

    while capture.isOpened():
        grabbed, frame = capture.read()  #read in single frame
        if grabbed == False:
            break

        #openCV processing goes here
        #
        
        cv2.imshow('frame',frame)  #Show the frame
        
        #Shown in a new window, To exit, push q on the keyboard
        if cv2.waitKey(20) & 0xFF == ord('q'):
            break

    capture.release()
    cv2.destroyAllWindows()  #close the windows automatically
                  
tempFile = "temp.ts"  #files are format ts, open cv can view them
videoURL = "https://www.youtube.com/watch?v=_9OBhtLA9Ig"

dl_stream(videoURL, tempFile, 3)
openCVProcessing(tempFile)

Answer 3

我在 VidGear Python 库中添加了 Youtube URL 源支持，该库仅通过提供其 URL 来自动将 YouTube 视频通过管道传输到 OpenCV。这是一个完整的python示例：

# import libraries
from vidgear.gears import CamGear
import cv2

stream = CamGear(source='https://youtu.be/dQw4w9WgXcQ', stream_mode = True, logging=True).start() # YouTube Video URL as input

# infinite loop
while True:
    
    frame = stream.read()
    # read frames

    # check if frame is None
    if frame is None:
        #if True break the infinite loop
        break
    
    # do something with frame here
    
    cv2.imshow("Output Frame", frame)
    # Show output window

    key = cv2.waitKey(1) & 0xFF
    # check for 'q' key-press
    if key == ord("q"):
        #if 'q' key-pressed break out
        break

cv2.destroyAllWindows()
# close output window

# safely close video stream.
stream.stop()

Code Source