Case Library / Use Case / Run Task Example¶
6. Case Library Operations¶
6.1 Create Case Library¶
API¶
"""
def create_category_api(self, cate_name:str):
'''
Create Case Library
@param cate_name: Case Library Name
@return:
'''
payload = {"parentId": "", "name": cate_name}
return self.send(self.post,self.categories_url,json=payload).json()
"""
Implementation¶
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):
'''
Create Case Library
@param cate_name: Case Library 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:
'''
Create Case Library
@param cate_name:
@return:
'''
if not cate_name:
cate_name = self.category_name
print(f"Case Library Directory Name: {cate_name}")
response = self.create_category_api(cate_name)
cate_id = response["data"]["categoryId"]
print(f"Case Library Directory ID: {cate_id}")
return cate_id
def main():
suite = Suite()
suite.create_category()
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
Return Value¶
{'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"}}
Case Library Directory ID: feeea7d0-439e-4796-a239-0f7d8f484d01
6.2 Get Case Library List¶
API¶
"""
def get_category_api(self):
"""
Get Case Library List
@return: response
"""
return self.send(self.get, self.categories_url).json()
"""
Implementation¶
def get_category_api(self):
"""
Get Case Library List
@return: response
"""
return self.send(self.get, categories_url).json()
def get_category(self,userid:str=None)->dict:
"""
Get Case Library
@param userid: User ID, desc: admin->builtIn
@return:
"""
cate_dict={"data":{},"parentId":{}}
response = self.get_category_api()
cate_dict["total"] = response["data"]["total"]
if not userid:
print("Fetching all current case library directory information")
for one in response["data"]["categories"]:
cate_dict["data"].setdefault(one["name"],one["id"])
cate_dict["parentId"].setdefault(one["name"],one["parentId"])
else:
print("Fetching case library directory info for user {}".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("Obtained Case Library->%s"%str(cate_dict))
return cate_dict
def main():
suite = Suite()
suite.create_category()
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
Return Value¶
Obtained Case Library->{'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 Get Case Library List¶
API¶
"""
def get_category_case_api(self,cate_id:str):
"""
Get All Use Cases in the Current Case Library
@param cate_name: Case Library Name
@return:
"""
return self.send(self.get,self.get_cases_url.format(category=cate_id)).json()
"""
Implementation¶
def get_category_case_api(self,cate_id:str):
"""
Get All Use Cases in the Current Case Library
@param cate_name: Case Library Name
@return:
"""
return self.send(self.get,get_cases_url.format(category=cate_id)).json()
def get_cases_category(self, cate_id: list):
"""
Get All Use Cases in the Current Case Library
@param cate_id: Case Library Directory 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 input 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 = []
# If the number of Use Cases is within 100
if cases_num <= page_size:
for i in range(cases_num):
update_case_id(response=response, page_size=i)
else:
# If the number of Use Cases exceeds 100, it fetches a fixed 100 Use Cases per page.
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 is integer division, indicating it has not reached the last page
# 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)
# Reached the last page. Since the number of Use Cases is not fixed, the number to fetch needs to be calculated separately.
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()
# The Case Library name can be specified by yourself.
cate_id = [cate_id_dict["data"][catgory_name]]
suite.get_cases_category(cate_id)
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main('Turn Conflict')
Return Value¶
{'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-5d897e35064","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-5d897e35064","lastModified":1642420795134,"vehicleId":["simonedriver"],"categoryId":"0863fa00-4618-11e9-803b-61333d8d0423"}],"page":1}}
6.4 Get Use Case ID for Subset Directory in Case Library¶
API¶
"""
def get_category_api(self):
"""
Get Case Library List
@return: response
"""
return self.send(self.get, categories_url).json()
"""
Implementation¶
def get_all_cate_id_list(self, cate_id) -> list:
"""
@param cate_id: Case Library 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
Usage Example¶
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")
Return Value¶
all_cate_list: ['dbaf8d6e-af07-44a6-b7c7-d8b7f2fd6616', 'c8557561-5729-43e1-9d65-91498f7a1129']
7. Use Case Operations¶
7.1 Get Use Case Information¶
API¶
"""
def get_case_detail_api(self, case_id:str):
"""
Get Use Case Information
@param params: /{case_id}/data
@return: response
"""
url = self.case_details_url.format(case_id=case_id)
return self.send(self.get, url).json()
"""
Implementation¶
def get_case_detail_api(self, case_id: str):
"""
Get Use Case Information
@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: Test Case ID
@return:
"""
response = self.get_case_detail_api(case_id)
return response
def main(catgory_name):
suite = Suite()
cate_id_dict = suite.get_category()
# The Case Library Name can be customized.
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])
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main('Turn Conflict')
Return Value¶
{'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 Use Case Import¶
API¶
"""
def import_case_api(self, payload: dict):
"""
Use Case Import
@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):
"""
File MD5 Check
@param file_path: File Path
@return:
"""
# global url
def file_md5(filename: str):
"""
MD5 encryption using file contents
@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()):
"""
Upload File
@param file_path: File Path
@param index: Upload Index
@param chunks: Total Chunks
@param category: Category
@param file_md5: File 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 encryption using file contents
@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
"""
Implementation¶
def file_md5_check(self, file_path: str, raw: bool = False):
"""
File MD5 Check
@param file_path: File Path
@return:
"""
# global url
def file_md5(filename: str):
"""
MD5 encryption using file contents
@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={}" # Verify file 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} File deleted successfully')
elif os.path.isdir(sim_file):
shutil.rmtree(sim_file)
# print(f'{sim_file} Directory deleted successfully')
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("Updated 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') # Bug when filename contains two .sim extensions
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("Reached here...")
zfile.extractall(path=extract_dir)
# print("Reached here 2...")
except Exception as e:
raise Exception(f"Failed to decompress {zip_file}, exception information: {e}")
temp = list(filter(lambda x: 'json' in x, os.listdir(extract_dir)))
# print("temp:",temp)
if temp == []: # From v3.3.0, Use Cases are folders after decompression, so the previous statement results in an empty temp.
for d in os.listdir(extract_dir):
# print("Contents of extract_dir:", d)
if os.path.isdir(os.path.join(extract_dir, d)):
temp.append(d)
# print("Temp test temp:", temp)
all_case_id = list(set(temp)) # deduplicate
# 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("Reached here 3...")
return old_case_id
all_case_id = list(map(lambda x: x.replace('.json', ''), temp))
all_case_id = list(set(all_case_id)) # deduplicate
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()):
"""
Upload File
@param file_path: File Path
@param index: Upload Index
@param chunks: Total Chunks
@param category: Type
@param file_md5: File MD5
@return:
"""
global url
# self.headers = {}
self.headers = {"ProjectId": 'default'} # Jianghuai 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 encryption using file contents
@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):
"""
Use Case Import
@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):
"""
Import Use Case to the specified Case Library Directory
@param file_path: Use Case File Path
@param file_name: Use Case File Name
@param cate_id: Case Library Directory ID
@return: [Important] The returned ID here is not the Use Case ID after successful import, but the original ID within the imported Use Case file.
"""
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"Importing Use Case does not support {file_name.split('.')[1]} format, Use Case interrupted."
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: Round up
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("Temp test import_case case_payload:", case_payload)
# print("-----import_case_api response -->>> ", response)
return case_id_list, response # [Important] The returned ID here is not the Use Case ID after successful import, but the original ID within the imported Use Case file.
def import_flow(self, category_name: str, file_path: str, file_name: str):
'''
File Import Flow
Importable file types are .xosc or .sim formats.
@param category_name: Case Library Name
@param file_path: File Path, e.g., r"D:\System
@param file_name: File Name, e.g., 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()
# The Case Library Name can be customized.
# 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")
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
Return Value¶
response:{"code":0,"data":{"id":"2222222222"}}
7.3 Export Use Cases¶
API¶
"""
def export_case_api(self, payload: dict) -> json:
"""
Export Test Case
@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:
"""
Download Test Case
@param pake_id: pake_id
@return:response
"""
return self.send(self.get, download_case_url.format(pake_id=pake_id)).content
"""
Implementation¶
def export_case_api(self, payload: dict) -> json:
"""
Export Test Case
@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:
"""
Download Test Case
@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:
"""
Get Binary Test Case Content
@param case_id: case_id_list
@return: bytes
"""
# Get 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: List of Case Names
@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: Case Library Name
@param downloadPath: Download Path
@param downloadName: Name of Download Package
@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)
# Get the actual format parameter
_format = inspect.getfullargspec(self.export_case)[3][0]
# Download file path
downloadPake = os.path.join(downloadPath, downloadName + _format)
with open(downloadPake, 'wb') as f:
f.write(downloadResponse)
print(f"Exported case path: {downloadPake}")
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
download_case_flow(category_name, downloadPath, downloadName)
Return Value¶
run_task response：{'code': 0, 'data': {'sessionId': '3f90ef1b-f5e1-4ab7-b24f-7f94b9d7f1cb', 'taskIds': ['d2973d5b-7fef-4d83-af5d-ec108db815df']}}
7.4 Use Case Scenario Editing¶
API¶
"""
def case_def_api(self, case_id: str):
"""
Use Case Summary
@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):
"""
Use Case Details
@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):
"""
Use Case Editing
@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):
"""
Get Ego Vehicle Information
@return:
"""
return self.send(self.get, get_vehicles_url).json()
"""
Implementation¶
def case_def_api(self, case_id: str):
"""
Use Case Summary
@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):
"""
Use Case Details
@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):
"""
Use Case Editing
@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):
"""
Get Ego Vehicle Information
@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"Found Ego Vehicle name for vehicle ID {vehicle_id}: {data[key]['name']}")
return data[key]["name"]
print(f"No Ego Vehicle found with vehicle ID {vehicle_id}")
return None
def update_case(self, case_id: str, **kwargs):
"""
Edit Use Case
@param case_id: Use Case ID
@param map_name: Map Name
@param map_id: Map ID
@param vehicle_id: Ego Vehicle ID
@param AbsoluteTargetSpeed: Ego Vehicle initial state absolute target speed
@param WorldPosition_x: Ego Vehicle initial state world position x
@param WorldPosition_y: Ego Vehicle initial state world position y
@param WorldPosition_z: Ego Vehicle initial state world position z
@return: Use Case with updated Ego Vehicle, map, or other information
"""
# Extract parameters (with default values)
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("Temp test original case_def:", case_def)
# print("Temp test original 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)
# Modify Ego Vehicle
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)
# Modify Map
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}
# Modify Ego Vehicle initial state world position
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
# Modify Ego Vehicle initial state absolute target speed
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("Temp test update_case response:", response)
if response['code'] != 0:
raise Exception("Use Case editing error, API returned: " + response)
return response
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
test = Suite()
test.update_case(case_id)
Return Value¶
update_case_api response {'code': 0, 'msg': 'ok', 'data': {'code': 0, 'msg': 'ok'}}
7.5 Save Replay Use Case¶
API¶
"""
def convert_replay_case_api(self, task_id: str, category_id: str, replay_case_name: str):
"""
Convert Replay Use Case
@param case_id: The Use Case ID to be converted
@param category_id: Case Library Directory ID to store the case
@param replay_case_name: Replay case name after conversion
@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))
"""
Implementation¶
def convert_replay_case_api(self, task_id: str, category_id: str, replay_case_name: str):
"""
Convert Replay Use Case
@param case_id: The Use Case ID to be converted
@param category_id: Case Library Directory ID to store the case
@param replay_case_name: Replay case name after conversion
@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))
Usage Example¶
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-8a66ec42525d", replay_case_name="TestCase")
Return Value¶
{code:0,success:True}
8. Task Operations¶
8.1 Run Task¶
API¶
"""
def create_task_api(self,payload:dict):
"""
Create 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()
"""
Implementation¶
Note: This feature differentiates between versions prior to and including 3.4, and versions 3.5 and later. Please be aware of the distinction.
def create_task_api(self,payload:dict):
"""
Create 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 test 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": "Ego Vehicle", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "Default Controller", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "Autonomous Driving Controller", "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": "Default Controller", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "Autonomous Driving Controller", "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"} # Legacy cloud: built-in cases may have issues
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("Requesting simone version API to extract version number: " + version)
cate_id_dict = suite.get_category()
# The Case Library name can be customized
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")
# Use Case naming logic
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)
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538","test_4-copy")
Return Value¶
run_task response：{'code': 0, 'data': {'sessionId': '3f90ef1b-f5e1-4ab7-b24f-7f94b9d7f1cb', 'taskIds': ['d2973d5b-7fef-4d83-af5d-ec108db815df']}}
8.2 End Task¶
API¶
"""
def stop_task_api(self, payload: dict):
"""
Task Stop 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 Stop API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_sessions_url, json=payload).json()
"""
Implementation¶
def stop_task_api(self, payload: dict):
"""
Task Stop 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 Stop 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):
"""
Stop tasks and sessions
@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()
# The Case Library Name can be customized.
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")
# Use Case naming logic
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])
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538", "test_4-copy")
Return Value¶
{'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 Pause Task & Resume¶
API¶
"""
def pause_task_api(self,task_id:str):
"""
Pause Task
@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):
"""
Resume Task After Pause
@param task_id:
@return:
"""
return self.send(self.post, self.resume_task_url.format(task_id=task_id),json={}).json()
"""
Implementation¶
def pause_task_api(self,task_id:str):
"""
Pause Task
@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):
"""
Resume Task After Pause
@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()
# The Case Library Name can be customized.
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")
# Use Case naming logic
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")
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538", "test_4-copy")
Return Value¶
{'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 Delete Task¶
API¶
"""
def delete_sessions_api(self, payload: dict):
"""
Sessions Delete 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 Delete API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_task_url, json=payload).json()
"""
Implementation¶
def delete_sessions_api(self, payload: dict):
"""
Sessions Delete 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 Delete 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):
"""
Delete tasks and sessions
@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()
# The Case Library Name can be customized.
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")
# Use Case naming logic
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)
Usage Example¶
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("Automation_5538", "test_4-copy")
Return Value¶
{'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 Determine if a Task Is Running¶
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]
"""
Implementation¶
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):
"""
Determine if the Use Case Has Finished Running
@param task_id: Task ID
@param case_name: Use 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)
Usage Example¶
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)
# Ego Vehicle Control Related Logic
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}")
# Use Case Name Logic
if case_name:
case_id_list = cases_dict[case_name]
print(f" case_name:{case_name}")
# Start Use Case
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])
# Wait for Use Case to Finish Running
# falg = suite.suite_queue_check()
# if falg is True:
# # Get Use Case Run Result
# 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")
Return Value¶
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"}}