You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

73 line
3.4KB

  1. # -*- coding: utf-8 -*-
  2. import time
  3. from traceback import format_exc
  4. import subprocess as sp
  5. from loguru import logger
  6. from exception.CustomerException import ServiceException
  7. from enums.ExceptionEnum import ExceptionType
  8. class PushStreamUtil:
  9. __slots__ = ('pullUrl', 'pushUrl', 'requestId', "status", "push_stream_sp", "start_time")
  10. def __init__(self, pullUrl=None, pushUrl=None, requestId=None):
  11. self.pullUrl = pullUrl
  12. self.pushUrl = pushUrl
  13. self.requestId = requestId
  14. self.status = True
  15. self.push_stream_sp = None
  16. self.start_time = time.time()
  17. def close_push_stream_sp(self):
  18. logger.info("开始关闭推流管道, requestId:{}", self.requestId)
  19. if self.push_stream_sp:
  20. logger.info("尝试关闭推流管道, requestId:{}", self.requestId)
  21. self.push_stream_sp.terminate()
  22. self.push_stream_sp.wait()
  23. if self.push_stream_sp and self.push_stream_sp.poll() is None:
  24. logger.error("尝试关闭推流管道失败, requestId:{}", self.requestId)
  25. # self.push_stream_sp.communicate(input=b"q\n", timeout=30)
  26. if self.push_stream_sp and self.push_stream_sp.poll() is not None:
  27. logger.info("尝试关闭推流管道成功, requestId:{}", self.requestId)
  28. self.push_stream_sp = None
  29. else:
  30. logger.info("推流管道已关闭, requestId:{}", self.requestId)
  31. # 构建 cv2
  32. def start_push_stream(self):
  33. try:
  34. if self.push_stream_sp:
  35. return
  36. if self.pullUrl is None or len(self.pullUrl) == 0:
  37. logger.error("推流地址不能为空, requestId:{}", self.requestId)
  38. raise ServiceException(ExceptionType.PULL_STREAM_URL_EXCEPTION.value[0],
  39. ExceptionType.PULL_STREAM_URL_EXCEPTION.value[1])
  40. if self.pushUrl is None or len(self.pushUrl) == 0:
  41. logger.error("推流地址不能为空, requestId:{}", self.requestId)
  42. raise ServiceException(ExceptionType.PUSH_STREAM_URL_EXCEPTION.value[0],
  43. ExceptionType.PUSH_STREAM_URL_EXCEPTION.value[1])
  44. command = ['ffmpeg', '-re', '-y', "-an", "-hide_banner"]
  45. if self.pullUrl.startswith("rtsp://"):
  46. command.extend(['-timeout', '20000000', '-rtsp_transport', 'tcp'])
  47. if self.pullUrl.startswith("rtmp://") or self.pullUrl.startswith("http"):
  48. command.extend(['-rw_timeout', '20000000'])
  49. command.extend(['-i', self.pullUrl,
  50. '-c:v', 'copy',
  51. '-b:v', '4000k',
  52. '-bufsize', '4000k',
  53. '-f', 'flv',
  54. self.pushUrl])
  55. logger.info("推流指令:{}, requestId:{}", command, self.requestId)
  56. self.push_stream_sp = sp.Popen(command, stdin=sp.PIPE, stdout=sp.PIPE, stderr=sp.PIPE, shell=False)
  57. except ServiceException as s:
  58. self.close_push_stream_sp()
  59. logger.error("构建p管道异常: {}, requestId:{}", s.msg, self.requestId)
  60. raise s
  61. except Exception as e:
  62. self.close_push_stream_sp()
  63. logger.error("初始化推流管道异常:{}, requestId:{}", format_exc(), self.requestId)
  64. raise e