# 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()
# 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)
4条答案
按热度按时间eyh26e7m1#
我相信您现在已经知道答案了,但是我将为搜索相同主题的其他人提供答案。
(可能与youtube_dl一起使用)。
应该就这样了。
8i9zcol22#
我在我的VidGear Python库中添加了Youtube URL源代码支持,它可以通过只提供YouTube视频的URL,自动将YouTube视频导入OpenCV。
w1jd8yoj3#
在100-120帧之后,@lee hannigan的回答突然出现在我的youtube上。
我用Pafy设计了一个方法,只抓取x个帧并将它们拼接在一起。但这最终导致拼接效果不佳,并给出了不连贯的结果。Pafy可能不是为实时流设计的,我找不到一种方法来无缝地将帧拼接在一起。
下面是最后起作用的,对guttentag_liu在this post上的回答做了一些修改。它需要更多的包,而且很长,但可以工作。因为文件是活的,它需要分块,因此保存到一个临时文件中。你可能可以在每个块上做openCV工作,然后最后保存到一个文件中,而不是重新打开。
n7taea2i4#
可能是因为Youtube不再提供喜欢/不喜欢的数量,所以第一个解决方案会出错。作为一个解决方案,你应该注解pafy包文件backend_youtube_dl.py中的第53和54行,如下图所示,之后第一个解决方案中的代码就可以工作了:
其次,你可以用OpenCV得到不能得到音频,它是一个计算机视觉库,不是多媒体。你应该尝试其他选择。