etl.py

import gradio as gr 
import pandas as pd
from gspread_dataframe import set_with_dataframe  

def load_data(gs, start_date, end_date, d_type):
    """
    load data from google sheet

    Args:
        gs (Gsheet instance ): GSheet instance with access to google sheet
        start_date (str): Date str format(YYYY-mm-dd)
        end_date (str): Date str format(YYYY-mm-dd)
        d_type (str):sheet type (prescription or diagnosis)

    Raises:
        gr.Error: 403 permission denied

    Returns:
        data (DataFrame): loaded data from google sheet
    """
    sheets_dict = {
        "diagnosis": "Diagnosis",
        "prescription": "Prescriptions"
    }

    # open data and super sheet
    try:
        data_sheet = gs.worksheet(sheets_dict[d_type])
    except:
        raise gr.Error(message='Permission denied. Please add IAM user to the sheet and try again')
    
    # read data from sheets
    all_data = pd.DataFrame(data_sheet.get_all_records())
    all_data.columns = all_data.columns.str.strip()

    # transform date fields
    all_data['date_cleaned'] = pd.to_datetime(all_data['date_cleaned'], infer_datetime_format=True)
    all_data['date_reviewed'] = pd.to_datetime(all_data['date_reviewed'], infer_datetime_format=True)

    data = all_data[(all_data['date_cleaned'] >= start_date) & (all_data['date_cleaned'] <= end_date)]

    return data

def preprocess_data(data, d_type):
    """
    preprocess loaded data from google sheet

    Args:
        data (DataFrame):  google sheet data as a dataframe
        d_type (Str): sheet type (prescription or diagnosis)

    Raises:
        gr.Error (ValueError): Wrong column

    Returns:
        Overlap (DataFrame): Overlap data
    """

    data.columns = data.columns.str.strip()

    overlap_cols = {
       "prescription": ["RX Norm [Super Generic]_x", "RX Norm [Super Generic]_y" ], 
       "diagnosis": ["ICD10 Diagnosis_x", "ICD10 Diagnosis_y"]
    }

    try:
        # extract the interns
        intern_1 = data['Intern'].unique()[0]
        intern_2 = data['Intern'].unique()[1]

        # intern 1 and intern 2 data 
        intern_1_data = data[data['Intern'] == intern_1.strip()]
        intern_2_data = data[data['Intern'] == intern_2.strip()]

        # extract the overlap
        overlap = pd.merge(intern_1_data, intern_2_data, on='Unstructured Name')

        # remove all white spaces 
        col = overlap_cols[d_type]
        overlap[col[0]] = overlap[col[0]].str.strip()
        overlap[col[1]] = overlap[col[1]].str.strip()
    except:
       raise gr.Error(message="No Intern Column")

    return overlap

def post_process(data, d_type):
    """
   post process data

    Args:
        data (DataFrame): matched data as dataframe
        d_type (str): sheet type (prescription or diagnosis)

    Returns:
        data (DataFrame): postprocessed data as dataframe
    """
    drop_cols = {
        "prescription": ['Type_y','Supervisor_y', 'date_cleaned_y', 'date_reviewed_y'], 
        "diagnosis": ['Supervisor_y', 'date_cleaned_y', 'date_reviewed_y']
    }
    post_cols = {
        "prescription": ['Unstructured Name', 'Intern_x', 'Status_x', 'RX Norm [Super Generic]_x', 
                            'Intern_y', 'Status_y', 'RX Norm [Super Generic]_y',  'match_status', 'Type_x', 
                            'Supervisor_x', 'date_cleaned_x', 'date_reviewed_x'],
        "diagnosis":   ['Unstructured Name', 'Intern_x', 'Status_x', 'ICD10 Diagnosis_x', 
                        'Intern_y', 'Status_y', 'ICD10 Diagnosis_y',  'match_status',
                        'Supervisor_x', 'date_cleaned_x', 'date_reviewed_x']
        }
    
    d_cols = drop_cols[d_type]
    p_cols = post_cols[d_type]

    data = data.drop(columns=d_cols)
    data = data[p_cols]

    data = data.rename(columns={
        'date_cleaned_x': 'date_cleaned',
        'date_reviewed_x': 'date_reviewed'
        })
    
    return data

def output_data(gc, data, week_no, dtype):
    """
    outputs data to a google sheet

    Args:
        gc (GSheet instance): Gsheet permission instance
        data (DataFrame):  Data to write to google sheet
        week_no (str): Week no e.g. week_1
        d_type (str): sheet type (prescription or diagnosis)
    
    Raises:
        gr.exception.Error: Sheet Error
    """
    out_sheet_dict = {
        "diagnosis": "https://docs.google.com/spreadsheets/d/1UJ2PHR62mcz11D2qeX-Wk2fs1357BW78o8g76lwG93w/edit#gid=2123373592", 
        "prescription": "https://docs.google.com/spreadsheets/d/1Fo9V6J_L9eWX3qEScITP2L8nI9exk_VslGMgk-1jCJw/edit#gid=1287467149"
    }

    outsheet_url = out_sheet_dict[dtype]
    out_gs = gc.open_by_url(outsheet_url)

    # write output to sheet 
    try:
        out_worksheet = out_gs.worksheet(week_no)
        # clear the worksheet 
        out_worksheet.clear()
    except:
        raise gr.Error(f"Result sheet for {week_no} does not exist")

    # write dataframe to work sheet
    set_with_dataframe(
        worksheet=out_worksheet, 
        dataframe=data, 
        include_index=False,
        include_column_header=True, 
        resize=True)