import base64
import copy
import hashlib
import math
import random
import shutil
import zipfile
from os.path import exists
from typing import Optional, Tuple, List
import fire
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('en_US')
version_url = "version"
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)
""" Initialization Configuration Interfaces """
accept_url = Prefix.UserServer + "/users/agreement" # Accept agreement
init_config_url = Prefix.TaskServer + "/system/initConfig" # Initialize configuration
init_serve_url = Prefix.TaskServer + "/system/initServices" # Initialize services
conn_master_url = Prefix.TaskServer + "/system/connectMaster" # Connect to master
start_serve_url = Prefix.TaskServer + "/system/start" # Start services
""" Multi-Vehicle Add Node """
add_worker_url = Prefix.TaskServer + "/worker" # Add node
""" Login Interfaces """
user_check_url = Prefix.UserServer + "/users/check"
health = Prefix.UserServer + "/health" # Check service health
cloud_login_url = Prefix.UserServer + "/users/login" # Cloud login
standalone_login_url = Prefix.UserServer + "/users/standalone" # Standalone login -- tested
login_check = Prefix.UserServer + "/users/check" # user check
""" Case Library Directory Interfaces """
categories_url = Prefix.CaseServer + "/categories" # Case library directory --tested
get_categories_url = Prefix.CaseServer + "/categories?withCases=false" # Get all case libraries --tested
# get_cases_url = "api-case/cases?page=1&filter={category}&pageSize=100"
get_cases_url = Prefix.CaseServer + "/cases?page={page}&filter={category}&pageSize=100" # Get all use cases in current case library --tested
delete_categories_url = Prefix.CaseServer + "/categories/{}" # Delete case library directory
""" File Upload Interfaces """
file_md5_url = Prefix.AssetServer + "/files/{}?raw={}" # Verify file MD5 -- done
upload_file_url = Prefix.AssetServer + "/files/{}/upload?raw={}" # Upload file -- done
""" Test Case Interfaces """
case_import_url = Prefix.AssetServer + "/cases/import" # Import use case
case_def_url = Prefix.CaseServer + "/cases/{case_id}" # Use case summary -- done
case_details_url = Prefix.CaseServer + "/cases/{case_id}/data" # Use case details -- done
delete_case_url = Prefix.CaseServer + "/trash" # Delete use case -- done
convert_replay_case_url = Prefix.TaskServer + "/tasks/{id}/convert2case" # Convert replay use case
update_case_url = "api-asset/cases/update" # Edit use case
""" Test Task Interfaces """
create_task_url = Prefix.TaskServer + "/tasks" # Create task --tested
stop_task_url = Prefix.TaskServer + "/tasks/stop" # Stop task --done
stop_sessions_url = Prefix.TaskServer + "/sessions/stop" # Stop session --done
delete_task_url = Prefix.TaskServer + "/tasks/delete" # Delete task --done
delete_sessions_url = Prefix.TaskServer + "/sessions/delete" # Delete session --done
result_url = Prefix.TaskServer + "/tasks/task?taskIds={task_id}" # Use case run result details -- tested
pause_task_url = Prefix.TaskServer + "/sessions/{task_id}/pause" # Pause task
resume_task_url = Prefix.TaskServer + "/sessions/{task_id}/resume" # Resume task after pause
task_result_url = Prefix.TaskServer + "/tasks/task?taskIds={id}" # Use case run result details -- tested
""" Ego Vehicle Interfaces """
import_vehicle_url = Prefix.AssetServer + "/vehicles/import" # Import ego vehicle --done
delete_vehicle_url = Prefix.AssetServer + "/vehicles/{vehicles_id}" # Delete ego vehicle --done
get_vehicle_url = Prefix.AssetServer + "/vehicles" # Get ego vehicle information --done
""" Map Interfaces """
import_map_url = Prefix.AssetServer + "/maps" # Import map --done
delete_map_url = Prefix.CaseServer + "/maps/trash" # Delete map --done
get_map_url = Prefix.CaseServer + "/maps?pageSize=100" # Get map information
road_service_url = Prefix.TaskServer + "/sce" # Start road network service
laneTypeIfInside_url = Prefix.TaskServer + "/sce/{map_id}/location/laneTypeIfInside" # Check if point is inside lane
""" Data-Driven Interfaces """
import_data_driven_url = Prefix.AssetServer + "/dd" # Import data-driven source
delete_data_dirven_url = Prefix.AssetServer + "/dd/{id}" # Delete data-driven source
""" Resource Interfaces """
import_asset_url = Prefix.AssetServer + "/assets" # Import resource
asset_repeat_url = Prefix.AssetServer + "/assets/isrepeat" # Check resource duplication
get_asset_url = Prefix.AssetServer + "/assets?schema={}" # Get resource information
delete_asset_url = Prefix.AssetServer + "/assets/{}" # Delete resource
""" Use Case Export """
export_case_url = Prefix.AssetServer + "/cases/export" # Export use case
download_case_url = Prefix.AssetServer + "/download/{pake_id}" # Download use case
""" Test Suite Interfaces """
get_suites_url = Prefix.AssetServer + "/suites?isProjectItem={}&opponent=openscenario" # Get test suite
run_suite_url = Prefix.TaskServer + "/tasks" # Run test suite
get_taskassemble_url = Prefix.TaskServer + "/tasks?finished=true&page=0&pageSize=12&own=true&expandedIds={task_set_id}" # Get test case collection
queue_status_url = Prefix.TaskServer + "/tasks/queue?own=true" # Suite run status check -- tested
""" Judgment Interfaces """
get_judgements_url = Prefix.AssetServer + "/cases/{caseId}/judgements"
# get_judgement_url = get_judgements_url + "/{judgementId}"
get_judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
""" Test Plan Interfaces """
testPlan_list_url = Prefix.TaskServer + "/testPlan?page=1&pageSize=20&options={options}" # Get test plan list
testPlan_id_url = Prefix.TaskServer + "/testPlan/{id}" # Get test plan / Update test plan (including rename)
create_testPlan_url = Prefix.TaskServer + "/testPlan" # Create test plan
del_testPlan_url = Prefix.TaskServer + "/testPlan/delete" # Delete multiple test plans
del_testPlan_single_url = Prefix.TaskServer + "/testPlan/{id}" # Delete single test plan
execute_testPlan_url = Prefix.TaskServer + "/testPlan/execute" # Execute test plan
clone_testPlan_url = Prefix.TaskServer +"/testPlan/{id}/clone" # Clone test plan
isrepeat_testPlan_url = Prefix.TaskServer +"/testPlan/isrepeat?name={name}&type={type}" # Check if test plan name exists
# f"api-asset/cases/{caseId}/judgements/{judgementId}"
# url = f"api-asset/cases/{caseId}/judgements"
class UrlCodeType:
unquote = "unquote"
quote = "quote"
class File:
"""
File base class
"""
def __init__(self, file_path: str):
if not exists(file_path):
raise FileNotFoundError
self._file_path = file_path
self._data = None
class FileReader(File):
"""
Read 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", "ProjectId": 'default'}
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:
"""
Send request
method: Request method
url: Request URL
kwargs: Request parameters
"""
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):
"""
Cloud login API
@param userinfo:
@return:
"""
return self.send(self.post, cloud_login_url, json=userinfo).json()
def check_username_api(self, username: str):
"""
Validate username format
@param username:
@return:
"""
payload = {'username': username}
return self.send(self.post, login_check, json=payload).json()
def getCheckToken(self) -> str:
"""
Validate 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):
"""
Padding function to make encrypted data byte length a multiple of 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 encryption
@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 encryption 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):
"""
Cloud version login method
@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):
"""
Standalone version login method
@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
# Login entry
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):
'''
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 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:
"""
# Initialize a dict with data and parentId for storing case library info
cate_dict = {"data": {}, "parentId": {}}
# Call API to get case library data
response = self.get_category_api()
# Get total number of case libraries
cate_dict["total"] = response["data"]["total"]
# If no userid provided, get all case library directory info
if not userid:
print("Getting all current case library directories")
# Iterate all case libraries and store name-ID, name-parentID pairs
for one in response["data"]["categories"]:
cate_dict["data"].setdefault(one["name"], one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
else:
# If userid provided, get only that user's case library directory info
print("Getting case library directories for user {}".format(userid))
# Iterate all case libraries and filter those belonging to the specified user
for one in response["data"]["categories"]:
if one["userId"] == userid:
# Store user case library name-ID, name-parentID pairs
cate_dict["data"].setdefault(one["name"], one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
# Print retrieved case library info
print("Retrieved case libraries -> %s" % str(cate_dict))
# Return dict containing case library info
return cate_dict
def get_category_case_api(self, cate_id: str):
"""
Get all use cases in 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 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 case count is within 100
if cases_num <= page_size:
for i in range(cases_num):
update_case_id(response=response, page_size=i)
else:
# If case count exceeds 100, paginate with fixed 100 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 == 0 means we haven't 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)
# On the last page, calculate the exact remaining case count
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):
"""
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)
print(f"Called get_case_detail: {response}")
return response
def get_version(self):
res = self.send(self.get, url= version_url).json()
version = res["data"]["profile"]["version"]
return version
# def run_task(self, caseid: list, vehicle_id: str = None):
# """
# :param caseid:
# :return:
# """
#
# def creat_task_name():
# return "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
#
# taskname = creat_task_name()
# payload = {"caseIds": caseid, "taskName": taskname, "speed": 1}
# if vehicle_id:
# payload.update({"overrideVehicleId": vehicle_id})
# response = self.send(self.post, url=create_task_url, json=payload).json()
# print(f"run_task response: {response}") # Empty taskIds indicates run case error
# data = response['data']
# return data['sessionId'], data['taskIds']
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": "Ego", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "Default Controller", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "Auto-Drive 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": "Auto-Drive 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"} # Caution: built-in cases on modern cloud may cause 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 stop_task_api(self, payload: dict):
"""
Stop 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):
"""
Stop 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):
"""
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 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 delete_sessions_api(self, payload: dict):
"""
Delete 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):
"""
Delete 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):
"""
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 import_vehicle_api(self, payload: dict):
"""
Import ego vehicle
@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:
"""
Import ego vehicle
@param file_path: Ego vehicle file
@return: Ego vehicle 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):
"""
Get ego vehicle information
@return:
"""
return self.send(self.get, get_vehicle_url).json()
def get_vehicle(self, userid: str = None):
"""
Get ego vehicle information based on user info
@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("Retrieved ego vehicle info ->: {}".format(str(vehicle_data)))
return vehicle_data
def get_vehicle_id(self, vehicle_name: list) -> list and dict:
"""
Get ego vehicle ID by ego vehicle name
@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):
"""
Delete ego vehicle
@param vehicles_id: Ego vehicle ID
@return:
"""
url = delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
def delete_vehicle(self, id: str):
"""
Delete ego vehicle method
@param id: Ego vehicle ID
@return:
"""
response = self.delete_vehicles_api(id)
def get_suite_api(self, isProjectItem : bool):
"""
@param isProjectItem : True(team) or False(personal)
@return:
"""
url = get_suites_url.format(isProjectItem)
return self.send(self.get, url=url).json()
def get_suite(self, suite_name: str = None , isProjectItem : bool = True):
"""
@param isProjectItem : true(team) or false(personal)
:return:
"""
if isProjectItem: isProjectItem = "true"
else: isProjectItem = "false"
response = self.get_suite_api(isProjectItem)
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)
try :
if not suite_name:
return suite_dict
else:
# print("suite_dict[suite_name]:",suite_dict[suite_name])
return suite_dict[suite_name]
except Exception as e:
print("suite_name does not exist")
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,
"type" : "worldsim",
"enableStateMachine" : False,
"vehicleConfig" : {
"Ego" : {
"instanceId" : "Ego" ,
"id" : "default",
"name" : "Ego",
"classId" :"MKZ",
"algorithms" : [
{
"id" : "Default",
"name" : "Default Controller",
"algorithmId": "SimOneDriver",
}
]
}
},
"withEvaluation" : False
}
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):
"""
Check if 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)
def get_taskassemble_api(self, task_set_id):
'''
Get test case collection
@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:
"""
Get task_id for all test cases in a task set
@param task_set_id: Task set 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"------------------------Element {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):
"""
Get result of a specified 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:
"""
Get use case name, caseId, task_id, and run result from task set
@param task_id_list: Task ID list
@param task_set_id: Task set ID
@return:
"""
task_result_list = []
print(f"task_id_list ===== >>> {task_id_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):
"""
Import map API
@param payload: Payload dictionary
@param files: File parameters
@return:
"""
self.headers = {"projectId": "default"}
return self.send(self.post, import_map_url, data=payload, files=files).json()
def import_map(self, xodr_path: str, thumbnail_path: str):
"""
Import map method
@param xodr_path: Map file
@param thumbnail_path: Thumbnail image for the map
@return:
"""
map_name = "map_" + str(random.randint(0, 9999))
payload = {"params": json.dumps(
{"category": "customized", "id": "", "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"import map {xodr_path}success")
else:
raise f"import map {xodr_path} fail"
def get_map_api(self):
"""
Get map
@return:
"""
return self.send(self.get, get_map_url).json()
def get_map_id(self, userid: str = None, map_name: str = None):
"""
Get map information by condition, default returns all maps
@param userid: User ID
@param map_name: 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):
"""
Delete map
@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):
"""
Batch delete maps
@param id: Map ID list
@return:
"""
print("Delete map list -> {}".format(str(id)))
payload = {"ids": id}
self.delete_map_api(payload)
def get_judgements_api(self, caseId, judgementId=None):
"""
Get judgment information API
@param caseId: case ID
@param judgementId: Judgment 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):
"""
Get judgment information
@param caseId: case ID
@param judgementId: Judgment ID
@return:
"""
response = self.get_judgements_api(caseId, judgementId)
print("Retrieved extended judgment info: ", 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": "Collision", "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": "Collision", "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("Update extended judgment info")
print("status_code:", response.status_code)
""" ——————————————————————————【Added 2024-07-21: Edit Case and Related APIs】———————————————————————— """
def update_case(self, case_id: str, **kwargs):
"""
Edit use case
@param case_id: case ID
@param map_name: Map name
@param map_id: Map ID
@param vehicle_id: Ego vehicle ID
@param AbsoluteTargetSpeed: Ego vehicle initial absolute target speed
@param WorldPosition_x: Ego vehicle initial world position x
@param WorldPosition_y: Ego vehicle initial world position y
@param WorldPosition_z: Ego vehicle initial world position z
@return: Use case after updating ego vehicle or map info
"""
# Extract parameters (set defaults)
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 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 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("Edit use case error, API returned: " + response)
return response
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 get_vehicle_name(self, vehicle_id: str):
res = self.get_vehicle_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 ID {vehicle_id}: {data[key]['name']}")
return data[key]["name"]
print(f"Ego vehicle with ID {vehicle_id} not found")
return None
def update_case_api(self, payload: dict):
"""
Edit use case API
@param payload:
@return:
"""
# print("********* update_case_api payload********",payload)
self.headers.update({"Content-Type": "application/json"})
return self.send(self.post, update_case_url, data=json.dumps(payload)).json()
def testPlan_list_api(self):
"""
Get test plan list
"""
options= {"search":"","types":[],"vehicleIds":[],"algorithms":[],"triggerTypes":[],"notifications":[],"category":"worldsim"}
json_str = json.dumps(options, indent=4, ensure_ascii=False) # Prettify with indentation, preserve non-ASCII characters
# encoded_url = quote(json_str, safe="/:?=") # Preserve path separators, query symbols, and equal signs
return self.send(self.get, url=testPlan_list_url.format(options=json_str)).json()
def test_testPlan_list(self):
"""
Get test plan list
"""
response = self.testPlan_list_api()
return response
def get_testPlan_api(self, PlanId):
"""
Get test plan
"""
return self.send(self.get, url= testPlan_id_url.format(id=PlanId)).json()
def test_get_testPlan(self, PlanId):
"""
Get test plan
@param PlanId: test plan ID
@return: response
"""
response = self.get_testPlan_api(PlanId)
return response
def create_testPlan_api(self, categoryIds, name, withEvaluation, evaluationPresetId, Default):
"""
Create test plan
"""
payload = {
"data": {"caseSuiteIds": [], "categoryIds": categoryIds, "name": name,
"type": "1", "vehicleConfig": {"Ego": {"instanceId": "Ego", "id": "default", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "Default Controller",
"algorithmId": Default},
{"id": "AutoDrive", "name": "Auto-Drive Controller",
"algorithmId": "SimOneDriver"}],
"name": "Ego"}}, "triggerType": "manual",
"withEvaluation": False, "overrideJudgementId": "-1", "evaluationPresetId": evaluationPresetId,
"evaluationOperatorId": "-1", "notification": {"type": "none", "condition": "any"},
"enableStateMachine": False}}
return self.send(self.post, create_testPlan_url, json=payload).json()
def test_create_testPlan(self, caseId : list ,test_case_title : str ,withEvaluation , evaluationPresetId , Default):
"""
@param caseId: case ID
@param test_case_title: test plan name
@param withEvaluation: whether to evaluate
@param evaluationPresetId: evaluation preset ID
@param Default: ego vehicle algorithm
@return: response
"""
response = self.create_testPlan_api(caseId, test_case_title,
withEvaluation = withEvaluation,
evaluationPresetId= 1,
Default="SimOneDriver")
return response
def update_testPlan_api(self, PlanId, categoryIds, name):
"""
Update test plan
"""
payload = {
"data": {"caseSuiteIds": [], "categoryIds": categoryIds, "name": name,
"type": "1", "vehicleConfig": {"Ego": {"instanceId": "Ego", "id": "default", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "Default Controller",
"algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "Auto-Drive Controller",
"algorithmId": "Manual"}],
"name": "Ego"}}, "triggerType": "manual",
"withEvaluation": False, "overrideJudgementId": "-1", "evaluationPresetId": 1,
"evaluationOperatorId": "-1", "notification": {"type": "none", "condition": "any"},
"enableStateMachine": False}}
return self.send(self.put, url=testPlan_id_url.format(id=PlanId), json=payload).json()
def test_update_testPlan(self, PlanId : str ,caseId : list, NewCaseTitle):
"""
@param PlanId: test plan ID
@param caseId: case ID
@param NewCaseTitle: New test plan name
@return: update_response
"""
update_response = self.update_testPlan_api(PlanId, caseId, NewCaseTitle)
return update_response
def delfel_testPlan_single_api(self, PlanID:str):
"""
Delete single test plan
"""
response = self.send(self.delete, url=del_testPlan_single_url.format(id=PlanID)).json()
return response
def delfel_testPlan_single(self,PlanId : str ):
"""
Delete single test plan
@param PlanId: test plan ID
@return: response
"""
response = self.delfel_testPlan_single_api(PlanId)
return response
def del_testPlan_api(self, ids:list):
"""
Delete multiple test plans
"""
payload = {"ids": ids}
return self.send(self.post, url=del_testPlan_url, json=payload).json()
def test_del_testPlan(self, planID):
"""
Delete multiple test plans
@param planID: test plan ID
@return: del_response
"""
del_response = self.del_testPlan_api([planID])
return del_response
def execute_testPlan_api(self, ids:list):
"""
Execute test plan
"""
payload = {"ids":ids}
return self.send(self.post, url=execute_testPlan_url, json=payload).json()
def test_execute_testPlan(self, PlanId : list):
"""
Execute test plan
@param planID: test plan ID
@return: response
"""
response = self.execute_testPlan_api(PlanId)
return response
def clone_testPlan_api(self, id):
"""
Clone test plan
"""
payload = {}
return self.send(self.post, url=clone_testPlan_url.format(id=id), json=payload).json()
def test_clone_testPlan(self, PlanId : str):
"""
Clone test plan
@param planID: test plan ID
@return: response
"""
response = self.clone_testPlan_api(PlanId)
return response
def isrepeat_testPlan_api(self, name, type):
"""
Check if test plan name is duplicated
"""
return self.send(self.get, url=isrepeat_testPlan_url.format(name=name, type=type)).json()
def test_isrepeat_testPlan(self, PlanName):
"""
Check if test plan name is duplicated
@param PlanName: test plan name
@return: response
"""
response = self.isrepeat_testPlan_api(PlanName + time.strftime("_%H:%M:%S", time.localtime()), "worldsim")
return response
def main(category_name: str, case_name: str = None, vehicle_name=None,version: str = None):
"""
Main entry function for running tasks
@param category_name: category_name
@param case_name:
@param vehicle_name:vehicle_name
@param version: version
@return:
"""
suite = Suite()
# suite.create_category() # Verified: create_category()
if not version:
version = suite.get_version()
print("Requested SimOne version API, extracted version: " + version)
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
# # Edit use case: params: case ID, initial ego vehicle absolute target speed, initial ego vehicle world position
# suite.update_case(case_id = "f1c2b1f0-9cce-11f0-b520-f5c71358353c", absolute_target_speed = 111.123, worldposition_x = -334.9999998954549, worldposition_y = -3.000050967810888, worldposition_z = 1.1)
# Ego vehicle control 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, version=version)
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 results
suite.get_task_result(task_ids)
SimOneUrl = "http://127.0.0.1:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
# LoginPolicy("cloud")
LoginPolicy("standalone")
if __name__ == '__main__':
# main("Turn Conflict", "Turn Conflict 8", None, "3.7.0")
fire.Fire(main)
# eg:
# 1. Run one case :python runSimOneCase.py --category_name='Beginner Cases' --case_name='Build Standard Case 2.0'
# 2. Run one case and switch the vehicle :python runSimOneCase.py --category_name='Beginner Cases' --case_name='Build Standard Case 2.0' --vehicle_name='Manual Control - Default'
# 3. Run all cases under the case category :python runSimOneCase.py --category_name='Beginner Cases'
# 4. Run all cases under the case category and switch the vehicle :python runSimOneCase.py --category_name='Beginner Cases' --vehicle_name='Manual Control - Default'