Code fragments of main.py

• # This script if for getting a report for the Leuko stations
• import requests
• import pandas as pd
• import numpy as np
• from redcap import Project
• from fpdf import FPDF
• from datetime import datetime
• import matplotlib.pyplot as plt
• import os
• import xnat
• redcap_address = str(os.environ['REDCAP_ADDRESS'])
• redcap_key = str(os.environ['REDCAP_KEY'])
• xnat_address = str(os.environ['XNAT_ADDRESS'])
• xnat_user = str(os.environ['XNAT_USERNAME'])
• xnat_key = str(os.environ['XNAT_KEY'])
• print(redcap_address)
• print(redcap_key)
• print(xnat_address)
• print(xnat_user)
• print(xnat_key)
• def connect_to_xnat_via_credential(username: str, password: str, server: str) -> object:
• """ This function is used to connect to an XNAT server via the credential :return: a connection object :param username: the username from the XNAT system you want to connect with :param password: the password from the XNAT system you want to connect with :param server: server address of the XNAT server """
• connection = xnat.connect(server=server, user=username, password=password)
• return connection
• def try_parsing_date(text):
• for fmt in ('%m-%Y', '%y', '%Y-%m-%d', '%m.%Y', '%m/%y', '%m/%Y'):
• try:
• return datetime.strptime(text, fmt).date()
• except ValueError: pass
• raise ValueError('no valid date format found')
• def convert_series_to_datetimes(colname):
• date_of_birth = data[colname]
• date_of_birth = date_of_birth.dropna()
• records = date_of_birth.index
• dob_list = date_of_birth.to_list()
• dob_dates = [try_parsing_date(date) for date in dob_list]
• df = pd.DataFrame(list(zip(records.to_list(), dob_dates)), columns=['record_id', colname])
• return df
• def export_to_redcap_via_pycap(api_url: str, api_key: str) -> pd.DataFrame:
• """ :param api_url: URL to the REDCAP API as String :param api_key: API Key for the Project :return: pandas dataframe of all records """
• project = Project(api_url, api_key)
• df = project.export_records(format_type="df", raw_or_label="label")
• return df
• def export_to_redcap_via_request(api_url: str, api_key: str) -> pd.DataFrame:
• """ """
• data = { 'token': api_key, 'content': 'record', 'action': 'export', 'format': 'csv', 'type': 'flat', 'csvDelimiter': '', 'records[0]': '2', 'rawOrLabel': 'raw', 'rawOrLabelHeaders': 'raw', 'exportCheckboxLabel': 'false', 'exportSurveyFields': 'false', 'exportDataAccessGroups': 'false', 'returnFormat': 'json' }
• r = requests.post(api_url, data=data)
• print('HTTP Status: ' + str(r.status_code))
• df = r.text
• return df
• def create_pdf(station_name, number_of_patients, number_of_mris, number_of_examinations, hist_plot_path, output_path): # save FPDF() class into a # variable pdf
• pdf = FPDF()
• # Add a page
• pdf.add_page()
• # set style and size of font
• # that you want in the pdf
• pdf.set_font("Arial", size=15)
• # create a cell
• pdf.cell(200, 10, txt="Train Report",ln = 1, align = 'C')
• pdf.cell(100, 10, ln=2, align='L', txt="This is the report for the Station {}".format(station_name))
• # add another cell
• pdf.cell(100, 10, txt="Number of Patients: {}".format(number_of_patients),ln=2, align='L')
• pdf.cell(100, 10, txt="Number of MRIs: {}".format(number_of_mris),ln=2, align='L')
• pdf.cell(100, 10, txt="Number of Examinations: {}".format(number_of_examinations),ln=2, align='L')
• # save the pdf with name .pdf
• pdf.image(name=hist_plot_path, x=0, y=70, w=200, h=150)
• pdf.output(output_path)
• # connect_to_xnat_via_credential(xnat_user, xnat_key, xnat_address)
• print("connection established")
• data = export_to_redcap_via_pycap(api_url=redcap_address, api_key=redcap_key)
• data_top = data.index
• # print(data_top)
• number_of_records = len(set(data_top.tolist()))
• instruments = data['redcap_repeat_instrument']
• instruments = instruments.replace(np.nan, 'Baseline')
• table_instrument = instruments.value_counts()
• dob_df = convert_series_to_datetimes('dob')
• doe_df = convert_series_to_datetimes('date_of_examination')
• records_first_exam = []
• date_first_exam = []
• for i, g in doe_df.groupby(by='record_id'):
• records_first_exam.append(g.iloc[0]['record_id'])
• date_first_exam.append(g.iloc[0]['date_of_examination'])
• doe_first_df = pd.DataFrame(list(zip(records_first_exam, date_first_exam)), columns=['record_id', 'date_of_examination'])
• final_df = pd.merge(dob_df, doe_first_df, how='inner', on='record_id')
• age = [final_df['date_of_examination'][patient].year - final_df['dob'][patient].year for patient in range(0,len(final_df['dob']))]
• age_file_path = './age_hist_plot.png'
• plt.rcParams.update({'figure.figsize':(7,5), 'figure.dpi':100})
• # Plot Histogram on x
• x = np.random.normal(size = 1000)
• plt.hist(age, bins=50)
• plt.gca().set(title='Age Frequency on first visit(n={})'.format(len(age)), ylabel='Frequency', xlabel='Age')
• plt.savefig(age_file_path)
• number_of_records = str(table_instrument['Baseline'])
• number_of_mri = str(table_instrument['MRI'])
• number_of_exams = str(table_instrument['Examination Data (use new sheet for every visit)'])
• create_pdf(station_name='leipzig',number_of_patients=number_of_records, number_of_mris=number_of_mri, number_of_examinations=number_of_exams, output_path="test_1.pdf", hist_plot_path=age_file_path)

Graph