# -*- coding: utf-8 -*- import time import cv2 import subprocess as sp import ffmpeg import numpy as np from loguru import logger ''' 获取视频信息 ''' def get_video_info(pullUrl): try: probe = ffmpeg.probe(pullUrl) if probe is None or probe.get("streams") is None: return # 视频大小 # format = probe['format'] # size = int(format['size'])/1024/1024 video_stream = next((stream for stream in probe['streams'] if stream.get('codec_type') == 'video'), None) if video_stream is None: logger.error("根据拉流地址未获取到视频流") return width = video_stream.get('width') height = video_stream.get('height') nb_frames = video_stream.get('nb_frames') fps = video_stream.get('r_frame_rate') # duration = video_stream.get('duration') bit_rate = video_stream.get('bit_rate') if width: width = int(width) if height: height = int(height) if nb_frames: all_frames = int(nb_frames) if fps: up, down = str(fps).split('/') fps = int(eval(up) / eval(down)) # if duration: # self.duration = float(video_stream['duration']) if bit_rate: bit_rate = int(bit_rate) / 1000 logger.info("视频信息, width:{}|height:{}|fps:{}|all_frames:{}|bit_rate:{}", width, height, fps, all_frames, bit_rate) except ffmpeg._run.Error as er: logger.error("获取视频信息异常: {}", er.stderr.decode(encoding='utf-8')) except Exception as e: logger.exception("获取视频信息异常：{}", e) ''' 拉取视频 ''' def build_pull_p(wh, pullUrl): try: command = [r'E:\liumeiti\ffmpeg\ffmpeg-master-latest-win64-gpl\bin\ffmpeg', '-re', '-y', '-c:v', 'h264_cuvid', '-resize', wh, '-i', pullUrl, '-f', 'rawvideo', '-an', '-'] return sp.Popen(command, stdout=sp.PIPE) except Exception as e: logger.exception("构建拉流管道异常：{}", e) return None def read(whr, whrz, pullUrl, h, w): result = None try: in_bytes = build_pull_p(whrz, pullUrl).stdout.read(whr) if in_bytes is not None and len(in_bytes) > 0: # result = (np.frombuffer(in_bytes, np.uint8).reshape([int(self.height), int(self.width), 3])) try: img = (np.frombuffer(in_bytes, np.uint8)).reshape((h, w)) except Exception as ei: logger.exception("视频格式异常：{}", ei) result = cv2.cvtColor(img, cv2.COLOR_YUV2BGR_NV12) except Exception as e: logger.exception("读流异常：{}", e) return result def close(self): if self.pull_p: if self.pull_p.stdout: self.pull_p.stdout.close() self.pull_p.terminate() self.pull_p.wait() logger.info("关闭拉流管道完成, requestId:{}", self.requestId) if self.p: if self.p.stdin: self.p.stdin.close() self.p.terminate() self.p.wait() # self.p.communicate() # self.p.kill() logger.info("关闭管道完成, requestId:{}", self.requestId) if self.or_video_file: self.or_video_file.release() logger.info("关闭原视频写入流完成, requestId:{}", self.requestId) if self.ai_video_file: self.ai_video_file.release() logger.info("关闭AI视频写入流完成, requestId:{}", self.requestId) # async def push_stream_write(self, frame): # self.p.stdin.write(frame.tostring()) # # async def push_stream(self, frame): # if self.p is None: # self.build_p() # try: # await self.push_stream_write(frame) # return True # except Exception as ex: # logger.exception("推流进管道异常：{}, requestId: {}", ex, self.requestId) # current_retry_num = 0 # while True: # try: # time.sleep(1) # self.p_push_retry_num += 1 # current_retry_num += 1 # if current_retry_num > 3 or self.p_push_retry_num > 600: # return False # self.build_p() # await self.push_stream_write(frame) # logger.info("构建p管道重试成功, 当前重试次数: {}, requestId: {}", current_retry_num, # self.requestId) # return True # except Exception as e: # logger.exception("构建p管道异常：{}, 开始重试, 当前重试次数：{}, requestId: {}", e, # current_retry_num, self.requestId) # return False # async def video_frame_write(self, or_frame, ai_frame): # if or_frame is not None: # self.or_video_file.write(or_frame) # if ai_frame is not None: # self.ai_video_file.write(ai_frame) # async def video_write(self, or_frame, ai_frame): # try: # self.build_write() # if or_frame is not None and len(or_frame) > 0: # await self.video_frame_write(or_frame, None) # if ai_frame is not None and len(ai_frame) > 0: # await self.video_frame_write(None, ai_frame) # return True # except Exception as ex: # ai_retry_num = 0 # while True: # try: # ai_retry_num += 1 # if ai_retry_num > 3: # logger.exception("重新写入离线分析后视频到本地,重试失败：{}, requestId: {}", e, self.requestId) # return False # if or_frame is not None and len(or_frame) > 0: # await self.or_video_file.write(or_frame) # if ai_frame is not None and len(ai_frame) > 0: # await self.ai_video_file.write(ai_frame) # logger.info("重新写入离线分析后视频到本地, 当前重试次数: {}, requestId: {}", ai_retry_num, # self.requestId) # return True # except Exception as e: # logger.exception("重新写入离线分析后视频到本地：{}, 开始重试, 当前重试次数：{}, requestId: {}", e, # ai_retry_num, self.requestId) # def build_write(self): # try: # if self.fps is None or self.width is None or self.height is None: # raise ServiceException(ExceptionType.VIDEO_CONFIG_EXCEPTION.value[0], # ExceptionType.VIDEO_CONFIG_EXCEPTION.value[1]) # if self.orFilePath is not None and self.or_video_file is None: # self.or_video_file = cv2.VideoWriter(self.orFilePath, cv2.VideoWriter_fourcc(*'mp4v'), self.fps, # (int(self.wn), int(self.hn))) # if self.or_video_file is None: # raise ServiceException(ExceptionType.OR_WRITE_OBJECT_EXCEPTION.value[0], # ExceptionType.OR_WRITE_OBJECT_EXCEPTION.value[1]) # if self.aiFilePath is not None and self.ai_video_file is None: # self.ai_video_file = cv2.VideoWriter(self.aiFilePath, cv2.VideoWriter_fourcc(*'mp4v'), self.fps, # (int(self.width), int(self.hn))) # if self.ai_video_file is None: # raise ServiceException(ExceptionType.AI_WRITE_OBJECT_EXCEPTION.value[0], # ExceptionType.AI_WRITE_OBJECT_EXCEPTION.value[1]) # except ServiceException as s: # logger.exception("构建文件写对象异常: {}, requestId:{}", s, self.requestId) # raise s # except Exception as e: # logger.exception("构建文件写对象异常: {}, requestId:{}", e, self.requestId) # raise e # def video_merge(self, frame1, frame2): # # frameLeft = cv2.resize(frame1, (int(self.width / 2), int(self.height / 2)), interpolation=cv2.INTER_LINEAR) # # frameRight = cv2.resize(frame2, (int(self.width / 2), int(self.height / 2)), interpolation=cv2.INTER_LINEAR) # # frame_merge = np.hstack((frameLeft, frameRight)) # frame_merge = np.hstack((frame1, frame2)) # return frame_merge # # def getP(self): # if self.p is None: # logger.error("获取管道为空, requestId:{}", self.requestId) # raise ServiceException(ExceptionType.PULL_PIPELINE_INIT_EXCEPTION.value[0], # ExceptionType.PULL_PIPELINE_INIT_EXCEPTION.value[1]) # return self.p # # def getCap(self): # if self.cap is None: # logger.error("获取cv2为空, requestId:{}", self.requestId) # raise ServiceException(ExceptionType.CV2_IS_NULL_EXCEPTION.value[0], # ExceptionType.CV2_IS_NULL_EXCEPTION.value[1]) # return self.cap # # def getOrVideoFile(self): # if self.or_video_file is None: # logger.error("获取原视频写入对象为空, requestId:{}", self.requestId) # raise ServiceException(ExceptionType.OR_WRITE_OBJECT_EXCEPTION.value[0], # ExceptionType.OR_WRITE_OBJECT_EXCEPTION.value[1]) # return self.or_video_file # # def getAiVideoFile(self): # if self.ai_video_file is None: # logger.error("获取AI视频写入对象为空, requestId:{}", self.requestId) # raise ServiceException(ExceptionType.AI_WRITE_OBJECT_EXCEPTION.value[0], # ExceptionType.AI_WRITE_OBJECT_EXCEPTION.value[1]) # return self.ai_video_file if __name__== "__main__": command = ['ffmpeg', '-rtsp_transport', 'tcp', '-i', 'rtsp://127.0.0.1:8554/video', # 指定输入文件 '-c', 'copy', '-f', 'flv', "rtmp://192.168.10.101:19350/rlive/stream_9?sign=f8a15b6n"] p = sp.Popen(command, shell=False) while True: time.sleep(2) print("pid", p.pid) print("poll", p.poll()) print("returncode", p.returncode) # p.terminate() # p.wait()