tuoheng_alg_move_20240715/concurrency/PullVideoStreamProcess2.py

# -*- coding: utf-8 -*-
import os
from multiprocessing import Process, Queue
from os import getpid
from time import time, sleep
from traceback import format_exc

import psutil
from loguru import logger

from util.LogUtils import init_log
from concurrency.FileUploadThread import ImageFileUpload
from entity.FeedBack import message_feedback
from enums.AnalysisStatusEnum import AnalysisStatus
from enums.ExceptionEnum import ExceptionType
from exception.CustomerException import ServiceException
from util.Cv2Utils import check_video_stream, build_video_info, pull_read_video_stream, clear_pull_p
from util.QueUtil import get_no_block_queue, put_queue, clear_queue, put_queue_result


class PullVideoStreamProcess2(Process):
    __slots__ = ("_command_queue", "_msg", "_context", "_fb_queue", "_pull_queue", "_image_queue", "_analyse_type",
                 "_frame_num")

    def __init__(self, *args):
        super().__init__()
        # 自带参数
        self._command_queue = Queue()

        # 传参
        self._msg, self._context, self._fb_queue, self._pull_queue, self._image_queue, self._analyse_type, \
            self._frame_num = args

    def sendCommand(self, result):
        try:
            self._command_queue.put(result, timeout=10)
        except Exception:
            logger.error("添加队列超时异常：{}, requestId:{}", format_exc(), self._msg.get("request_id"))
            raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                   ExceptionType.SERVICE_INNER_EXCEPTION.value[1])

    @staticmethod
    def start_File_upload(*args):
        fb_queue, context, msg, image_queue, analyse_type = args
        image_thread = ImageFileUpload(fb_queue, context, msg, image_queue, analyse_type)
        image_thread.setDaemon(True)
        image_thread.start()
        return image_thread

    @staticmethod
    def check(start_time, service_timeout, request_id, image_thread):
        if time() - start_time > service_timeout:
            logger.error("分析超时, requestId: {}", request_id)
            raise ServiceException(ExceptionType.ANALYSE_TIMEOUT_EXCEPTION.value[0],
                                   ExceptionType.ANALYSE_TIMEOUT_EXCEPTION.value[1])
        # 检测图片上传线程是否正常运行
        if image_thread is not None and not image_thread.is_alive():
            logger.error("未检测到图片上传线程活动，图片上传线程可能出现异常, requestId:{}", request_id)
            raise Exception("未检测到图片上传线程活动，图片上传线程可能出现异常!")


class OnlinePullVideoStreamProcess2(PullVideoStreamProcess2):
    __slots__ = ()

    def run(self):
        # 避免循环调用性能影响, 优先赋值
        context, msg, analyse_type = self._context, self._msg, self._analyse_type
        request_id, base_dir, env = msg["request_id"], context['base_dir'], context['env']
        pull_url, frame_num = msg["pull_url"], self._frame_num
        pull_stream_timeout = int(context["service"]["cv2_pull_stream_timeout"])
        read_stream_timeout = int(context["service"]["cv2_read_stream_timeout"])
        service_timeout = int(context["service"]["timeout"])
        command_queue, pull_queue, image_queue = self._command_queue, self._pull_queue, self._image_queue
        fb_queue = self._fb_queue
        image_thread, pull_p = None, None
        width, height, width_height_3, all_frames, w_2, h_2 = None, None, None, 0, None, None
        frame_list, frame_index_list = [], []
        ex = None
        ex_status = True
        full_timeout = None
        try:
            # 初始化日志
            init_log(base_dir, env)
            logger.info("开启实时视频拉流进程, requestId:{}", request_id)

            # 开启图片上传线程
            image_thread = self.start_File_upload(fb_queue, context, msg, image_queue, analyse_type)

            # 初始化拉流工具类
            cv2_init_num, init_pull_num, concurrent_frame = 0, 1, 1
            start_time, pull_start_time, read_start_time = time(), None, None
            while True:
                # 检测任务执行是否超时、图片上传线程是否正常
                self.check(start_time, service_timeout, request_id, image_thread)
                command_msg = get_no_block_queue(command_queue)
                if command_msg is not None:
                    if 'stop' == command_msg.get("command"):
                        logger.info("开始停止实时拉流进程, requestId:{}", request_id)
                        break
                    if 'stop_ex' == command_msg.get("command"):
                        logger.info("开始停止实时拉流进程, requestId:{}", request_id)
                        ex_status = False
                        break
                # 检测视频信息是否存在或拉流对象是否存在
                if check_video_stream(width, height):
                    if len(frame_list) > 0:
                        put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
                        frame_list, frame_index_list = [], []
                    logger.info("开始重新获取视频信息: {}次, requestId: {}", cv2_init_num, request_id)
                    if pull_start_time is None:
                        pull_start_time = time()
                    pull_stream_init_timeout = time() - pull_start_time
                    if pull_stream_init_timeout > pull_stream_timeout:
                        logger.info("开始拉流超时, 超时时间：{}, 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_2, h_2 = build_video_info(pull_url, request_id)
                    if width is None:
                        sleep(1)
                    continue
                pull_start_time, cv2_init_num = None, 1
                frame, pull_p, width, height = pull_read_video_stream(pull_p, pull_url, width, height, width_height_3,
                                                                      w_2, h_2, request_id)
                if frame is None:
                    if len(frame_list) > 0:
                        put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
                        frame_list, frame_index_list = [], []
                    logger.info("获取帧为空, 开始重试: {}次, requestId: {}", init_pull_num, request_id)
                    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, read_start_time = 1, None
                if pull_queue.full():
                    logger.info("pull拉流队列满了:{}, requestId: {}", os.getppid(), request_id)
                    if full_timeout is None:
                        full_timeout = time()
                    if time() - full_timeout > 180:
                        logger.error("拉流队列阻塞异常, requestId: {}", request_id)
                        raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                               ExceptionType.SERVICE_INNER_EXCEPTION.value[1])
                    if psutil.Process(getpid()).ppid() == 1:
                        clear_pull_p(pull_p, request_id)
                        ex_status = False
                        for q in [command_queue, pull_queue, image_queue]:
                            clear_queue(q)
                        if image_thread and image_thread.is_alive():
                            put_queue(image_queue, (2, "stop"), timeout=1)
                            image_thread.join(120)
                        logger.info("检测到父进程异常停止, 请检测服务器资源是否负载过高, requestId: {}", request_id)
                        put_queue(self._fb_queue, message_feedback(request_id,
                                                                   AnalysisStatus.FAILED.value,
                                                                   self._analyse_type,
                                                                   ExceptionType.NO_RESOURCES.value[0],
                                                                   ExceptionType.NO_RESOURCES.value[1]))
                        break
                    del frame
                    continue
                full_timeout = None
                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, is_ex=True)
                    frame_list, frame_index_list = [], []
                concurrent_frame += 1
                del frame
        except ServiceException as s:
            logger.error("实时拉流异常: {}, 队列大小:{}, requestId:{}", s.msg, pull_queue.qsize(), request_id)
            ex = s.code, s.msg
        except Exception:
            logger.error("实时拉流异常: {}, 队列大小:{}, requestId:{}", format_exc(), pull_queue.qsize(), request_id)
            ex = ExceptionType.SERVICE_INNER_EXCEPTION.value[0], ExceptionType.SERVICE_INNER_EXCEPTION.value[1]
        finally:
            clear_pull_p(pull_p, request_id)
            del frame_list, frame_index_list
            if ex_status:
                if ex:
                    code, msg = ex
                    r = put_queue_result(pull_queue, (1, code, msg), timeout=10)
                else:
                    r = put_queue_result(pull_queue, (2,), timeout=10)
                if r:
                    c_time = time()
                    while time() - c_time < 180:
                        command_msg = get_no_block_queue(command_queue)
                        if command_msg is not None:
                            if 'stop' == command_msg.get("command"):
                                logger.info("开始停止实时拉流进程, requestId:{}", request_id)
                                if image_thread and image_thread.is_alive():
                                    put_queue(image_queue, (2, "stop"), timeout=1)
                                    logger.info("停止图片上传线程, requestId:{}", request_id)
                                    image_thread.join(120)
                                    logger.info("停止图片上传线程结束, requestId:{}", request_id)
                                break
            for q in [command_queue, pull_queue, image_queue]:
                clear_queue(q)
            if image_thread and image_thread.is_alive():
                put_queue(image_queue, (2, "stop"), timeout=1)
                logger.info("停止图片上传线程, requestId:{}", request_id)
                image_thread.join(120)
                logger.info("停止图片上传线程结束, requestId:{}", request_id)
            logger.info("实时拉流线程结束, 图片队列: {}, 拉流队列: {}, 图片进程的状态: {} requestId: {}",
                        image_queue.qsize(), pull_queue.qsize(), image_thread.is_alive(), request_id)


class OfflinePullVideoStreamProcess2(PullVideoStreamProcess2):
    __slots__ = ()

    def run(self):
        msg, context, frame_num, analyse_type = self._msg, self._context, self._frame_num, self._analyse_type
        request_id, base_dir, env, pull_url = msg["request_id"], context['base_dir'], context['env'], msg["pull_url"]
        ex, service_timeout = None, int(context["service"]["timeout"])
        command_queue, pull_queue, image_queue, fb_queue = self._command_queue, self._pull_queue, self._image_queue, \
            self._fb_queue
        image_thread, pull_p = None, None
        width, height, width_height_3, all_frames, w_2, h_2 = None, None, None, 0, None, None
        frame_list, frame_index_list = [], []
        ex_status = True
        full_timeout = None
        try:
            # 初始化日志
            init_log(base_dir, env)
            logger.info("开启离线视频拉流进程, requestId:{}", request_id)

            # 开启图片上传线程
            image_thread = self.start_File_upload(fb_queue, context, msg, image_queue, analyse_type)

            # 初始化拉流工具类
            cv2_init_num = 0
            concurrent_frame = 1
            start_time = time()
            while True:
                # 检测任务执行是否超时、图片上传线程是否正常
                self.check(start_time, service_timeout, request_id, image_thread)
                command_msg = get_no_block_queue(command_queue)
                if command_msg is not None:
                    if 'stop' == command_msg.get("command"):
                        logger.info("开始停止离线拉流进程, requestId:{}", request_id)
                        break
                    if 'stop_ex' == command_msg.get("command"):
                        logger.info("开始停止离线拉流进程, requestId:{}", request_id)
                        ex_status = False
                        break
                # 检测视频信息是否存在或拉流对象是否存在
                if check_video_stream(width, height):
                    if len(frame_list) > 0:
                        put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1)
                        frame_list, frame_index_list = [], []
                    logger.info("开始重新获取视频信息: {}次, requestId: {}", cv2_init_num, request_id)
                    if cv2_init_num > 3:
                        clear_pull_p(pull_p, request_id)
                        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
                    sleep(1)
                    width, height, width_height_3, all_frames, w_2, h_2 = build_video_info(pull_url, request_id)
                    continue
                if pull_queue.full():
                    logger.info("pull拉流队列满了:{}, requestId: {}", os.getppid(), request_id)
                    if full_timeout is None:
                        full_timeout = time()
                    if time() - full_timeout > 300:
                        logger.error("拉流队列阻塞超时, 请检查父进程是否正常！requestId: {}", request_id)
                        raise ServiceException(ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
                                               ExceptionType.SERVICE_INNER_EXCEPTION.value[1])
                    if psutil.Process(getpid()).ppid() == 1:
                        clear_pull_p(pull_p, request_id)
                        ex_status = False
                        for q in [command_queue, pull_queue, image_queue]:
                            clear_queue(q)
                        if image_thread and image_thread.is_alive():
                            put_queue(image_queue, (2, "stop"), timeout=1)
                            image_thread.join(120)
                        logger.info("检测到父进程异常停止, 请检测服务器资源是否负载过高, requestId: {}", request_id)
                        put_queue(self._fb_queue, message_feedback(request_id,
                                                                   AnalysisStatus.FAILED.value,
                                                                   self._analyse_type,
                                                                   ExceptionType.NO_RESOURCES.value[0],
                                                                   ExceptionType.NO_RESOURCES.value[1]))
                        break
                    continue
                full_timeout = None
                frame, pull_p, width, height = pull_read_video_stream(pull_p, pull_url, width, height, width_height_3,
                                                                      w_2, h_2, request_id)
                if frame is None:
                    logger.info("总帧数: {}, 当前帧数: {}, requestId: {}", all_frames, concurrent_frame, request_id)
                    clear_pull_p(pull_p, request_id)
                    if len(frame_list) > 0:
                        put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=2, is_ex=False)
                        frame_list, frame_index_list = [], []
                    # 允许100帧的误差
                    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
                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, is_ex=True)
                    frame_list, frame_index_list = [], []
                concurrent_frame += 1
                del frame
        except ServiceException as s:
            logger.error("实时拉流异常: {}, 队列大小:{}, requestId:{}", s.msg, pull_queue.qsize(), request_id)
            ex = s.code, s.msg
        except Exception:
            logger.error("实时拉流异常: {}, 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)
            del frame_list, frame_index_list
            if ex_status:
                if ex:
                    code, msg = ex
                    r = put_queue_result(pull_queue, (1, code, msg), timeout=10)
                else:
                    r = put_queue_result(pull_queue, (2,), timeout=10)
                if r:
                    c_time = time()
                    while time() - c_time < 180:
                        command_msg = get_no_block_queue(command_queue)
                        if command_msg is not None:
                            if 'stop' == command_msg.get("command"):
                                logger.info("开始停止离线拉流进程, requestId:{}", request_id)
                                if image_thread and image_thread.is_alive():
                                    put_queue(image_queue, (2, "stop"), timeout=1)
                                    logger.info("停止图片上传线程, requestId:{}", request_id)
                                    image_thread.join(120)
                                    logger.info("停止图片上传线程结束, requestId:{}", request_id)
                                break
            for q in [command_queue, pull_queue, image_queue]:
                clear_queue(q)
            if image_thread and image_thread.is_alive():
                put_queue(image_queue, (2, "stop"), timeout=1)
                logger.info("停止图片上传线程, requestId:{}", request_id)
                image_thread.join(120)
                logger.info("停止图片上传线程结束, requestId:{}", request_id)
            logger.info("离线拉流线程结束, 图片队列: {}, 拉流队列: {}, 图片进程的状态: {} requestId: {}",
                        image_queue.qsize(), pull_queue.qsize(), image_thread.is_alive(), request_id)