<jupyter_start><jupyter_text>PRMT-2039 ["HYPOTHESIS"] Table showing distribution number of messages per pathway and statusWe have so far assumed that a particular transfer status is associated with a particular number of messages:Eg Pending would typically be expected to have 3, Integrated would usually expect to see 4.However, we have seen fewer that 3 for pending in inspection of some Vision to Vision transfers.By generating a full table:- for 6 months of transfers, - broken down by status and supplier pathway, - showing what % of transfers had what number of messages, we can:- check the degree to which this assumption holds. - Identify areas (eg by pathway and/or status) for further investigation- This may allow us to redefine our statuses from their current 4<jupyter_code>import pandas as pd
import numpy as np
# Using data generated from branch PRMT-1742-duplicates-analysis.
# This is needed to correctly handle duplicates.
# Once the upstream pipeline has a fix for duplicate EHRs, then we can go back to using the main output.
transfer_file_location = "s3://prm-gp2gp-data-sandbox-dev/transfers-duplicates-hypothesis/"
transfer_files = [
    "9-2020-transfers.parquet",
    "10-2020-transfers.parquet",
    "11-2020-transfers.parquet",
    "12-2020-transfers.parquet",
    "1-2021-transfers.parquet",
    "2-2021-transfers.parquet"
]

transfer_input_files = [transfer_file_location + f for f in transfer_files]
transfers_raw = pd.concat((
    pd.read_parquet(f)
    for f in transfer_input_files
))

# In the data from the PRMT-1742-duplicates-analysis branch, these columns have been added , but contain only empty values.
transfers_raw = transfers_raw.drop(["sending_supplier", "requesting_supplier"], axis=1)


# Given the findings in PRMT-1742 - many duplicate EHR errors are misclassified, the below reclassifies the relevant data
has_at_least_one_successful_integration_code = lambda errors: any((np.isnan(error) or error==15 for error in errors))
successful_transfers_bool = transfers_raw['request_completed_ack_codes'].apply(has_at_least_one_successful_integration_code)
transfers = transfers_raw.copy()
transfers.loc[successful_transfers_bool, "status"] = "INTEGRATED"

# Correctly interpret certain sender errors as failed.
# This is explained in PRMT-1974. Eventaully this will be fixed upstream in the pipeline. 
pending_sender_error_codes=[6,7,10,24,30,23,14,99]
transfers_with_pending_sender_code_bool=transfers['sender_error_code'].isin(pending_sender_error_codes)
transfers_with_pending_with_error_bool=transfers['status']=='PENDING_WITH_ERROR'
transfers_which_need_pending_to_failure_change_bool=transfers_with_pending_sender_code_bool & transfers_with_pending_with_error_bool
transfers.loc[transfers_which_need_pending_to_failure_change_bool,'status']='FAILED'

# Add integrated Late status
eight_days_in_seconds=8*24*60*60
transfers_after_sla_bool=transfers['sla_duration']>eight_days_in_seconds
transfers_with_integrated_bool=transfers['status']=='INTEGRATED'
transfers_integrated_late_bool=transfers_after_sla_bool & transfers_with_integrated_bool
transfers.loc[transfers_integrated_late_bool,'status']='INTEGRATED LATE'

# If the record integrated after 28 days, change the status back to pending or pending with error.
# This is to handle each month consistentently and to always reflect a transfers status 28 days after it was made.
# TBD how this is handled upstream in the pipeline
twenty_eight_days_in_seconds=28*24*60*60
transfers_after_month_bool=transfers['sla_duration']>twenty_eight_days_in_seconds
transfers_pending_at_month_bool=transfers_after_month_bool & transfers_integrated_late_bool
transfers.loc[transfers_pending_at_month_bool,'status']='PENDING'
transfers_with_early_error_bool=(~transfers.loc[:,'sender_error_code'].isna()) |(~transfers.loc[:,'intermediate_error_codes'].apply(len)>0)
transfers.loc[transfers_with_early_error_bool & transfers_pending_at_month_bool,'status']='PENDING_WITH_ERROR'

# Supplier name mapping
supplier_renaming = {
    "EGTON MEDICAL INFORMATION SYSTEMS LTD (EMIS)":"EMIS",
    "IN PRACTICE SYSTEMS LTD":"Vision",
    "MICROTEST LTD":"Microtest",
    "THE PHOENIX PARTNERSHIP":"TPP",
    None: "Unknown"
}

asid_lookup_file = "s3://prm-gp2gp-data-sandbox-dev/asid-lookup/asidLookup-Mar-2021.csv.gz"
asid_lookup = pd.read_csv(asid_lookup_file)
lookup = asid_lookup[["ASID", "MName", "NACS","OrgName"]]

transfers = transfers.merge(lookup, left_on='requesting_practice_asid',right_on='ASID',how='left')
transfers = transfers.rename({'MName': 'requesting_supplier', 'ASID': 'requesting_supplier_asid', 'NACS': 'requesting_ods_code','OrgName':'requesting_practice_name'}, axis=1)
transfers = transfers.merge(lookup, left_on='sending_practice_asid',right_on='ASID',how='left')
transfers = transfers.rename({'MName': 'sending_supplier', 'ASID': 'sending_supplier_asid', 'NACS': 'sending_ods_code','OrgName':'sending_practice_name'}, axis=1)

transfers["sending_supplier"] = transfers["sending_supplier"].replace(supplier_renaming.keys(), supplier_renaming.values())
transfers["requesting_supplier"] = transfers["requesting_supplier"].replace(supplier_renaming.keys(), supplier_renaming.values())
# Import the message data - output of PRMT-2038 (see 33-PRMT-2038-generate-new-fields-raw-splunk-data notebook)
interaction_file_name='s3://prm-gp2gp-data-sandbox-dev/extra-fields-data-from-splunk/Sept_20_Feb_21_conversations_interaction_messages.parquet'
message_lists=pd.read_parquet(interaction_file_name)
# Merge the datasets
combined_transfers=transfers.merge(message_lists,left_on='conversation_id',right_index=True ,how='left')
# Add in the number of messages as the main field
combined_transfers['interaction length']=combined_transfers['interaction name'].apply(len)
# Generate a table to count the transfers of each message length broken down by pathway/status
message_count_table=combined_transfers.pivot_table(index=['requesting_supplier','sending_supplier','status'],columns='interaction length',values='conversation_id',aggfunc='count').fillna(0).astype(int)
message_count_table
# For each pathway/status, what is the percentage in each message length?
message_count_table_percentages=message_count_table.div(message_count_table.sum(axis=1),axis=0).multiply(100).round(2)
message_count_table_percentages.reset_index()
# Uncomment to output CSV to S3 location
# message_count_table_percentages.reset_index().to_csv('s3://prm-gp2gp-data-sandbox-dev/notebook-outputs/34--PRMT-2039-message-length-counts-by-pathway-and-status.csv')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10088522588522589, 'fraction_numerical': 0.020909645909645908, 'mean_word_length': 8.001373626373626, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '54928', 'license': 'Apache-2.0', 'n_tokens_mistral': 2208, 'n_tokens_neox': 2069, 'n_words': 454, 'path': 'notebooks/34-PRMT-2039--table-number-messages-by-status-pathway.ipynb', 'repo_name': 'nhsconnect/prm-gp2gp-data-sandbox', 'size': '6552'}