# -*- coding: utf-8 -*- # ------- 全局配置 ------- import sys reload(sys) sys.setdefaultencoding("utf-8") import os import time import oss2 import logging import json import subprocess import io from utils import json_utils path1 = '/mnt/disk001/dcl_data_process/src/python2/pjibot_delivery/' path2 = '/mnt/disk001/dcl_data_process/src/python3/pjibot_outdoor/' logging.basicConfig(filename=path1 + 'log/xosc-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/xosc-errorBag.json" # 1 创建阿里云对象 auth = oss2.Auth('n8glvFGS25MrLY7j', 'xZ2Fozoarpfw0z28FUhtg8cu0yDc5d') endpoint = 'oss-cn-beijing-gqzl-d01-a.ops.gqzl-cloud.com' bucket = oss2.Bucket(auth, endpoint, 'pji-bucket1') # ------- 全局配置 ------- def generate_xosc(parse_prefix, local_parse_dir, local_delete_list): try: os.chdir(path2) command2 = 'python3 jiqiren_outdoor.py {} 0'.format(local_parse_dir[:-1]) # 配送机器人0 巡检机器人1 logging.info("进入目录 %s 调用命令2: %s", path2, command2) process = subprocess.Popen(command2, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() # 等待进程执行完成并获取输出 if process.returncode == 0: logging.info("命令输出: %s", stdout.decode("utf-8")) else: logging.error("命令执行失败，错误码: %d", process.returncode) logging.error("命令错误输出: %s", stderr.decode("utf-8")) local_xosc_path2 = local_parse_dir + 'simulation/xosc/openx_outdoor0.xosc' bucket.put_object_from_file(parse_prefix + 'scenario.xosc', local_xosc_path2) bucket.put_object_from_file(parse_prefix + 'scenario_hmi.xosc', local_xosc_path2) logging.info("上传 scenario_hmi.xosc 成功: %s" % str(parse_prefix + 'scenario.xosc')) # ------- 处理 output.json - 开始 ------- output_json_path = local_parse_dir + 'simulation/xosc/output.json' callback_json_oss_key = parse_prefix+'callback.json' callback_json_local = local_parse_dir+'/callback.json' try: # 1 解析 output.json if os.path.exists(output_json_path): outputs = json_utils.parse_json_to_string_array(output_json_path) # 2 将 output.json 添加到 callback.json 的 check 字段 bucket.get_object_to_file(callback_json_oss_key, callback_json_local) with io.open(callback_json_local, 'r', encoding='utf-8') as f: data = json.load(f) if 'check' not in data: data['check'] = [] data['check'].extend(outputs) json_data = json.dumps(data, ensure_ascii=False, indent=4) with io.open(callback_json_local, 'w', encoding='utf-8') as f: f.write(unicode(json_data)) bucket.put_object_from_file(callback_json_oss_key, callback_json_local) except Exception as e3: # todo 可能没有callback.json，已经处理成 callback_done.json了，暂时不管 logging.exception("处理 output.json报错: %s" % str(e3)) pass # ------- 处理 output.json - 结束 ------- # 处理删除 local_delete_list.append(local_xosc_path2) local_delete_list.append(callback_json_local) except Exception as e: error_bag_list = json_utils.parse_json_to_string_array(error_bag_json) error_bag_list.append(parse_prefix) json_utils.list_to_json_file(error_bag_list, error_bag_json) logging.exception("生成xosc报错: %s" % str(e)) if __name__ == '__main__': while True: try: logging.info("开始新一轮扫描") local_delete_list = [] oss_delete_list = [] prefix_list = [] # 2 获取已经上传完成的所有目录并分组 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] + '/' xosc_done = False csv1_done = False csv2_done = False for obj3 in oss2.ObjectIterator(bucket, prefix=str(parse_prefix_full)): if '/scenario_hmi.xosc' in str(obj3.key): # 仿真使用的 scenario_hmi.xosc 所以必须有 xosc_done = True if '/objects_pji.csv' in str(obj3.key): csv1_done = True if '/ego_pji.csv' in str(obj3.key): csv2_done = True if xosc_done: continue if not csv1_done: continue if not csv2_done: continue error_bag_list = json_utils.parse_json_to_string_array(error_bag_json) if str(parse_prefix_full) in error_bag_list: continue logging.info("需要生成 scenario_hmi.xosc: %s" % str(parse_prefix_full)) local_dir_full = path1 + parse_prefix_full if not os.path.exists(local_dir_full): os.makedirs(local_dir_full) bucket.get_object_to_file(parse_prefix_full + 'objects_pji.csv',local_dir_full + 'objects_pji.csv') bucket.get_object_to_file(parse_prefix_full + 'ego_pji.csv', local_dir_full + 'ego_pji.csv') generate_xosc(parse_prefix_full, local_dir_full, local_delete_list) except Exception as e: logging.exception("局部异常处理: %s", str(e)) # 删除本地临时文件 if len(local_delete_list) > 0: for local_delete in local_delete_list: try: os.remove(local_delete) except Exception as e: logging.exception("删除本地临时文件报错: %s" % str(e)) time.sleep(sleep_time) except Exception as e: logging.exception("全局异常处理: %s", str(e))