dcl_data_process/src/python2/pjisuv/xosc-pjisuv.py

# -*- coding: utf-8 -*-
import os
import time
import oss2
import logging
import json

path1 = '/mnt/disk001/dcl_data_process/src/python2/pjisuv/'
path2 = '/mnt/disk001/dcl_data_process/src/python3/pjisuv/'

logging.basicConfig(filename=path1 + 'log/xosc.log', level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')

key1 = 'pjisuv/'
sleep_time = 60  # 每多少秒扫描一次

error_bag_json = "/mnt/disk001/dcl_data_process/src/python2/pjisuv/errorBag.json"


def parse_json_to_string_array(file_path):
    try:
        # 打开并读取JSON文件
        with open(file_path, 'r', encoding='utf-8') as file:
            # 解析JSON内容
            data = json.load(file)

        # 检查数据是否是一个列表，并且列表中的元素是否是字符串
        if isinstance(data, list):
            for item in data:
                if not isinstance(item, str):
                    raise ValueError("JSON数组中的元素不是字符串")
            return data
        else:
            return []
    except Exception as e:
        return []


def list_to_json_file(data, file_path):
    """
    将列表转换为JSON格式并写入指定的文件路径。
    如果文件已存在，则覆盖它。

    参数:
    data (list): 要转换为JSON的列表。
    file_path (str): 要写入JSON数据的文件路径。
    """
    # 将列表转换为JSON格式的字符串
    json_data = json.dumps(data, ensure_ascii=False, indent=4)

    # 以写入模式打开文件，如果文件已存在则覆盖
    with open(file_path, 'w', encoding='utf-8') as file:
        # 将JSON字符串写入文件
        file.write(json_data)


def generate_xosc(parse_prefix, local_parse_dir, local_delete_list):
    try:
        os.chdir(path2)
        # todo simulation_hmi 实际生成的 orig，因为多功能车没有 hmi
        command2 = 'python3 simulation_hmi.py ' + local_parse_dir[:-1]  # 生成orig
        logging.info("调用命令2: %s" % str(command2))
        os.system(command2)
        local_xosc_path2 = local_parse_dir + 'simulation_orig/xosc/openx0.xosc'
        bucket.put_object_from_file(parse_prefix + 'scenario_orig.xosc', local_xosc_path2)
        bucket.put_object_from_file(parse_prefix + 'scenario_hmi.xosc',
                                    local_xosc_path2)  # todo 将orig也上传成hmi，因为多功能车暂时没有hmi
        logging.info("上传 scenario_orig.xosc 成功: %s" % str(parse_prefix + 'scenario_orig.xosc'))
        local_delete_list.append(local_xosc_path2)
    except Exception as e:
        error_bag_list = parse_json_to_string_array(error_bag_json)
        error_bag_list.append(parse_prefix)
        list_to_json_file(error_bag_list, error_bag_json)
        logging.exception(parse_prefix, "生成 xosc 报错，将其添加到错误bag列表。", str(e))


# ------- 获取合并之后的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, 'open-bucket')
    while True:
        try:
            logging.info("开始新一轮扫描")
            prefix_list = []
            # 2 获取已经上传完成的所有目录并分组
            all_prefix_set = set()
            for obj1 in oss2.ObjectIterator(bucket, prefix=key1):
                # 获取csv
                if 'data_parse' in str(obj1.key) and str(obj1.key).count('/') == 4:
                    all_prefix_set.add('/'.join(str(obj1.key).split('/')[:-1]) + '/')
            for obj2_key in all_prefix_set:
                try:
                    local_delete_list = []
                    oss_delete_list = []
                    xosc1_done = False
                    xosc2_done = False
                    csv1_done = False
                    csv2_done = False
                    for obj3 in oss2.ObjectIterator(bucket, prefix=obj2_key):
                        if '/scenario_orig.xosc' in str(obj3.key):
                            xosc1_done = True
                        # if '/scenario_hmi.xosc' in str(obj3.key):
                        #     xosc2_done = True
                        if '/pos_orig.csv' in str(obj3.key):
                            csv1_done = True
                        # if '/pos_hmi.csv' in str(obj3.key):
                        #     csv2_done = True
                    if xosc1_done or not csv1_done:
                        continue
                    error_bag_list = parse_json_to_string_array(error_bag_json)
                    if str(obj2_key) in error_bag_list:
                        continue

                    logging.info("需要生成xosc: %s" % obj2_key)
                    local_dir_full = path1 + obj2_key
                    if not os.path.exists(local_dir_full):
                        os.makedirs(local_dir_full)
                    # todo 下载单个csv生成两个视频？
                    bucket.get_object_to_file(obj2_key + 'pos_orig.csv', local_dir_full + 'pos_orig.csv')
                    # bucket.get_object_to_file(obj2_key+'pos_hmi.csv', local_dir_full+'pos_hmi.csv')
                    # local_delete_list.append(local_dir_full + 'pos_orig.csv') # todo 不删除了 做测试用
                    # local_delete_list.append(local_dir_full+'pos_hmi.csv')
                    generate_xosc(obj2_key, local_dir_full, 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:
                                logging.exception("删除本地临时文件报错: %s" % str(e))
                except Exception as e:
                    logging.exception("局部异常处理: %s", str(e))
            time.sleep(sleep_time)
        except Exception as e:
            logging.exception("全局异常处理: %s", str(e))