|
- # -*- coding: utf-8 -*-
- from queue import Queue
- from threading import Thread
- from time import time, sleep
- from traceback import format_exc
-
- from loguru import logger
-
- from enums.ExceptionEnum import ExceptionType
- from enums.RecordingStatusEnum import RecordingStatus
- from exception.CustomerException import ServiceException
- from util.Cv2Utils import check_video_stream, clear_pull_p, build_video_info2, pull_read_video_stream2
- from util.QueUtil import put_queue, get_no_block_queue, clear_queue, put_queue_result
-
-
- class PullStreamThread(Thread):
- __slots__ = ('_command', '_pull_queue', '_hb_queue', '_fb_queue', '_msg', '_context')
-
- def __init__(self, *args):
- super().__init__()
- self._msg, self._context, self._pull_queue, self._hb_queue, self._fb_queue, self._frame_num = args
- self._command = Queue()
-
- def sendEvent(self, result):
- put_queue(self._command, result, timeout=10, is_ex=False)
-
-
- class RecordingPullStreamThread(PullStreamThread):
-
- def run(self):
- msg, context, frame_num = self._msg, self._context, self._frame_num
- request_id, pull_url = msg["request_id"], msg['pull_url']
- service = context["service"]
- pull_stream_timeout = int(service["recording_pull_stream_timeout"])
- read_stream_timeout = int(service["cv2_read_stream_timeout"])
- service_timeout = int(service["timeout"])
- command_queue, pull_queue, fb_queue, hb_queue = self._command, self._pull_queue, self._fb_queue, self._hb_queue
- width, height, width_height_3, all_frames, w, h = None, None, None, 0, None, None
- read_start_time, pull_p, ex = None, None, None
- frame_list, frame_index_list = [], []
- stop_ex = True
- pull_stream_start_time = time()
- try:
- logger.info("录屏拉流线程开始启动, requestId: {}", request_id)
- cv2_init_num, init_pull_num, concurrent_frame = 0, 1, 1
- start_time = time()
- while True:
- # 检查任务是否超时
- if time() - start_time > service_timeout:
- logger.error("录屏拉流超时, requestId: {}", request_id)
- raise ServiceException(ExceptionType.TASK_EXCUTE_TIMEOUT.value[0],
- ExceptionType.TASK_EXCUTE_TIMEOUT.value[1])
- # 最终停止拉流
- event = get_no_block_queue(command_queue)
- if event is not None:
- # 当接收到停止指令，说明不会再处理视频帧了, 直接退出
- if 'stop' == event.get("command"):
- if len(frame_list) > 0:
- put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
- logger.info("录屏拉流线程开始停止, requestId: {}", request_id)
- break
- # 主进程异常，停止子线程
- if 'stop_ex' == event.get("command"):
- logger.info("录屏异常拉开始停止拉流线程, requestId: {}", request_id)
- stop_ex = False
- break
- # 如果是离线拉流
- if pull_url.startswith('http'):
- if check_video_stream(width, height):
- logger.info("开始重新获取视频信息: {}次, requestId: {}", cv2_init_num, request_id)
- # 当是离线地址重试3次还是拉取不到视频流，关闭拉流管道，返回失败信息
- # 目前改为等待5分钟
- # if cv2_init_num > 3:
- if time() - start_time > 300:
- logger.info("离线拉流重试失败, 重试次数: {}, requestId: {}", cv2_init_num, request_id)
- raise ServiceException(ExceptionType.OR_VIDEO_ADDRESS_EXCEPTION.value[0],
- ExceptionType.OR_VIDEO_ADDRESS_EXCEPTION.value[1])
- cv2_init_num += 1
- width, height, width_height_3, all_frames, w, h = build_video_info2(pull_url, request_id)
- if width is not None:
- put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RUNNING.value[0]}, timeout=2)
- else:
- # if cv2_init_num < 2:
- if time() - start_time < 300:
- put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RETRYING.value[0]}, timeout=2)
- continue
- # 当离线视频时, 队列满了, 等待1秒后再试
- if pull_queue.full():
- logger.info("pull拉流队列满了: {}, requestId: {}", pull_queue.qsize(), request_id)
- sleep(1)
- continue
- # 如果是实时拉流
- else:
- if check_video_stream(width, height):
- logger.info("开始重新获取视频信息: {}次, requestId: {}", cv2_init_num, request_id)
- pull_stream_init_timeout = time() - pull_stream_start_time
- if len(frame_list) > 0:
- put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
- frame_list, frame_index_list = [], []
- if pull_stream_init_timeout > pull_stream_timeout:
- logger.error("开始拉流超时, 超时时间：{}, requestId:{}", pull_stream_init_timeout, request_id)
- raise ServiceException(ExceptionType.PULLSTREAM_TIMEOUT_EXCEPTION.value[0],
- ExceptionType.PULLSTREAM_TIMEOUT_EXCEPTION.value[1])
- cv2_init_num += 1
- width, height, width_height_3, all_frames, w, h = build_video_info2(pull_url, request_id)
- if width is not None:
- put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RUNNING.value[0]}, timeout=1)
- else:
- if cv2_init_num < 3:
- put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RETRYING.value[0]}, timeout=1)
- sleep(1)
- continue
- pull_stream_start_time = time()
- cv2_init_num = 1
- frame, pull_p, width, height = pull_read_video_stream2(pull_p, pull_url, width, height,
- width_height_3, w, h, request_id)
- if frame is None:
- if pull_url.startswith('http'):
- clear_pull_p(pull_p, request_id)
- logger.info("总帧数: {}, 当前帧数: {}, requestId: {}", all_frames, concurrent_frame, request_id)
- if len(frame_list) > 0:
- put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
- if concurrent_frame < all_frames - 100:
- logger.info("离线拉流异常结束：requestId: {}", request_id)
- raise ServiceException(ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[0],
- ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[1])
- logger.info("离线拉流线程结束, requestId: {}", request_id)
- break
- else:
- logger.info("获取帧为空, 开始重试: {}次, requestId: {}", init_pull_num, request_id)
- if len(frame_list) > 0:
- put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
- frame_list, frame_index_list = [], []
- if read_start_time is None:
- read_start_time = time()
- pull_stream_read_timeout = time() - read_start_time
- if pull_stream_read_timeout > read_stream_timeout:
- logger.info("拉流过程中断了重试超时, 超时时间: {}, requestId: {}", pull_stream_read_timeout,
- request_id)
- raise ServiceException(ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[0],
- ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[1])
- init_pull_num += 1
- continue
- init_pull_num = 1
- read_start_time = None
- if pull_queue.full():
- sleep(1)
- logger.info("pull拉流队列满了:{}, requestId: {}", pull_queue.qsize(), request_id)
- continue
- frame_list.append(frame)
- frame_index_list.append(concurrent_frame)
- if len(frame_list) >= frame_num:
- put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
- frame_list, frame_index_list = [], []
- concurrent_frame += 1
- del frame
- except ServiceException as s:
- ex = s.code, s.msg
- except Exception:
- logger.exception("实时拉流异常: {}, requestId:{}", format_exc(), request_id)
- ex = ExceptionType.SERVICE_INNER_EXCEPTION.value[0], ExceptionType.SERVICE_INNER_EXCEPTION.value[1]
- finally:
- clear_pull_p(pull_p, request_id)
- if stop_ex:
- if ex:
- error_code, error_msg = ex
- result = put_queue_result(pull_queue, (1, error_code, error_msg), timeout=3)
- else:
- result = put_queue_result(pull_queue, (2,), timeout=3)
- if result:
- # 3分钟超时时间
- cr_time = time()
- while time() - cr_time < 180:
- event = get_no_block_queue(command_queue)
- if event is not None:
- # 当接收到停止指令，说明不会再处理视频帧了, 直接退出
- if 'stop' == event.get("command"):
- logger.info("录屏拉流线程开始停止, requestId: {}", request_id)
- break
- sleep(1)
- clear_queue(command_queue)
- clear_queue(pull_queue)
- clear_queue(hb_queue)
- del frame_list, frame_index_list
- logger.info("录屏拉流线程结束, requestId: {}", request_id)
|
