Spaces:
Build error
Build error
| 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. | |
| """) | |