This code demonstrates how to use the Pallas APIs by providing Python functions that interact with the various endpoints. It includes examples for registering a new user, logging in to obtain an authentication token, and making API calls to different endpoints (e.g., suspecting, riskscore, radiology).
The sample payloads provided for each endpoint are formatted as JSON objects. It's important to note that while the code sample processes these payloads, it assumes that the input data is a list of JSON objects. Each object within the list represents an individual input for a separate API call. This allows multiple API calls to be made sequentially using a single authentication token obtained from the login function.
For example, the 'input_suspecting_payload.json' file contains a list of JSON objects, with each object representing the data structure necessary for one API call to the suspecting endpoint. The code iterates over this list, making individual API calls for each JSON object. This approach ensures that each patient's data is processed separately, and the results for each call are stored in an output file.
The same logic applies to the 'input_riskscore_payload.json' and 'input_radiology_payload.json' files, where each file contains a list of JSON objects for their respective endpoints.
By structuring the input data as lists of JSON objects, the code provides flexibility to handle multiple entries efficiently, ensuring that each entry is processed independently. This design is particularly useful when dealing with batch processing of multiple patient records or test cases.
# This function registers a new user with the API.
# Parameters:
# - username: The desired username for the new account.
# - password: The desired password for the new account.
# - registration_key: A special key required for registration.
def register(username, password, registration_key):
url = 'https://pallas.respecthealth.us/register/'
payload = {
'username': username,
'password': password,
'registration_key': registration_key
}
response = requests.post(url, data=payload)
return response.json()
# This function logs in a user and retrieves an authentication token.
# Parameters:
# - username: The username of the account.
# - password: The password of the account.
def login(username, password):
url = 'https://pallas.respecthealth.us/api_token/'
payload = {
'username': username,
'password': password
}
response = requests.post(url, data=payload)
auth_token = response.json()['access_token']
return auth_token
# This function calls the suspecting endpoint of the API.
# Parameters:
# - auth_token: The authentication token obtained from the login function.
# - data: A JSON object containing patient data for analysis.
def call_suspecting_endpoint(auth_token, data):
url = 'https://pallas.respecthealth.us/suspecting/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
return response.json()
[
{
"AnonymizedPtId": "DUMMY_ID",
"egfr0": {
"value": 120,
"date_value": "2024-03-31"
},
"egfr1": {
"value": 110,
"date_value": "2024-03-15"
},
"inr": {
"value": 1.0,
"date_value": "2024-03-20"
},
"pltcnt": {
"value": 200000,
"date_value": "2024-03-25"
},
"hivTiters": {
"value": 1.5,
"date_value": "2024-03-10"
},
"albumin": {
"value": 4.5,
"date_value": "2024-03-05"
},
"prealbumin": {
"value": 30,
"date_value": "2024-03-01"
},
"serumCalcium": {
"value": 9.8,
"date_value": "2024-02-28"
},
"serumPhosphorus": {
"value": 3.5,
"date_value": "2024-02-25"
},
"bloodGlucose": {
"value": 95,
"date_value": "2024-02-20"
},
"hba1c": {
"value": 5.6,
"date_value": "2024-02-18"
},
"microalbumincrRatio": {
"value": 25,
"date_value": "2024-02-15"
},
"phq9": {
"value": 2,
"date_value": "2024-01-01"
},
"smkstatus": {
"value": "Never",
"date_value": "2024-01-01"
},
"o2Sat": {
"value": 97,
"date_value": "2024-01-01"
},
"bmi": {
"value": 22.5,
"date_value": "2024-02-10"
},
"MedicationListInput": [
"acetaminophen",
"biotin",
"celexa",
"diflucan",
"glipizide",
"jardiance",
"metformin hcl",
"multiple electrolytes ph ",
"nystatin ointment",
"citric buffered normal saline"
],
"problemsListInput": [
{
"problem": "Cutaneous abscess of limb, unspecified",
"date_value": "2023-09-11"
},
{
"problem": "Candidiasis of skin and nail",
"date_value": "2023-09-11"
},
{
"problem": "TYPE 2 DIABETES MELLITUS WITHOUT COMPLICATIONS",
"date_value": "2023-09-11"
}
]
}
]
# This function calls the riskscore endpoint of the API.
# Parameters:
# - auth_token: The authentication token obtained from the login function.
# - data: A JSON object containing patient data for risk score calculation.
def call_riskscore_endpoint(auth_token, data):
url = 'https://pallas.respecthealth.us/riskscore/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
return response.json()
[
{
"AnonymizedPtId": "DUMMY_ID",
"age": 65,
"gender": "F",
"icd10_conditions": ["D6869", "I70", "E1151"],
"months_post_graft": 6
}
]
# This function calls the suspecting radiology endpoint of the API.
# Parameters:
# - auth_token: The authentication token obtained from the login function.
# - data: A JSON object containing radiology report data for analysis.
def call_radiology_endpoint(auth_token,data):
url = 'https://pallas.respecthealth.us/suspecting_radiology/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
response.raise_for_status() # Raise an exception for HTTP errors
return response.json()
[
{
"AnonymizedPtId": "DUMMY_ID",
"report": "your_radiology_report"
}
]
# This function calls the suspecting priority endpoint of the API.
# Parameters:
# - auth_token: The authentication token obtained from the login function.
# - data: A JSON object containing radiology report data for analysis.
def call_priority_endpoint(auth_token,data):
url = 'https://pallas.respecthealth.us/suspecting_priority/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
response.raise_for_status() # Raise an exception for HTTP errors
return response.json()
[
{
"AnonymizedPtId": "DUMMY_ID",
"icd10_conditions": []"D6869","G20","E1151"]
}
]
# This script demonstrates how to use the login and endpoint functions together.
import requests
import json
# Login function to get the authentication token
def login(username, password):
url = 'https://pallas.respecthealth.us/api_token'
payload = {
'username': username,
'password': password
}
response = requests.post(url, data=payload)
auth_token = response.json()['access_token']
return auth_token
# Function to call the suspecting endpoint
def call_suspecting_endpoint(auth_token, data):
url = 'https://pallas.respecthealth.us/suspecting/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
return response.json()
# Function to call the riskscore endpoint
def call_riskscore_endpoint(auth_token, data):
url = 'https://pallas.respecthealth.us/riskscore/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
return response.json()
# Function to call the priority endpoint
def call_priority_endpoint(auth_token,data):
url = 'https://pallas.respecthealth.us/suspecting_priority/'
headers = {
'accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.post(url, headers=headers, json=data)
response.raise_for_status() # Raise an exception for HTTP errors
return response.json()
# Load input data for suspecting endpoint from input_suspecting_payload.json
with open('input_suspecting_payload.json', 'r') as file:
input_suspecting_data = json.load(file)
# Load input data for riskscore endpoint from input_riskscore_payload.json
with open('input_riskscore_payload.json', 'r') as file:
input_riskscore_data = json.load(file)
# Load input data for radiology endpoint from input_radiology_payload.json
with open ('input_radiology_payload.json', 'r') as file:
input_radiology_data = json.load(file)
with open ('input_priority_payload.josn','r') as file:
input_priority_data = json.load(file)
# Get the authentication token by logging in
auth_token = login('your_username', 'your_password')
# Lists to store the output data
output_suspecting_data = []
output_riskscore_data = []
output_radiology_data = []
output_priority_data = []
# Iterate over each item in the suspecting input data and call the endpoint
for item in input_suspecting_data:
try:
response = call_suspecting_endpoint(auth_token, item)
output_suspecting_data.append({
"input": item,
"output": response
})
except requests.exceptions.HTTPError as err:
output_suspecting_data.append({
"input": item,
"output": str(err)
})
except Exception as err:
output_suspecting_data.append({
"input": item,
"output": str(err)
})
# Iterate over each item in the riskscore input data and call the endpoint
for item in input_riskscore_data:
try:
response = call_riskscore_endpoint(auth_token, item)
output_riskscore_data.append({
"input": item,
"output": response
})
except requests.exceptions.HTTPError as err:
output_riskscore_data.append({
"input": item,
"output": str(err)
})
except Exception as err:
output_riskscore_data.append({
"input": item,
"output": str(err)
})
# Iterate over each item in the radiology input data and call the endpoint
for item in input_radiology_data:
try:
response = call_radiology_endpoint(auth_token, item)
output_radiology_data.append({
"input": item,
"output": response
})
except requests.exceptions.HTTPError as err:
output_radiology_data.append({
"input": item,
"output": str(err)
})
print(f"HTTP error occurred: {err}")
print(f"Response content: {err.response.content}")
except Exception as err:
output_radiology_data.append({
"input": item,
"output": str(err)
})
print(f"Other error occurred: {err}")
for item in input_priority_data:
try:
response = call_priority_endpoint(auth_token, item)
output_priority_data.append({
"input": item,
"output": response
})
except requests.exceptions.HTTPError as err:
output_priority_data.append({
"input": item,
"output": str(err)
})
print(f"HTTP error occurred: {err}")
print(f"Response content: {err.response.content}")
except Exception as err:
output_priority_data.append({
"input": item,
"output": str(err)
})
print(f"Other error occurred: {err}")
# Save the responses to output_suspecting.json
with open('output_suspecting.json', 'w') as file:
json.dump(output_suspecting_data, file, indent=4)
# Save the responses to output_riskscore.json
with open('output_riskscore.json', 'w') as file:
json.dump(output_riskscore_data, file, indent=4)
# Save the responses to output_radiology.json
with open('output_radiology.json', 'w') as file:
json.dump(output_radiology_data, file, indent=4)
# Save the responses to output_priority.json
with open('output_priority.json', 'w') as file:
json.dump(output_priority_data, file, indent=4)
print("Responses have been written to output_suspecting.json, output_riskscore.json, and output_radiology.json")