from copy import deepcopy
def create_batch(BATCH_NAME):
base_url = f"https://api.playment.io/v1/projects/{PROJECT_ID}/batch"
DATA = {"name":BATCH_NAME}
response = requests.post(base_url, headers={'x-api-key': CLIENT_KEY}, json=DATA)
response_data = response.json()
if response.status_code >= 500:
raise Exception(f"Something went wrong at Playment's end {response.status_code}")
if 400 <= response.status_code < 500:
raise Exception(f"{response_data['error']['message']} {response.status_code}")
base_url = f"https://api.playment.io/v1/projects/{PROJECT_ID}/jobs"
response = requests.post(base_url, headers={'x-api-key': CLIENT_KEY}, json=DATA)
response_data = response.json()
if response.status_code >= 500:
raise Exception(f"Something went wrong at Playment's end {response.status_code}")
if 400 <= response.status_code < 500:
raise Exception(f"{response_data['error']['message']} {response.status_code}")
# Details for creating JOBS,
# PROJECT_ID ->> ID of project in which job needed to be created
# CLIENT_KEY ->> secret client key to create JOBS
# WORK_FLOW_ID ->> You can ask this from playment side
# BATCH_ID ->> The batch in which JOB needed to be created
## Job creation payload structure
payload_data_structure ={
"reference_id": "", # reference_id is the unique reference id for the job
"ego_pose": { # ego_pose is the ego vehicle's pose
"position": { # ego vehicle's position
"heading": { # ego vehicle's heading in quaternion format
"sensor_meta": [] # sensor_meta contains meta data for lidar and camera sensors
"work_flow_id": "", # this is the id of the workflow in which the jobs are to be created
"batch_id": "" # this is the batch_id of the batch in which jobs are to be created
sensor_data_structure = {
"sensor_id": "", # This is the sensor's id
"data_url": "", # sensor's data's URL
"sensor_pose": { # This are the sensor's extrinsic values
"position": { # sensor's position
"heading": { # sensor's heading in quaternion format
lidar_sensor_meta_structure = { # lidar sensor's metadata. There can only be one lidar sesnor
"id": "", # lidar sensor's ID
"name": "", # should be kept same as lidar sensor's ID
"state": "editable", # should be kept as editable only
"modality": "lidar", # moality specifies the type of sensor [lidar/camera]
"primary_view": True # should be kept True for lidar sensor
camera_sensor_meta_structure = { # camer sensor's metadata. There can be multiple camera sensors
"id": "", # camera sensor's ID
"name": "", # should be kept same as camera sensor's ID
"state": "editable", # should be kept as editable only
"modality": "camera", # moality specifies the type of sensor [lidar/camera]
"camera_model": "brown_conrady", # specifies the model of camera
"primary_view": False, # should be kept as False for all camera sensors
"intrinsics": { # camera sensor's intrinsic values
if __name__ == "__main__":
## Set sensor id and data url (Only 1 lidar can be present per job)
# Here lidar's sensor_id is the key and pcd file for each job is in the list
"https://example.com/pcd_url_1",
"https://example.com/pcd_url_2"
## Set camera sensor ids and data urls (1 or more cameras can be present per job)
# Here camera's sensor_id is the key and image url for each job is in the list corresponding to that camera
"https://example.com/image_url_1",
"https://example.com/image_url_2" ],
"https://example.com/image_url_3",
"https://example.com/image_url_4"
lidar_sensor_id = 'lidar' # set your lidar sensor id
# validate the number of urls are equal for each sensor
for camera_sensor_id, camer_sensor_urls in camera_data.items():
assert len(camer_sensor_urls) == len(lidar_data), f"Number of URLs are not equal for {camera_sensor_id}"
numer_of_jobs = len(lidar_data)
for i in range(numer_of_jobs):
# define the payload for job creation
payload_data = deepcopy(payload_data_structure)
payload_data['reference_id'] = lidar_data[i].replace('.pcd', '') # you can set unique reference id as per your wish
payload_data['work_flow_id'] = WORK_FLOW_ID
payload_data['batch_id'] = BATCH_ID
# define lidar sensor meta and add it to payload data
lidar_sensor_meta = deepcopy(lidar_sensor_meta_structure)
lidar_sensor_meta['id'] = lidar_sensor_id
lidar_sensor_meta['name'] = lidar_sensor_id
payload_data['data']['sensor_data']['sensor_meta'].append(lidar_sensor_meta)
# define lidar sensor data and add it to payload data
lidar_sensor_data = deepcopy(sensor_data_structure)
lidar_sensor_data['sensor_id'] = lidar_sensor_id
lidar_sensor_data['data_url'] = lidar_data[i]
# set appropriate lidar position and heading, corrently defaulted it to (0,0,0),(1,0,0,0)
lidar_sensor_data['sensor_pose']['position']['x'] = 0
lidar_sensor_data['sensor_pose']['position']['y'] = 0
lidar_sensor_data['sensor_pose']['position']['z'] = 0
lidar_sensor_data['sensor_pose']['heading']['w'] = 1
lidar_sensor_data['sensor_pose']['heading']['x'] = 0
lidar_sensor_data['sensor_pose']['heading']['y'] = 0
lidar_sensor_data['sensor_pose']['heading']['z'] = 0
payload_data['data']['sensor_data']['sensors'].append(lidar_sensor_data)
for camera_sensor_id, camera_sensor_urls in camera_data.items():
# define camera sensor meta and add it to payload
camera_sensor_meta = deepcopy(camera_sensor_meta_structure)
camera_sensor_meta['id'] = camera_sensor_id
camera_sensor_meta['name'] = camera_sensor_id
payload_data['data']['sensor_data']['sensor_meta'].append(camera_sensor_meta)
# define camera sensor data and add it to payload data
camera_sensor_data = deepcopy(sensor_data_structure)
camera_sensor_data['sensor_id'] = camera_sensor_id
camera_sensor_data['data_url'] = camera_sensor_urls[i]
# set appropriate camera position and heading, corrently defaulted it to (0,0,0),(1,0,0,0)
camera_sensor_data['sensor_pose']['position']['x'] = 0
camera_sensor_data['sensor_pose']['position']['y'] = 0
camera_sensor_data['sensor_pose']['position']['z'] = 0
camera_sensor_data['sensor_pose']['heading']['w'] = 1
camera_sensor_data['sensor_pose']['heading']['x'] = 0
camera_sensor_data['sensor_pose']['heading']['y'] = 0
camera_sensor_data['sensor_pose']['heading']['z'] = 0
payload_data['data']['sensor_data']['sensors'].append(camera_sensor_data)
#print(json.dumps(payload_data))
Upload_jobs(payload_data)