Spaces:
Sleeping
Sleeping
File size: 8,689 Bytes
7aac402 22ea3d2 7aac402 d0e9868 7aac402 df56ffc 7aac402 d0e9868 7aac402 d0e9868 7aac402 d4955ac d0e9868 d4955ac df56ffc d0e9868 df56ffc d0e9868 df56ffc d4955ac df56ffc 7aac402 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 |
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)
|