import streamlit as st import pandas as pd import requests import json from datetime import datetime from pdfminer.high_level import extract_text import regex as re st.title("FHIR Converter") # Dropdown for conversion type conversion_type = st.selectbox( "Select Conversion Type", ["CSV to FHIR", "CCDA to FHIR", "Clinical Notes to FHIR"] ) def extract_entries_from_bundles(bundles): """ Extracts the 'entry' data from a list of FHIR bundles, including only the first Patient resource encountered. Parameters: - bundles: List of JSON strings, each representing a FHIR bundle. Returns: - List of all entries from the provided bundles with only the first Patient resource included. """ combined_entries = [] patient_included = False # Flag to track if the first Patient has been included # Iterate through each bundle for bundle_str in bundles: try: # Parse the JSON bundle bundle = json.loads(bundle_str) # Ensure the bundle contains the 'entry' key if 'entry' in bundle: for entry in bundle['entry']: resource = entry.get('resource', {}) # Check the resource type resource_type = resource.get('resourceType') # Handle Patient resources if resource_type == 'Patient': if not patient_included: # Include the first Patient resource combined_entries.append(entry) patient_included = True continue # Include all other resource types combined_entries.append(entry) else: print("Warning: Bundle does not contain 'entry' key.") except json.JSONDecodeError as e: print(f"Error decoding JSON bundle: {e}") return combined_entries def create_combined_bundle(entries): """ Creates a new FHIR bundle containing all the entries. Parameters: - entries: List of FHIR entries to include in the new bundle. Returns: - JSON string representing the new combined FHIR bundle. """ combined_bundle = { "resourceType": "Bundle", "type": "collection", "entry": entries } return json.dumps(combined_bundle, indent=2) if conversion_type == "CSV to FHIR": res_type = st.selectbox( "Select Resource Type", ["Patient", "CarePlan", "MedicationRequest", "Immunization", "Encounter", "Claim", "Procedure", "Observation"] ) # File uploader for CSV file uploaded_file = st.file_uploader("Upload CSV", type=["csv"]) if uploaded_file is not None: # Display the uploaded CSV file df = pd.read_csv(uploaded_file) st.write("Uploaded CSV:") st.dataframe(df) # Placeholders for progress bar and time left progress_bar = st.progress(0) time_placeholder = st.empty() items_placeholder = st.empty() # Variables to store the result and state result = st.empty() conversion_successful = False # Button to trigger conversion if st.button("Convert"): # Convert the CSV to FHIR using the FastAPI endpoint start_time = datetime.now() total_items = len(df) try: files = {"file": uploaded_file.getvalue()} data = {"res_type": res_type} response = requests.post("https://fhir-api-9jsn.onrender.com/convert", files=files, data=data, stream=True) result_text = "" for chunk in response.iter_lines(): if chunk: chunk_data = chunk.decode("utf-8") json_chunk = json.loads(chunk_data) if "progress" in json_chunk: progress = json_chunk["progress"] current_item = json_chunk["current_item"] total_items = json_chunk["total_items"] elapsed_time = datetime.now() - start_time items_per_second = current_item / elapsed_time.total_seconds() time_left = (total_items - current_item) / items_per_second progress_bar.progress(int(progress)) time_placeholder.text(f"Time elapsed: {elapsed_time}") items_placeholder.text(f"Items processed: {current_item}/{total_items}") elif "output" in json_chunk: result_text = json_chunk["output"] elif "error" in json_chunk: st.error(f"An error occurred: {json_chunk['error']}") break if result_text: result.success("Conversion successful!") result.text_area("FHIR Output", result_text, height=400) conversion_successful = True except Exception as e: st.error(f"An error occurred: {str(e)}") # Display validator button if conversion_successful: st.markdown("[Go to FHIR Validator](https://validator.fhir.org)", unsafe_allow_html=True) elif conversion_type == "CCDA to FHIR": res_type = st.selectbox( "Select Resource Type", ["Patient", "Encounter", "Observation", "MedicationRequest", "Procedure", "Immunization", "Condition"] ) cda_content = st.text_area("Enter CCDA Content in XML format") # Variables to store the result and state result = st.empty() conversion_successful = False # Button to trigger conversion if st.button("Convert"): try: data = {"res_type": res_type, "cda_content": cda_content} response = requests.post("https://fhir-api-9jsn.onrender.com/convert_ccda", json=data) if response.status_code == 200: result_text = response.json().get("output", "") if result_text: result.success("Conversion successful!") result.text_area("FHIR Output", result_text, height=400) conversion_successful = True else: st.error(f"An error occurred: {response.json().get('error', 'Unknown error')}") except Exception as e: st.error(f"An error occurred: {str(e)}") if conversion_successful: st.markdown("[Go to FHIR Validator](https://validator.fhir.org)", unsafe_allow_html=True) elif conversion_type == "Clinical Notes to FHIR": uploaded_pdf = st.file_uploader("Upload PDF with Clinical Notes", type=["pdf"]) # Variables to store the result and state result = st.empty() conversion_successful = False if uploaded_pdf is not None and st.button("Convert"): try: # Extract text from PDF full_text = extract_text(uploaded_pdf) pattern = r'(?