(
output_imgs_queue,
temp_dir,
result_dir,
work_id,
audio_path,
result_queue,
width,
height,
fps,
watermark_switch=0,
digital_auth=0,
temp_queue=None,
)
| 25 | |
| 26 | |
| 27 | def write_video_gradio( |
| 28 | output_imgs_queue, |
| 29 | temp_dir, |
| 30 | result_dir, |
| 31 | work_id, |
| 32 | audio_path, |
| 33 | result_queue, |
| 34 | width, |
| 35 | height, |
| 36 | fps, |
| 37 | watermark_switch=0, |
| 38 | digital_auth=0, |
| 39 | temp_queue=None, |
| 40 | ): |
| 41 | output_mp4 = os.path.join(temp_dir, "{}-t.mp4".format(work_id)) |
| 42 | fourcc = cv2.VideoWriter_fourcc(*"mp4v") |
| 43 | result_path = os.path.join(result_dir, "{}-r.mp4".format(work_id)) |
| 44 | video_write = cv2.VideoWriter(output_mp4, fourcc, fps, (width, height)) |
| 45 | print("Custom VideoWriter init done") |
| 46 | try: |
| 47 | while True: |
| 48 | state, reason, value_ = output_imgs_queue.get() |
| 49 | if type(state) == bool and state == True: |
| 50 | logger.info( |
| 51 | "Custom VideoWriter [{}]视频帧队列处理已结束".format(work_id) |
| 52 | ) |
| 53 | logger.info( |
| 54 | "Custom VideoWriter Silence Video saved in {}".format( |
| 55 | os.path.realpath(output_mp4) |
| 56 | ) |
| 57 | ) |
| 58 | video_write.release() |
| 59 | break |
| 60 | else: |
| 61 | if type(state) == bool and state == False: |
| 62 | logger.error( |
| 63 | "Custom VideoWriter [{}]任务视频帧队列 -> 异常原因:[{}]".format( |
| 64 | work_id, reason |
| 65 | ) |
| 66 | ) |
| 67 | raise CustomError(reason) |
| 68 | for result_img in value_: |
| 69 | video_write.write(result_img) |
| 70 | if video_write is not None: |
| 71 | video_write.release() |
| 72 | if watermark_switch == 1 and digital_auth == 1: |
| 73 | logger.info( |
| 74 | "Custom VideoWriter [{}]任务需要水印和数字人标识".format(work_id) |
| 75 | ) |
| 76 | if width > height: |
| 77 | command = 'ffmpeg -y -i {} -i {} -i {} -i {} -filter_complex "overlay=(main_w-overlay_w)-10:(main_h-overlay_h)-10,overlay=(main_w-overlay_w)-10:10" -c:a aac -crf 15 -strict -2 {}'.format( |
| 78 | audio_path, |
| 79 | output_mp4, |
| 80 | GlobalConfig.instance().watermark_path, |
| 81 | GlobalConfig.instance().digital_auth_path, |
| 82 | result_path, |
| 83 | ) |
| 84 | logger.info("command:{}".format(command)) |
nothing calls this directly
no outgoing calls
no test coverage detected