Spaces:
Sleeping
Sleeping
| 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'(?<!\n)\n\n(?![@#\$%\^&\*\(\)\[\]\{\};:,\.])' | |
| paragraphs = re.split(pattern, full_text) | |
| paragraphs.pop(len(paragraphs)-1) | |
| total_paragraphs = len(paragraphs) | |
| if total_paragraphs == 0: | |
| st.warning("No paragraphs found in the PDF.") | |
| start_time = datetime.now() | |
| result_list = [] | |
| for i, paragraph in enumerate(paragraphs): | |
| data = {"note_content": paragraph} | |
| response = requests.post("https://fhir-api-9jsn.onrender.com/convert_notes", json=data) | |
| if response.status_code == 200: | |
| result_list.append(response.json().get("output", "")) | |
| else: | |
| st.error(f"An error occurred at paragraph {i + 1}: {response.json().get('error', 'Unknown error')}") | |
| break | |
| entries = extract_entries_from_bundles(result_list) | |
| combined_bundle = create_combined_bundle(entries) | |
| if combined_bundle: | |
| result.success("Conversion successful!") | |
| result.text_area("FHIR Output", combined_bundle, 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) | |