This example demonstrates importing requisition(s) from a CSV file
In [ ]:
import dateutil.parser
import csv
from ovation.session import connect_lab
from tqdm import tqdm_notebook as tqdm
In [ ]:
user = input('Email: ')
In [ ]:
s = connect_lab(user, api='https://lab-services-staging.ovation.io/') # use api='https://lab-services.ovation.io' for production
Several functions that we'll use for the import…
In [ ]:
def get_project_by_name(s, project_name, organization_id=0):
projects = s.get(s.path('project'),
params={'organization_id': organization_id}).projects
project = list(filter(lambda p: p.name == project_name, projects))[0]
return project
def create_container(s, organization_id, container_type, container_barcode=None):
container_data = {'type': container_type}
if container_barcode is not None:
container_data['barcode'] = container_barcode
container = s.post(s.path('container'),
data={'container': container_data},
params={'organization_id': organization_id})
return container
def create_requisition(s, project=None,
organization_id=0,
identifier=None,
template_name=None,
container_type='Tube',
container_barcode=None,
container_position='A01',
custom_attributes={},
sample_date_received=None,
patient_mrn=None,
patient_gender=None,
patient_dob=None,
patient_first_name=None,
patient_last_name=None,
panel=None,
physician_email=None,
sample_collection_date=None
):
# Create sample container
container = create_container(s, organization_id, container_type, container_barcode)
# Create the requisition
sample_data = {"identifier": identifier,
"sample_states": [{"container_id": container.id,
"position": container_position}]}
if sample_date_received:
sample_data["date_received"] = sample_date_received
else:
sample_data["received"] = False
gender = {}
if patient_gender.lower == 'f' or patient_gender.lower() == 'female':
gender['female'] = True
elif patient_gender.lower == 'm' or patient_gender.lower() == 'male':
gender['male'] = True
requisition_data = {"identifier": identifier, # Any unique (within organization) identifier
"template": template_name,
"custom_attributes": custom_attributes,
"sample_collection_date": sample_collection_date,
"samples": [sample_data],
"patient": {
"identifier": patient_mrn,
"first_name": patient_first_name,
"last_name": patient_last_name,
"date_of_birth": patient_dob,
"gender": gender
},
"requested_tests": {
panel: True
},
"complete": False,
"organization_id": organization_id,
"project_id": project.id
}
if physician_email:
requisition_data["physician"] = {"contact_email": physician_email}
req = s.post(s.path('requisition'),
data={'requisition': requisition_data},
params={'organization_id': organization_id,
"project_id": project.id})
return req
def import_requisition(s, row, organization_id=0, template_name=None, project=None, physician_email=None):
assert project is not None
assert template_name is not None
assert row is not None
identifier = row['accession']
container_type = 'Tube',
container_position = 'A01',
container_barcode = row['tube no']
custom_attributes = {'date shipped': row['date shipped']} # Add Institution, Department, etc.
patient_mrn = row['mrn']
patient_gender = row['gender']
dob = dateutil.parser.parse(row['dob']) if len(row['dob']) > 0 else None
patient_dob = dob.date().isoformat() if dob else None
patient_first_name = row['first name']
patient_last_name = row['last name']
panel = row['test mnemonic'] # Must be a panel code!
sample_collection_date = dateutil.parser.parse(row['collect date']) if len(row['collect date']) > 0 else None
sample_collection_date = sample_collection_date.date().isoformat() if sample_collection_date else None
return create_requisition(s, project,
organization_id=organization_id,
identifier=identifier,
template_name=template_name,
container_type='Tube',
container_barcode=container_barcode,
container_position='A01',
custom_attributes=custom_attributes,
sample_date_received=None,
patient_mrn=patient_mrn,
patient_gender=patient_gender,
patient_dob=patient_dob,
patient_first_name=patient_first_name,
patient_last_name=patient_last_name,
panel=panel,
physician_email=physician_email,
sample_collection_date=sample_collection_date)
Collect required information, then import each requisition in the provided CSV.
In [ ]:
organization_id = int(input('Organization ID: '))
In [ ]:
csv_path = input('CSV path: ')
In [ ]:
template_name = input("Requisition template name: ")
In [ ]:
project_name = input('Project name: ')
In [ ]:
project = get_project_by_name(s, project_name, organization_id=organization_id)
In [ ]:
with open(csv_path, newline='') as csvfile:
reader = csv.DictReader(csvfile)
for row in tqdm(reader, unit='requisitions'):
import_requisition(s, row, organization_id=organization_id, template_name=template_name, project=project, physician_email=None)