dcl_data_process/archive/2camera-pjibot_delivery.py

# -*- coding: utf-8 -*-
# ------- 全局配置 -------
import os
import time
import oss2
import logging
from resource import parse_robot_image
from utils import json_utils
path1 = '/mnt/disk001/dcl_data_process/src/python2/pjibot_delivery/'
logging.basicConfig(filename=path1 + 'log/2camera-pjibot_delivery.log', level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')
key1 = 'pjibot_delivery/'
sleep_time = 10  # 每多少秒扫描一次
error_bag_json = "/mnt/disk001/dcl_data_process/src/python2/pjibot_delivery/2camera-errorBag.json"
# 1 创建阿里云对象
auth = oss2.Auth('n8glvFGS25MrLY7j', 'xZ2Fozoarpfw0z28FUhtg8cu0yDc5d')
endpoint = 'oss-cn-beijing-gqzl-d01-a.ops.gqzl-cloud.com'
bucket = oss2.Bucket(auth, endpoint, 'open-bucket')
# ------- 全局配置 -------

def parse_to_mp4(merged_bag_file_path, parse_prefix1, local_parse_dir2, local_delete_list3):
    mp4_file_name = 'camera'
    local_mp4_file_path1 = local_mp4_dir + '/' + mp4_file_name + '.mp4'
    oss_csv_object_key1 = parse_prefix1 + mp4_file_name + '.mp4'
    try:
        flag, local_mp4_dir = parse_robot_image.parse(merged_bag_file_path, local_parse_dir2 + '/camera/')
        if flag:
            bucket.put_object_from_file(oss_csv_object_key1, local_mp4_file_path1)
            logging.info("上传 camera.mp4 成功: %s", str(oss_csv_object_key1))
        else: 
            # 如果没有图像话题，也添加到错误列表，防止后面重复扫描生成
            json_utils.add_error(parse_prefix1,error_bag_json)
            logging.info("没有图像话题: %s", merged_bag_file_path)
    except Exception as e2:
        json_utils.add_error(parse_prefix1,error_bag_json)
        logging.exception("生成摄像头视频报错: %s", e2)
    finally:
        local_delete_list3.append(local_mp4_file_path1)
        local_delete_list.append(local_merged_bag_path)

def judge(parse_prefix_full):
    camera_done = False
    for obj2 in oss2.ObjectIterator(bucket, prefix=parse_prefix_full):
        if '/camera.mp4' in str(obj2.key):
            camera_done = True
    if camera_done:
        return True
    
def end_delete(local_delete_list):
    if len(local_delete_list) > 0:
        for local_delete in local_delete_list:
            try:
                os.remove(local_delete)
            except Exception as e:
                pass

# ------- 获取合并之后的bag包，解析出csv -------
if __name__ == '__main__':
    try:
        while True:
            logging.info("开始新一轮扫描")
            local_delete_list = []
            oss_delete_list = []
            prefix_list = []
            for obj1 in oss2.ObjectIterator(bucket, prefix=key1):
                try:
                    merged_bag_object_key = str(obj1.key)
                    if 'data_merge' in str(obj1.key) and str(obj1.key).endswith('.bag'):
                        merged_bag_object_key_split = merged_bag_object_key.split('/')
                        merged_prefix = '/'.join(merged_bag_object_key_split[:-1])
                        parse_prefix = merged_prefix.replace('data_merge', 'data_parse')
                        parse_prefix_full = merged_bag_object_key.replace('data_merge', 'data_parse')[:-4] + '/'
                        if judge():
                            continue
                        local_merged_bag_path = path1 + 'camera/' + merged_bag_object_key
                        local_merged_dir = '/'.join(local_merged_bag_path.split('/')[:-1])
                        local_parse_dir = local_merged_dir.replace('data_merge', 'data_parse')
                        if not os.path.exists(local_merged_dir):
                            os.makedirs(local_merged_dir)
                        if not os.path.exists(local_parse_dir):
                            os.makedirs(local_parse_dir)
                        merged_bag_full_name = merged_bag_object_key_split[-1]
                        merged_bag_name = merged_bag_full_name.split('.')[0]
                        bucket.get_object_to_file(merged_bag_object_key, local_merged_bag_path)
                        logging.info("生成视频: %s" % str(obj1.key))
                        parse_to_mp4(local_merged_bag_path, parse_prefix_full, local_parse_dir, local_delete_list)
                except Exception as e:
                    logging.exception("局部异常处理: %s" % str(e))
                    continue
                end_delete()
                time.sleep(sleep_time)
    except Exception as e:
        logging.exception("全局错误处理: %s" % str(e))