# -*- coding:utf-8 -*- import cv2 import time import redis import queue import sys import threading from logger import cusLogger r_db = None log = cusLogger("rtsp_process").getLogger() class jpgSaveThread(threading.Thread): def __init__(self,invoker,prefix, filePath, fps, frameSize, dQueue): threading.Thread.__init__(self) self.invoker = invoker # 存储文件前缀 self.prefix = prefix # 存储路径 self.filePath = filePath # 帧率 self.fps = fps # 帧大小（width,height) self.frameSize = frameSize self.dQueue = dQueue log.debug(self.invoker + '——视频片段存储') def run(self): log.debug(self.invoker + "-BEGIN" + str(time.time())) # 这里采用h264编码 fourcc = cv2.VideoWriter_fourcc(*"X264") # 命名里不要使用不允许的字符， fname = self.filePath + self.prefix + time.strftime("%Y%m%d%H%M%S", time.localtime()) + '.mp4' out = cv2.VideoWriter(fname, fourcc, self.fps, self.frameSize) while not self.dQueue.empty(): try: frame = self.dQueue.get() #print(frame.size) #out.write(frame) cv2.imwrite('./img/' + time.strftime("%Y%m%d%H%M%S", time.localtime()) + '.jpg', frame) except Exception as e: log.info(e) # out.release() # out = None log.debug(self.invoker + "-END" + str(time.time())) class videoSaveThread(threading.Thread): def __init__(self, invoker, prefix, filePath, fps, video, frameSize, dQueue, queueSize): threading.Thread.__init__(self) self.invoker = invoker # 存储文件前缀 self.prefix = prefix # 存储路径 self.filePath = filePath # 帧率 self.fps = fps # 传入缓存队列 self.dQueue = dQueue self.queueSize = queueSize # 缓存队列大小 self.frameSize = frameSize # frame帧尺寸 # 传入cv2videowriter对象 self.out = video log.debug(self.invoker + '——视频片段存储') def run(self): framecount = 0 log.debug(self.invoker + "-BEGIN" + str(time.time())) # # 这里采用h264编码 # fourcc = cv2.VideoWriter_fourcc(*"MP4V") # # 命名里不要使用不允许的字符， # fname = self.filePath + self.prefix + time.strftime("%Y%m%d%H%M%S", time.localtime()) + '.mp4' # out = cv2.VideoWriter(fname, fourcc, self.fps, self.frameSize) # while not self.dQueue.empty(): # try: # frame = self.dQueue.get() # #print(frame.size) # out.write(frame) # except Exception as e: # log.info(e) # out.release() # out = None # log.debug(self.invoker + "-END" + str(time.time())) while framecount < self.queueSize: try: if(self.dQueue.empty()): print("kong") else: frame = self.dQueue.get() #print(frame.size) frame = cv2.resize(frame, self.frameSize) self.out.write(frame) framecount += 1 #print(count) except Exception as e: log.info(e) # out.release() # out = None # log.debug(self.invoker + "-END" + str(time.time())) class rtspThread(threading.Thread): def __init__(self, thName, rtspUrl, prefix, filePath, fps, frameSize, queueSize, timer, eventKey, eventValue): threading.Thread.__init__(self) # name self.thName = thName # rtsp流地址 self.rtspUrl = rtspUrl # 帧率 self.fps = fps # 帧大小（width,height) self.frameSize = frameSize # 存储队列大小，fps * 10秒 self.queueSize = queueSize # 总视频帧数 = fps * 存储时间 self.sumframesize = fps * timer # redis的key self.eventKey = eventKey # key对应的value self.eventValue = eventValue # 存储文件前缀 self.prefix = prefix # 存储路径 self.filePath = filePath # 存储视频帧大小 self.videoframesize = (1920, 1080) # cv2writer对象 self.video = None def run(self): # 视频缓存队列 videoSaveQueue = [queue.Queue(self.queueSize), queue.Queue(self.queueSize)] # 图片缓存队列 jpgSaveQueue = [queue.Queue(self.queueSize), queue.Queue(self.queueSize)] # 队列索引编号 curQueueIdx = 0 # 错误标志位 closed = False # 错误计数 failCnt = 0 #存视频标志 saveFlag = False # 视频段落计数 sectionCnt = 0 # 获取cap对象 cap = cv2.VideoCapture(self.rtspUrl) winName = self.rtspUrl + "_" + str(time.time()) while True: if cap.isOpened(): res, reframe = cap.read() if res: closed = False failCnt = 0 frame = reframe cv2.namedWindow(winName, flags=cv2.WINDOW_FREERATIO) cv2.imshow(winName, frame) cv2.waitKey(10) else: failCnt += 1 if(failCnt >= 15): closed = True # try: # if not videoSaveQueue[curQueueIdx].full(): # videoSaveQueue[curQueueIdx].put(frame) # except: # pass try: if jpgSaveQueue[curQueueIdx].full(): jpgSaveQueue[curQueueIdx].get_nowait() jpgSaveQueue[curQueueIdx].put_nowait(frame) except: pass if (r_db.exists(self.eventKey) and int(r_db.get(self.eventKey)) == self.eventValue): # 删除键值 r_db.delete(self.eventKey) # 存储标志位 saveFlag = True # 这里采用h264编码 fourcc = cv2.VideoWriter_fourcc(*"MP4V") # 命名里不要使用不允许的字符， fname = self.filePath + self.prefix + time.strftime("%Y%m%d%H%M%S", time.localtime()) + '.mp4' # 创建cv2video对象 out = cv2.VideoWriter(fname, fourcc, self.fps, self.videoframesize) self.video = out if saveFlag: try: if not videoSaveQueue[curQueueIdx].full(): videoSaveQueue[curQueueIdx].put(frame) else: print(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) print(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) # 存视频 videoSaveProc = videoSaveThread(self.thName, self.prefix, self.filePath, self.fps, self.video, self.videoframesize, videoSaveQueue[curQueueIdx], self.queueSize) videoSaveProc.start() sectionCnt += 1 # 存图 # jpgSaveSaveProc = jpgSaveThread(self.thName, self.prefix, self.filePath, self.fps, # self.frameSize, jpgSaveQueue[curQueueIdx]) # jpgSaveSaveProc.start() # 切换缓存队列 curQueueIdx = (curQueueIdx + 1) % 2 except: pass if sectionCnt <= int(self.sumframesize/self.queueSize): pass else: # 释放资源 saveFlag = False sectionCnt = 0 self.video.release() self.video = None log.debug(self.prefix + "-END" + str(time.time())) else: closed = True if closed: # 休眠10秒钟 重连 log.info("reconnect: " + self.rtspUrl) time.sleep(5) cap.open(self.rtspUrl) if __name__ == '__main__': r_db = redis.StrictRedis(host='localhost', port=6379, db=0) #r_db.set('key', 1) try: # 如果有多个流要同时取存，可以创建多个线程 thread1 = rtspThread("thread_vs", "rtsp://admin:glodon123@192.168.116.13:554/Streaming/Channels/101", # "e://2021-01-03-08-43-43.mp4", "ch1_", "",# 为空串时，默认保存在当前目录 25, (3840, 2160), 50, # 25*6s=150,需要存30s,分5段存 30, "key", 1) thread1.start() # thread2 = rtspThread("thread_ir", # "rtsp://admin:12345@x.x.x./stream/main/ch2", # "ch2_", # "",# 为空串时，默认保存在当前目录 # 16, # (1920, 1080), # 150, # "key", # 2) # thread2.start() except Exception as e: log.info(e) cv2.destroyAllWindows()