# -*- coding: utf-8 -*- import os import time import oss2 from resource import parse_pji_image from utils import json_utils import logging path1 = '/mnt/disk001/dcl_data_process/src/python2/pjibot/' logging.basicConfig(filename=path1 + 'log/camera-pjibot_guide.log', level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s') key1 = 'pjibot/' sleep_time = 30 # 每多少秒扫描一次 error_bag_json = "/mnt/disk001/dcl_data_process/src/python2/pjibot/camera-errorBag.json" def get_paths(local_parse_dir,parse_prefix): local_mp4_dir1 = local_parse_dir + '/camera1/' mp4_file_name1 = 'camera1' local_mp4_file_path1 = local_mp4_dir1 + mp4_file_name1 + '.mp4' oss_csv_object_key1 = parse_prefix + 'camera.mp4' local_mp4_dir2 = local_parse_dir + '/camera2/' mp4_file_name2 = 'camera2' local_mp4_file_path2 = local_parse_dir + '/camera2/' + mp4_file_name2 + '.mp4' oss_csv_object_key2 = parse_prefix + mp4_file_name2 + '.mp4' return local_mp4_dir1,local_mp4_file_path1,oss_csv_object_key1, local_mp4_dir2,local_mp4_file_path2, oss_csv_object_key2 def parse_to_mp4(merged_bag_file_path, parse_prefix, local_parse_dir, local_delete_list): try: local_mp4_dir1,local_mp4_file_path1,oss_csv_object_key1, local_mp4_dir2,local_mp4_file_path2, oss_csv_object_key2 = get_paths(local_parse_dir,parse_prefix) parse_pji_image.parse1('/ob_camera_01/color/image_raw', merged_bag_file_path,local_mp4_dir1) bucket.put_object_from_file(oss_csv_object_key1, local_mp4_file_path1) logging.info("上传 camera1.mp4 成功: %s" , str(oss_csv_object_key1)) parse_pji_image.parse2('/ob_camera_02/color/image_raw', merged_bag_file_path,local_mp4_dir2) bucket.put_object_from_file(oss_csv_object_key2, local_mp4_file_path2) logging.info("上传 camera2.mp4 成功: %s" , str(oss_csv_object_key1)) local_delete_list.append(local_mp4_file_path1) local_delete_list.append(local_mp4_file_path2) except Exception as e2: logging.exception("生成摄像头视频报错,添加到 camera-errorBag.json: %s", e2) json_utils.add_error(parse_prefix,error_bag_json) # 当出现异常时执行的代码 # ------- 获取合并之后的bag包,解析出csv ------- if __name__ == '__main__': # 1 创建阿里云对象 auth = oss2.Auth('n8glvFGS25MrLY7j', 'xZ2Fozoarpfw0z28FUhtg8cu0yDc5d') endpoint = 'oss-cn-beijing-gqzl-d01-a.ops.gqzl-cloud.com' bucket = oss2.Bucket(auth, endpoint, 'pji-bucket1') while True: logging.info("开始新一轮扫描") try: local_delete_list = [] oss_delete_list = [] prefix_list = [] # 2 获取已经上传完成的所有目录并分组 for obj1 in oss2.ObjectIterator(bucket, prefix=key1): # 获取合并后的包 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] + '/' camera_done = False camera2_done = False for obj2 in oss2.ObjectIterator(bucket, prefix=parse_prefix_full): if '/camera.mp4' in str(obj2.key): camera_done = True if '/camera2.mp4' in str(obj2.key): camera2_done = True if camera_done and camera2_done: continue error_bag_list = json_utils.parse_json_to_string_array(error_bag_json) if parse_prefix_full in error_bag_list: continue logging.info("%s 等待处理到: %s", merged_bag_object_key, parse_prefix_full) 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] try: bucket.get_object_to_file(merged_bag_object_key, local_merged_bag_path) except Exception as e: logging.exception("下载合并后的bag包失败: %s" % str(e)) local_delete_list.append(local_merged_bag_path) # 2 生成 pos_orig.csv 和 pos_hmi.csv parse_to_mp4(local_merged_bag_path, parse_prefix_full, local_parse_dir, 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 time.sleep(sleep_time) except Exception as e: logging.exception("全局错误处理: %s" % str(e))