Restful API 运行控制
提供基于 HTTP 的 Restful API，支持 Web 服务接口，控制 Simone 界面的运行流程，比如可控制创建案例库，导入导出案例，运行停止案例等。
接口文档：
使用参考手册
1. 步骤描述
-
创建一个python文件 依次将接口路径、依赖、请求方法粘贴进去
-
顺序粘贴对应的 功能实现 模块代码
-
登录功能 必须要有并且在首位调用所有的操作都需要依赖登录功能携带的token
-
粘贴调用示例并稍微根据自己的环境做一些参数修改比如 请求路径修改 ， 使用案例库案例名称修改 等，即可成功调用API
-
修改以下几处位置：
# 路径地址与云端账户需要修改为自己对应的账户
SimOneUrl = "http://127.0.0.1:30083"
loginData = {
"username": 'admin',
"password": 'admin',
}
if __name__ == '__main__':
# 云端登录
LoginPolicy("cloud")
# 单机版本登录
LoginPolicy("standalone")
备注： 附录参考代码是根据步骤描述中的顺序进行粘贴并可以成功运行的代码示例，如有疑惑请查看附录参考代码！！
2. 接口路径与依赖库
import base64
import copy
import inspect
import random
from os.path import exists
from Crypto.Cipher import AES
from faker import Faker
from requests import request, Response
import json
from urllib.parse import urljoin
import os
import time
# 路径地址与云端账户需要修改为自己对应的账户
SimOneUrl = "http://127.0.0.1:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
fake = Faker('zh_CN')
interval = 10
# header
header = {
"Content-Type": "application/json",
}
# Environment Path
class EnvPath:
token = "TOKEN"
class Prefix:
UserServer = "api-user"
CaseServer = "api-case"
AssetServer = "api-asset"
TaskServer = "api-task"
# token
_token = os.environ.get(EnvPath.token)
""" 登录相关接口 """
user_check_url = Prefix.UserServer + "/users/check"
health = Prefix.UserServer + "/health" # 检查服务是否正常
cloud_login_url = Prefix.UserServer + "/users/login" # 云端登录
standalone_login_url = Prefix.UserServer + "/users/standalone" # 单机版登录
login_check = Prefix.UserServer + "/users/check" # user check
""" 案例库目录相关接口 """
categories_url = Prefix.CaseServer + "/categories" # 案例库目录
get_categories_url = Prefix.CaseServer + "/categories?withCases=false" # 获取所有案例库
get_cases_url = "api-case/cases?page=1&filter={category}&pageSize=100"
# get_cases_url = Prefix.CaseServer + "/cases?page={page}&filter={category}&pageSize=100" # 获取当前案例库的所有案例
delete_categories_url = Prefix.CaseServer + "/categories/{}" # 删除案例库目录
""" 上传文件相关接口 """
file_md5_url = Prefix.AssetServer + "/files/{}?raw={}" # 验证文件MD5
upload_file_url = Prefix.AssetServer + "/files/{}/upload?raw={}" # 上传文件
""" 测试案例相关接口 """
case_import_url = Prefix.AssetServer + "/cases/import" # 导入案例
case_details_url = Prefix.CaseServer + "/cases/{case_id}/data" # 案例详情
delete_case_url = Prefix.CaseServer + "/trash" # 删除案例
convert_replay_case_url = Prefix.TaskServer + "/tasks/{id}/convert2case" # 转换回放案例
case_def_url = Prefix.CaseServer + "/cases/{case_id}" # 案例概要
update_case_url = Prefix.AssetServer + "/cases/update" # 编辑案例 # doing
""" 测试任务相关接口 """
create_task_url = Prefix.TaskServer + "/tasks" # 创建task
stop_task_url = Prefix.TaskServer + "/tasks/stop" # 停止task
stop_sessions_url = Prefix.TaskServer + "/sessions/stop" # 会话停止
delete_task_url = Prefix.TaskServer + "/tasks/delete" # 删除task
delete_sessions_url = Prefix.TaskServer + "/sessions/delete" # 删除停止
result_url = Prefix.TaskServer + "/tasks/task?taskIds={task_id}" # 删除停止 案例运行结果详情
task_info = Prefix.TaskServer + "/tasks/task" # 删除停止 案例运行结果详情
pause_task_url = Prefix.TaskServer + "/sessions/{task_id}/pause" # 任务暂停
resume_task_url = Prefix.TaskServer + "/sessions/{task_id}/resume" # 任务暂停后开始
task_result_url = Prefix.TaskServer + "/tasks/task?taskIds={id}" # 删除停止 案例运行结果详情
""" 主车相关接口 """
import_vehicle_url = Prefix.AssetServer + "/vehicles/import" # 导入主车
delete_vehicle_url = Prefix.AssetServer + "/vehicles/{vehicles_id}" # 删除主车
get_vehicle_url = Prefix.AssetServer + "/vehicles" # get获取主车信息 post创建主车
get_vehicles_url = Prefix.AssetServer + "/vehicles"
""" 地图相关接口 """
import_map_url = Prefix.AssetServer + "/maps" # 导入地图
delete_map_url = Prefix.CaseServer + "/maps/trash" # 删除地图
get_map_url = Prefix.CaseServer + "/maps?pageSize=100" # 获取地图相关信息
road_service_url = Prefix.TaskServer + "/sce" # 启动路网服务
laneTypeIfInside_url = Prefix.TaskServer + "/sce/{map_id}/location/laneTypeIfInside" # 监测点是否在车道内
""" 数据驱动相关接口 """
import_data_driven_url = Prefix.AssetServer + "/dd" # 导入数据驱动源
delete_data_dirven_url = Prefix.AssetServer + "/dd/{id}" # 删除数据驱动源
""" 资源相关接口 """
import_asset_url = Prefix.AssetServer + "/assets" # 资源导入
asset_repeat_url = Prefix.AssetServer + "/assets/isrepeat" # 资源查重
get_asset_url = Prefix.AssetServer + "/assets?schema={}" # 获取资源信息
delete_asset_url = Prefix.AssetServer + "/assets/{}" # 删除对应资源
""" 案例导出相关 """
export_case_url = Prefix.AssetServer + "/cases/export" # 案例导出
download_case_url = Prefix.AssetServer + "/{pake_id}" # 案例下载
""" 测试案例集相关 """
get_suites_url = Prefix.AssetServer + "/suites" # 获取测试集
run_suite_url = Prefix.TaskServer + "/tasks" # 运行测试集
get_task_list_url = Prefix.TaskServer + "/tasks?finished=true&page=0&pageSize=12&own=true&expandedIds={task_set_id}" # 获取测试案例集合
queue_status_url = Prefix.TaskServer + "/tasks/queue?own=true" # 套件运行状态检查
""" 案例判定相关 """
get_judgements_url = Prefix.AssetServer + "/cases/{caseId}/judgements" # 获取更新 所有判定信息
get_judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}" # 获取更新 单个判定信息
"""测试报告导出相关"""
export_report_url = Prefix.TaskServer + "/sessions/{sessions_id}/report" # 导出测试报告
download_report_url = "/report/{sessions_id}_zh-Hans.pdf" # 下载测试报告
3. 接口请求方法
class RquestApi():
def __init__(self):
super(RquestApi, self).__init__()
self._headers = {"Content-Type": "application/json"}
self._token = os.environ.get(EnvPath.token)
self._host_ip = SimOneUrl
self.get = "GET"
self.post = "POST"
self.delete = "DELETE"
self.put = "PUT"
def send(self, method: str, url: str, **kwargs) -> Response:
"""
发送请求
method: 请求方法
url: 请求地址
kwargs: 请求参数
"""
self.headers.update({"Authorization": self.token})
complete_url = urljoin(self._host_ip, url)
data = dict()
data['method'] = method
data['url'] = complete_url
data['headers'] = self.headers
data.update(kwargs)
print(data)
response = request(verify=False, **data)
try:
print("response:%s" % str(response.content.decode("utf-8")))
self._status_code = response.status_code
self._response_code = response.json().get("code")
except:
pass
return response
@property
def token(self):
return self._token
@property
def headers(self):
return self._headers
@headers.setter
def headers(self, value):
self._headers = value
4. 云端登录
云端API
"""
def cloud_login_api(self, userinfo: dict):
"""
接口名称：云端版登录
请求参数：userinfo
请求方式：post
请求路径：cloud_login_url = Prefix.UserServer + "/users/login"
参数示例：userinfo: body {"username":xxx, "password": xxx}
"""
return self.send(self.post, self.cloud_login_url, json=userinfo).json()
def check_username_api(self, username:str):
"""
接口名称：用户校验
请求参数：username
请求方式：post
请求路径：cloud_login_url = Prefix.UserServer + "/users/check"
"""
payload = {'username': username}
return self.send(self.post, self.login_check, json=payload).json()
"""
功能实现
class LoginBusiness(RquestApi):
def cloud_login_api(self, userinfo: dict):
"""
云端登录API
@param userinfo:
@return:
"""
return self.send(self.post, cloud_login_url, json=userinfo).json()
def check_username_api(self, username: str):
"""
校验用户名称是否合格
@param username:
@return:
"""
payload = {'username': username}
return self.send(self.post, login_check, json=payload).json()
def getCheckToken(self) -> str:
"""
校验token是否合格
@return:
"""
getCheckUrl = os.path.join(SimOneUrl, user_check_url)
payload = {'username': loginData['username']}
response = self.send(self.post, getCheckUrl, json=payload)
checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
return checkToken
def pad(self, text):
"""
填充函数，使被加密数据的字节码长度是block_size的整数倍
@param text:
@return:
"""
length = AES.block_size
count = len(text.encode('utf-8'))
add = length - (count % length)
entext = text + (chr(add) * add)
return entext
def str_aes(self, string: str, key: str):
"""
aes 加密
@param string:
@param key:
@return:
"""
aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
res = aes.encrypt(self.pad(string).encode("utf8"))
msg = str(base64.b64encode(res), encoding="utf8")
return msg
def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
'''
@param string:
@param username:
@param key: AES加密的key
@return:
'''
check_token = self.check_username_api(username)
target_string = check_token["data"]["checkToken"] + string
msg = self.str_aes(target_string, key)
return msg
@property
def _cloud_std_login(self):
"""
云端版本登陆方式
@return:
"""
try:
new_login_data = copy.deepcopy(loginData)
new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
username=new_login_data['username'])
print('cloud login data:' + str(new_login_data))
response = self.cloud_login_api(userinfo=new_login_data)
print("-------", response)
print('cloud login request result:' + str(response))
token = response['data']['token']
os.environ[EnvPath.token] = token
print(token)
return token
except Exception as e:
print(str(e))
raise print('get token error ')
# 登录入口
class LoginPolicy():
standalone = "standalone"
cloud = "cloud"
def __new__(cls, env_name):
if env_name == cls.standalone:
return LoginBusiness()._enterprise_std_login
elif env_name == cls.cloud:
return LoginBusiness()._cloud_std_login
else:
return print("----------input error---------------")
云端调用示例
SimOneUrl = "http://127.0.0.1:30083"
loginData = {
"username": 'admin',
"password": 'admin',
}
if __name__ == '__main__':
LoginPolicy("cloud")
正确返回结果
cloud login request result:{'code': 0, 'data': {'token': 'f2e3e458-264b-44fb-9f95-319a978ddec3'}}
5. 单机版登录
单机版API
"""
def enterprise_login_api(self):
"""
接口名称：企业单机版本登录
请求参数：无
请求方式：get
请求路径：SimOneUrl + standalone_login_url
参数示例：userinfo: body {"username":xxx, "password": xxx}
"""
getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
response = self.send(self.get, url=getLoginUrl)
return response
"""
功能实现
class LoginBusiness(RquestApi):
def cloud_login_api(self, userinfo: dict):
"""
云端登录API
@param userinfo:
@return:
"""
return self.send(self.post, cloud_login_url, json=userinfo).json()
def check_username_api(self, username: str):
"""
校验用户名称是否合格
@param username:
@return:
"""
payload = {'username': username}
return self.send(self.post, login_check, json=payload).json()
def getCheckToken(self) -> str:
"""
校验token是否合格
@return:
"""
getCheckUrl = os.path.join(SimOneUrl, user_check_url)
payload = {'username': loginData['username']}
response = self.send(self.post, getCheckUrl, json=payload)
checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
return checkToken
def pad(self, text):
"""
填充函数，使被加密数据的字节码长度是block_size的整数倍
@param text:
@return:
"""
length = AES.block_size
count = len(text.encode('utf-8'))
add = length - (count % length)
entext = text + (chr(add) * add)
return entext
def str_aes(self, string: str, key: str):
"""
aes 加密
@param string:
@param key:
@return:
"""
aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
res = aes.encrypt(self.pad(string).encode("utf8"))
msg = str(base64.b64encode(res), encoding="utf8")
return msg
def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
'''
@param string:
@param username:
@param key: AES加密的key
@return:
'''
check_token = self.check_username_api(username)
target_string = check_token["data"]["checkToken"] + string
msg = self.str_aes(target_string, key)
return msg
def enterprise_login_api(self):
"""
@return:
"""
getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
response = self.send(self.get, url=getLoginUrl)
return response
@property
def _enterprise_std_login(self):
"""
单机版本登录方式
@return:
"""
try:
getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
response = self.send(self.get, url=getLoginUrl)
token = json.loads(response.content.decode("utf-8"))['data']['token']
os.environ[EnvPath.token] = token
except Exception as e:
print(str(e))
token = ""
# raise AssertionError('get token error ')
return token
@property
def _cloud_std_login(self):
"""
云端版本登陆方式
@return:
"""
try:
new_login_data = copy.deepcopy(loginData)
new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
username=new_login_data['username'])
print('cloud login data:' + str(new_login_data))
response = self.cloud_login_api(userinfo=new_login_data)
print("-------", response)
print('cloud login request result:' + str(response))
token = response['data']['token']
os.environ[EnvPath.token] = token
print(token)
return token
except Exception as e:
print(str(e))
raise print('get token error ')
class LoginPolicy():
standalone = "standalone"
cloud = "cloud"
def __new__(cls, env_name):
if env_name == cls.standalone:
return LoginBusiness()._enterprise_std_login
elif env_name == cls.cloud:
return LoginBusiness()._cloud_std_login
else:
return print("----------input error---------------")
调用示例
if __name__ == '__main__':
LoginPolicy("standalone")
返回结果
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-user/users/standalone', 'headers': {'Content-Type': 'application/json', 'Authorization': None}}
response:{"code":0,"data":{"token":"b557f84a-070d-4ce3-8f70-8c0021428bed"}}
6. 案例库相关
6.1 创建案例库
API
"""
def create_category_api(self, cate_name:str):
'''
创建案例库
@param cate_name: 案例库名
@return:
'''
payload = {"parentId": "", "name": cate_name}
return self.send(self.post,self.categories_url,json=payload).json()
"""
功能实现
class Suite(RquestApi):
def __init__(self):
super(Suite, self).__init__()
self._cyclical = 0
self.category_name = "Automation_" + str(random.randint(0,9999))
def create_category_api(self, cate_name:str):
'''
创建案例库
@param cate_name: 案例库名
@return:
'''
payload = {"parentId": "", "name": cate_name}
return self.send(self.post,categories_url,json=payload).json()
def create_category(self, cate_name:str = None)->str:
'''
创建案例库
@param cate_name:
@return:
'''
if not cate_name:
cate_name = self.category_name
print(f"案例库目录名称：{cate_name}")
response = self.create_category_api(cate_name)
cate_id = response["data"]["categoryId"]
print(f"案例库目录 id ：{cate_id}")
return cate_id
def main():
suite = Suite()
suite.create_category()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-case/categories', 'headers': {'Content-Type': 'application/json', 'Authorization': 'a2b6beb8-f582-4b15-b063-d92c506452dd'}, 'json': {'parentId': '', 'name': 'Automation_5538'}}
response:{"code":0,"data":{"name":"Automation_5538","categoryId":"feeea7d0-439e-4796-a239-0f7d8f484d01"}}
案例库目录 id ：feeea7d0-439e-4796-a239-0f7d8f484d01
6.2 获取案例库列表
API
"""
def get_category_api(self):
"""
获取案例库列表
@return: response
"""
return self.send(self.get, self.categories_url).json()
"""
功能实现
def get_category_api(self):
"""
获取案例库列表
@return: response
"""
return self.send(self.get, categories_url).json()
def get_category(self,userid:str=None)->dict:
"""
获取案例库
@param userid: 用户id desc：admin->builtIn
@return:
"""
cate_dict={"data":{},"parentId":{}}
response = self.get_category_api()
cate_dict["total"] = response["data"]["total"]
if not userid:
print("获取当前所有的案例库目录信息")
for one in response["data"]["categories"]:
cate_dict["data"].setdefault(one["name"],one["id"])
cate_dict["parentId"].setdefault(one["name"],one["parentId"])
else:
print("获取用户{}的案例库目录信息".format(userid))
for one in response["data"]["categories"]:
if one["userId"] == userid:
cate_dict["data"].setdefault(one["name"],one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
print("获取到的案例库->%s"%str(cate_dict))
return cate_dict
def main():
suite = Suite()
suite.create_category()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
获取到的案例库->{'data': {'智能网联场景库': 'f58fdcf4-c36b-11eb-81fd-1831bf08c354', 'V2X泛化': 'f58fdcf4-c36b-11eb-81fd-1831bf08c354', '前向碰撞预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '交叉路口碰撞预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '盲区变道预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '异常车辆提醒': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '车辆失控预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '道路危险状况提示': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '限速预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '弱势交通参与者碰撞预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '绿波车速引导': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '车内标牌': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '左转辅助': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '前方拥堵提醒': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '紧急车辆提醒': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '闯红灯预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '逆向超车预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', '紧急制动预警': 'fbdc25a4-f982-4db7-9e39-7ca68b5acd76', 'new': '', 'Automation_5538': ''}, 'total': 15}
6.3 获取案例库列表
API
"""
def get_category_case_api(self,cate_id:str):
"""
获取当前案例库所有案例
@param cate_name: 案例库名
@return:
"""
return self.send(self.get,self.get_cases_url.format(category=cate_id)).json()
"""
功能实现
def get_category_case_api(self,cate_id:str):
"""
获取当前案例库所有案例
@param cate_name: 案例库名
@return:
"""
return self.send(self.get,get_cases_url.format(category=cate_id)).json()
def get_cases_category(self, cate_id: list):
"""
获取当前案例库的所有案例
@param cate_id: 案例库目录id
@return:
"""
def url_code(url: str, way: str):
from urllib import parse
if way == "unquote":
return parse.unquote(url)
elif way == "quote":
return parse.quote(url)
page_size = 100
# print("get all case id from api by category_id")
_filter = {"keyword": "", "categories": cate_id}
string = json.dumps(_filter).replace(" ", "")
filter = url_code(string, UrlCodeType.quote).replace("%", "%25")
response = self.send(self.get, get_cases_url.format(page="1", category=filter)).json()
# get case id
# print("get_cases_category 入参cate_id:%s" % cate_id)
cases_num = response['data']['total']
if cases_num >= page_size:
if cases_num % page_size == 0:
page = cases_num // page_size
else:
page = cases_num // page_size + 1
else:
page = 1
def update_case_id(response, page_size):
case_name = response['data']['caseDefs'][page_size]['name']
case_id = response['data']['caseDefs'][page_size]['id']
case_id_list.append(case_id)
cases_dict.setdefault(case_name, [case_id])
cases_dict = {}
case_id_list = []
# 如果案例数量是100以内
if cases_num <= page_size:
for i in range(cases_num):
update_case_id(response=response, page_size=i)
else:
# 如果案例数量超过100了，固定获取案例数量为100个就翻页
for i in range(page):
response = self.send(self.get, self.get_cases_url.format(page=str(i + 1), category=filter)).json()
# cases_num % page_size 是取整表示没有到达最后一页
# if i == 0 or i > 0 and cases_num % page_size == 0:
if i + 1 != page:
# print("xxx page_size:",page_size)
for j in range(page_size):
# print("xxx j:",j)
update_case_id(response=response, page_size=j)
# 到最后一页了，因为案例数量不固定所以需要单独计算获取的个数
else:
for j in range(cases_num % page_size):
update_case_id(response=response, page_size=j)
return cases_dict, case_id_list
def main(catgory_name):
suite = Suite()
cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
cate_id = [cate_id_dict["data"][catgory_name]]
suite.get_cases_category(cate_id)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main('转向冲突')
返回值
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-case/cases?page=1&filter=%257B%2522keyword%2522%253A%2522%2522%252C%2522categories%2522%253A%255B%25220863fa00-4618-11e9-803b-61333d8d0423%2522%255D%257D&pageSize=100', 'headers': {'Content-Type': 'application/json', 'Authorization': '5022f578-187b-47ca-bed8-8d452c8738c5'}}
response:{"code":0,"data":{"total":1,"caseDefs":[{"thumbnail":null,"data":{"schema":"casedef","vehicles":[{"id":"simonedriver","name":"SimOneDriver控制","classId":"MKZ","physicalGradeSensor":0}],"id":"25834d95-d7c2-4f5f-af0e-5d8937e35064","opponent":"openscenario","userId":"18392614332","mapId":"Junction5","categoryId":"0863fa00-4618-11e9-803b-61333d8d0423","name":"转向冲突8","tags":["十字路口","机动车","转向"],"lastModified":1642420795018,"selected":false,"builtIn":true,"created":1642420795018,"version":{"major":3,"minor":0,"build":0,"patch":2},"createUsername":"standalone","createUserRole":0,"revision":{"id":0,"userId":"standalone","timestamp":1608881493476},"updateUsername":"standalone","notes":"本车转向，与前车发生冲突","joinway":"import","thumbnail":"","mapName":"五车道路口"},"id_":97,"updateUserId":"18392614332","created":1642420795134,"builtIn":true,"terminal":null,"userId":"18392614332","tags":["转向","机动车","十字路口"],"total":1,"invalidedFlag":"","deleteTime":0,"opponent":"openscenario","name":"转向冲突8","editRevision":"5712fa29-188d-4653-81d2-b5e56baaeb45","mapId":"Junction5","id":"25834d95-d7c2-4f5f-af0e-5d8937e35064","lastModified":1642420795134,"vehicleId":["simonedriver"],"categoryId":"0863fa00-4618-11e9-803b-61333d8d0423"}],"page":1}}
6.4 获取案例库中子集目录案例ID
API
"""
def get_category_api(self):
"""
获取案例库列表
@return: response
"""
return self.send(self.get, categories_url).json()
"""
功能实现
def get_all_cate_id_list(self, cate_id) -> list:
"""
@param cate_id: 案例库ID
@return: cate_id_list
"""
all_cate_list = [cate_id]
response = self.get_category_api()
def get_cate_id(current_cate_id):
categories = response["data"]["categories"]
for category in categories:
if current_cate_id == category["parentId"] and category["id"] not in all_cate_list:
all_cate_list.append(category["id"])
get_cate_id(category["id"])
get_cate_id(cate_id)
print("all_cate_list:", all_cate_list)
return all_cate_list
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
# main("new", "test_4")
test = Suite()
test.get_all_cate_id_list("dbaf8d6e-af07-44a6-b7c7-d8b7f2fd6616")
返回值
all_cate_list: ['dbaf8d6e-af07-44a6-b7c7-d8b7f2fd6616', 'c8557561-5729-43e1-9d65-91498f7a1129']
7. 案例相关
7.1 获取案例信息
API
"""
def get_case_detail_api(self, case_id:str):
"""
获取案例信息
@param params: /{case_id}/data
@return: response
"""
url = self.case_details_url.format(case_id=case_id)
return self.send(self.get, url).json()
"""
功能实现
def get_case_detail_api(self, case_id: str):
"""
获取案例信息
@param : /{case_id}/data
@return: response
"""
url = case_details_url.format(case_id=case_id)
return self.send(self.get, url).json()
def get_case_detail(self, case_id: str):
"""
@param case_id: 测试案例id
@return:
"""
response = self.get_case_detail_api(case_id)
return response
def main(catgory_name):
suite = Suite()
cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
cate_id = [cate_id_dict["data"][catgory_name]]
# suite.get_cases_category(cate_id)
cases_dict, case_id_list = suite.get_cases_category(cate_id)
suite.get_case_detail(case_id_list[0])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main('转向冲突')
返回值
{'code': 0, 'data': {'schema': 'casedata', 'caseId': '25834d95-d7c2-4f5f-af0e-5d8937e35064', 'sensors': {'byId': {}, 'allIds': []}, 'judgements': {'byId': {'timeout': {'id': 'timeout', 'name': '超时', 'type': 'timeout', 'enabled': True, 'scope': {'type': 'global', 'position': {'x': 0, 'y': 0, 'z': 0}, 'heading': {'x': 0, 'y': 0, 'z': 0, 'w': 1}, 'size': {'x': 10, 'y': 10, 'z': 0}}, 'conditions': [{'variable': 'timeout', 'value': 600}], 'settings': {'action': 'failure', 'logLevel': 'error', 'logInfo': ''}, 'builtIn': True, 'lock': True, 'userId': 'admin', 'schema': 'judgement'}, 'collision': {'id': 'collision', }..........}
7.2 案例导入
API
"""
def import_case_api(self, payload: dict):
"""
案例导入
@param payload:
@return:
"""
return self.send(self.post, case_import_url, data=json.dumps(payload)).json()
def file_md5_check(self, file_path: str, raw: bool = False):
"""
文件md5检查
@param file_path: 文件路径
@return:
"""
# global url
def file_md5(filename: str):
"""
以文件内容md5加密
@param filename:
@return:
"""
with open(filename, 'rb') as f:
contents = f.read()
return hashlib.md5(contents).hexdigest()
file_md5 = file_md5(file_path)
# print("file_md5:", file_md5)
if raw:
url = file_md5_url.format(file_md5, "true")
else:
url = file_md5_url.format(file_md5, "false")
# print("file_md5:",file_md5)
# print(self.send(self.get, url).json())
return self.send(self.get, url).json()
def upload_file(self, file_path: str, index: int, chunks: int, category: str, file_md5: str, raw: bool = False,
uuid: str = Faker('zh_CN').uuid4()):
"""
上传文件
@param file_path: 文件路径
@param index: 上传序列
@param chunks: chunks总数
@param category: 类别
@param file_md5: 文件MD5
@return:
"""
global url
self.headers = {}
file_size = os.path.getsize(file_path)
file_name = file_path.split(os.sep)[-1]
files = {'file': open(os.path.join(file_path), 'rb')}
if not file_md5:
def file_md5(filename: str):
"""
以文件内容md5加密
@param filename:
@return:
"""
with open(filename, 'rb') as f:
contents = f.read()
return hashlib.md5(contents).hexdigest()
file_md5 = file_md5(file_path)
if raw:
url = upload_file_url.format(file_md5, "true")
else:
url = upload_file_url.format(file_md5, "false")
upload_payload = {"params": json.dumps({"filename": file_name,
"index": index,
"chunks": chunks,
"filesize": file_size,
"uploadId": uuid,
"category": category})}
response = self.send(self.post, url, data=upload_payload, files=files).json()
# print(f"-----upload_file-- url:{url} upload_payload:{upload_payload} files:{files} response:{response}")
print("uuid：", uuid)
return uuid, response
"""
功能实现
def file_md5_check(self, file_path: str, raw: bool = False):
"""
文件md5检查
@param file_path: 文件路径
@return:
"""
# global url
def file_md5(filename: str):
"""
以文件内容md5加密
@param filename:
@return:
"""
with open(filename, 'rb') as f:
contents = f.read()
return hashlib.md5(contents).hexdigest()
file_md5 = file_md5(file_path)
# print("file_md5_check file_md5:", file_md5)
# file_md5_url = "api-asset/files/{}?raw={}" # 验证文件MD5
if raw:
url = file_md5_url.format(file_md5, "true")
else:
url = file_md5_url.format(file_md5, "false")
response = self.send(self.get, url).json()
# print(f"-----file_md5_check：{url} get response:",response)
return response
def remove_file(self, sim_path: str):
"""
remove files that except (.sim .zip .xosc)
:param sim_path:
"""
for item in os.listdir(sim_path):
sim_file = os.path.join(sim_path, item)
try:
if os.path.isfile(sim_file):
os.remove(sim_file)
# print(f'{sim_file} 文件删除成功')
elif os.path.isdir(sim_file):
shutil.rmtree(sim_file)
# print(f'{sim_file} 文件夹删除成功')
except OSError as e:
print(f'Error: {e.filename} - {e.strerror}.')
def get_caseid(self, sim_path: str, sim_list: list):
"""
get the caseid before import according to
:param sim_path: string, the path to the .sim file
:sim_list run_sim: list, cases that need to be run
:return: list, caseid before import
"""
# print("sim_path:",sim_path)
# print("sim_list:",sim_list)
file_list = os.listdir(sim_path)
# print("file_list:",file_list)
target_run = list()
for item in file_list:
for run_sim_item in sim_list:
# if ('31map.sim' in item) and ('all_sim' in choose_run):
if os.path.join(sim_path, run_sim_item + '.sim') == os.path.join(sim_path, item):
target_run.append(item)
# print("target_run:",target_run)
if not target_run:
target_run = file_list
# print("更新后target_run：",target_run)
extract_dir = os.path.join(sim_path, "extract_dir")
# print("extract_dir:",extract_dir)
if not os.path.exists(extract_dir):
os.makedirs(extract_dir)
for item in target_run:
if item == "extract_dir":
continue
# zip_file = os.path.join(sim_path, item).replace('.sim', '.zip') # 当文件名中有两个.sim时会有BUG
zip_file = os.path.join(extract_dir, item[:-4] + ".zip") if item.endswith(".sim") else os.path.join(
extract_dir, item)
# print("zip_file:",zip_file)
try:
if item.endswith(".sim") or item.endswith(".zip"):
# print("try item:",item)
shutil.copy(os.path.join(sim_path, item), zip_file)
except Exception as e:
pass
try:
with zipfile.ZipFile(zip_file) as zfile:
# print("执行到这。。。")
zfile.extractall(path=extract_dir)
# print("执行到这2。。。")
except Exception as e:
raise Exception(f"解压{zip_file}失败，异常信息：{e}")
temp = list(filter(lambda x: 'json' in x, os.listdir(extract_dir)))
# print("temp:",temp)
if temp == []: # 从3.3.0版本起案例解压后是文件夹，则上一条语句执行后temp为空
for d in os.listdir(extract_dir):
# print("extract_dir目录下的内容：",d)
if os.path.isdir(os.path.join(extract_dir, d)):
temp.append(d)
# print("临时测试temp:",temp)
all_case_id = list(set(temp)) # 去重
# print("all_case_id:",all_case_id)
old_case_id = list(filter(lambda x: 'vehicle' not in x, all_case_id))
# print("old_case_id:",old_case_id)
self.remove_file(extract_dir)
# print("执行到这3。。。")
return old_case_id
all_case_id = list(map(lambda x: x.replace('.json', ''), temp))
all_case_id = list(set(all_case_id)) # 去重
old_case_id = list(filter(lambda x: 'vehicles' not in x, all_case_id))
self.remove_file(extract_dir)
return old_case_id
def upload_file(self, file_path: str, index: int, chunks: int, category: str, file_md5: str, raw: bool = False,
uuid: str = Faker('zh_CN').uuid4()):
"""
上传文件
@param file_path: 文件路径
@param index: 上传序列
@param chunks: chunks总数
@param category: 类别
@param file_md5: 文件MD5
@return:
"""
global url
# self.headers = {}
self.headers = {"ProjectId": 'default'} # 江淮3.5.0
file_size = os.path.getsize(file_path)
file_name = file_path.split(os.sep)[-1]
files = {'file': open(os.path.join(file_path), 'rb')}
if not file_md5:
def file_md5(filename: str):
"""
以文件内容md5加密
@param filename:
@return:
"""
with open(filename, 'rb') as f:
contents = f.read()
return hashlib.md5(contents).hexdigest()
file_md5 = file_md5(file_path)
if raw:
url = upload_file_url.format(file_md5, "true")
else:
url = upload_file_url.format(file_md5, "false")
upload_payload = {"params": json.dumps({"filename": file_name,
"index": index,
"chunks": chunks,
"filesize": file_size,
"uploadId": uuid,
"category": category})}
response = self.send(self.post, url, data=upload_payload, files=files).json()
# print(f"-----upload_file-- url:{url} upload_payload:{upload_payload} files:{files} response:{response}")
print("uuid：", uuid)
return uuid, response
def import_case_api(self, payload: dict):
"""
案例导入
@param payload:
@return:
"""
# print("********* import_case_api payload********",payload)
self.headers.update({"Content-Type": "application/json;charset=UTF-8"})
return self.send(self.post, case_import_url, data=json.dumps(payload)).json()
def import_case(self, file_path: str, file_name: str, cate_id: str):
"""
导入案例到指定的案例库目录
@param file_path: 案例文件路径
@param file_name: 案例文件名
@param cate_id: 案例库目录id
@return: [重要] 此处返回的并不是导入成功后的案例ID，而是导入案例文件里的原ID
"""
case_file_path = os.path.join(file_path, file_name)
filename = file_name.split(".")[0]
file_size = os.path.getsize(case_file_path)
md5_res = self.file_md5_check(case_file_path)
md5_res_info = md5_res["data"]
file_md5 = md5_res_info["id"]
global ext
global path_ext
if file_name.split(".")[1] == "xosc":
ext = "xosc"
case_id_list = [f"{filename}"]
includes = case_id_list
overrides_id = case_id_list[0]
overrides_name = filename
overrides_path = filename + ".xosc"
elif file_name.split(".")[1] == "sim":
ext = "sim"
case_id_list = self.get_caseid(file_path, [filename])
includes = case_id_list
overrides_id = case_id_list[0]
overrides_name = filename
overrides_path = case_id_list[0] + "/case/" + filename + ".json"
elif file_name.split(".")[1] == "zip":
ext = "zip"
case_id_list = self.get_caseid(file_path, [filename])
print("case_id_list:", case_id_list)
includes = case_id_list[0] + "/case/" + filename
overrides_id = includes
overrides_name = includes
overrides_path = includes + ".xosc"
else:
raise f"导入案例不支持{file_name.split('.')[1]}格式,用例中断"
case_payload = {"category": cate_id,
"ext": ext,
"file": file_md5,
"id": "2222222222",
"includes": includes,
"includesMap": True,
"includesStopTrigger": True,
"includesVehicle": True,
"name": filename,
# "overrides": [],
"overrides": [{
"id": overrides_id,
"missingMap": False,
"missingVehicle": False,
"name": overrides_name,
"path": overrides_path,
"replaceMap": False,
"replaceVehicle": True
}],
"categoryInfo": {},
"needRemoveCases": {}
}
def round_up(data):
"""
func:向上取整
Args:
data: float
Returns: int
"""
return math.ceil(data)
chunk_size = 2 * 1024 * 1024
if md5_res_info['size'] == 0:
chunks = round_up(file_size / chunk_size)
for i in range(chunks):
self.upload_file(file_path=case_file_path, index=i, chunks=chunks, category="case", file_md5=file_md5)
response = self.import_case_api(payload=case_payload)
else:
response = self.import_case_api(payload=case_payload)
# print("临时测试 import_case case_payload：", case_payload)
# print("-----import_case_api response -->>> ", response)
return case_id_list, response # [重要] 此处返回的并不是导入成功后的案例ID，而是导入案例文件里的原ID
def import_flow(self, category_name: str, file_path: str, file_name: str):
'''
导入文件流程
可导入的文件类型为 .xosc 或者 .sim格式
@param category_name: 案例库名称
@param file_path: 文件路径比如 r"D:\System
@param file_name:文件名称比如 123.xosc or 123.sim
'''
cate_id = self.get_category()
cate_idname = cate_id["data"][category_name]
self.import_case(file_path, file_name, cate_idname)
def main():
suite = Suite()
# cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
# cate_id = [cate_id_dict["data"][catgory_name]]
# suite.get_cases_category(cate_id)
# cases_dict, case_id_list = suite.get_cases_category(cate_id)
# suite.get_case_detail(case_id_list[0])
suite.import_flow("new",r"C:\Users\Administrator\Desktop\test", "test.xosc")
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"data":{"id":"2222222222"}}
7.3 案例导出
API
"""
def export_case_api(self, payload: dict) -> json:
"""
导出测试案例
@param payload:dict
@return:response
"""
return self.send(self.post, export_case_url, data=json.dumps(payload)).json()
def download_case_api(self, pake_id: str) -> bytes:
"""
下载测试案例
@param pake_id: pake_id
@return:response
"""
return self.send(self.get, download_case_url.format(pake_id=pake_id)).content
"""
功能实现
def export_case_api(self, payload: dict) -> json:
"""
导出测试案例
@param payload:dict
@return:response
"""
return self.send(self.post, export_case_url, data=json.dumps(payload)).json()
def download_case_api(self, pake_id: str) -> bytes:
"""
下载测试案例
@param pake_id: pake_id
@return:response
"""
return self.send(self.get, download_case_url.format(pake_id=pake_id)).content
def download_case(self, case_id: list) -> bytes:
"""
获取二进制测试案例内容
@param case_id: case_id_list
@return: bytes
"""
# 取pake_id
pake_id = self.export_case(case_id)["data"]["id"]
print(f"download_case_bytes：{self.download_case_api(pake_id=pake_id)}")
return self.download_case_api(pake_id=pake_id)
def export_case(self, case_id: list, _format: str = ".sim", vehicle: bool = True, includesMap: bool = True,
includesObstacle: bool = True):
"""
@param case_id: case_id_list
@param _format: sim or xosc
@param vehicle: True or False
@return: response
"""
if _format == ".sim" or _format == ".xosc":
payload = {
"caseIds": case_id,
"includesVehicle": vehicle,
"includesMap": includesMap, # 3.5.0
"includesObstacle": includesObstacle,
"format": _format,
"pathinfo": [{"id": caseId, "parentId": None} for caseId in case_id]
}
response = self.export_case_api(payload)
return response
else:
raise Exception(f"-------------input format={_format} error----------------")
def get_all_cate_id_list(self, cate_id) -> list:
"""
@param cate_name_list: 案例名称列表
@return: cate_id_list
"""
all_cate_list = [cate_id]
response = self.get_category_api()
def get_cate_id(current_cate_id):
categories = response["data"]["categories"]
for category in categories:
if current_cate_id == category["parentId"] and category["id"] not in all_cate_list:
all_cate_list.append(category["id"])
get_cate_id(category["id"])
get_cate_id(cate_id)
all_cate_list = [j for i in all_cate_list for j in i]
print("all_cate_list:", all_cate_list)
return all_cate_list
def download_case_flow(self, category_name: str, downloadPath: str, downloadName: str) -> None:
"""
@param cate_name: 案例库名称
@param downloadPath: 下载路径
@param downloadName: 下载包的名称
@return: None
"""
cate_id_dict = self.get_category()
cate_id_str = [cate_id_dict["data"][category_name]]
# cate_method=CategoryBusiness()
cate_id_list = self.get_all_cate_id_list(cate_id_str)
case_id_list = self.get_cases_category(cate_id_list)[1]
downloadResponse = self.download_case(case_id_list)
# 取 format 的实际参数
_format = inspect.getfullargspec(self.export_case)[3][0]
# 下载的文件路径
downloadPake = os.path.join(downloadPath, downloadName + _format)
with open(downloadPake, 'wb') as f:
f.write(downloadResponse)
print(f"导出的案例路径为：{downloadPake}")
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538","test_4-copy")
返回值
run_task response：{'code': 0, 'data': {'sessionId': '3f90ef1b-f5e1-4ab7-b24f-7f94b9d7f1cb', 'taskIds': ['d2973d5b-7fef-4d83-af5d-ec108db815df']}}
7.4 案例场景编辑
API
"""
def case_def_api(self, case_id: str):
"""
案例概要
@param case_id:
@return:
"""
return self.send(self.get, case_def_url.format(case_id=case_id)).json()
def case_data_api(self, case_id: str):
"""
案例详情
@param case_id:
@return:
"""
return self.send(self.get, case_details_url.format(case_id=case_id)).json()
def update_case_api(self, payload: dict):
"""
案例编辑
@param payload:
@return:
"""
self.headers.update({"Content-Type": "application/json"})
return self.send(self.post, update_case_url, data=json.dumps(payload)).json()
def get_vehicles_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get, get_vehicles_url).json()
"""
功能实现
def case_def_api(self, case_id: str):
"""
案例概要
@param case_id:
@return:
"""
return self.send(self.get, case_def_url.format(case_id=case_id)).json()
def case_data_api(self, case_id: str):
"""
案例详情
@param case_id:
@return:
"""
return self.send(self.get, case_details_url.format(case_id=case_id)).json()
def update_case_api(self, payload: dict):
"""
案例编辑
@param payload:
@return:
"""
self.headers.update({"Content-Type": "application/json"})
return self.send(self.post, update_case_url, data=json.dumps(payload)).json()
def get_vehicles_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get, get_vehicles_url).json()
def get_vehicle_name(self, vehicle_id: str):
res = self.get_vehicles_api()
print(f"get_vehicle_api response: {res}")
data = res["data"]["list"]["byId"]
for key in data:
if key == vehicle_id:
print(f"找到主车ID为{vehicle_id}的主车名称：{data[key]['name']}")
return data[key]["name"]
print(f"没有找到主车ID为{vehicle_id}的主车")
return None
def update_case(self, case_id: str,
map_name: str = None,
map_id: str = None,
vehicle_id: str = None,
main_vehicle_speed: int = None):
"""
编辑案例
@param case_id: 案例ID
@param map_name: 地图名称
@param map_id: 地图ID
@param vehicle_id: 主车ID
@param main_vehicle_speed: 主车速度
@return: 更新主车或地图后的案例
"""
case_def = self.case_def_api(case_id=case_id)
case_data = self.case_data_api(case_id=case_id)
print("临时测试 原case_def：", case_def)
print("临时测试 原case_data：", case_data)
print("update_case case_id:", case_id)
print("update_case map_name:", map_name)
print("update_case map_id:", map_id)
print("update_case vehicle_id:", vehicle_id)
if vehicle_id:
vehicle_name = self.get_vehicle_name(vehicle_id)
case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"][
"name"] = vehicle_name # 3.5.0
case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"]["Properties"][
"Property"][0]["value"] = vehicle_id # 3.4.0 没有加这一行 此行待确认是否可以删除
for key in case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"]["Properties"][
"Property"]:
if key["name"] == 'model':
key["value"] = vehicle_id
print(key)
if key["name"] == 'name':
key["value"] = vehicle_name
print(key)
if map_id:
case_def["data"]["data"]["mapId"] = map_id
case_def["data"]["data"]["mapName"] = map_name.split(".")[0]
case_data["data"]["openSCENARIO"]["RoadNetwork"]["LogicFile"]["filepath"] = map_id + "." + \
map_name.split(".")[1]
case_payload = {"casedef": case_def["data"]["data"],
"casedata": case_data["data"],
"needRefresh": True}
if main_vehicle_speed:
privates = case_data["data"]["openSCENARIO"]['Storyboard']['Init']['Actions']['Private']
for private in privates:
if private["entityRef"] == "Ego":
for PrivateAction in private["PrivateAction"]:
if PrivateAction.get('LongitudinalAction'):
PrivateAction['LongitudinalAction']['SpeedAction']['SpeedActionTarget'][
'AbsoluteTargetSpeed']['value'] = main_vehicle_speed
response = self.update_case_api(payload=case_payload)
# print("临时测试 update_case response",response)
if response['code'] != 0:
raise Exception("编辑案例异常，接口返回：" + response)
return response
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
test = Suite()
test.update_case(case_id,map_name,map_id,vehicle_id,main_vehicle_speed)
返回值
update_case_api response {'code': 0, 'msg': 'ok', 'data': {'code': 0, 'msg': 'ok'}}
7.5 保存回放案例
API
"""
def convert_replay_case_api(self, task_id: str, category_id: str, replay_case_name: str):
"""
转换回放案例
@param case_id: 需要转转换的案例ID
@param category_id: 存放案例库目录ID
@param replay_case_name: 转换后的回放案例名称
@return:
"""
payload = {
"name": replay_case_name,
"categoryId": category_id,
"opponent": "record",
"ego2npc": False,
"notes": "",
"tags": ""
}
return self.send(self.get, convert_replay_case_url.format(id=task_id), data=json.dumps(payload))
"""
功能实现
def convert_replay_case_api(self, task_id: str, category_id: str, replay_case_name: str):
"""
转换回放案例
@param case_id: 需要转转换的案例ID
@param category_id: 存放案例库目录ID
@param replay_case_name: 转换后的回放案例名称
@return:
"""
payload = {
"name": replay_case_name,
"categoryId": category_id,
"opponent": "record",
"ego2npc": False,
"notes": "",
"tags": ""
}
return self.send(self.get, convert_replay_case_url.format(id=task_id), data=json.dumps(payload))
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
suite = Suite()
suite.convert_replay_case_api(task_id="4c3f93ff-3f4a-4bea-ac04-507759fed4d2",
category_id="ab21e34c-9df8-433f-89e7-8a66ec42529d", replay_case_name="TestCase")
返回值
{code:0,success:True}
8. 运行任务相关
8.1 运行任务
API
"""
def create_task_api(self,payload:dict):
"""
创建task
@param payload: eg {"caseIds": ["3dd03c09-3ba9-4f7a-b988-fde88296095a"],
"taskName": "task_name"}
@return:
"""
return self.send(self.post,self.create_task_url,json=payload).json()
"""
功能实现
注意: 此功能 3.4 以前包括3.4的版本 和 3.5 以后的版本有所区分请注意
def create_task_api(self,payload:dict):
"""
创建task
@param payload: eg {"caseIds": ["3dd03c09-3ba9-4f7a-b988-fde88296095a"],
"taskName": "task_name"}
@return:
"""
return self.send(self.post,create_task_url,json=payload).json()
def run_task(self, caseid: list, vehicle_id: str = None, withEvaluation: bool = False, version: float = 3.5):
"""
运行测试任务集合
Run task set
@param caseid: caseid
@param vehicle_id: vehicle_id
@param withEvaluation: withEvaluation
@param version: version
@return:
"""
task_name = "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
if version >= 3.5:
payload = {"caseIds": caseid,
"taskName": task_name,
"withEvaluation": withEvaluation,
"controllers": [{"id": "Default", "name": "默认控制器", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器", "algorithmId": "SimOneDriver"}]}
else:
payload = {"caseIds": caseid, "taskName": task_name, "speed": 1}
if vehicle_id:
payload.update({"overrideVehicleId": vehicle_id})
response = self.send(self.post, url=create_task_url, json=payload).json()
data = response['data']
return data['sessionId'], data['taskIds']
def main(catgory_name,case_name):
suite = Suite()
cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
cate_id = [cate_id_dict["data"][catgory_name]]
suite.get_cases_category(cate_id)
cases_dict, case_id_list = suite.get_cases_category(cate_id)
suite.get_case_detail(case_id_list[0])
# suite.import_flow("new",r"C:\Users\Administrator\Desktop\test", "test.xosc")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
# print(f" case_name:{case_name}")
suite.run_task(caseid=case_id_list)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538","test_4-copy")
返回值
run_task response：{'code': 0, 'data': {'sessionId': '3f90ef1b-f5e1-4ab7-b24f-7f94b9d7f1cb', 'taskIds': ['d2973d5b-7fef-4d83-af5d-ec108db815df']}}
8.2 结束任务
API
"""
def stop_task_api(self, payload: dict):
"""
task停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_task_url, json=payload).json()
def stop_sessions_api(self, payload: dict):
"""
sessions停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_sessions_url, json=payload).json()
"""
功能实现
def stop_task_api(self, payload: dict):
"""
task停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_task_url, json=payload).json()
def stop_sessions_api(self, payload: dict):
"""
sessions停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_sessions_url, json=payload).json()
def stop_task(self, task_id: list, sessions_id: list):
"""
停止任务和会话
@param task_id:
@param sessions_id:
@return:
"""
task_payload = {"ids": task_id}
self.stop_task_api(task_payload)
sessions_id = {"ids": sessions_id}
self.stop_sessions_api(sessions_id)
def main(catgory_name, case_name):
suite = Suite()
cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
cate_id = [cate_id_dict["data"][catgory_name]]
suite.get_cases_category(cate_id)
cases_dict, case_id_list = suite.get_cases_category(cate_id)
suite.get_case_detail(case_id_list[0])
# suite.import_flow("new",r"C:\Users\Administrator\Desktop\test", "test.xosc")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
# print(f" case_name:{case_name}")
session_id,task_ids = suite.run_task(caseid=case_id_list)
time.sleep(10)
suite.stop_task(task_ids, [session_id])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538", "test_4-copy")
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/tasks/stop', 'headers': {'Content-Type': 'application/json', 'Authorization': 'ae04e5b4-6a7a-4ad4-b476-33f946e717d5'}, 'json': {'ids': ['8d18f2f6-fd8b-48ae-86a0-0a3bbee1763c']}}
response:{"code":0,"data":{}}
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/sessions/stop', 'headers': {'Content-Type': 'application/json', 'Authorization': 'ae04e5b4-6a7a-4ad4-b476-33f946e717d5'}, 'json': {'ids': ['fc25f9de-7a21-4f0d-8906-a830b7e799ab']}}
response:{"code":0,"data":{}}
8.3 暂停任务&重新开始
API
"""
def pause_task_api(self,task_id:str):
"""
暂停任务
@param task_id:
@return:
"""
return self.send(self.post, self.pause_task_url.format(task_id=task_id),json={}).json()
def resume_task_api(self,task_id:str):
"""
暂停任务后开始
@param task_id:
@return:
"""
return self.send(self.post, self.resume_task_url.format(task_id=task_id),json={}).json()
"""
功能实现
def pause_task_api(self,task_id:str):
"""
暂停任务
@param task_id:
@return:
"""
return self.send(self.post, pause_task_url.format(task_id=task_id),json={}).json()
def resume_task_api(self,task_id:str):
"""
暂停任务后开始
@param task_id:
@return:
"""
return self.send(self.post, resume_task_url.format(task_id=task_id),json={}).json()
def main(catgory_name, case_name):
suite = Suite()
cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
cate_id = [cate_id_dict["data"][catgory_name]]
suite.get_cases_category(cate_id)
cases_dict, case_id_list = suite.get_cases_category(cate_id)
suite.get_case_detail(case_id_list[0])
# suite.import_flow("new",r"C:\Users\Administrator\Desktop\test", "test.xosc")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
# print(f" case_name:{case_name}")
session_id ,task_ids = suite.run_task(caseid=case_id_list)
time.sleep(5)
suite.stop_task(task_ids, [session_id])
for i in task_ids:
suite.pause_task_api("c686a2de-47fa-4604-b354-36f8fe2b9d75")
time.sleep(5)
for i in task_ids:
suite.resume_task_api("c686a2de-47fa-4604-b354-36f8fe2b9d75")
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538", "test_4-copy")
返回值
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-case/cases/f5bbfb2c-619a-4812-90c2-75a84247a4db/data', 'headers': {'Content-Type': 'application/json', 'Authorization': 'beff0aaa-a060-4e9e-9f54-58362549b3f0'}}
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/sessions/c686a2de-47fa-4604-b354-36f8fe2b9d75/resume', 'headers': {'Content-Type': 'application/json', 'Authorization': 'beff0aaa-a060-4e9e-9f54-58362549b3f0'}, 'json': {}}
response:{"code":0,"data":{}}
8.4 删除任务
API
"""
def delete_sessions_api(self, payload: dict):
"""
sessions删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_sessions_url, json=payload).json()
def delete_task_api(self, payload: dict):
"""
task删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_task_url, json=payload).json()
"""
功能实现
def delete_sessions_api(self, payload: dict):
"""
sessions删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_sessions_url, json=payload).json()
def delete_task_api(self, payload: dict):
"""
task删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_task_url, json=payload).json()
def delete_task(self, sessions_id: list, task_id: list):
"""
删除任务和会话
@param sessions_id:
@param task_id:
@return:
"""
sessions_id = {"ids": sessions_id}
self.delete_sessions_api(sessions_id)
task_payload = {"ids": task_id}
self.delete_task_api(task_payload)
def main(catgory_name, case_name):
suite = Suite()
cate_id_dict = suite.get_category()
# 案例库名称可以自己指定
cate_id = [cate_id_dict["data"][catgory_name]]
suite.get_cases_category(cate_id)
cases_dict, case_id_list = suite.get_cases_category(cate_id)
suite.get_case_detail(case_id_list[0])
# suite.import_flow("new",r"C:\Users\Administrator\Desktop\test", "test.xosc")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
# print(f" case_name:{case_name}")
session_id ,task_ids = suite.run_task(caseid=case_id_list)
time.sleep(5)
# suite.stop_task(task_ids, [session_id])
suite.delete_task([session_id],task_ids)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538", "test_4-copy")
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/sessions/delete', 'headers': {'Content-Type': 'application/json', 'Authorization': '5a3662a3-2fbb-4790-b40d-42f600e63801'}, 'json': {'ids': ['bc344bf1-4bd3-493e-a4d3-95f98dcc3e48']}}
response:{"code":0,"data":{}}
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/tasks/delete', 'headers': {'Content-Type': 'application/json', 'Authorization': '5a3662a3-2fbb-4790-b40d-42f600e63801'}, 'json': {'ids': ['18a0fafe-d1e2-404b-b5e8-b4f3ea4c8e86']}}
response:{"code":0,"msg":"success"}
8.5 判断任务是否正在运行
API
"""
def get_result(self, task_id: list, index: int = None):
"""
:param task_id:
:return:
"""
task_id_str = ",".join(task_id)
# print(task_id_str)
response = self.send(self.get, url=result_url.format(task_id=task_id_str)).json()
data = response["data"]["tasks"]
result = []
for one in data:
result.append({"pass": one["pass"], "is_ended": True if one["key"] == "ended" else False})
if not index:
return result
else:
return result[index]
"""
功能实现
def get_result(self, task_id: list, index: int = None):
"""
:param task_id:
:return:
"""
task_id_str = ",".join(task_id)
# print(task_id_str)
response = self.send(self.get, url=result_url.format(task_id=task_id_str)).json()
data = response["data"]["tasks"]
result = []
for one in data:
result.append({"pass": one["pass"], "is_ended": True if one["key"] == "ended" else False})
if not index:
return result
else:
return result[index]
def is_ended(self, task_id: list, case_name=None):
"""
判断案例是否运行结束
@param task_id: 任务id
@param case_name: 案例名称
@return:
"""
interval = 10
while (1):
result = self.get_result(task_id, -1)
if result["is_ended"]:
print('Case status：end of run')
if case_name:
print(f'Name of the current ending case->{case_name}')
return False
else:
print('Case status：in progress')
if case_name:
print(f'Name of the current running case->{case_name}')
time.sleep(interval)
调用示例
def main(category_name: str, case_name: str = None, vehicle_name=None):
"""
@param category_name: category_name
@param case_name:
@param vehicle_name:vehicle_name
@return:
"""
suite = Suite()
suite.get_category()
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
# 主车控制相关逻辑
if vehicle_name:
vehicle_dict = suite.get_vehicle_id([vehicle_name])[1]
if vehicle_name not in vehicle_dict.keys():
print(f"vehicle_name：{vehicle_name} inexistence")
return
vehicle_id = vehicle_dict[vehicle_name]
else:
vehicle_id = None
print(f"vehicle_name:{vehicle_name},vehicle_id:{vehicle_id}")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
print(f" case_name:{case_name}")
# 启动案例
session_id, task_ids = suite.run_task(case_id_list, vehicle_id=vehicle_id)
time.sleep(10)
# suite.stop_task(task_id=task_ids, sessions_id=[session_id])
# 等待案例运行完毕
# falg = suite.suite_queue_check()
# if falg is True:
# # 获取案例运行结果
# suite.get_task_result(task_ids)
suite.is_ended(task_ids)
SimOneUrl = "http://172.31.9.85:8088/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("new", "test")
返回值
response:{"code":0,"data":{"tasks":[{"parentId":"-1","loading":false,"deleted":null,"deleteTime":null,"createAt":"2024-06-19T06:22:14.000Z","updateAt":"2024-06-19T06:22:23.000Z","id":"39e2f0e7-5315-47da-b4d1-79b4427d0beb","userId":"standalone","caseId":"6b6fcd60-2d1f-11ef-905e-8f06be1a01ef","case":{"schema":"casedef"}}
9. 主车相关
9.1 导入主车
API
"""
def delete_sessions_api(self, payload: dict):
"""
sessions删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_sessions_url, json=payload).json()
def delete_task_api(self, payload: dict):
"""
task删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_task_url, json=payload).json()
"""
功能实现
class File:
"""
文件父类
"""
def __init__(self, file_path: str):
if not exists(file_path):
raise FileNotFoundError
self._file_path = file_path
self._data = None
class FileReader(File):
"""
读取文件
"""
def __init__(self, file_path: str):
super(FileReader, self).__init__(file_path)
def read_byte(self) -> bytes:
with open(self._file_path, 'rb') as f:
return f.read()
def read_str(self) -> str:
with open(self._file_path, 'r', encoding='utf-8') as f:
return f.read()
def read_json(self) -> json:
with open(self._file_path, 'r', encoding='utf-8') as f:
return json.load(f)
def import_vehicle_api(self,payload:dict):
"""
导入主车
@param payload: eg {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
@return:
"""
return self.send(self.post,import_vehicle_url,json=payload).json()
def get_vehicle_data(file_path: str):
vehicle_info = FileReader(file_path).read_json()
vehicle_id = vehicle_info.get("id")
return vehicle_info, vehicle_id
def import_vehicle(self, file_path:str)->str:
"""
导入主车
@param file_path: 主车文件
@return: 主车id
"""
vehicle_data,vehicle_id = self.get_vehicle_data(file_path)
payload = {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
result = self.import_vehicle_api(payload)
return result["data"]["id"]
def main(suite_name:str=None):
suite = Suite()
# suite.get_suite(suite_name)
suite.import_vehicle('c:/test.json')
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/sessions/delete', 'headers': {'Content-Type': 'application/json', 'Authorization': '5a3662a3-2fbb-4790-b40d-42f600e63801'}, 'json': {'ids': ['bc344bf1-4bd3-493e-a4d3-95f98dcc3e48']}}
response:{"code":0,"data":{}}
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/tasks/delete', 'headers': {'Content-Type': 'application/json', 'Authorization': '5a3662a3-2fbb-4790-b40d-42f600e63801'}, 'json': {'ids': ['18a0fafe-d1e2-404b-b5e8-b4f3ea4c8e86']}}
response:{"code":0,"msg":"success"}
9.2 删除主车
API
"""
def delete_vehicles_api(self, vehicles_id: str):
"""
删除主车
@param vehicles_id: 主车id
@return:
"""
url = self.delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
"""
功能实现
def delete_vehicles_api(self, vehicles_id: str):
"""
删除主车
@param vehicles_id: 主车id
@return:
"""
url = delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
def delete_vehicle(self,id:str):
"""
删除主车
@param id: 主车id
@return:
"""
self.delete_vehicles_api(id)
def main(category_name:str,case_name:str=None,vehicle_name:str=None):
suite = Suite()
suite.get_category()
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
print(case_name)
# 主车控制相关逻辑
if vehicle_name:
vehicle_dict = suite.get_vehicle_id([vehicle_name])[1]
if vehicle_name not in vehicle_dict.keys():
print(f"vehicle_name：{vehicle_name} inexistence")
return
vehicle_id = vehicle_dict[vehicle_name]
else:
vehicle_id = None
print(f"vehicle_name:{vehicle_name},vehicle_id:{vehicle_id}")
suite.delete_vehicle(vehicle_id)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("转向冲突", "转向冲突8", "test")
返回值
response:{"code":0,"data":{"id":"cce9e320-0db0-11ef-98b6-f5137907bac2"}}
9.3 获取主车信息
API
"""
def get_vehicle_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get,self.get_vehicle_url).json()
"""
功能实现
def get_vehicle_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get, get_vehicle_url).json()
def get_vehicle(self, userid: str = None):
"""
根据用户信息获取主车信息
@param userid:
@return:
"""
global vehicle_data
vehicle_data = {}
res = self.get_vehicle_api()
# print(f"get_vehicle_api response ->: {res}")
data = res["data"]["list"]["byId"]
id_list = res["data"]["list"]["allIds"]
if userid:
for id in id_list:
if data[id]["userId"] == userid:
vehicle_data.setdefault(data[id]["name"], data[id]["id"])
else:
for id in id_list:
vehicle_data.setdefault(data[id]["userId"], {}).update({data[id]["name"]: data[id]["id"]})
print("获取到的主车信息->：{}".format(str(vehicle_data)))
return vehicle_data
def get_vehicle_id(self, vehicle_name: list) -> list and dict:
"""
根据主车名称获取主车ID
@param vehicle_name:
@return:
"""
vehicle_id_dict = {}
vehicle_id_list = []
vehicle_list = []
for i in vehicle_name:
for k, v in self.get_vehicle().items():
vehicle_list.append(v)
for vehicle_dict in vehicle_list:
# print("vehicle_dict", vehicle_dict)
vehicle_dict_key = dict(vehicle_dict).items()
for vehicle_dict_k, vehicle_dict_v in vehicle_dict_key:
if i in vehicle_dict_k:
vehicle_name_id = vehicle_dict[i]
vehicle_id_list.append(vehicle_name_id)
vehicle_id_dict.setdefault(i, vehicle_name_id)
print("vehicle_id_list->:", vehicle_id_list, "\nvehicle_id_dict->:", vehicle_id_dict)
return vehicle_id_list, vehicle_id_dict
def main(category_name:str,case_name:str=None,vehicle_name:str=None):
suite = Suite()
suite = Suite()
suite.get_category()
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
# 主车控制相关逻辑
if vehicle_name:
vehicle_dict = suite.get_vehicle_id([vehicle_name])[1]
if vehicle_name not in vehicle_dict.keys():
print(f"vehicle_name：{vehicle_name} inexistence")
return
vehicle_id = vehicle_dict[vehicle_name]
else:
vehicle_id = None
print(f"vehicle_name:{vehicle_name},vehicle_id:{vehicle_id}")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
print(f" case_name:{case_name}")
# 启动案例
session_id, task_ids = suite.run_task(case_id_list, vehicle_id=vehicle_id)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("转向冲突", "转向冲突8", "手动控制-默认")
返回值
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-asset/vehicles', 'headers': {'Content-Type': 'application/json', 'Authorization': 'e5033d43-2368-4fd0-9c4b-299c4d3ba8d9'}}
10. 测试集相关
10.1 获取测试集
API
"""
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
"""
运行测试集
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
def get_suite(self,suite_name:str=None):
"""
:return:
"""
response = self.get_suite_api()
data = response['data']
suite_dict = {}
for id in data['allIds']:
suite_dict.setdefault(data['byId'][id]['name'], {}).update({"caseIds": data['byId'][id]['caseIds']})
print("suite_dict",suite_dict)
if not suite_name:
return suite_dict
else:
#print("suite_dict[suite_name]:",suite_dict[suite_name])
return suite_dict[suite_name]
def main(suite_name:str=None):
suite = Suite()
suite.get_suite(suite_name)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
response:{"code":0,"data":{"byId":{"ea6d0670-0db5-11ef-98b6-f5137907bac2":{"schema":"suite","id":"ea6d0670-0db5-11ef-98b6-f5137907bac2","userId":"standalone","parentId":"","name":"test","created":1715226044506,"lastModified":1715226044506,"createUserRole":1,"caseIds":["f5bbfb2c-619a-4812-90c2-75a84247a4db"]}},"allIds":["ea6d0670-0db5-11ef-98b6-f5137907bac2"]}}
suite_dict {'test': {'caseIds': ['f5bbfb2c-619a-4812-90c2-75a84247a4db']}}
10.2 运行测试集
API
"""
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
def run_suite_api(self, payload:dict):
"""
run_suite
@param payload: payload
@return:
"""
return self.send(self.post, url=self.run_suite_url, json=payload).json()
"""
功能实现
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
def get_suite(self,suite_name:str=None):
"""
:return:
"""
response = self.get_suite_api()
data = response['data']
suite_dict = {}
for id in data['allIds']:
suite_dict.setdefault(data['byId'][id]['name'], {}).update({"caseIds": data['byId'][id]['caseIds']})
print("suite_dict",suite_dict)
if not suite_name:
return suite_dict
else:
#print("suite_dict[suite_name]:",suite_dict[suite_name])
return suite_dict[suite_name]
def run_suite_api(self, payload: dict):
"""
run_suite
@param payload: payload
@return:
"""
return self.send(self.post, url=run_suite_url, json=payload).json()
def run_suite(self,suite_name:str,taskName:str =None,vehicle_id:str=None):
if taskName is None:
def creat_task_name():
return "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
taskName = creat_task_name()
caseid_list=self.get_suite(suite_name)['caseIds']
payload = {'caseIds': caseid_list,
'taskName': taskName}
if vehicle_id:
# vehicle_module = VehicleBusiness()
# vehicle_module.get_vehicle()
payload.update({"overrideVehicleId":vehicle_id})
response=self.run_suite_api(payload)
data = response['data']
return data['sessionId'], data['taskIds']
def main(suite_name:str=None):
suite = Suite()
# suite.get_suite(suite_name)
suite.run_suite(suite_name)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/tasks', 'headers': {'Content-Type': 'application/json', 'Authorization': '7d4c315b-dbd1-4f4b-8552-ff66be9346ec'}, 'json': {'caseIds': ['f5bbfb2c-619a-4812-90c2-75a84247a4db'], 'taskName': 'taskName_2024-05-09_11:49:56'}}
response:{"code":0,"data":{"sessionId":"5b4fc9fb-8da4-42f4-8eef-7e4522b85dd4","taskIds":["0b933b59-b4d6-49e8-b656-de52a8561fb6"]}}
10.3 获取运行案例集信息
API
"""
def suite_queue_api(self):
"""
suite_queue
@return:
"""
return self.send(self.get, url=self.queue_status_url).json()
"""
功能实现
def suite_queue_api(self):
"""
suite_queue
@return:
"""
return self.send(self.get, url=queue_status_url).json()
def suite_queue_check(self,n=10):
"""
@param n:cycle index
@return:
"""
if n==1:
print("-----------------------The test case run fail--------------------------------------")
return False
while(1):
try:
assert self.suite_queue_api()
response = self.suite_queue_api()
assert response["code"]==0
except Exception as e:
return -1
queue_info = response["data"]
running,pending,waiting = queue_info["running"],queue_info["pending"],queue_info["waiting"]
if running == 0 and pending == 0 and waiting == 0:
print("-----------------------The test case run finish--------------------------------------")
return True
else:
time.sleep(interval)
def main(suite_name:str=None):
suite = Suite()
# suite.get_suite(suite_name)
suite.run_suite(suite_name)
suite.suite_queue_check()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-task/tasks/queue?own=true', 'headers': {'Content-Type': 'application/json', 'Authorization': 'f083c671-f6fe-4d64-b775-3f0539eca7c9'}}
response:{"code":0,"data":{"total":2,"running":1,"pending":1,"waiting":1,"finished":0,"estimatedTaskDuration":87.22495652173913}}
10.4 获取任务集中的task_id
API
"""
def get_taskassemble_api(self):
'''
获取测试案例集
@param
@return:
'''
return self.send(self.get, get_taskassemble_url).json()
"""
功能实现
def get_taskassemble_api(self):
'''
获取测试案例集
@param
@return:
'''
return self.send(self.get, get_taskassemble_url).json()
def get_task_id_of_task_set(self) -> list:
"""
获取任务集合中所有测试案例的task_id
@return: list
"""
response = self.get_taskassemble_api()
_task_id_list = response["data"]["list"]
index = 0
task_set_id_list = []
for i in _task_id_list:
index += 1
task_set_id_list.append(i["id"])
print("task_set_id_list", task_set_id_list)
return task_set_id_list
def main(suite_name:str=None):
suite = Suite()
suite.run_suite(suite_name)
suite.get_task_id_of_task_set()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
task_set_id_list [id1,id2,...]
11. 地图相关
11.1 导入地图
API
"""
def import_map_api(self, payload: dict, files: dict):
"""
导入地图API
@param payload:传参字典
@param files: 传参文件
@return:
"""
self.headers = {}
return self.send(self.post, import_map_url, data=payload, files=files).json()
"""
功能实现
def import_map_api(self, payload: dict, files: dict):
"""
导入地图API
@param payload:传参字典
@param files: 传参文件
@return:
"""
self.headers = {}
return self.send(self.post, import_map_url, data=payload, files=files).json()
def import_map(self, xodr_path: str, thumbnail_path: str):
"""
导入地图方法
@param xodr_path: 地图文件
@param thumbnail_path: 地图对应的图像
@return:
"""
payload = {"params": json.dumps(
{"category": "customized", "id": "new_" + Faker('zh_CN').uuid4(),
"name": os.path.splitext(os.path.basename(xodr_path))[0], "size": 512,
"ppm": 10, "bgColor": "#dddddd", "reproject": True, "reprojectOrigin": False, "reprojectOriginLat": 0,
"reprojectOriginLng": 0, "tags": [], "notes": "",
"header": {"minX": -210.20535534122396, "minY": -149.68815701999185, "minZ": -1.862645149230957e-9,
"maxX": 237.95535534122394, "maxY": 135.43815701999196, "maxZ": 2.7940070024635385e-9,
"centerX": 97.12480158531203, "centerY": 24.463606820251727, "centerZ": 100,
"localEnuExt": "6378137,0,0;0,1,0;0,0,1;1,0,0"}})}
files = {'xodr': open(xodr_path, 'rb'), "thumbnail": open(thumbnail_path, 'rb')}
result = self.import_map_api(payload, files)
if result["code"] == 0:
print(f"导入地图{xodr_path}成功，id:{result['data']['mapId']}")
else:
print(f"导入地图{xodr_path}失败，返回结果：{result}")
raise f"导入地图{xodr_path}失败，返回结果：{result}"
res_dict = {"code": result["code"], "mapId": result["data"]["mapId"],
"name": os.path.basename(os.path.basename(xodr_path))}
return res_dict
def main():
suite = Suite()
suite.import_map(r"D:\Project\SimOneAutomation\simone_automation\Sysdata\Public\import\Map\xodr\标志牌测试.xodr",
r"D:\Project\SimOneAutomation\simone_automation\Sysdata\Public\import\Map\img\import_map.png"
)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-asset/maps', 'headers': {'Authorization': '92400a5e-0e20-4d34-bfb3-05f9344f4889'}, 'data': {'params': '{"category": "customized", "id": "", "name": "\\u6807\\u5fd7\\u724c\\u6d4b\\u8bd5", "size": 512, "ppm": 10, "bgColor": "#dddddd", "reproject": true, "reprojectOrigin": false, "reprojectOriginLat": 0, "reprojectOriginLng": 0, "tags": [], "notes": "", "header": {"minX": -210.20535534122396, "minY": -149.68815701999185, "minZ": -1.862645149230957e-09, "maxX": 237.95535534122394, "maxY": 135.43815701999196, "maxZ": 2.7940070024635385e-09, "centerX": 97.12480158531203, "centerY": 24.463606820251727, "centerZ": 100, "localEnuExt": "6378137,0,0;0,1,0;0,0,1;1,0,0"}}'}, 'files': {'xodr': <_io.BufferedReader name='D:\\Project\\SimOneAutomation\\simone_automation\\Sysdata\\Public\\import\\Map\\xodr\\标志牌测试.xodr'>, 'thumbnail': <_io.BufferedReader name='D:\\Project\\SimOneAutomation\\simone_automation\\Sysdata\\Public\\import\\Map\\img\\import_map.png'>}}
response:{"code":0,"data":{"mapId":"01ca5bba-d804-4d32-900f-742fa958d47f"}}
import map D:\Project\SimOneAutomation\simone_automation\Sysdata\Public\import\Map\xodr\标志牌测试.xodrsuccess
11.2 获取地图信息
API
"""
def get_map_api(self):
"""
获取地图
@return:
"""
return self.send(self.get, self.get_map_url).json()
"""
功能实现
def get_map_api(self):
"""
获取地图
@return:
"""
return self.send(self.get, self.get_map_url).json()
def get_map_id(self,userid:str=None,map_name:str=None):
"""
按条件获取地图信息，默认获取所有地图
@param userid: 用户id
@param map_name: 地图名字
@return:
"""
global map_data
map_data={}
try:
result = self.get_map_api()
# print(len(result['data']['maps']))
map_info = result['data']['maps']
if userid:
for map in map_info:
if map["userId"] == userid:
map_data.setdefault(map["name"],map["id"])
elif map_name:
for map in map_info:
if map["name"] == map_name:
map_data = {map["name"]: map["id"]}
else:
for map in map_info:
map_data.setdefault(map["userId"],{}).update({map["name"]: map["id"]})
return map_data
except Exception as e:
print(e)
def main():
suite = Suite()
suite.get_map_id()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"data"...}
11.3 删除地图
API
"""
def delete_map_api(self,payload:dict):
"""
删除地图
@param payload: eg：{ids: ["27a0bbc4-5d4d-48c1-9187-62df794dadae"]}
@return:
"""
return self.send(self.post, delete_map_url, json=payload).json()
"""
功能实现
def delete_map_api(self,payload:dict):
"""
删除地图
@param payload: eg：{ids: ["27a0bbc4-5d4d-48c1-9187-62df794dadae"]}
@return:
"""
return self.send(self.post, delete_map_url, json=payload).json()
def delete_map(self,id:list):
"""
批量删除地图
@param id: 地图id列表
@return:
"""
print("删除的地图列表-》{}".format(str(id)))
payload = {"ids": id}
self.delete_map_api(payload)
def main():
suite = Suite()
suite.delete(["'92400a5e-0e20-4d34-bfb3-05f9344f4889'"])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"data"...}
12. 判定相关
12.1 获取案例判定信息
API
"""
def get_judgements_api(self, caseId, judgementId=None):
"""
获取判定信息API
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
if judgementId:
url = get_judgement_url.format(caseId=caseId,judgementId=judgementId)
else:
url = get_judgements_url.format(caseId=caseId)
response = self.send(self.get, url).json()
return response
"""
功能实现
def get_judgements_api(self, caseId, judgementId=None):
"""
获取判定信息API
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
if judgementId:
url = get_judgement_url.format(caseId=caseId,judgementId=judgementId)
else:
url = get_judgements_url.format(caseId=caseId)
response = self.send(self.get, url).json()
return response
def get_judgements_method(self, caseId:str, judgementId:str = None):
"""
获取判定信息
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
response = self.get_judgements_api(caseId,judgementId)
print("得到的扩展判定信息：",response["data"][1])
return response
def main():
suite = Suite()
suite.get_judgements_api("ebd4cf06-b66f-4e03-9294-5f60f55d453a")
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"msg":"success","data":[{"id":"timeout","name":"超时","type":"timeout","category":"general","enabled":true,"scope":{"type":"global","position":{"x":0,"y":0,"z":0},"heading":{"x":0,"y":0,"z":0,"w":1},"size":{"x":10,"y":10,"z":0}},"conditions":[{"variable":"timeout","value":600}],"settings":{"action":"failure","logLevel":"error","logInfo":""},"builtIn":true,"lock":true,"userId":"admin","schema":"judgement"},{"id":"collision","name":"碰撞","type":"collision","category":"general","enabled":true,"scope":{"type":"global","position":{"x":0,"y":0,"z":0},"heading":{"x":0,"y":0,"z":0,"w":1},"size":{"x":10,"y":10,"z":0}},"conditions":[],"settings":{"action":"failure","logLevel":"error","logInfo":""},"builtIn":true,"lock":true,"userId":"admin","schema":"judgement"}]}
12.2 更新判定信息
API
"""
def update_judgement_method(self, caseId, judgementId="collision"):
__collision = {"schema": "judgement", "settings": {"logLevel": "error", "action": "failure", "logInfo": ""},
"scope": {"size": {"x": 10, "y": 10, "z": 0}, "heading": {"w": 1, "x": 0, "y": 0, "z": 0},
"position": {"x": 0, "y": 0, "z": 0},
"type": "global"}, "builtIn": True,
"name": "碰撞", "lock": True, "id": "collision",
"type": "collision", "category": "general",
"conditions": [], "userId": "admin",
"enabled": True}
payload = json.dumps(__collision)
judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
response = self.send(self.put, judgement_url.format(caseId=caseId, judgementId=judgementId), data=payload)
print("更新扩展判定信息")
print("status_code:", response.status_code)
"""
功能实现
def update_judgement_method(self, caseId, judgementId="collision"):
__collision = {"schema": "judgement", "settings": {"logLevel": "error", "action": "failure", "logInfo": ""},
"scope": {"size": {"x": 10, "y": 10, "z": 0}, "heading": {"w": 1, "x": 0, "y": 0, "z": 0},
"position": {"x": 0, "y": 0, "z": 0},
"type": "global"}, "builtIn": True,
"name": "碰撞", "lock": True, "id": "collision",
"type": "collision", "category": "general",
"conditions": [], "userId": "admin",
"enabled": True}
payload = json.dumps(__collision)
judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
response = self.send(self.put, judgement_url.format(caseId=caseId, judgementId=judgementId), data=payload)
print("更新扩展判定信息")
print("status_code:", response.status_code)
def main(category_name: str, case_name: str):
"""
category_name:案例库名称
case_name：案例名称
reset:是否重置判定
"""
LoginPolicy("standalone")
suite = Suite()
suite.get_category()
# 获取caseid逻辑
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
case_id = cases_dict[case_name][0]
suite.get_judgements_method(case_id)
suite.update_judgement_method(case_id)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("new", "test_4")
返回值
更新扩展判定信息
status_code: 202
13. 任务运行结果相关
13.1 获取运行案例集信息
API
"""
def get_task_result_api(self,task_id:str):
"""
获取指定task的结果
@param task_id:
@return:
"""
return self.send(self.get,task_result_url.format(id=task_id)).json()
"""
功能实现
def get_task_result_api(self,task_id:str):
"""
获取指定task的结果
@param task_id:
@return:
"""
return self.send(self.get,task_result_url.format(id=task_id)).json()
def get_task_result(self, task_id_list: list = None) -> list:
"""
获取任务集中的 案例名称，caseId，task_id,运行结果
@param task_id_list: 任务id列表
@param task_set_id: 任务集合id
@return:
"""
task_result_list = []
for task_id in task_id_list:
result = self.get_task_result_api(task_id)
case_name = result["data"]["tasks"][0]["case"]["name"]
case_id = result["data"]["tasks"][0]["caseId"]
task_id = result["data"]["tasks"][0]["id"]
task_result = result["data"]["tasks"][0]["pass"]
if task_result != True: task_result = False
task_result_list.append({"case_name": case_name,
"case_id": case_id,
"task_id": task_id,
"task_result": task_result})
print("task_result_list->", task_result_list)
return task_result_list
def main():
suite = Suite()
# 入参id taskid 可以用 run_task方法中的返回值获取
suite.get_task_result(["51ec8581-4563-49c4-8ce1-ba96962b278b"])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
task_result_list-> [{'case_name': 'test_4-copy', 'case_id': 'f5bbfb2c-619a-4812-90c2-75a84247a4db', 'task_id': '51ec8581-4563-49c4-8ce1-ba96962b278b', 'task_result': False}]
13.2 获取任务集列表信息
API
"""
def get_task_list_api(self):
"""
获取任务列表集合
"""
response = self.send(self.get, url=get_task_list_url).json()
"""
功能实现
def get_task_list_api(self):
"""
获取任务列表集合
"""
response = self.send(self.get, url=get_task_list_url).json()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
test = Suite()
test.get_task_list_api()
返回值
response:{"code":0,"data":{"list":[{"parentId":null,"deleted":null,"deleteTime":null,"createAt":"2024-05-16T02:34:13.000Z","updateAt":"2024-05-16T02:35:08.000Z",....}
13.3 下载测试报告
API
"""
def export_report_api(self, sessions_id: str):
"""
导出测试报告API
@param sessions_id: sessions_id
@return: Response
"""
self.headers.update({"Language": "zh-Hans"})
return self.send(self.get, export_report_url.format(sessions_id=sessions_id)).json()
def download_report_api(self, sessions_id: str):
"""
下载测试报告API
@param sessions_id: sessions_id
@return: Response
"""
return self.send(self.get, download_report_url.format(sessions_id=sessions_id)).content
"""
功能实现
def export_report_api(self, sessions_id: str):
"""
导出测试报告API
@param sessions_id: sessions_id
@return: Response
"""
self.headers.update({"Language": "zh-Hans"})
return self.send(self.get, export_report_url.format(sessions_id=sessions_id)).json()
def download_report_api(self, sessions_id: str):
"""
下载测试报告API
@param sessions_id: sessions_id
@return: Response
"""
return self.send(self.get, download_report_url.format(sessions_id=sessions_id)).content
def download_report(self, sessions_id: str, download_path: str, download_name: str) -> None:
"""
下载测试报告
@param sessions_id: sessions_id
@return: Response
"""
self.export_report_api(sessions_id)
pdf_data = self.download_report_api(sessions_id)
download_pathname = os.path.join(download_path, download_name)
with open(download_pathname, mode="wb") as f:
f.write(pdf_data)
调用示例
LoginPolicy("standalone")
if __name__ == '__main__':
test = Suite()
# 测试任务集ID
test.download_report("54c0806b-ba80-4821-ae25-49f7c284f486", r"C:\Users\Administrator\Desktop\test", "test.pdf")
返回值
response:{"code":0,"data":{"success":true}}
测试报告二进制数据
14. 附录参考代码
import base64
import copy
import hashlib
import math
import random
import shutil
import zipfile
from datetime import datetime
from os.path import exists
from uuid import uuid4
from Crypto.Cipher import AES
from faker import Faker
from requests import request, Response
import json
from urllib.parse import urljoin
import os
import time
fake = Faker('zh_CN')
interval = 10
# header
header = {
"Content-Type": "application/json",
}
# Environment Path
class EnvPath:
token = "TOKEN"
class Prefix:
UserServer = "api-user"
CaseServer = "api-case"
AssetServer = "api-asset"
TaskServer = "api-task"
# token
_token = os.environ.get(EnvPath.token)
""" 登录相关接口 """
user_check_url = Prefix.UserServer + "/users/check"
health = Prefix.UserServer + "/health" # 检查服务是否正常
cloud_login_url = Prefix.UserServer + "/users/login" # 云端登录
standalone_login_url = Prefix.UserServer + "/users/standalone" # 单机版登录
login_check = Prefix.UserServer + "/users/check" # user check
""" 案例库目录相关接口 """
categories_url = Prefix.CaseServer + "/categories" # 案例库目录
get_categories_url = Prefix.CaseServer + "/categories?withCases=false" # 获取所有案例库
get_cases_url = "api-case/cases?page=1&filter={category}&pageSize=100"
get_cases_url = Prefix.CaseServer + "/cases?page={page}&filter={category}&pageSize=100" # 获取当前案例库的所有案例
delete_categories_url = Prefix.CaseServer + "/categories/{}" # 删除案例库目录
""" 上传文件相关接口 """
file_md5_url = Prefix.AssetServer + "/files/{}?raw={}" # 验证文件MD5
upload_file_url = Prefix.AssetServer + "/files/{}/upload?raw={}" # 上传文件
""" 测试案例相关接口 """
case_import_url = Prefix.AssetServer + "/cases/import" # 导入案例
case_details_url = Prefix.CaseServer + "/cases/{case_id}/data" # 案例详情
delete_case_url = Prefix.CaseServer + "/trash" # 删除案例
convert_replay_case_url = Prefix.TaskServer + "/tasks/{id}/convert2case" # 转换回放案例
""" 测试任务相关接口 """
create_task_url = Prefix.TaskServer + "/tasks" # 创建task
stop_task_url = Prefix.TaskServer + "/tasks/stop" # 停止task
stop_sessions_url = Prefix.TaskServer + "/sessions/stop" # 会话停止
delete_task_url = Prefix.TaskServer + "/tasks/delete" # 删除task
delete_sessions_url = Prefix.TaskServer + "/sessions/delete" # 删除停止
result_url = Prefix.TaskServer + "/tasks/task?taskIds={task_id}" # 删除停止 案例运行结果详情
pause_task_url = Prefix.TaskServer + "/sessions/{task_id}/pause" # 任务暂停
resume_task_url = Prefix.TaskServer + "/sessions/{task_id}/resume" # 任务暂停后开始
task_result_url = Prefix.TaskServer + "/tasks/task?taskIds={id}" # 删除停止 案例运行结果详情
""" 主车相关接口 """
import_vehicle_url = Prefix.AssetServer + "/vehicles/import" # 导入主车
delete_vehicle_url = Prefix.AssetServer + "/vehicles/{vehicles_id}" # 删除主车
get_vehicle_url = Prefix.AssetServer + "/vehicles" # 获取主车信息
""" 地图相关接口 """
import_map_url = Prefix.AssetServer + "/maps" # 导入地图
delete_map_url = Prefix.CaseServer + "/maps/trash" # 删除地图
get_map_url = Prefix.CaseServer + "/maps?pageSize=100" # 获取地图相关信息
road_service_url = Prefix.TaskServer + "/sce" # 启动路网服务
laneTypeIfInside_url = Prefix.TaskServer + "/sce/{map_id}/location/laneTypeIfInside" # 监测点是否在车道内
""" 数据驱动相关接口 """
import_data_driven_url = Prefix.AssetServer + "/dd" # 导入数据驱动源
delete_data_dirven_url = Prefix.AssetServer + "/dd/{id}" # 删除数据驱动源
""" 资源相关接口 """
import_asset_url = Prefix.AssetServer + "/assets" # 资源导入
asset_repeat_url = Prefix.AssetServer + "/assets/isrepeat" # 资源查重
get_asset_url = Prefix.AssetServer + "/assets?schema={}" # 获取资源信息
delete_asset_url = Prefix.AssetServer + "/assets/{}" # 删除对应资源
""" 案例导出相关 """
export_case_url = Prefix.AssetServer + "/cases/export" # 案例导出
download_case_url = Prefix.AssetServer + "/download/{pake_id}" # 案例下载
""" 测试案例集相关 """
get_suites_url = Prefix.AssetServer + "/suites" # 获取测试集
run_suite_url = Prefix.TaskServer + "/tasks" # 运行测试集
get_taskassemble_url = Prefix.TaskServer + "/tasks?finished=true&page=0&pageSize=12&own=true&expandedIds={task_set_id}" # 获取测试案例集合
queue_status_url = Prefix.TaskServer + "/tasks/queue?own=true" # 套件运行状态检查
""" 案例判定相关 """
get_judgements_url = Prefix.AssetServer + "/cases/{caseId}/judgements"
# get_judgement_url = get_judgements_url + "/{judgementId}"
get_judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
# f"api-asset/cases/{caseId}/judgements/{judgementId}"
# url = f"api-asset/cases/{caseId}/judgements"
class UrlCodeType:
unquote = "unquote"
quote = "quote"
class File:
"""
文件父类
"""
def __init__(self, file_path: str):
if not exists(file_path):
raise FileNotFoundError
self._file_path = file_path
self._data = None
class FileReader(File):
"""
读取文件
"""
def __init__(self, file_path: str):
super(FileReader, self).__init__(file_path)
def read_byte(self) -> bytes:
with open(self._file_path, 'rb') as f:
return f.read()
def read_str(self) -> str:
with open(self._file_path, 'r', encoding='utf-8') as f:
return f.read()
def read_json(self) -> json:
with open(self._file_path, 'r', encoding='utf-8') as f:
return json.load(f)
class RquestApi():
def __init__(self):
super(RquestApi, self).__init__()
self._headers = {"Content-Type": "application/json"}
self._token = os.environ.get(EnvPath.token)
self._host_ip = SimOneUrl
self.get = "GET"
self.post = "POST"
self.delete = "DELETE"
self.put = "PUT"
def send(self, method: str, url: str, **kwargs) -> Response:
"""
发送请求
method: 请求方法
url: 请求地址
kwargs: 请求参数
"""
self.headers.update({"Authorization": self.token})
complete_url = urljoin(self._host_ip, url)
data = dict()
data['method'] = method
data['url'] = complete_url
data['headers'] = self.headers
data.update(kwargs)
print(data)
response = request(verify=False, **data)
try:
print("response:%s" % str(response.content.decode("utf-8")))
self._status_code = response.status_code
self._response_code = response.json().get("code")
except:
pass
return response
@property
def token(self):
return self._token
@property
def headers(self):
return self._headers
@headers.setter
def headers(self, value):
self._headers = value
class LoginBusiness(RquestApi):
def cloud_login_api(self, userinfo: dict):
"""
云端登录API
@param userinfo:
@return:
"""
return self.send(self.post, cloud_login_url, json=userinfo).json()
def check_username_api(self, username: str):
"""
校验用户名称是否合格
@param username:
@return:
"""
payload = {'username': username}
return self.send(self.post, login_check, json=payload).json()
def getCheckToken(self) -> str:
"""
校验token是否合格
@return:
"""
getCheckUrl = os.path.join(SimOneUrl, user_check_url)
payload = {'username': loginData['username']}
response = self.send(self.post, getCheckUrl, json=payload)
checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
return checkToken
def pad(self, text):
"""
填充函数，使被加密数据的字节码长度是block_size的整数倍
@param text:
@return:
"""
length = AES.block_size
count = len(text.encode('utf-8'))
add = length - (count % length)
entext = text + (chr(add) * add)
return entext
def str_aes(self, string: str, key: str):
"""
aes 加密
@param string:
@param key:
@return:
"""
aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
res = aes.encrypt(self.pad(string).encode("utf8"))
msg = str(base64.b64encode(res), encoding="utf8")
return msg
def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
'''
@param string:
@param username:
@param key: AES加密的key
@return:
'''
check_token = self.check_username_api(username)
target_string = check_token["data"]["checkToken"] + string
msg = self.str_aes(target_string, key)
return msg
@property
def _cloud_std_login(self):
"""
云端版本登陆方式
@return:
"""
try:
new_login_data = copy.deepcopy(loginData)
new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
username=new_login_data['username'])
print('cloud login data:' + str(new_login_data))
response = self.cloud_login_api(userinfo=new_login_data)
print("-------", response)
print('cloud login request result:' + str(response))
token = response['data']['token']
os.environ[EnvPath.token] = token
print(token)
return token
except Exception as e:
print(str(e))
raise print('get token error ')
@property
def _enterprise_std_login(self):
"""
单机版本登录方式
@return:
"""
try:
getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
response = self.send(self.get, url=getLoginUrl)
token = json.loads(response.content.decode("utf-8"))['data']['token']
os.environ[EnvPath.token] = token
except Exception as e:
print(str(e))
token = ""
# raise AssertionError('get token error ')
return token
# 登录入口
class LoginPolicy():
standalone = "standalone"
cloud = "cloud"
def __new__(cls, env_name):
if env_name == cls.standalone:
return LoginBusiness()._enterprise_std_login
elif env_name == cls.cloud:
return LoginBusiness()._cloud_std_login
else:
return print("----------input error---------------")
class Suite(RquestApi):
def __init__(self):
super(Suite, self).__init__()
self._cyclical = 0
self.category_name = "Automation_" + str(random.randint(0, 9999))
def create_category_api(self, cate_name: str):
'''
创建案例库
@param cate_name: 案例库名
@return:
'''
payload = {"parentId": "", "name": cate_name}
return self.send(self.post, categories_url, json=payload).json()
def create_category(self, cate_name: str = None) -> str:
'''
创建案例库
@param cate_name:
@return:
'''
if not cate_name:
cate_name = self.category_name
print(f"案例库目录名称：{cate_name}")
response = self.create_category_api(cate_name)
cate_id = response["data"]["categoryId"]
print(f"案例库目录 id ：{cate_id}")
return cate_id
def get_category_api(self):
"""
获取案例库列表
@return: response
"""
return self.send(self.get, categories_url).json()
def get_category(self, userid: str = None) -> dict:
"""
获取案例库
@param userid: 用户id desc：admin->builtIn
@return:
"""
cate_dict = {"data": {}, "parentId": {}}
response = self.get_category_api()
cate_dict["total"] = response["data"]["total"]
if not userid:
print("获取当前所有的案例库目录信息")
for one in response["data"]["categories"]:
cate_dict["data"].setdefault(one["name"], one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
else:
print("获取用户{}的案例库目录信息".format(userid))
for one in response["data"]["categories"]:
if one["userId"] == userid:
cate_dict["data"].setdefault(one["name"], one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
print("获取到的案例库->%s" % str(cate_dict))
return cate_dict
def get_category_case_api(self, cate_id: str):
"""
获取当前案例库所有案例
@param cate_name: 案例库名
@return:
"""
return self.send(self.get, get_cases_url.format(category=cate_id)).json()
def get_cases_category(self, cate_id: list):
"""
获取当前案例库的所有案例
@param cate_id: 案例库目录id
@return:
"""
def url_code(url: str, way: str):
from urllib import parse
if way == "unquote":
return parse.unquote(url)
elif way == "quote":
return parse.quote(url)
page_size = 100
# print("get all case id from api by category_id")
_filter = {"keyword": "", "categories": cate_id}
string = json.dumps(_filter).replace(" ", "")
filter = url_code(string, UrlCodeType.quote).replace("%", "%25")
response = self.send(self.get, get_cases_url.format(page="1", category=filter)).json()
# get case id
# print("get_cases_category 入参cate_id:%s" % cate_id)
cases_num = response['data']['total']
if cases_num >= page_size:
if cases_num % page_size == 0:
page = cases_num // page_size
else:
page = cases_num // page_size + 1
else:
page = 1
def update_case_id(response, page_size):
case_name = response['data']['caseDefs'][page_size]['name']
case_id = response['data']['caseDefs'][page_size]['id']
case_id_list.append(case_id)
cases_dict.setdefault(case_name, [case_id])
cases_dict = {}
case_id_list = []
# 如果案例数量是100以内
if cases_num <= page_size:
for i in range(cases_num):
update_case_id(response=response, page_size=i)
else:
# 如果案例数量超过100了，固定获取案例数量为100个就翻页
for i in range(page):
response = self.send(self.get, self.get_cases_url.format(page=str(i + 1), category=filter)).json()
# cases_num % page_size 是取整表示没有到达最后一页
# if i == 0 or i > 0 and cases_num % page_size == 0:
if i + 1 != page:
# print("xxx page_size:",page_size)
for j in range(page_size):
# print("xxx j:",j)
update_case_id(response=response, page_size=j)
# 到最后一页了，因为案例数量不固定所以需要单独计算获取的个数
else:
for j in range(cases_num % page_size):
update_case_id(response=response, page_size=j)
return cases_dict, case_id_list
def get_case_detail_api(self, case_id: str):
"""
获取案例信息
@param : /{case_id}/data
@return: response
"""
url = case_details_url.format(case_id=case_id)
return self.send(self.get, url).json()
def get_case_detail(self, case_id: str):
"""
@param case_id: 测试案例id
@return:
"""
response = self.get_case_detail_api(case_id)
return response
def file_md5_check(self, file_path: str, raw: bool = False):
"""
文件md5检查
@param file_path: 文件路径
@return:
"""
# global url
def file_md5(filename: str):
"""
以文件内容md5加密
@param filename:
@return:
"""
with open(filename, 'rb') as f:
contents = f.read()
return hashlib.md5(contents).hexdigest()
file_md5 = file_md5(file_path)
# print("file_md5_check file_md5:", file_md5)
# file_md5_url = "api-asset/files/{}?raw={}" # 验证文件MD5
if raw:
url = file_md5_url.format(file_md5, "true")
else:
url = file_md5_url.format(file_md5, "false")
response = self.send(self.get, url).json()
# print(f"-----file_md5_check：{url} get response:",response)
return response
def remove_file(self, sim_path: str):
"""
remove files that except (.sim .zip .xosc)
:param sim_path:
"""
for item in os.listdir(sim_path):
sim_file = os.path.join(sim_path, item)
try:
if os.path.isfile(sim_file):
os.remove(sim_file)
# print(f'{sim_file} 文件删除成功')
elif os.path.isdir(sim_file):
shutil.rmtree(sim_file)
# print(f'{sim_file} 文件夹删除成功')
except OSError as e:
print(f'Error: {e.filename} - {e.strerror}.')
def get_caseid(self, sim_path: str, sim_list: list):
"""
get the caseid before import according to
:param sim_path: string, the path to the .sim file
:sim_list run_sim: list, cases that need to be run
:return: list, caseid before import
"""
# print("sim_path:",sim_path)
# print("sim_list:",sim_list)
file_list = os.listdir(sim_path)
# print("file_list:",file_list)
target_run = list()
for item in file_list:
for run_sim_item in sim_list:
# if ('31map.sim' in item) and ('all_sim' in choose_run):
if os.path.join(sim_path, run_sim_item + '.sim') == os.path.join(sim_path, item):
target_run.append(item)
# print("target_run:",target_run)
if not target_run:
target_run = file_list
# print("更新后target_run：",target_run)
extract_dir = os.path.join(sim_path, "extract_dir")
# print("extract_dir:",extract_dir)
if not os.path.exists(extract_dir):
os.makedirs(extract_dir)
for item in target_run:
if item == "extract_dir":
continue
# zip_file = os.path.join(sim_path, item).replace('.sim', '.zip') # 当文件名中有两个.sim时会有BUG
zip_file = os.path.join(extract_dir, item[:-4] + ".zip") if item.endswith(".sim") else os.path.join(
extract_dir, item)
# print("zip_file:",zip_file)
try:
if item.endswith(".sim") or item.endswith(".zip"):
# print("try item:",item)
shutil.copy(os.path.join(sim_path, item), zip_file)
except Exception as e:
pass
try:
with zipfile.ZipFile(zip_file) as zfile:
# print("执行到这。。。")
zfile.extractall(path=extract_dir)
# print("执行到这2。。。")
except Exception as e:
raise Exception(f"解压{zip_file}失败，异常信息：{e}")
temp = list(filter(lambda x: 'json' in x, os.listdir(extract_dir)))
# print("temp:",temp)
if temp == []: # 从3.3.0版本起案例解压后是文件夹，则上一条语句执行后temp为空
for d in os.listdir(extract_dir):
# print("extract_dir目录下的内容：",d)
if os.path.isdir(os.path.join(extract_dir, d)):
temp.append(d)
# print("临时测试temp:",temp)
all_case_id = list(set(temp)) # 去重
# print("all_case_id:",all_case_id)
old_case_id = list(filter(lambda x: 'vehicle' not in x, all_case_id))
# print("old_case_id:",old_case_id)
self.remove_file(extract_dir)
# print("执行到这3。。。")
return old_case_id
all_case_id = list(map(lambda x: x.replace('.json', ''), temp))
all_case_id = list(set(all_case_id)) # 去重
old_case_id = list(filter(lambda x: 'vehicles' not in x, all_case_id))
self.remove_file(extract_dir)
return old_case_id
def upload_file(self, file_path: str, index: int, chunks: int, category: str, uploadId: str, raw: bool = False):
"""
上传文件
@param file_path: 文件路径
@param index: 上传序列
@param chunks: chunks总数
@param category: 类别
@param file_md5: 文件MD5
@return:
"""
# self.headers = {}
self.headers = {"ProjectId": 'default'} # 江淮3.5.0
file_size = os.path.getsize(file_path)
file_name = file_path.split(os.sep)[-1]
f = open(f'{file_path}', 'rb')
files = [('file', (f'{file_name}', f, 'application/octet-stream'))]
if raw:
url = upload_file_url.format(uploadId, "true")
else:
url = upload_file_url.format(uploadId, "false")
upload_payload = {"params": json.dumps({"filename": file_name,
"index": index,
"chunks": chunks,
"filesize": file_size,
"uploadId": uploadId,
"category": category})}
response = self.send(self.post, url, data=upload_payload, files=files).json()
f.close()
# print(f"-----upload_file-- url:{url} upload_payload:{upload_payload} files:{files} response:{response}")
# print("uuid：", uploadId)
return uploadId
def import_case_api(self, payload: dict):
"""
案例导入
@param payload:
@return:
"""
# print("********* import_case_api payload********",payload)
self.headers.update({"Content-Type": "application/json;charset=UTF-8"})
return self.send(self.post, case_import_url, data=json.dumps(payload)).json()
def import_case(self, file_path: str, file_name: str, cate_id: str, override_map_id: str = ""):
"""
导入案例到指定的案例库目录
@param override_map_id:案例当中需要关联的地图ID
@param file_path: 案例文件路径
@param file_name: 案例文件名
@param cate_id: 案例库目录ID
@return: [重要] 此处返回的并不是导入成功后的案例ID，而是导入案例文件里的原ID
"""
case_file_path = os.path.join(file_path, file_name)
filename = file_name.split(".")[0]
file_size = os.path.getsize(case_file_path)
print("file_size : ", file_size)
md5_res = self.file_md5_check(case_file_path)
md5_res_info = md5_res["data"]
# file_md5 = md5_res_info["id"]
global ext
global path_ext
if file_name.split(".")[1] == "xosc":
ext = "xosc"
case_id_list = [f"{filename}"]
includes = case_id_list
overrides_id = case_id_list[0]
overrides_name = filename
overrides_path = filename + ".xosc"
elif file_name.split(".")[1] == "sim":
ext = "sim"
case_id_list = self.get_caseid(file_path, [filename])
includes = case_id_list
overrides_id = case_id_list[0]
overrides_name = filename
overrides_path = case_id_list[0] + "/case/" + filename + ".json"
elif file_name.split(".")[1] == "zip":
ext = "zip"
case_id_list = self.get_caseid(file_path, [filename])
print("case_id_list:", case_id_list)
includes = case_id_list[0] + "/case/" + filename
overrides_id = includes
overrides_name = includes
overrides_path = includes + ".xosc"
else:
raise f"导入案例不支持{file_name.split('.')[1]}格式,用例中断"
def round_up(data):
"""
func:向上取整
Args:
data: float
Returns: int
"""
return math.ceil(data)
upload_id = str(uuid4())
chunk_size = 2 * 1024 * 1024
# if md5_res_info['size'] == 0:
chunks = round_up(file_size / chunk_size)
self.upload_file(file_path=case_file_path, index=0, chunks=1, category="case", uploadId=upload_id)
if override_map_id:
overrides = [{"id": f"{filename}",
"name": f"{filename}",
"replaceMap": True,
"replaceVehicle": True,
"missingMap": True,
"overrideMapId": f"{override_map_id}",
"missingVehicle": True,
"path": f"{filename}",
"vehicleIds": ["default"]}]
else:
overrides = []
case_payload = {
"category": cate_id,
"includesMap": True,
"includesVehicle": True,
"includesStopTrigger": True,
"includes": includes,
"categoryInfo": {},
"needRemoveCases": {},
"id": upload_id,
"file": upload_id,
"name": filename,
"ext": ext,
"overrides": overrides
}
response = self.import_case_api(payload=case_payload)
return case_id_list, response # [重要] 此处返回的并不是导入成功后的案例ID，而是导入案例文件里的原ID
def import_flow(self, category_name: str, file_path: str, file_name: str,map_name:str=""):
'''
导入文件流程
可导入的文件类型为 .xosc 或者 .sim格式
@param map_name: 需要关联的地图名称
@param category_name: 案例库名称
@param file_path: 文件路径比如 r"D:\System
@param file_name:文件名称比如 123.xosc or 123.sim
'''
cate_id = self.get_category()
cate_idname = cate_id["data"][category_name]
if map_name:
map_id_dict = self.get_map_id(map_name=map_name)
print("map_id_dict -> ",map_id_dict)
map_id = map_id_dict[f"{map_name}"]
self.import_case(file_path, file_name, cate_idname,override_map_id=map_id)
else:
self.import_case(file_path, file_name, cate_idname)
def create_task_api(self, payload: dict):
"""
创建task
@param payload: eg {"caseIds": ["3dd03c09-3ba9-4f7a-b988-fde88296095a"],
"taskName": "task_name"}
@return:
"""
return self.send(self.post, create_task_url, json=payload).json()
def run_task(self, caseid: list, vehicle_id: str = None, withEvaluation: bool = False, version: float = 3.5):
"""
运行测试任务集合
Run task set
@param caseid: caseid
@param vehicle_id: vehicle_id
@param withEvaluation: withEvaluation
@param version: version
@return:
"""
task_name = "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
if version >= 3.5:
payload = {"caseIds": caseid,
"taskName": task_name,
"withEvaluation": withEvaluation,
"controllers": [{"id": "Default", "name": "默认控制器", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器", "algorithmId": "SimOneDriver"}]}
else:
payload = {"caseIds": caseid, "taskName": task_name, "speed": 1}
if vehicle_id:
payload.update({"overrideVehicleId": vehicle_id})
response = self.send(self.post, url=create_task_url, json=payload).json()
data = response['data']
return data['sessionId'], data['taskIds']
def stop_task_api(self, payload: dict):
"""
task停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_task_url, json=payload).json()
def stop_sessions_api(self, payload: dict):
"""
sessions停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_sessions_url, json=payload).json()
def stop_task(self, task_id: list, sessions_id: list):
"""
停止任务和会话
@param task_id:
@param sessions_id:
@return:
"""
task_payload = {"ids": task_id}
self.stop_task_api(task_payload)
sessions_id = {"ids": sessions_id}
self.stop_sessions_api(sessions_id)
def pause_task_api(self, task_id: str):
"""
暂停任务
@param task_id:
@return:
"""
return self.send(self.post, pause_task_url.format(task_id=task_id), json={}).json()
def resume_task_api(self, task_id: str):
"""
暂停任务后开始
@param task_id:
@return:
"""
return self.send(self.post, resume_task_url.format(task_id=task_id), json={}).json()
def delete_sessions_api(self, payload: dict):
"""
sessions删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_sessions_url, json=payload).json()
def delete_task_api(self, payload: dict):
"""
task删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_task_url, json=payload).json()
def delete_task(self, sessions_id: list, task_id: list):
"""
删除任务和会话
@param sessions_id:
@param task_id:
@return:
"""
sessions_id = {"ids": sessions_id}
self.delete_sessions_api(sessions_id)
task_payload = {"ids": task_id}
self.delete_task_api(task_payload)
def import_vehicle_api(self, payload: dict):
"""
导入主车
@param payload: eg {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
@return:
"""
return self.send(self.post, import_vehicle_url, json=payload).json()
def get_vehicle_data(self, file_path: str):
vehicle_info = FileReader(file_path).read_json()
vehicle_id = vehicle_info.get("id")
return vehicle_info, vehicle_id
def import_vehicle(self, file_path: str) -> str:
"""
导入主车
@param file_path: 主车文件
@return: 主车id
"""
vehicle_data, vehicle_id = self.get_vehicle_data(file_path)
payload = {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
result = self.import_vehicle_api(payload)
return result["data"]["id"]
def get_vehicle_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get, get_vehicle_url).json()
def get_vehicle(self, userid: str = None):
"""
根据用户信息获取主车信息
@param userid:
@return:
"""
global vehicle_data
vehicle_data = {}
res = self.get_vehicle_api()
# print(f"get_vehicle_api response ->: {res}")
data = res["data"]["list"]["byId"]
id_list = res["data"]["list"]["allIds"]
if userid:
for id in id_list:
if data[id]["userId"] == userid:
vehicle_data.setdefault(data[id]["name"], data[id]["id"])
else:
for id in id_list:
vehicle_data.setdefault(data[id]["userId"], {}).update({data[id]["name"]: data[id]["id"]})
print("获取到的主车信息->：{}".format(str(vehicle_data)))
return vehicle_data
def get_vehicle_id(self, vehicle_name: list) -> list and dict:
"""
根据主车名称获取主车ID
@param vehicle_name:
@return:
"""
vehicle_id_dict = {}
vehicle_id_list = []
vehicle_list = []
for i in vehicle_name:
for k, v in self.get_vehicle().items():
vehicle_list.append(v)
for vehicle_dict in vehicle_list:
# print("vehicle_dict", vehicle_dict)
vehicle_dict_key = dict(vehicle_dict).items()
for vehicle_dict_k, vehicle_dict_v in vehicle_dict_key:
if i in vehicle_dict_k:
vehicle_name_id = vehicle_dict[i]
vehicle_id_list.append(vehicle_name_id)
vehicle_id_dict.setdefault(i, vehicle_name_id)
print("vehicle_id_list->:", vehicle_id_list, "\nvehicle_id_dict->:", vehicle_id_dict)
return vehicle_id_list, vehicle_id_dict
def delete_vehicles_api(self, vehicles_id: str):
"""
删除主车
@param vehicles_id: 主车id
@return:
"""
url = delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
def delete_vehicle(self, id: str):
"""
删除主车
@param id: 主车id
@return:
"""
self.delete_vehicles_api(id)
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
def get_suite(self, suite_name: str = None):
"""
:return:
"""
response = self.get_suite_api()
data = response['data']
suite_dict = {}
for id in data['allIds']:
suite_dict.setdefault(data['byId'][id]['name'], {}).update({"caseIds": data['byId'][id]['caseIds']})
print("suite_dict", suite_dict)
if not suite_name:
return suite_dict
else:
# print("suite_dict[suite_name]:",suite_dict[suite_name])
return suite_dict[suite_name]
def run_suite_api(self, payload: dict):
"""
run_suite
@param payload: payload
@return:
"""
return self.send(self.post, url=run_suite_url, json=payload).json()
def run_suite(self, suite_name: str, taskName: str = None, vehicle_id: str = None):
if taskName is None:
def creat_task_name():
return "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
taskName = creat_task_name()
caseid_list = self.get_suite(suite_name)['caseIds']
payload = {'caseIds': caseid_list,
'taskName': taskName}
if vehicle_id:
# vehicle_module = VehicleBusiness()
# vehicle_module.get_vehicle()
payload.update({"overrideVehicleId": vehicle_id})
response = self.run_suite_api(payload)
data = response['data']
return data['sessionId'], data['taskIds']
def suite_queue_api(self):
"""
suite_queue
@return:
"""
return self.send(self.get, url=queue_status_url).json()
def suite_queue_check(self, n=10):
"""
@param n:cycle index
@return:
"""
if n == 1:
print("-----------------------The test case run fail--------------------------------------")
return False
while (1):
try:
assert self.suite_queue_api()
response = self.suite_queue_api()
assert response["code"] == 0
except Exception as e:
return -1
queue_info = response["data"]
running, pending, waiting = queue_info["running"], queue_info["pending"], queue_info["waiting"]
if running == 0 and pending == 0 and waiting == 0:
print("-----------------------The test case run finish--------------------------------------")
return True
else:
time.sleep(interval)
def get_result(self, task_id: list, index: int = None):
"""
:param task_id:
:return:
"""
task_id_str = ",".join(task_id)
# print(task_id_str)
response = self.send(self.get, url=result_url.format(task_id=task_id_str)).json()
data = response["data"]["tasks"]
result = []
for one in data:
result.append({"pass": one["pass"], "is_ended": True if one["key"] == "ended" else False})
if not index:
return result
else:
return result[index]
def is_ended(self, task_id: list, case_name=None):
"""
判断案例是否运行结束
@param task_id: 任务id
@param case_name: 案例名称
@return:
"""
interval = 10
while (1):
result = self.get_result(task_id, -1)
if result["is_ended"]:
print('Case status：end of run')
if case_name:
print(f'Name of the current ending case->{case_name}')
return False
else:
print('Case status：in progress')
if case_name:
print(f'Name of the current running case->{case_name}')
time.sleep(interval)
def get_taskassemble_api(self, task_set_id):
'''
获取测试案例集
@param
@return:
'''
return self.send(self.get, get_taskassemble_url.format(task_set_id=task_set_id)).json()
def get_task_id_of_task_set(self, task_set_id: str) -> list:
"""
获取任务集合中所有测试案例的task_id
@param task_set_id: 任务集合id
@return: list
"""
response = self.get_taskassemble_api(task_set_id)
_task_id_list = response["data"]["list"]
# print("---------------------------------task_id_list-----------------------")
# print(_task_id_list)
index = 0
task_id_list = []
for i in _task_id_list:
index += 1
# print(f"------------------------第{index}个元素----------------")
# print(i)
if i['parentId'] == task_set_id:
# print(i)
task_id_list.append(i["id"])
# print("task_id_list",task_id_list)
return task_id_list
def get_task_result_api(self, task_id: str):
"""
获取指定task的结果
@param task_id:
@return:
"""
return self.send(self.get, task_result_url.format(id=task_id)).json()
def get_task_result(self, task_id_list: list = None) -> list:
"""
获取任务集中的 案例名称，caseId，task_id,运行结果
@param task_id_list: 任务id列表
@param task_set_id: 任务集合id
@return:
"""
task_result_list = []
for task_id in task_id_list:
result = self.get_task_result_api(task_id)
case_name = result["data"]["tasks"][0]["case"]["name"]
case_id = result["data"]["tasks"][0]["caseId"]
task_id = result["data"]["tasks"][0]["id"]
task_result = result["data"]["tasks"][0]["pass"]
if task_result != True: task_result = False
task_result_list.append({"case_name": case_name,
"case_id": case_id,
"task_id": task_id,
"task_result": task_result})
print("task_result_list->", task_result_list)
return task_result_list
def import_map_api(self, payload: dict, files: dict):
"""
导入地图API
@param payload:传参字典
@param files: 传参文件
@return:
"""
self.headers = {}
return self.send(self.post, import_map_url, data=payload, files=files).json()
def import_map(self, xodr_path: str, thumbnail_path: str):
"""
导入地图方法
@param xodr_path: 地图文件
@param thumbnail_path: 地图对应的图像
@return:
"""
payload = {"params": json.dumps(
{"category": "customized", "id": "new_" + Faker('zh_CN').uuid4(),
"name": os.path.splitext(os.path.basename(xodr_path))[0], "size": 512,
"ppm": 10, "bgColor": "#dddddd", "reproject": True, "reprojectOrigin": False, "reprojectOriginLat": 0,
"reprojectOriginLng": 0, "tags": [], "notes": "",
"header": {"minX": -210.20535534122396, "minY": -149.68815701999185, "minZ": -1.862645149230957e-9,
"maxX": 237.95535534122394, "maxY": 135.43815701999196, "maxZ": 2.7940070024635385e-9,
"centerX": 97.12480158531203, "centerY": 24.463606820251727, "centerZ": 100,
"localEnuExt": "6378137,0,0;0,1,0;0,0,1;1,0,0"}})}
files = {'xodr': open(xodr_path, 'rb'), "thumbnail": open(thumbnail_path, 'rb')}
result = self.import_map_api(payload, files)
if result["code"] == 0:
print(f"导入地图{xodr_path}成功，id:{result['data']['mapId']}")
else:
print(f"导入地图{xodr_path}失败，返回结果：{result}")
raise f"导入地图{xodr_path}失败，返回结果：{result}"
res_dict = {"code": result["code"], "mapId": result["data"]["mapId"],
"name": os.path.basename(os.path.basename(xodr_path))}
return res_dict
def get_map_api(self):
"""
获取地图
@return:
"""
return self.send(self.get, get_map_url).json()
def get_map_id(self, userid: str = None, map_name: str = None):
"""
按条件获取地图信息，默认获取所有地图
@param userid: 用户id
@param map_name: 地图名字
@return:
"""
global map_data
map_data = {}
try:
result = self.get_map_api()
# print(len(result['data']['maps']))
map_info = result['data']['maps']
if userid:
for map in map_info:
if map["userId"] == userid:
map_data.setdefault(map["name"], map["id"])
elif map_name:
for map in map_info:
if map["name"] == map_name:
map_data = {map["name"]: map["id"]}
else:
for map in map_info:
map_data.setdefault(map["userId"], {}).update({map["name"]: map["id"]})
return map_data
except Exception as e:
print(e)
def delete_map_api(self, payload: dict):
"""
删除地图
@param payload: eg：{ids: ["27a0bbc4-5d4d-48c1-9187-62df794dadae"]}
@return:
"""
return self.send(self.post, delete_map_url, json=payload).json()
def delete_map(self, id: list):
"""
批量删除地图
@param id: 地图id列表
@return:
"""
print("删除的地图列表-》{}".format(str(id)))
payload = {"ids": id}
self.delete_map_api(payload)
def get_judgements_api(self, caseId, judgementId=None):
"""
获取判定信息API
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
if judgementId:
url = get_judgement_url.format(caseId=caseId, judgementId=judgementId)
else:
url = get_judgements_url.format(caseId=caseId)
response = self.send(self.get, url).json()
return response
def get_judgements_method(self, caseId: str, judgementId: str = None):
"""
获取判定信息
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
response = self.get_judgements_api(caseId, judgementId)
print("得到的扩展判定信息：", response["data"][1])
return response
def update_judgement_api(self, caseId, judgementId="collision", payload: dict = None):
"""
@param caseId: caseId
@param judgementId: judgementId
@return:
"""
if not payload:
payload = {"schema": "judgement", "settings": {"logLevel": "error", "action": "failure", "logInfo": ""},
"scope": {"size": {"x": 10, "y": 10, "z": 0}, "heading": {"w": 1, "x": 0, "y": 0, "z": 0},
"position": {"x": 0, "y": 0, "z": 0},
"type": "global"}, "builtIn": True,
"name": "碰撞", "lock": True, "id": "collision",
"type": "collision", "category": "general",
"conditions": [], "userId": "admin",
"enabled": True}
response = self.send(self.put, get_judgement_url.format(caseId=caseId, judgementId=judgementId),
data=json.dumps(payload))
return response
def update_judgement_method(self, caseId, judgementId="collision"):
__collision = {"schema": "judgement", "settings": {"logLevel": "error", "action": "failure", "logInfo": ""},
"scope": {"size": {"x": 10, "y": 10, "z": 0}, "heading": {"w": 1, "x": 0, "y": 0, "z": 0},
"position": {"x": 0, "y": 0, "z": 0},
"type": "global"}, "builtIn": True,
"name": "碰撞", "lock": True, "id": "collision",
"type": "collision", "category": "general",
"conditions": [], "userId": "admin",
"enabled": True}
payload = json.dumps(__collision)
judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
response = self.send(self.put, judgement_url.format(caseId=caseId, judgementId=judgementId), data=payload)
print("更新扩展判定信息")
print("status_code:", response.status_code)
def main(category_name: str, case_name: str = None, vehicle_name=None):
"""
@param category_name: category_name
@param case_name:
@param vehicle_name:vehicle_name
@return:
"""
suite = Suite()
suite.get_category()
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
# 主车控制相关逻辑
if vehicle_name:
vehicle_dict = suite.get_vehicle_id([vehicle_name])[1]
if vehicle_name not in vehicle_dict.keys():
print(f"vehicle_name：{vehicle_name} inexistence")
return
vehicle_id = vehicle_dict[vehicle_name]
else:
vehicle_id = None
print(f"vehicle_name:{vehicle_name},vehicle_id:{vehicle_id}")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
print(f" case_name:{case_name}")
# 启动案例
session_id, task_ids = suite.run_task(case_id_list, vehicle_id=vehicle_id)
time.sleep(10)
# suite.stop_task(task_id=task_ids, sessions_id=[session_id])
# 等待案例运行完毕
falg = suite.suite_queue_check()
if falg is True:
# 获取案例运行结果
suite.get_task_result(task_ids)
SimOneUrl = "http://127.0.0.1:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("入门案例", "构建标准案例2.0")
1 用户登录接口文档
SimOne Web User Server
1.1 登录接口
Path： /api-user/user/login
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必须 |
类型 |
示例 |
|---|---|---|---|---|---|
|
username |
query |
账号名称 |
Yes |
string |
zhangsan |
|
password |
query |
鉴权token，有效期30s |
Yes |
string |
qKP1Cgv3NysooGPKHRiiKGqQEjLjkbBFy2FqbkaY6ew |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
token |
当前用户鉴权Token |
Yes |
string |
备注:
密码是通过一定加密算法计算得到的需要联系售后支持以获取生成方法
1.2 获取用户登录token
Path： /api-user/user/check
Method： POST
Headers： Headers
请求参数
|
参数名称 |
位于 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
username |
body |
用户id |
Yes |
string |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
checkToken |
当前用户鉴权Token |
Yes |
string |
1.3 单机版获取authToken
Path： /api-user/user/standalone
Method： GET
Headers： Headers
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
token |
当前用户鉴权Token |
Yes |
string |
1.4 登出
Path： /api-user/user/logout
Method： POST
Headers： Headers
返回数据
无
1.5 判断是否为云端环境
Path： /api-user/user/isCloud
Method： GET
Headers： Headers
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
isCloud |
是否云端环境 |
Yes |
boolean |
通用请求头
Headers
|
参数名称 |
参数值 |
是否必填 |
示例 |
|---|---|---|---|
|
Content-Type |
application/json |
Yes |
|
|
Authorization |
string |
Yes |
“6a82749b-7d4e-46a3-a932-e5ca21f34f16” |
2 任务服务接口文档
SimOne Web Task Server
2.1 服务状态接口
2.1.1 获取服务健康状态
Path： /api-task/health
Method： GET
Headers： Headers
请求参数
无
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
version |
版本 |
Yes |
string |
|
current |
当前状态 |
Yes |
number |
|
maximum |
最大状态 |
No |
number |
|
message |
当前状态信息 |
No |
string |
|
level |
服务当前等级 |
No |
“info” | “error” | “warning” |
|
code |
服务错误代码 |
No |
number |
2.2 任务接口
2.2.1 获取运行的任务或回放的任务id
Path： /api-task/sessions/isPlay
Method： GET
Headers： Headers
请求参数
无
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
testingSessionIds |
运行的任务id |
Yes |
string[] |
|
replaySessionIds |
回放的任务id |
Yes |
string[] |
2.2.2 获取回放任务状态
Path： /api-task/sessions/replayTasks
Method： GET
Headers： Headers
请求参数
无
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
replayTasks |
回放任务状态 |
Yes |
Record<string, ReplayTaskInfo > |
2.2.3 获取任务队列
Path： /api-task/sessions/queue
Method： GET
Headers： Headers
请求参数
无
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
queue |
排队信息 |
Yes |
Record<string, number> |
2.2.4 恢复暂停任务集
Path： /api-task/sessions/:id/resume
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.2.5 暂停任务
Path： /api-task/sessions/:id/pause
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.2.6 获取任务报告
Path： /api-task/sessions/:id/report
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务集id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
success |
是否成功 |
Yes |
boolean |
备注
返回 success true 之后使用/report/:id_zh-Hans.pdf地址进行下载，例sessionId为123456，下载地址为：原网页网址/report/123456_zh-Hans.pdf
2.2.7 任务下一帧
Path： /api-task/sessions/step
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
body |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.2.8 修改仿真速度
Path： /api-task/sessions/speed
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
body |
任务id |
Yes |
string |
|
speed |
body |
任务速度 |
Yes |
number |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.2.9 任务跳转
Path： /api-task/sessions/skipTo
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
body |
任务id |
Yes |
string |
|
seconds |
body |
任务时间 |
Yes |
number |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
无
2.2.10 更新主车控制
Path： /api-task/sessions/updateVehicleControl
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
body |
任务id |
Yes |
string |
|
egoId |
body |
主车id |
Yes |
string |
|
vehicleControl |
body |
主车控制 |
Yes |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
无
2.2.11 获取回放任务数据
Path： /api-task/sessions/:id/replayChartData
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
timestamp |
时间戳 |
Yes |
string |
|
taskId |
任务id |
Yes |
string |
|
caseId |
案例id |
Yes |
string |
|
taskUuid |
任务uuid |
No |
string |
|
timeMs |
时间 |
No |
number |
|
durationMs |
持续时间 |
No |
number |
|
fps |
帧率 |
No |
number |
|
speed |
速度 |
No |
number |
|
mainvehicles |
主车 |
No |
2.2.12 获取回放任务状态
Path： /api-task/sessions/:id/taskState
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
taskId |
任务id |
Yes |
string |
|
isPlay |
是否是回放 |
Yes |
number |
2.2.13 停止任务集
Path： /api-task/sessions/stop
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
ids |
body |
任务集id |
Yes |
string[] |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.2.14 删除任务集
Path： /api-task/sessions/delete
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
ids |
body |
任务集id |
Yes |
string[] |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3 任务运行时接口
2.3.1 获取任务列表
Path： /api-task/tasks/
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
finished |
query |
是否已完成 |
Yes |
boolean |
|
page |
query |
页数 |
Yes |
string |
|
pageSize |
query |
每页数量 |
Yes |
string |
|
own |
query |
是否是自己的任务 |
Yes |
boolean |
|
expandedIds |
query |
展开的任务id |
Yes |
string[] |
|
filters |
query |
筛选 |
No |
string[] |
|
searchFilter |
query |
搜索 |
No |
string |
|
startTime |
query |
开始时间 |
No |
number |
|
endTime |
query |
结束时间 |
No |
number |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
list |
任务列表 |
Yes |
|
|
page |
页数 |
Yes |
number |
|
total |
总页数 |
Yes |
number |
|
pageSize |
每页数量 |
Yes |
number |
2.3.2 获取任务详情
Path： /api-task/tasks/task
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
taskIds |
query |
展开的任务id |
Yes |
string[] |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
tasks |
任务列表 |
Yes |
TaskListItem [] |
2.3.3 获取任务数量
Path： /api-task/tasks/count
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
own |
query |
是否是自己的任务 |
Yes |
boolean |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
count |
任务数量 |
Yes |
number |
2.3.4 获取任务队列
Path： /api-task/tasks/queue
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
own |
query |
是否是自己的任务 |
Yes |
boolean |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
total |
任务总数 |
Yes |
number |
|
running |
运行任务数量 |
Yes |
number |
|
pending |
即将运行任务数量 |
Yes |
number |
|
waiting |
等待任务数量 |
Yes |
number |
|
finished |
结束任务数量 |
Yes |
number |
|
estimatedTaskDuration |
预计任务时间 |
Yes |
number |
2.3.5 清除全部任务（需要管理员权限）
Path： /api-task/tasks/clean
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
ids |
body |
任务id集合 |
Yes |
string[] |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.6 获取指定id的任务
Path： /api-task/tasks/:id/task
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.7 创建任务
Path： /api-task/tasks/
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
caseIds |
body |
案例id |
Yes |
string[] |
|
taskName |
body |
任务名称 |
Yes |
string |
|
overrideVehicleId |
body |
覆盖主车id |
No |
string |
|
overrideJudgementId |
body |
覆盖判定id |
No |
string |
|
evaluationPresetId |
body |
评价id |
No |
string |
|
withEvaluation |
body |
是否评价 |
No |
boolean |
|
algorithmName |
body |
算法名称 |
No |
string |
|
algorithmVersion |
body |
算法版本 |
No |
string |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
sessionId |
sessionId |
Yes |
string |
|
taskIds |
任务id |
Yes |
string[] |
2.3.8 回放指定任务
Path： /api-task/tasks/:id/replay
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
|
RTC |
query |
是否RTC任务 |
No |
boolean |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
taskId |
任务id |
Yes |
string |
2.3.9 查看任务运行可视化
Path： /api-task/tasks/:id/observe
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.10 dump任务
Path： /api-task/tasks/:id/dump
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
|
vizConfig |
body |
每页数量 |
Yes |
|
|
outputConfig |
body |
每页数量 |
Yes |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.11 获取任务数据
Path： /api-task/tasks/:id/data
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
success |
是否成功 |
Yes |
boolean |
2.3.12 创建回放案例
Path： /api-task/tasks/convert2case
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
name |
path |
案例名 |
Yes |
string |
|
categoryId |
body |
保存目录id |
Yes |
string |
|
opponent |
body |
保存类型，回放案例或标准案例 |
No |
“record” | “openscenario” |
|
ego2npc |
body |
选择保存为标准案例时，主车是否转换为对手车 |
No |
boolean |
|
tags |
body |
标签 |
No |
string[] |
|
notes |
body |
案例备注 |
No |
string |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
|
404 |
Not Found |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
id |
id |
Yes |
string |
2.3.12 停止任务
Path： /api-task/tasks/stop
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
all |
query |
是否停止全部任务 |
No |
boolean |
|
ids |
body |
任务id |
Yes |
string[] |
|
isReplay |
body |
是否是回放 |
No |
boolean |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.13 停止回放任务
Path： /api-task/tasks/stopReplay
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
ids |
body |
任务id |
Yes |
string[] |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.14 删除任务
Path： /api-task/tasks/delete
Method： POST
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
taskIds |
body |
任务id |
Yes |
string[] |
返回码
|
Code |
说明 |
|---|---|
|
202 |
Ok |
返回数据
无
2.3.15 获取任务是否结束
Path： /api-task/tasks/:id/hasFinished
Method： GET
Headers： Headers
请求参数
|
字段名 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回码
|
Code |
说明 |
|---|---|
|
200 |
Ok |
返回数据
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
hasFinished |
任务是否结束 |
Yes |
boolean |
通用请求头
Headers
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
Authorization |
string |
用于权限验证的token |
Yes |
相关类型
2.2 任务接口相关类型
ReplayTaskInfo
回放任务信息
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
status |
任务状态 |
Yes |
string |
|
mode |
任务模式 |
Yes |
“replay” | “dump” |
|
RTC |
是否RTC任务 |
No |
boolean |
VehicleControl
主车控制输入信息
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
vehicleId |
主车id |
No |
number |
|
throttlepercentage |
油门百分比 |
No |
number |
|
steeringpercentage |
转向百分比 |
No |
number |
|
brakepercentage |
刹车百分比 |
No |
number |
|
handBrake |
手刹状态 |
No |
boolean |
|
isManualGear |
是否手动挡 |
No |
boolean |
|
manualGear |
手动挡位 |
No |
number |
|
gearImmediate |
立即换挡 |
No |
boolean |
|
activeInput |
活跃输入 |
No |
number |
|
acceleration |
加速度 |
No |
number |
|
speed |
速度 |
No |
number |
|
signalLights |
信号灯 |
No |
number |
MainVehicleUIData
热区主车信息
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
id |
id |
No |
number | null |
|
speed |
速度 |
No |
number | null |
|
throttle |
油门 |
No |
number | null |
|
steering |
转向 |
No |
number | null |
|
brake |
刹车 |
No |
number | null |
|
rpm |
转速 |
No |
number | null |
|
gear |
档位 |
No |
number | null |
|
velocityx |
速度x方向值 |
No |
number | null |
|
velocityy |
速度y方向值 |
No |
number | null |
|
velocityz |
速度z方向值 |
No |
number | null |
|
accelerationx |
加速度x方向值 |
No |
number | null |
|
accelerationy |
加速度y方向值 |
No |
number | null |
|
accelerationz |
加速度z方向值 |
No |
number | null |
|
posx |
位置x方向值 |
No |
number | null |
|
posy |
位置y方向值 |
No |
number | null |
|
posy |
位置z方向值 |
No |
number | null |
|
rotationRate |
旋转比率 |
No |
number | null |
2.3 任务时运行接口相关类型
VizConfig
可视化配置
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
duration |
开始时间，结束时间 |
Yes |
[number, number] |
|
sensors |
传感器 |
Yes |
{ [key: string]: boolean } |
OutputConfig
dump输出配置
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
dumpHz |
dump频率 |
Yes |
number |
|
h264 |
H.264编码 |
Yes |
boolean |
|
bitrate |
码率 |
Yes |
number |
|
bbox3D |
3D边界框 |
Yes |
boolean |
TaskListItem
任务项简要信息
|
字段名 |
说明 |
是否必填 |
类型 |
|---|---|---|---|
|
id |
任务id |
Yes |
string |
|
parentId |
sessionId |
Yes |
string | null |
|
loading |
是否加载中 |
No |
boolean |
|
userId |
用户名 |
No |
string |
|
userName |
用户姓名 |
No |
string |
3 评价服务接口文档
SimOne Web Evaluation Server
3.1 测试集接口
3.1.1 获取分页测试集
Path： /api-evaluation/suites
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
page |
query |
当前页码 |
No |
number |
|
pageSize |
query |
每页数量 |
No |
number |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.1.2 创建测试集
Path： /api-evaluation/suites
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
suite |
body |
包含id,name等测试集信息 |
Yes |
{ “suite” : Suite } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.1.3 获取单个测试集
Path： /api-evaluation/suites/{id}
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
测试集id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.1.4 更新单个测试集
Path： /api-evaluation/suites/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
测试集id |
Yes |
string |
|
data |
body |
需要更新的测试集信息 |
Yes |
{ “data” : Suite } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.1.5 批量删除测试集
Path： /api-evaluation/suites/remove
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
ids |
body |
测试集id列表 |
Yes |
{ “ids” : [ string ] } |
返回数据
|
Code |
说明 |
|---|---|
|
204 |
No content |
3.2 任务接口
3.2.1 获取指定测试集下的任务列表
Path： /api-evaluation/tasks
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
suite |
query |
测试集id |
Yes |
string |
|
page |
query |
当前页码 |
No |
number |
|
pageSize |
query |
每页数量 |
No |
number |
|
sort |
query |
排序类型
|
No |
string |
|
order |
query |
升降序
|
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.2.2 创建任务
Path： /api-evaluation/tasks
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
task |
body |
包含id,name等任务信息 |
Yes |
{ “task” : Task } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.2.3 获取单个任务详情
Path： /api-evaluation/tasks/{id}
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.2.4 更新单个任务
Path： /api-evaluation/tasks/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
任务id |
Yes |
string |
|
data |
body |
需要更新的任务信息 |
Yes |
{ “data” : Task } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.2.5 批量删除任务
Path： /api-evaluation/tasks/remove
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
ids |
body |
任务id列表 |
Yes |
{ “ids” : [ string ] } |
返回数据
|
Code |
说明 |
|---|---|
|
204 |
No content |
3.3 评价预设接口
3.3.1 获取所有预设
Path： /api-evaluation/presets/all
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
type |
query |
预设类型 |
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.2 获取分页预设
Path： /api-evaluation/presets
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
page |
query |
当前页码 |
No |
number |
|
pageSize |
query |
每页数量 |
No |
number |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.3 创建一个预设
Path： /api-evaluation/presets
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
preset |
body |
包含name，data等预设信息 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.4 获取单个预设
Path： /api-evaluation/presets/{id}
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
预设id |
Yes |
number |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.5 更新单个预设
Path： /api-evaluation/presets/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
预设id |
Yes |
number |
|
data |
body |
要更新的预设信息对象 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.6 删除单个预设
Path： /api-evaluation/presets/{id}
Method： DELETE
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
预设id |
Yes |
number |
返回数据
|
Code |
说明 |
|---|---|
|
204 |
No content |
3.3.7 克隆单个预设
Path： /api-evaluation/presets/clone
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
body |
预设id |
Yes |
number |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.8 导入单个预设
Path： /api-evaluation/presets/import
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
preset |
body |
要更新的预设信息 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.3.9 判断预设名是否重复
Path： /api-evaluation/presets/isrepeat
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
preset |
body |
预设信息 |
No |
{ “name” : string, “type?” : string, “id?” : string } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
{ “isNameRepeat” : boolean; “uniqueName” : string } |
3.4 评价报告接口
3.4.1 创建评价报告
Path： /api-evaluation/evaluations
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
evaluation |
body |
评价报告信息 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.4.2 获取单个评价报告
Path： /api-evaluation/evaluations
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
taskId |
query |
任务id，任务id和测试集id只能传一个 |
No |
string |
|
suiteId |
query |
测试集id |
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
3.4.3 批量获取评价报告状态
Path： /api-evaluation/evaluations/status
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
taskIds |
query |
任务id字符串，taskIds和suiteId只能传一个 |
No |
string |
|
suiteIds |
query |
测试集id字符串 |
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
object |
3.4.4 删除单个评价报告
Path： /api-evaluation/evaluations/{id}
Method： DELETE
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
报告id |
Yes |
string |
返回数据
|
Code |
说明 |
|---|---|
|
204 |
No content |
3.4.5 触发规控评价
Path： /api-evaluation/evaluations/evaluate
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
config |
body |
包含测试集id和预设id |
Yes |
{ suiteId: string; presetId: number } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
object |
3.4.6 查看评价服务是否正常
Path： /api/health
Method： GET
Headers： Headers
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
OK |
{ “code” : number, “data” : { “message” : string } } |
通用请求头
Headers
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
Authorization |
string |
用于权限验证的token |
Yes |
相关类型
3.1 测试集接口相关类型
Suite
测试集
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
测试集id |
Yes |
|
name |
string |
测试集名称 |
Yes |
|
withEvaluation |
boolean |
是否能评价 |
Yes |
|
presetId |
number |
当前预设 |
No |
|
algorithmName |
string |
当前算法名称 |
No |
|
algorithmVersion |
string |
当前算法版本 |
No |
|
startTime |
number |
测试集开始时间 |
Yes |
|
endTime |
number |
测试集结束时间 |
No |
SuiteResponse
测试集响应
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
total |
number |
总数 |
Yes |
|
suites |
[ Suite ] |
测试集列表 |
Yes |
3.2 任务接口相关类型
Task
任务
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
任务id |
Yes |
|
name |
string |
任务名称 |
Yes |
|
suiteId |
string |
所属测试集id |
Yes |
|
map |
string |
地图id |
No |
|
mapName |
string |
地图名 |
No |
|
vehicle |
string |
主车id |
No |
|
controller |
string |
控制器id |
No |
|
algorithmName |
string |
算法名称 |
No |
|
algorithmVersion |
string |
算法版本 |
No |
|
presetId |
number |
当前预设id |
No |
|
startTime |
number |
任务开始时间 |
Yes |
|
endTime |
number |
任务结束时间 |
No |
TaskResponse
任务响应
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
total |
number |
总数 |
Yes |
|
tasks |
[ Task ] |
任务列表 |
Yes |
3.3 评价预设接口相关类型
Preset
评价预设
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
name |
string |
预设名称 |
Yes |
|
type |
string |
预设类型 |
No |
|
data |
[ object ] |
预设数据 |
Yes |
PresetResponse
评价预设响应
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
code |
number |
状态码 |
Yes |
|
data |
预设数据 |
Yes |
PresetsResponse
评价预设集响应
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
code |
number |
状态码 |
Yes |
|
data |
{ presets: [ Preset ] , total: number } |
预设数据 |
Yes |
PresetImportResponse
评价预设导入响应
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
code |
number |
状态码 |
Yes |
|
data |
{ reason: string } |
失败原因 |
Yes |
3.4 评价报告接口相关类型
Evaluation
评价报告
|
参数名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
name |
string |
报告名称 |
Yes |
|
type |
string |
报告枚举类型：
|
Yes |
|
status |
string |
报告状态 |
Yes |
|
suiteId |
string |
测试集id |
Yes |
|
taskId |
string |
任务id |
No |
|
presetId |
number |
预设id |
Yes |
|
presetName |
string |
预设名称 |
No |
|
presetVersionId |
number |
预设版本id |
Yes |
|
score |
number |
评价得分 |
No |
|
data |
object |
报告数据 |
Yes |
4 资源服务接口文档
SimOne Web Asset Server
4.1 资源接口
4.1.1 获取资源列表
Path： /api-asset/assets
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
schema |
query |
资源类型
Enum:
|
No |
string |
|
category |
query |
目录id |
No |
string |
|
vehicleOnly |
query |
为”true”时，只获取主车编辑器页面使用的资源 |
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.1.2 创建资源
Path： /api-asset/assets
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
data
：资源数据；
|
No |
{ “data” : Asset , “sourceId” : string, “overwriteId” : string } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.1.3 获取指定资源
Path： /api-asset/assets/{id}
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
需要获取资源 id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.1.4 修改资源
Path： /api-asset/assets/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
需要更新的资源 id |
Yes |
string |
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.1.5 删除资源
Path： /api-asset/assets/{id}
Method： DELETE
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
要删除的资源 id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.1.6 资源判重
Path： /api-asset/assets/isrepeat
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.1.7 克隆资源
Path： /api-asset/assets/{id}/clone
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
要克隆资源的 id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2 案例接口
4.2.1 创建案例
Path： /api-asset/cases
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
CaseCreateRequest extends PartialCaseDef |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.2 已完成任务另存为案例
Path： /api-asset/cases/vr
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.3 更新案例
Path： /api-asset/cases/update
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.4 克隆案例
Path： /api-asset/cases/clone
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.5 移动案例
Path： /api-asset/cases/move
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.6 导出案例
Path： /api-asset/cases/export
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.7 导入案例
Path： /api-asset/cases/import
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.2.8 获取传感器数量超载的案例
Path： /api-asset/cases/checkSensors
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
[ { “id” : string, “name” : string } ] |
4.2.9 删除案例
Path： /api-case/cases/trash
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
{ “ids” : [string] } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
204 or 200 |
Ok |
{} |
4.3 地图接口
4.3.1 创建地图 / 更新地图
Path： /api-asset/maps
Method： POST
Headers： FormDataHeaders
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
formdata |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
object |
4.4 文件接口
4.4.1 上传文件
Path： /api-asset/files/{md5}/upload
Method： POST
Headers： FormDataHeaders
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
md5 |
path |
文件 md5 |
Yes |
string |
|
raw |
query |
为”true”时表示上传的是数据驱动源 |
Yes |
string |
|
formData |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “code” : double } |
4.4.2 替换指定目录下的文件
Path： /api-asset/files/replace
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
sourceId |
query |
文件名称 |
Yes |
string |
|
targetId |
query |
案例 id or 主车模型 id |
Yes |
string |
|
category |
query |
资源类型
Enum:
|
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “code” : double } |
4.5 主车接口
4.5.1 获取主车列表
Path： /api-asset/vehicles
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
page |
query |
页码 |
No |
string |
|
pageSize |
query |
每页数量 |
No |
string |
|
search |
query |
搜索关键字 |
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.2 创建主车
Path： /api-asset/vehicles
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.3 获取指定主车
Path： /api-asset/vehicles/{id}/vehicle
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
主车 id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.4 更新主车
Path： /api-asset/vehicles/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.5 删除主车
Path： /api-asset/vehicles/{id}
Method： DELETE
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
删除的主车 id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
object |
4.5.6 判断主车是否重复
Path： /api-asset/vehicles/isrepeat
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.7 克隆主车
Path： /api-asset/vehicles/{id}/clone
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
主车id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.8 更新主车资源文件
Path： /api-asset/vehicles/{vehicleId}/files
Method： POST
Headers： FormDataHeaders
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
vehicleId |
path |
主车id |
Yes |
string |
|
formData |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.5.9 获取主车数量
Path： /api-asset/vehicles/count
Method： GET
Headers： Headers
请求参数
无
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “count” : double } |
4.5.10 导入主车
Path： /api-asset/vehicles/import
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.6 测试集接口
4.6.1 获取测试集列表
Path： /api-asset/suites
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
opponent |
query |
案例类型
Enum:
|
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.6.2 创建测试集
Path： /api-asset/suites
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.6.3 获取测试集
Path： /api-asset/suites/{id}
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
测试集id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.6.4 更新测试集
Path： /api-asset/suites/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
测试集id |
Yes |
string |
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.6.5 删除测试集
Path： /api-asset/suites/{id}
Method： DELETE
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
测试集id |
Yes |
string |
|
opponent |
query |
案例类型
Enum:
|
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.6.6 移动测试集
Path： /api-asset/suites/move
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{} |
4.6.7 测试集判重
Path： /api-asset/suites/isrepeat
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.7 页面目录接口
4.7.1 获取目录列表
Path： /api-asset/categories
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
pageType |
query |
页面类型
Enum:
|
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.8 对手元素接口
4.8.1 获取对手元素列表
Path： /api-asset/obstacles
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
schema |
query |
Enum:
|
Yes |
string |
|
category |
query |
目录 id |
Yes |
string |
|
page |
query |
页码 |
No |
string |
|
pageSize |
query |
每页数量 |
No |
string |
|
search |
query |
搜索关键字 |
No |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
4.8.2 创建自定义对手元素
Path： /api-asset/obstacles
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “npc” : ObstacleNpc } |
4.8.3 获取指定对手元素
Path： /api-asset/obstacles/{id}
Method： GET
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
对手元素id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “npc” : ObstacleNpc } |
4.8.4 更新对手元素
Path： /api-asset/obstacles/{id}
Method： PUT
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
对手元素id |
Yes |
string |
|
requestBody |
body |
请求体 |
Yes |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “npc” : ObstacleNpc } |
4.8.5 删除自定义元素
Path： /api-asset/obstacles/{id}
Method： DELETE
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
id |
path |
需要删除元素的 id |
Yes |
string |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
{ “id” : string, “schema” : string } |
4.8.6 对手元素判重
Path： /api-asset/obstacles/isrepeat
Method： POST
Headers： Headers
请求参数
|
参数名称 |
传参格式 |
说明 |
是否必填 |
类型 |
|---|---|---|---|---|
|
requestBody |
body |
请求体 |
Yes |
{ “name” : string; “id” ?: string } |
返回数据
|
Code |
说明 |
类型 |
|---|---|---|
|
200 |
Ok |
通用请求头
Headers
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
Authorization |
string |
用于权限验证的 token |
Yes |
|
accept |
application/json |
客户端接收JSON数据 |
No |
FormDataHeaders
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
Authorization |
string |
用于权限验证的 token |
Yes |
|
Content-Type |
multipart/form-data |
客户端发送的数据类型 |
Yes |
|
accept |
application/json |
客户端接收JSON数据 |
No |
相关类型
4.1 资源接口相关类型
AssetList
系统资源列表
Record<string, [ Asset ]>
Asset
资源
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
ID |
Yes |
|
name |
string |
名称 |
Yes |
|
userId |
string |
用户id |
Yes |
|
schema |
string |
类型 |
Yes |
|
category |
string |
目录 |
Yes |
|
thumbnail |
string |
缩略图文件路径 |
No |
|
builtIn |
boolean |
是否内置 |
No |
|
external |
boolean |
是否是外挂元素 |
No |
PartialAsset
Partial< Asset >
表示 Asset 的每个键值对都是可选的
AssetBaseResponse
返回资源的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
资源数据 |
Yes |
AssetSaveRequest
修改资源的请求体
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
更新的资源的数据 |
Yes |
|
|
sourceId |
string |
当资源包含文件时，已上传的文件名称 |
No |
|
overwrite |
boolean |
是否重新启用已删除的资源 |
No |
|
needRefresh |
boolean |
更新后是否刷新页面列表 |
No |
IsRepeatResponse
是否重复的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
isRepeat |
boolean |
是否重复 |
Yes |
AssetIsRepeatRequest
资源是否重复的请求体
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
schema |
string |
资源类型
Enum:
|
Yes |
|
name |
string |
资源名称 |
Yes |
GeneralResponse
返回id的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
被操作对象的id |
Yes |
4.2 案例接口相关类型
Environment
环境
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
timeOfDay |
double |
当前时间 |
Yes |
|
lightIntensity |
double |
光照强度 |
Yes |
|
ambientLightIntensity |
double |
环境光 |
Yes |
|
artificialLightIntensity |
double |
人造光 |
Yes |
|
heightAngle |
double |
高度角 |
Yes |
|
cloudDensity |
double |
云密度 |
Yes |
|
rainDensity |
double |
雨密度 |
Yes |
|
fogDensity |
double |
雾密度 |
Yes |
|
snowDensity |
double |
雪密度 |
Yes |
|
ground |
{ “enable” : boolean, “humidityLevel” : double, “dirtyLevel” : double, “adhesionCoefficient” : double } |
地面参数 |
Yes |
|
id |
string |
环境id |
Yes |
|
name |
string |
环境名称 |
Yes |
|
schema |
string |
资源类型
Enum:
|
Yes |
|
thumbnail |
string |
缩略图路径 |
No |
|
builtIn |
boolean |
是否为内置 |
No |
|
classId |
string |
类型 |
Yes |
|
time |
string |
时间 |
Yes |
EnvironmentList
案例环境列表
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byId |
{ [id]: Environment } |
环境实例id和环境数据作为键值对的对象 |
Yes |
|
allIds |
[ string ] |
环境实例id列表 |
Yes |
CaseSensorList
案例路测传感器列表
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byId |
{ [id]: Sensor } |
传感器实例id和传感器数据作为键值对的对象 |
Yes |
|
allIds |
[ string ] |
传感器实例id列表 |
Yes |
JudgementList
案例判定列表
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byId |
object |
判定实例id和数据作为键值对的对象 |
Yes |
|
allIds |
[ string ] |
判定实例id |
Yes |
SpecialAreaList
案例特殊区域列表
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byId |
object |
特殊区域实例id和数据作为键值对的对象 |
Yes |
|
allIds |
[ string ] |
特殊区域实例id |
Yes |
MapGenConfigItem
案例地图配置
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
enable |
boolean |
是否自动生成道路周边场景 |
Yes |
|
distance |
double |
间距 |
No |
|
distanceToRoadVerge |
double |
距路边的距离 |
No |
MapGenConfig
案例地图配置列表（键值对）
Record<string, MapGenConfigItem >
CommunicationConfig
通信公共参数设置
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
frequencyBand |
double |
频段 |
Yes |
|
bandwidth |
double |
带宽 |
Yes |
|
maxRadiu |
double |
最大覆盖范围半径 |
Yes |
|
protocol |
string |
协议
Enum:
|
Yes |
|
model |
string |
无线通信传输模型
Enum:
|
Yes |
|
psSwitch |
boolean |
是否启用无线通信性能仿真 |
Yes |
CaseData
案例数据
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
environments |
环境配置 |
Yes |
|
|
sensors |
路侧传感器配置 |
Yes |
|
|
judgements |
判定配置 |
Yes |
|
|
specialAreas |
特殊区域配置 |
Yes |
|
|
mapGenConfig |
地图配置 |
No |
|
|
communicationParams |
通信公共参数设置 |
No |
|
|
schema |
string |
类型
Enum:
|
Yes |
|
caseId |
string |
案例id |
Yes |
CaseCreateOptions
创建案例选项
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
envId |
string |
指定环境id |
No |
|
mapGenConfig |
地图配置 |
No |
|
|
autoMapScene |
boolean |
省略 |
No |
|
scenario |
{ “vehicleId” : string, “content” : string } |
省略 |
No |
|
taskId |
string |
已完成任务存为案例时使用，任务id |
No |
|
changeType |
boolean |
已完成任务存为案例时使用，表示是否把主车转换为对手车 |
No |
CaseCreateRequest
创建案例请求体
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
casedata |
案例数据 |
No |
|
|
options |
创建案例选项 |
No |
|
|
needRefresh |
boolean |
创建完成后是否更新页面 |
No |
CaseSaveResponse
更新案例的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
code |
double |
成功时为0 |
No |
|
msg |
string |
返回信息 |
No |
CaseDefVehicle
案例主车配置
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
案例主车资源id |
Yes |
|
name |
string |
主车名称 |
Yes |
|
classId |
string |
主车模型id |
Yes |
|
physicalGradeSensor |
double |
物理传感器数量 |
Yes |
|
category |
string |
省略 |
No |
|
instanceId |
string |
案例配置主车实例id |
No |
|
instanceName |
string |
案例中的实例名称 |
No |
|
instanceColor |
string |
实例配色 |
No |
CaseDef
案例
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
schema |
string |
类型
Enum:
|
Yes |
|
id |
string |
案例id |
Yes |
|
name |
string |
案例名称 |
Yes |
|
userId |
string |
用户名称 |
Yes |
|
created |
double |
创建时间 |
Yes |
|
lastModified |
double |
修改时间 |
Yes |
|
opponent |
object |
案例类型 |
Yes |
|
categoryId |
string |
目录id |
No |
|
mapId |
string |
地图id |
Yes |
|
mapName |
string |
地图名称 |
No |
|
vehicles |
[ CaseDefVehicle ] |
主车信息列表 |
No |
|
vehicleId |
[ string ] |
主车id列表 |
No |
|
tags |
[ string ] |
标签列表 |
No |
|
notes |
string |
备注 |
No |
|
builtIn |
boolean |
是否内置的 |
No |
|
joinway |
string |
省略 |
No |
|
thumbnail |
string |
缩略图 |
No |
|
roadSidePhysicalGradeSensor |
double |
路侧物理传感器数量 |
No |
PartialCaseDef
Partial< CaseDef >
表示 CaseDef 的每个键值对都是可选的
CaseSaveRequest
更新案例请求体
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
casedef |
案例信息 |
No |
|
|
casedata |
案例配置数据 |
Yes |
|
|
needRefresh |
boolean |
更新后是否更新页面 |
No |
CaseCloneResponse
案例克隆响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
clonedIds |
[ string ] |
复制成功后案例 id |
Yes |
CaseCloneRequest
案例克隆请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
ids |
[ string ] |
需要复制的案例的 id |
Yes |
|
categoryId |
string |
复制进文件夹的 id |
Yes |
CaseMoveResponse
移动案例响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
movedIds |
[ string ] |
案例id列表 |
Yes |
CaseMoveRequest
移动案例请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
ids |
[ string ] |
需要移动的案例的 id |
Yes |
|
categoryId |
string |
目的文件夹的 id |
Yes |
IRow
案例目录
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
目录id |
Yes |
|
parentId |
string |
父目录id |
No |
CaseExportRequest
导出案例请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
caseIds |
[ string ] |
案例id列表 |
Yes |
|
includesVehicle |
boolean |
是否包含主车 |
Yes |
|
includesMap |
boolean |
是否包含地图 |
Yes |
|
format |
string |
导出的案例格式
Enum:
|
Yes |
|
pathinfo |
[ IRow ] |
导出案例的目录信息 |
Yes |
CaseImportResponse
导入案例响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
导入文件的md5码 |
Yes |
|
isFixed |
boolean |
是否修复格式错误的案例 |
No |
|
reasons |
[ [ string ] ] |
错误原因 |
No |
Recordstringstring
键值对（值的类型是字符串）
Reacord<string, string>
CaseOverrideInfo
导入案例时的主车和地图信息
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
上传文件内包含的案例（包含相对路径） |
Yes |
|
name |
string |
名称 |
Yes |
|
replaceMap |
boolean |
是否替换地图 |
Yes |
|
replaceVehicle |
boolean |
是否替换主车 |
Yes |
|
missingMap |
boolean |
是否缺失地图 |
Yes |
|
missingVehicle |
boolean |
是否缺失主车 |
Yes |
|
overrideMapId |
string |
要替换为的地图id |
Yes |
|
overrideVehicleId |
string |
要替换为的主车id |
Yes |
|
mapId |
string |
可忽略 |
No |
|
mapName |
string |
可忽略 |
No |
|
vehicleIds |
[ string ] |
可忽略 |
No |
|
path |
string |
可忽略 上传文件内包含的案例（包含相对路径）+ ext |
No |
CaseImportRequest
导入案例的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
上传 file 文件时的 uploadId（uuid） |
Yes |
|
file |
string |
file 的 md5 （在服务端用作临时文件名称） |
Yes |
|
category |
string |
目录文件夹 |
Yes |
|
includesVehicle |
boolean |
是否包含主车 |
Yes |
|
includesMap |
boolean |
是否包含地图 |
Yes |
|
includes |
[ string ] |
上传文件内包含哪些案例（包含相对路径） |
Yes |
|
needRemoveCases |
传空对象就可以{} |
Yes |
|
|
overrides |
[ CaseOverrideInfo ] |
传空数组就可以[] |
Yes |
|
categoryInfo |
传空对象就可以{} |
Yes |
|
|
includesStopTrigger |
boolean |
添加默认案例结束条件（仿真时间>60s） |
No |
|
excludeVehicleIds |
[ string ] |
传空数组就可以[] |
No |
|
excludeMapIds |
[ string ] |
传空数组就可以[] |
No |
|
replaceVehiclesMap |
被替换的主车 |
No |
|
|
replaceMapIds |
[ string ] |
被替换的地图（从资源库删除） |
No |
|
name |
string |
上传文件的名称 |
Yes |
|
ext |
string |
file 后缀 （zip、sim、xosc） |
Yes |
|
ignoreNameChecking |
boolean |
标签列表 |
No |
|
tags |
[ string ] |
标签列表 |
No |
|
notes |
string |
备注 |
No |
|
needFix |
boolean |
是否需要验证失败的案例 |
No |
|
needFixCases |
[ string ] |
验证失败并需要修复的案例 |
No |
CheckSensorsRequest
检测案例配置的传感器是否过载的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
caseIds |
[ string ] |
需要运行的案例 id |
Yes |
|
maxWay |
double |
系统支持的传感器数量 |
Yes |
|
vehicleId |
string |
运行案例使用的主车 id, 不指定主车传”-1” |
Yes |
4.3 地图接口相关类型
CreateMapParams
创建地图参数
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
category |
string |
地图类型（目录）
Enum:
|
Yes |
|
id |
string |
默认是“” |
Yes |
|
name |
string |
名称 |
Yes |
|
size |
double |
从xodr解析获取 |
Yes |
|
ppm |
double |
从xodr解析获取 |
Yes |
|
bgColor |
string |
从xodr解析获取 |
Yes |
|
reproject |
boolean |
从xodr解析获取 |
Yes |
|
reprojectOrigin |
boolean |
从xodr解析获取 |
Yes |
|
reprojectOriginLat |
double |
从xodr解析获取 |
Yes |
|
reprojectOriginLng |
double |
从xodr解析获取 |
Yes |
|
tags |
[ string ] |
标签列表 |
No |
|
notes |
string |
备注 |
No |
MapCreateRequest
创建地图请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
params |
JSON 需要字符化 |
Yes |
|
|
xodr |
File |
xodr 文件 |
Yes |
|
thumbnail |
File |
地图缩略图文件 |
No |
4.4 文件接口相关类型
UploadFileParams
上传文件参数
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
uploadId |
string |
uuid |
Yes |
|
category |
string |
业务类型 |
Yes |
|
filename |
string |
文件名称 |
Yes |
|
index |
double |
chunks序号 |
Yes |
|
chunks |
double |
分几次上传 |
Yes |
|
filesize |
double |
当前chunk大小 |
Yes |
FileUploadRequest
上传文件请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
params |
JSON 需要字符化 |
Yes |
|
|
file |
File |
文件 |
Yes |
4.5 主车接口相关类型
VehicleList
主车列表
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byId |
{[id]: VehicleItem } |
主车id和数据作为键值对的对象 |
Yes |
|
allIds |
[ string ] |
主车资源id |
Yes |
BasePageResponseVehicleList
获取主车列表的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
list |
主车列表 |
Yes |
|
|
page |
double |
当前页码 |
Yes |
|
total |
double |
主车总数 |
Yes |
VehicleCreateRequest
创建主车的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
name |
string |
主车名称 |
Yes |
|
classId |
string |
主车模型 id |
Yes |
|
overwriteId |
string |
要覆盖的主车 id |
No |
DataBlock
键值对
Reacord<string, any>
VehicleDynamics
主车动力学
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
ID |
Yes |
|
name |
string |
名称 |
Yes |
|
schema |
string |
资源类型
|
Yes |
|
builtIn |
boolean |
是否内置 |
No |
|
type |
string |
动力学类型
Enum:
|
Yes |
|
params |
动力学参数 |
Yes |
VehicleCtl
主车控制器
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
ID |
Yes |
|
type |
string |
类型 |
Yes |
|
name |
string |
名称 |
Yes |
|
params |
控制器参数 |
Yes |
|
|
aideId |
string |
辅助控制器id |
No |
|
builtIn |
boolean |
是否内置 |
No |
SensorOutput
传感器输出设置
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byType |
{[selected]: any} |
输出配置 |
Yes |
|
selected |
string |
输出格式 |
Yes |
Sensor
传感器
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
ID |
Yes |
|
name |
string |
名称 |
Yes |
|
schema |
string |
资源类型
Enum:
|
Yes |
|
category |
string |
目录id |
Yes |
|
thumbnail |
string |
缩略图路径 |
No |
|
builtIn |
boolean |
是否内置 |
No |
|
classId |
string |
类型 |
Yes |
|
x |
double |
x方向位置 |
Yes |
|
y |
double |
y方向位置 |
Yes |
|
z |
double |
z方向位置 |
Yes |
|
roll |
double |
倾斜角 |
Yes |
|
pitch |
double |
倾斜角 |
Yes |
|
yaw |
double |
倾斜角 |
Yes |
|
params |
传感器参数 |
Yes |
|
|
includeIds |
[ string ] |
融合传感器时，融合了哪些传感器 |
No |
|
output |
传感器输出设置 |
No |
|
|
disabled |
boolean |
是否禁用 |
No |
|
presetId |
string |
非内置时，表示原始传感器id |
No |
VehicleItem
主车
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
ID |
Yes |
|
name |
string |
名称 |
Yes |
|
thumbnail |
string |
缩略图路径 |
No |
|
builtIn |
boolean |
是否是内置 |
No |
|
schema |
string |
类型
Enum:
|
Yes |
|
classId |
string |
模型id |
Yes |
|
driver |
string |
忽略
Enum:
|
Yes |
|
dynamics |
动力学参数 |
Yes |
|
|
controller |
控制器 |
Yes |
|
|
category |
string |
主车类型
Enum:
|
Yes |
|
sensors |
[ Sensor ] |
传感器配置 |
Yes |
PartialVehicleItem
Partial< VehicleItem >
表示 VehicleItem 的每个键值对都是可选的
VehicleSaveResponse
更新主车的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
更新后的数据 |
Yes |
VehicleSaveRequest
更新主车的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
主车信息 |
Yes |
|
|
overwrite |
boolean |
如果当前主车已删除，是否重新启用 |
No |
|
needRefresh |
boolean |
更新后是否刷新页面 |
No |
IsRepeatRequest
主车或测试集名称是否重复的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
name |
string |
名称 |
Yes |
VehiclesCreateRequest
导入主车的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
vehicleData |
导入主车的数据 |
Yes |
VehicleReplaceRequest
更新主车资源文件的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
params |
{ “type” : “controller” or “dynamics” } |
替换控制器或者动力学相关的资源，传参时需要字符串化 |
Yes |
|
file |
File |
要替换的资源文件 |
Yes |
4.6 测试集接口相关类型
SuiteList
测试集列表
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
byId |
{[id]: Suite } |
suiteid和数据的键值对 |
Yes |
|
allIds |
[ string ] |
测试集id列表 |
Yes |
SuiteCreateRequest
创建测试集的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
parentId |
string |
父目录id |
Yes |
|
name |
string |
名称 |
Yes |
Suite
测试集
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
schema |
string |
类型
Enum:
|
Yes |
|
id |
string |
ID |
Yes |
|
name |
string |
名称 |
Yes |
|
userId |
string |
用户id |
Yes |
|
parentId |
string |
父目录id |
Yes |
|
caseIds |
[ string ] |
包含哪些用例 |
No |
PartialSuite
Partial< Suite >
表示 Suite 的每个键值对都是可选的
SuiteSaveResponse
更新测试集的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
更新后的测试集 |
Yes |
SuiteSaveRequest
更新测试集的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
要更新的测试集 |
Yes |
|
|
caseIds |
[ string ] |
要操作的案例id列表 |
No |
|
delete |
boolean |
表示删除或者添加caseIds |
No |
4.7 页面目录接口相关类型
CategoryMoveRequest
移动测试集的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
dragId |
string |
要移动的测试集id |
Yes |
|
dropId |
string |
移动到相对这个测试集的某个位置 |
Yes |
|
state |
string |
相对dropId的位置
Enum:
|
Yes |
4.8 对手元素接口相关类型
ObstacleNpc
对手元素
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
id |
string |
ID |
Yes |
|
category |
string |
类型 |
Yes |
|
schema |
string |
元素分类
Enum:
|
Yes |
|
assetId |
double |
资源id |
No |
|
name |
string |
名称 |
No |
|
uicategory |
string |
目录id |
No |
|
oscSubCategory |
string |
osc类型 |
No |
|
resourceType |
string |
元素资源的类型
Enum:
|
No |
|
resourceMd5 |
string |
资源文件md5 |
No |
|
thumbnail |
string |
缩略图路径 |
Yes |
|
interpo |
string |
忽略
Enum:
|
Yes |
|
semantic |
string |
语义类型
Enum:
|
Yes |
|
builtIn |
boolean |
是否内置 |
No |
|
modelPath |
string |
外挂模型路径 |
No |
|
aniPath |
string |
外挂动画路径 |
No |
|
dimensions |
{ “length” : double, “width” : double, “height” : double } |
尺寸 |
No |
|
originToCenter |
{ “x” : double, “y” : double, “z” : double } |
原点位置 |
No |
PartialObstacleNpc
Partial< ObstacleNpc >
表示 ObstacleNpc 的每个键值对都是可选的
BasePageResponseObstacleNpcArray
获取对手元素列表的响应
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
list |
[ ObstacleNpc ] |
对手元素列表 |
Yes |
|
page |
double |
当前页码 |
Yes |
|
total |
double |
总数 |
Yes |
ObstacleCreateRequest
创建对手元素的请求
|
名称 |
类型 |
说明 |
是否必填 |
|---|---|---|---|
|
data |
元素参数 |
Yes |
|
|
sourceId |
string |
已上传的文件 md5 |
Yes |
|
thumbnailId |
string |
已上传的缩略图的 md5 |
No |