import streamlit as st st.title("SMART FHIR Kits SDC HL7") st.markdown(""" HAPI FHIR: The HAPI FHIR project provides an open-source reference implementation of the FHIR specification. They offer a public test server that you can use to test your FHIR applications. You can access the server at https://hapi.fhir.org. Smile CDR: Smile CDR is a commercial FHIR server that offers a free test server that you can use for development and testing. You can access the server at https://smilecdr.com/free-fhir-test-server. Aidbox: Aidbox is another commercial FHIR server that offers a free test server for development and testing. You can sign up for a free account at https://aidbox.app/signup. Simplifier: Simplifier is an online platform for FHIR development that provides a free test server for FHIR R4 and STU3. You can sign up for a free account at https://simplifier.net. IBM FHIR Sandbox: IBM offers a free FHIR sandbox environment that you can use for development and testing. You can access the sandbox at https://ibm-fhir-server.mybluemix.net. """) st.markdown(""" import hl7apy from fhir.resources import Bundle, Patient, Observation from fhirclient.models.fhirreference import FHIRReference import streamlit as st # Create a sample HL7 v2.x message msg = hl7apy.Message("ORU^R01") msg.msh.msh_3 = "LAB" msg.msh.msh_4 = "LAB" msg.msh.msh_5 = "TEST" msg.pid.pid_3 = "1234" msg.pid.pid_5 = "Doe^John" msg.pid.pid_7 = "19800101" msg.obx = [] obx = hl7apy.Segment("OBX") obx.obx_2 = "ST" obx.obx_3 = "GLU" obx.obx_5 = "100" msg.obx.append(obx) # Convert HL7 v2.x message to FHIR resources patient = Patient(name=[{"given": ["John"], "family": "Doe"}], birthDate="1980-01-01") observation = Observation(code={"coding": [{"system": "http://loinc.org", "code": "2339-0", "display": "GLUCOSE"}]}, valueQuantity={"value": 100, "unit": "mg/dL"}, subject=FHIRReference({"reference": f"Patient/{patient.id}"})) bundle = Bundle(type="collection", entry=[{"resource": patient}, {"resource": observation}]) # Display the HL7 v2.x message and FHIR resources in the Streamlit app st.write("HL7 v2.x message:") st.code(str(msg)) st.write("FHIR resources:") st.code(bundle.json()) """) st.markdown(""" import requests import streamlit as st from fhir.resources import QuestionnaireResponse from fhirclient.models.fhirreference import FHIRReference # Set up the LOINC search API endpoint loinc_search_url = "https://search.loinc.org/search" # Set up the FHIR server base URL fhir_server_url = "http://hapi.fhir.org/baseR4" # Define the Exercise Assessment questionnaire ID exercise_questionnaire_id = "exercise-questionnaire" # Define the Exercise Assessment questionnaire response ID prefix exercise_response_prefix = "exercise-response" # Define the Exercise Assessment observation code exercise_observation_code = "8867-4" # Set the Streamlit app title and page layout st.set_page_config(page_title="Exercise Assessment", layout="wide") st.title("Exercise Assessment") # Define a function to search for LOINC codes def search_loinc_codes(query): params = { "sa": "true", "co": "true", "ec": "true", "df": "true", "loinc_num": query } response = requests.get(loinc_search_url, params=params) if response.ok: return response.json()["hits"] else: return [] # Display a search box for LOINC codes query = st.text_input("Enter a LOINC code:") # Search for LOINC codes and display the results if query: st.write(f"Search results for '{query}':") results = search_loinc_codes(query) for result in results: st.write(f"{result['code']} - {result['display']}") st.write(f"{result['system']}#{result['code']}") # Allow the user to select a LOINC code if len(results) == 1: selected_code = results[0]["code"] else: selected_code = st.selectbox("Select a LOINC code:", [result["code"] for result in results]) # Render the Exercise Assessment using the selected LOINC code st.write(f"Selected LOINC code: {selected_code}") exercise_questionnaire_response_id = f"{exercise_response_prefix}-{selected_code}" exercise_questionnaire_response = QuestionnaireResponse( questionnaire=FHIRReference(f"Questionnaire/{exercise_questionnaire_id}"), status="in-progress", subject=FHIRReference("Patient/example") ) exercise_questionnaire_response.identifier = [{"value": exercise_questionnaire_response_id}] exercise_questionnaire_response.item = [ { "linkId": "1", "text": "How many minutes of aerobic exercise did you do today?", "type": "integer" }, { "linkId": "2", "text": "How many minutes of strength training did you do today?", "type": "integer" } ] st.write("Exercise Assessment:") st.json(exercise_questionnaire_response.as_json()) # Save the Exercise Assessment to the FHIR server fhir_client = FHIRClient(settings={"app_id": "my_web_app", "api_base": fhir_server_url}) fhir_client.create(exercise_questionnaire_response) """) st.markdown(""" from hl7apy.parser import parse_message from fhirpy import SyncFHIRClient from fhirpy.base.exceptions import OperationOutcome from fhir.resources import Bundle, Patient, Observation from fhirclient.models.fhirreference import FHIRReference from fhirclient.models.codeableconcept import CodeableConcept from fhirclient.models.fhirsearch import FHIRSearch from fhirclient.models.observation import Observation as FhirObservation from fhirclient.models.questionnaire import QuestionnaireResponse as FhirQuestionnaireResponse from fhirclient.models.fhirabstractbase import FHIRValidationError import webbrowser # Set up the FHIR server base URL fhir_server_url = "https://fhirtest.uhn.ca/baseDstu3" # Set up the SMART on FHIR launch URL smart_launch_url = "https://launch.smarthealthit.org/v/r4/sim/eyJhIjoiMSIsImYiOiI5LjUuMTQwMDkiLCJlIjoiMi4wLjAiLCJzIjoibmV3LXNzbCIsInQiOiJkYXRhc2V0In0/fhir" # Define the LOINC code for the test observation test_observation_loinc = "29463-7" # Define the Exercise Assessment questionnaire ID exercise_questionnaire_id = "exercise-questionnaire" # Define the Exercise Assessment questionnaire response ID prefix exercise_response_prefix = "exercise-response" # Define the Exercise Assessment observation code exercise_observation_code = "8867-4" # Define the SMART on FHIR launch parameters smart_launch_params = { "iss": fhir_server_url, "launch": "12345", "patient": "Patient/123", "scope": "patient/*.read", "aud": "https://example.com/fhir" } # Create a FHIR client client = SyncFHIRClient(fhir_server_url) # Receive an HL7 v2.x message hl7_message = b"MSH|^~\&|HIS|FHIRCLIENT|HIS|FHIRCLIENT|20230101010101||ORU^R01|123|P|2.5.1||||||||\nPID|||1234^^^^MR||DOE^JOHN|||||||||||||||\nOBR|1|1234||^^^29463-7^GLU^L|||20230101010101|||||||||||||||||||||||||||||\nOBX|1|ST|29463-7^GLU^L|1|100|mg/dL|||||F\n" message = parse_message(hl7_message) # Convert the HL7 v2.x message to FHIR resources patient = Patient( id=message.pid.pid_3.value, name=[{"given": [message.pid.pid_5.value.split("^")[1]], "family": message.pid.pid_5.value.split("^")[0]}], birthDate=message.pid.pid_7.value ) observation = Observation( code=CodeableConcept( coding=[{"system": "http://loinc.org", "code": test_observation_loinc, "display": "GLUCOSE"}], text="Glucose" ), valueQuantity={ "value": float(message.obx[0].obx_5.value), "unit": message.obx[0].obx_6.value ), subject=FHIRReference({"reference": f"Patient/{patient.id}"}) ) Create a bundle with the Patient and Observation resources bundle = Bundle(type="collection", entry=[{"resource": patient}, {"resource": observation}]) Save the bundle to the FHIR server try: response = client.create(bundle) st.write("Observation saved to FHIR server:") st.json(response.as_json()) except OperationOutcome as error: st.write("Error saving observation to FHIR server:") st.json(error.as_json()) Render the Exercise Assessment using the FHIR resources exercise_questionnaire_response_id = f"{exercise_response_prefix}-{observation.code.coding[0].code}" exercise_questionnaire_response = FhirQuestionnaireResponse( questionnaire=FHIRReference(f"Questionnaire/{exercise_questionnaire_id}"), status="in-progress", subject=FHIRReference({"reference": f"Patient/{patient.id}"}) ) exercise_questionnaire_response.identifier = [{"value": exercise_questionnaire_response_id}] exercise_questionnaire_response.item = [ { "linkId": "1", "text": "How many minutes of aerobic exercise did you do today?", "type": "integer" }, { "linkId": "2", "text": "How many minutes of strength training did you do today?", "type": "integer" } ] Save the Exercise Assessment to the FHIR server try: response = client.create(exercise_questionnaire_response) st.write("Exercise Assessment saved to FHIR server:") st.json(response.as_json()) except (OperationOutcome, FHIRValidationError) as error: st.write("Error saving Exercise Assessment to FHIR server:") st.json(error.as_json()) Generate the SMART on FHIR launch URL with launch parameters smart_launch_url_with_params = f"{smart_launch_url}?{'&'.join([f'{key}={value}' for key, value in smart_launch_params.items()])}" Display the SMART on FHIR launch URL st.write("SMART on FHIR launch URL:") st.write(smart_launch_url_with_params) Open the SMART on FHIR UI in a web browser webbrowser.open(smart_launch_url_with_params) This program receives an HL7 v2.x message, converts it to FHIR resources (a Patient and an Observation), saves the resources to the FHIR server, and then renders the Exercise Assessment questionnaire using the saved resources. The program then saves the Exercise Assessment to the FHIR server and generates a SMART on FHIR launch URL with launch parameters. Finally, the program displays the launch URL and opens the SMART on FHIR UI in a web browser. Note that in order to run this program, you'll need to have the `hl7apy`, `fhirpy`, `fhir.resources`, `fhirclient`, and `webbrowser` packages installed, and you'll need to update the `fhir_server_url`, `smart_launch_url`, `test_observation_loinc`, `exercise_questionnaire_id`, and `exercise_response_prefix` variables to match your environment. """)