| import pandas as pd | |
| 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 json_report_date_insertion(json_data, df_list, logging_df): | |
| biomarker_details = json_data['patient_level']['biomarkers']['details'] | |
| details_attributes = (attribute for biomarker_detail in biomarker_details for attribute in biomarker_detail['attribute']) | |
| for attribute in details_attributes: | |
| attribute_details = attribute['attribute_details'] | |
| attribute['attribute_details'], logging_df = report_date_insertion(attribute_details, df_list, logging_df) | |
| return json_data, logging_df | |