186 lines
11 KiB
Python
186 lines
11 KiB
Python
# -*- 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)
|