案例库 / 案例 / 运行任务示例¶
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()[0]
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__':
download_case_flow(category_name, downloadPath, downloadName)
返回值¶
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, **kwargs):
"""
编辑案例
@param case_id: 案例ID
@param map_name: 地图名称
@param map_id: 地图ID
@param vehicle_id: 主车ID
@param AbsoluteTargetSpeed: 主车初始状态绝对目标速度
@param WorldPosition_x: 主车初始状态世界位置x
@param WorldPosition_y: 主车初始状态世界位置y
@param WorldPosition_z: 主车初始状态世界位置z
@return: 更新主车或地图等信息后的案例
"""
# 提取参数（设置默认值）
map_name = kwargs.get("map_name")
map_id = kwargs.get("map_id")
vehicle_id = kwargs.get("vehicle_id")
absolute_target_speed = kwargs.get("absolute_target_speed", 0.0)
worldposition_x = kwargs.get("worldposition_x", 0.0)
worldposition_y = kwargs.get("worldposition_y", 0.0)
worldposition_z = kwargs.get("worldposition_z", 0.0)
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
case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"]["Properties"][
"Property"][0]["value"] = vehicle_id
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 worldposition_x:
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('TeleportAction'):
PrivateAction['TeleportAction']['Position']['WorldPosition']['x'] = worldposition_x
PrivateAction['TeleportAction']['Position']['WorldPosition']['y'] = worldposition_y
PrivateAction['TeleportAction']['Position']['WorldPosition']['z'] = worldposition_z
# 修改主车初始状态绝对目速度
if absolute_target_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'] = absolute_target_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)
返回值¶
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, case_ids: List[str], vehicle_id: Optional[str] = None, withEvaluation: bool = False, version: str = "3.5.0", type: str = "worldsim")-> Tuple[str, List[str]]:
"""
运行测试任务集合
Run task set
@param caseid: caseid
@param vehicle_id: vehicle_id
@param withEvaluation: withEvaluation
@param version: version
@return:
"""
task_name = "AutoTest_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
payload = {}
if version >= "3.6.3":
payload = {"caseIds": case_ids, "taskName": task_name,
"type": type, "withEvaluation": False, "vehicleConfig": {
"Ego": {"instanceId": "Ego", "id": "default", "name": "主车", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "默认控制器", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器", "algorithmId": "SimOneDriver"}]}},
"notification": {"type": "none", "condition": "any"}, "enableStateMachine": False, "priority": 0}
elif version >= "3.5.0":
payload = {"caseIds": case_ids,
"taskName": task_name,
"withEvaluation": withEvaluation,
"controllers": [{"id": "Default", "name": "默认控制器", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器", "algorithmId": "SimOneDriver"}]}
elif version < "3.5.0":
payload = {"caseIds": case_ids, "taskName": task_name, "speed": 1}
# elif version > "3.1.1":
# payload = {"caseIds": case_ids, "taskName": task_name, "speed": 1}
# else:
# payload = {"caseIds":case_ids,"taskName":task_name,"withEvaluation":True,"type":"worldsim"} # 现代云端 内置案例有坑
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()
if not version:
version = suite.get_version()
print("请求simone version接口提取版本号为：" + version)
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(case_id_list, vehicle_id=vehicle_id, version=version)
调用示例¶
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"}}