Create multiprocessing_in_date_ingestion.py

multiprocessing_in_date_ingestion.py

@@ -0,0 +1,86 @@
+import pandas as pd
+from multiprocessing import Pool
+
+def get_encounter_dates(df_list, pif_key):
+    encounter_dates = []
+    for df in df_list:
+        df['encounter_date'].fillna('', inplace=True)
+        df_date = df.loc[df['pif_key'].astype(str) == str(pif_key), 'encounter_date'].values
+        if len(df_date) > 0:
+            encounter_dates.extend(df_date)
+    return encounter_dates
+
+def get_latest_date(encounter_dates):
+    if encounter_dates:
+        return max(encounter_dates)
+    return ''
+
+def create_date_insert_dict(ingested_date):
+    return {
+        'attribute_name': 'report_date',
+        'attribute_method': 'cv',
+        'attribute_normalized_prediction': '',
+        'attribute_prediction': str(ingested_date),
+        'attribute_version': 'v2_090523',
+        'attribute_vocab': '',
+        'attribute_code': '',
+        'date_of_service': ''
+    }
+
+def add_logging_entry(logging_df, pif_key, json_report_date_exists, encounter_dates, ingested_date, multiple_date, old_date):
+    logging_df = logging_df.append({
+        'pif_key': pif_key,
+        'json_report_date_exists': json_report_date_exists,
+        'encounter_dates': encounter_dates,
+        'ingested_date': ingested_date,
+        'multiple_date': multiple_date,
+        'old_date': old_date
+    }, ignore_index=True)
+    return logging_df
+
+def date_dict(df_list, pif_key):
+    encounter_dates = get_encounter_dates(df_list, pif_key)
+    ingested_date = get_latest_date(encounter_dates)
+    return create_date_insert_dict(ingested_date), encounter_dates, ingested_date
+
+def report_date_insertion(dict_list, df_list, logging_df):
+    col_names = {col['attribute_name'] for col in dict_list}
+    pif_key = next((col['attribute_prediction'] for col in dict_list if col['attribute_name'] == 'pif_key'), None)
+    
+    if 'report_date' not in col_names and pif_key is not None:
+        date_insert_dict, encounter_dates, ingested_date = date_dict(df_list, pif_key)
+        dict_list.insert(1, date_insert_dict)
+        logging_df = add_logging_entry(logging_df, pif_key, False, encounter_dates, ingested_date, len(encounter_dates) > 1, 'missing' if not ingested_date else '')
+
+    elif 'report_date' in col_names and pif_key is not None:
+        date_insert_dict, encounter_dates, ingested_date = date_dict(df_list, pif_key)
+        if ingested_date:
+            for report_date_idx, tm in enumerate(dict_list):
+                if tm['attribute_name'] == 'report_date':
+                    old_date = tm['attribute_prediction']
+                    break
+            dict_list.pop(report_date_idx)
+            dict_list.insert(1, date_insert_dict)
+            logging_df = add_logging_entry(logging_df, pif_key, True, encounter_dates, ingested_date, len(encounter_dates) > 1, old_date)
+        else:
+            for report_date_idx, tm in enumerate(dict_list):
+                if tm['attribute_name'] == 'report_date':
+                    old_date = tm['attribute_prediction']
+                    break
+            logging_df = add_logging_entry(logging_df, pif_key, True, None, '', False, old_date)
+
+    return dict_list, logging_df
+
+def process_biomarker_detail(biomarker_detail):
+    attributes = biomarker_detail['attribute']
+    for attribute in attributes:
+        attribute_details = attribute['attribute_details']
+        attribute['attribute_details'], logging_df = report_date_insertion(attribute_details, df_list, logging_df)
+    return biomarker_detail
+
+def json_report_date_insertion(json_data, df_list, logging_df):
+    biomarker_details = json_data['patient_level']['biomarkers']['details']
+    with Pool() as pool:
+        updated_biomarker_details = pool.map(process_biomarker_detail, biomarker_details)
+    json_data['patient_level']['biomarkers']['details'] = updated_biomarker_details
+    return json_data, logging_df