The Kernel API allows you to automate repeated tasks programmatically.
To get started, have your Organization Owner generate an API key on the Organization Settings page.
External/Customer facing API
Routes for:
- All dataset IDs + metadata in a study
- Run a pipeline, given a dataset ID and pipeline name
- Status for most recent pipeline run and signed urls if present, given dataset ID and pipeline name
- Run pipelines across dataset IDs in batch
- Status and signed urls for a pipeline batch
NOTE:
Each route requires an API key in the header, such as:
headers={"Authorization": f"{api_key}"}An Organization Owner can generate an API key on the Organization Settings page.
Common ways to use Kernel API
To find your study_id:
Navigate to the study in portal and notice the URL looks like:
https://portal.kernel.com/organizations/2d1fee1c-c2bc-4833-b560-afc673140f42/studies/b249e4b9-9108-4ce4-a7ce-41cc3bbd2bf2/datasets
The bold part b249e4b9-9108-4ce4-a7ce-41cc3bbd2bf2 is your study_id to use in the following routes.
List all dataset IDs and metadata in a study:
GET /api/vi/study/{study_id}/datasets
- Parameters:
- study_id (str) - study id
- Returns:
- All dataset IDs and metadata in a study
- Return Type:
- Dict[str, List[dict]]
{
"datasets": [
{
"id": "00000000000000000000000000000000",
"meta": {}, # possible keys include "description", "name", "experiment"
"participant": {
"id": "00000000000000000000000000000000"
"participant_id": "00000000000000000000000000000000"
"created_at": 0.0, # seconds since epoch
"active": True, # participant active status
"pending": False, # participant pending status
"status": "active" # participant status in the study,
},
"created_date": 0.0, # seconds since epoch
"started_at": 0.0, # seconds since epoch
"stopped_at": 0.0, # seconds since epoch
}
]
}Examples
>>> response = requests.get("https://api.kernel.com/api/v1/study/000000000000000/datasets")
>>> assert response.status_code == 200
>>> datasets = response.json()["datasets"]
>>> dataset_ids = [dataset["id"] for dataset in datasets]Available pipelines:
# Available pipelines are:
"analysis_eeg"
"analysis_nirs_epoched"
"analysis_nirs_glm"
"analysis_task"
"qc_eeg"
"qc_nirs_basic"
"qc_nirs"
"qc_syncbox"
"reconstruction"
"pipeline_snirf_gated"
"snirf_hb_moments"
"snirf_moments"Get the status of the most recent pipeline run for a dataset and asset URLs if available:
GET /api/vi/study/{study_id}/dataset/{dataset_id}/pipeline/{pipeline_name]/status
- Parameters:
- study_id (str) - study id
- dataset_id (str) - dataset id
- pipeline_name (str) - pipeline name
- Returns:
- Status of the most recent pipeline run and asset URLs if available
- Return Type:
- dict
{
"job_id": "00000000000000000000000000000000",
"status": "SUCCEEDED",
"signed_urls": {
"urls": {
"filename1": "https://someurl"
}, "sizes": {
"filename1": 123.1
},
"batch_job_id": "00000000000000000000000000000000"
"execution_id": "00000000000000000000000000000000"
}
}Examples
>>> response = requests.get("https://api.kernel.com/api/v1/study/000000000000000/dataset/000000000000000/pipeline/name/status")
>>> assert response.status_code == 200
>>> status = response.json()
>>> assert status["status"] == "SUCCEEDED"Run a pipeline for a dataset:
POST /api/vi/study/{study_id}/dataset/{dataset_id}/pipeline/{pipeline_name}
- Parameters:
- study_id (str) - study id
- dataset_id (str) - dataset id
- pipeline_name (str) - pipeline name
- Returns:
- Job ID if the pipeline run
- Return Type:
- Dict
Response json structure:
{
"job_id": "00000000000000000000000000000000"
}Examples
>>> response = requests.post("https://api.kernel.com/api/v1/study/00000000000000/dataset/00000000000000/pipeline/name")
>>> assert response.status_code == 200
>>> job_id = response.json()["job_id"]Run a pipeline batch for many datasets within a study:
POST /api/v1/study/{study_id}/pipeline/{pipeline_name}/batch
- Parameters:
- study_id (str) - study id
- pipeline_name (str) - pipeline name
- Headers
- Content-Type: application/json
- Request Body
{
"dataset_ids": [ // MAX: 500
"00000000000000000000000000000000",
"00000000000000000000000000000001",
"00000000000000000000000000000002",
...
]
}- Returns:
- Pipeline Batch ID
- Return Type:
- Dict
Response json structure:
{
"pipeline_batch_id": "00000000000000000000000000000000"
}Examples
>>> response = requests.post("https://api.kernel.com/api/v1/study/00000000000000000000000000000000/pipeline/pipeline_name/batch", json={"dataset_ids": ["00000000000000000000000000000001", "00000000000000000000000000000002"]})
>>> assert response.status_code == 200
>>> pipeline_batch_id = response.json()["pipeline_batch_id"]Get the status of the most recent pipeline run for a dataset and asset URLs if available:
GET /api/v1/study/{study_id}/pipeline/batch/{pipeline_batch_id}/status
- Parameters:
- study_id (str) - study id
- pipeline_batch_id (str) - Pipeline Batch ID
- Returns:
- Status of the pipeline batch
- Return Type:
- dict
json structure (in progress)::
{
"pipeline_batch_id": "00000000000000000000000000000000",
"complete": false,
"stats": {
"total": 10,
"not_analyzable": 2,
"analyzable": 8,
"succeeded": 3,
"failed": 1,
"pending": 4
}
}
json structure (complete)::
{
"pipeline_batch_id": "00000000000000000000000000000000",
"complete": true,
"stats": {
"total": 10,
"not_analyzable": 2,
"analyzable": 8,
"succeeded": 7,
"failed": 1,
"pending": 0
},
"results_url": "https://signed-s3-url..."
}Results
Downloading the results_url retrieves a JSON file:
{
"statuses": {
"dataset_id_1": "SUCCEEDED",
"dataset_id_2": "FAILED",
...
},
"stats": {
"total": 10,
"not_analyzable": 2,
"analyzable": 8,
"succeeded": 3,
"failed": 1,
"pending": 4
},
"signed_urls": {
"dataset_id_1": {
"urls": {
"filename1": "https://someurl"
},
"sizes": {
"filename1": 123.1
},
"batch_job_id": "00000000000000000000000000000000",
"execution_id": "00000000000000000000000000000000"
}
}
}Examples
>>> response = requests.get("https://api.kernel.com/api/v1/study/00000000000000000000000000000000/pipeline/batch/00000000000000000000000000000000/status")
>>> assert response.status_code == 200
>>> status = response.json()