# -*-ing:utf-8-*-
import os
import sys
from datetime import datetime
import traceback
import json
import math
import pandas as pd
import subprocess
import time
import multiprocessing # 导入进程模块
from functools import partial # 导入偏函数
import warnings
warnings.filterwarnings("ignore")
from openx import Scenario, formatThree, formatTwo, change_CDATA
import osgbGenerator
from utils import smooth_1, get_coordinate_cow, get_map_name, reverse_gyb_gps
class Batchrun:
def __init__(self, path, keyFileName):
""""初始化方法"""
self.path = path
self.keyFileName = keyFileName
def getFile(self, path, keyFileName):
'''
获得可以进行轨迹提取的所有片段文件夹路径
Parameters
----------
path : TYPE
DESCRIPTION.
Returns
-------
FileList : TYPE
可以进行轨迹提取的所有片段文件夹路径.
'''
files = os.listdir(path) # 得到文件夹下的所有文件，包含文件夹名称
FileList = []
if keyFileName not in files:
for name in files:
if os.path.isdir(path + '/' + name):
FileList.extend(self.getFile(path + '/' + name, keyFileName)) # 回调函数，对所有子文件夹进行搜索
else:
FileList.append(path)
FileList = list(set(FileList))
return FileList
def generateScenarios_raw(self, absPath, param):
'''
原始自然驾驶场景还原
'''
# 读取自车数据
posPath = os.path.join(absPath, param) # 用来生成场景中自车和他车的路径
posdata = pd.read_csv(posPath)
# 读取仿真地图数据
offset_x = -457000.0
offset_y = -4400000.0
# offset_h = 90 # 初始headinga 和 VTD 有 90偏角
offset_h = 0
# 对感知数据进行后处理
# 对公园北环路进行坐标转换
# 根据pos文件规范时间戳
posdata['Time'] = posdata['Time'].apply(lambda x: (x // 100) * 100)
# 绝对高程与相对高程的差值
posdata['altitude'] = 19.27
# posdata['Type'].replace({8:2, 10:2}, inplace=True)
pos_ego = posdata.loc[posdata['ID'] == -1, ['Time', 'East', 'North', 'HeadingAngle', 'altitude', 'Type']]
# pos_ego[['East', 'North']] = pos_ego.apply(get_coordinate_cow, axis=1)
# pos_ego = posdata.loc[posdata['ID'] == -1, ['Time', 'position_x', 'position_y', 'HeadingAngle', 'Type']]
pos_ego = pos_ego.reset_index(drop=True)
# offset_x = -float(pos_ego.at[0, 'East']) # 初始East 设为0
# offset_y = -float(pos_ego.at[0, 'North']) # 初始North 设为0
start_time = pos_ego.at[0, 'Time']
ego_points, gps_time, ego_type = Scenario.getXoscPosition(pos_ego, 'Time', 'East', 'North', 'HeadingAngle',
'altitude', 'Type', offset_x, offset_y, offset_h,
start_time)
# ego_points, gps_time, ego_type = Scenario.getXoscPosition(pos_ego, 'Time', 'position_x', 'position_y', 'HeadingAngle', 'Type', offset_x, offset_y, offset_h, start_time)
# 读取目标数据
pos_obs = posdata # 排除非目标物的物体
# pos_obs = posdata[posdata['Type'] != 10] # 排除非目标物的物体
pos_obs = pos_obs.loc[
pos_obs['ID'] != -1, ['Time', 'ID', 'East', 'North', 'HeadingAngle', 'AbsVel', 'altitude', 'Type']]
# pos_obs = pos_obs.loc[pos_obs['ID'] != -1, ['Time', 'ID', 'position_x', 'position_y', 'HeadingAngle', 'Type']]
# !!!!自定义修改
# pos_obs.loc[pos_obs['ID'] == 1200, 'type'] = 7
# pos_obs.loc[pos_obs['ID'] == 1197, 'type'] = 7
# pos_obs.loc[pos_obs['ID'] == 1200, 'position_x'] += 1
# pos_obs.loc[pos_obs['ID'] == 1200, 'position_y'] -= 2
# pos_obs.loc[pos_obs['ID'] == 1197, 'position_x'] += 1
# pos_obs.loc[pos_obs['ID'] == 1197, 'position_y'] -= 1
# pos_obs.loc[pos_obs['ID'] == 1203, 'type'] = 4
# pos_obs.loc[pos_obs['ID'] == 1203, 'position_x'] = 456319
# pos_obs.loc[pos_obs['ID'] == 1203, 'position_y'] = 4397979
# pos_obs.loc[pos_obs['ID'] == 1203, 'HeadingAngle'] = 1
# offset_h = 30
# pos_obs.loc[pos_obs['ID'] == 1194, 'type'] = 7
# pos_obs.loc[pos_obs['ID'] == 1188, 'type'] = 2
# pos_obs.loc[pos_obs['ID'] == 1189, 'type'] = 2
# pos_obs.loc[pos_obs['ID'] == 1195, 'type'] = 2
# obsPath = os.path.join(absPath, 'obs.csv') # 用来筛选场景中不需要的他车
# obsdata = pd.read_csv(obsPath)
# obsdata = obsdata[(obsdata['ObjectPosY'] < 5) & (obsdata['ObjectPosY'] > -5) & (obsdata['ObjectPosX'] > -10) & (obsdata['ObjectPosX'] < 100)] # 排除车道线范围外且前向距离较远的目标物
# idlist = obsdata['ObjectID'].tolist() # 筛选出符合条件的的目标物ID
# pos_obs = pos_obs[(pos_obs['ID'].isin(idlist))]
pos_obs = pos_obs.reset_index(drop=True)
groups = pos_obs.groupby('ID')
object_points = []
for key, value in groups:
if len(value) < 5:
continue
# value = smooth_1(value)
object_points.append(
Scenario.getXoscPosition(value, 'Time', 'East', 'North', 'HeadingAngle', 'altitude', 'Type', offset_x,
offset_y, offset_h, start_time))
# object_points.append(Scenario.getXoscPosition(value, 'Time', 'position_x', 'position_y', 'HeadingAngle', 'type', offset_x, offset_y, offset_h,start_time))
ego_speed = 5
period = math.ceil(gps_time[-1] - gps_time[0])
work_mode = 0 # 0为CICV车端数据
hour = int(absPath.split('/')[-1].split('-')[3])
if hour + 8 >= 24:
hour = hour - 16
else:
hour = hour + 8
time_of_day = hour * 3600
# 没有路灯防止天过暗
if time_of_day >= 64800:
time_of_day = 64800
s = Scenario(ego_points, object_points, gps_time, ego_speed, work_mode, period, absPath, time_of_day)
filename = absPath + '/simulation' + '_' + param.split("_")[1].split(".")[0]
files = s.generate(filename)
change_CDATA(files[0][0])
print(files)
# 生成osgb
# osgbGenerator.formatTwo(filename)
# 修改xosc路径
# formatThree(filename)
# # # 生成每个场景的描述文件 json
# # getLabel(output_path, scenario_series['场景编号'], scenario_series['场景名称'])
# 拷贝到vtd路径下
# vtd_path = os.path.join('/home/lxj/VIRES/VTD.2021.3/Data/Projects/Current/Scenarios/myScenario', datetime.now().strftime("%Y%m%d%H%M%S"))
# if not os.path.exists(vtd_path):
# os.makedirs(vtd_path)
# os.system('cp '+ files[0][0] + ' ' + vtd_path + '/')
def multiRun(self, path, param):
files = self.getFile(path, self.keyFileName)
print('程序开始，%s 个数据包' % len(files))
t1 = time.time()
# 无参数时，使用所有cpu核; 有参数时，使用CPU核数量为参数值
pool = multiprocessing.Pool(processes=10)
pfunc = partial(self.generateScenarios_raw, param)
pool.map(pfunc, files)
# 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
pool.close()
pool.join()
t2 = time.time()
print("程序结束，并行执行时间：%s s" % int(t2 - t1))
def batchRun(self, path, param):
files = self.getFile(path, self.keyFileName)
print('程序开始，%s 个数据包' % len(files))
for di, absPath in enumerate(sorted(files)):
print(absPath)
# try:
self.generateScenarios_raw(absPath, param)
# except:
# print('Augmentation failed!!!!')
# error = {'time':datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],'traceback':traceback.format_exc()}
# with open('error.log','a+') as f:
# json.dump(error, f, indent=4)
# f.write('\n')
# print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
return None
def GenerateVideo(self, root_path):
# 批量生成视频
imagelist = self.getFile(root_path, 'image')
for item in imagelist:
strs = item.split('/')
respath = os.path.join(item, "video.mp4")
# respath = os.path.join(res_path,strs[-3],strs[-2],"video.mp4")
print('---------------------')
# os.makedirs(os.path.join(res_path,strs[-3],strs[-2]))
command = "ffmpeg -f image2 -r 10 -pattern_type glob -i '" + item + "/image/*.jpg" + "' -y '" + respath + "'"
print(command)
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
process.wait()
if __name__ == "__main__":
# rootPath = "/media/hancheng/Simulation5/pjisuv_pjisuv95_data_merge_2024-01-31-07-53-50_TTC_11" # 跟车
rootPath = sys.argv[1] # 跟车
# 生成场景
a = Batchrun(rootPath, "pos_orig.csv")
a.batchRun(rootPath, 'pos_orig.csv')
# a.batchRun(rootPath, 'pos_hmi.csv')
# a.multiRun(rootPath, 0)
# 生成视频
# a.GenerateVideo(rootPath)