Spaces:

awacke1
/

SMART-FHIR-Assessment-Observation-SDKs

Runtime error

App Files Files Community

SMART-FHIR-Assessment-Observation-SDKs / app.py

awacke1

Update app.py

f9462db over 1 year ago

raw history blame contribute delete

No virus

23.6 kB

	import streamlit as st

	# pip install streamlit fhir.resources==6.5.0 fhirclient smart.models sdcclient
	# FHIR resources functions
	from datetime import datetime
	from fhir.resources.activitydefinition import ActivityDefinition
	from fhir.resources.medicationrequest import MedicationRequest
	from fhir.resources.observation import Observation
	from fhir.resources.diagnosticreport import DiagnosticReport
	from fhir.resources.questionnaireresponse import QuestionnaireResponse
	from fhir.resources.careplan import CarePlan
	from fhir.resources.goal import Goal
	from fhir.resources.condition import Condition
	from fhir.resources.patient import Patient
	#from fhir.resources.patient import HumanName
	from fhir.resources.humanname import HumanName
	from fhir.resources.procedure import Procedure
	from fhir.resources.encounter import Encounter
	from fhir.resources.organization import Organization
	from fhir.resources.practitioner import Practitioner
	from fhir.resources.practitionerrole import PractitionerRole
	from fhir.resources.healthcareservice import HealthcareService
	from fhir.resources.location import Location
	from fhir.resources.immunization import Immunization
	from fhir.resources.documentreference import DocumentReference
	from fhir.resources.medicationdispense import MedicationDispense
	from fhir.resources.medicationstatement import MedicationStatement
	from fhir.resources.appointment import Appointment
	from fhir.resources.schedule import Schedule
	from fhir.resources.slot import Slot
	from fhir.resources.patient import Patient
	from fhir.resources.questionnaire import Questionnaire
	from fhir.resources.activitydefinition import ActivityDefinition
	from fhir.resources.measure import Measure
	from fhir.resources.plandefinition import PlanDefinition
	from fhir.resources.careteam import CareTeam
	from fhir.resources.person import Person
	from fhir.resources.group import Group
	from fhir.resources.practitioner import Practitioner
	from fhir.resources.practitionerrole import PractitionerRole
	from fhir.resources.healthcareservice import HealthcareService
	from fhir.resources.location import Location
	from fhir.resources.immunization import Immunization
	from fhir.resources.documentreference import DocumentReference
	from fhir.resources.medicationdispense import MedicationDispense
	from fhir.resources.medicationstatement import MedicationStatement
	from fhir.resources.appointment import Appointment
	from fhir.resources.schedule import Schedule
	from fhir.resources.slot import Slot
	from fhir.resources.servicerequest import ServiceRequest
	from fhir.resources.communicationrequest import CommunicationRequest
	from fhir.resources.annotation import Annotation


	import json
	from typing import List, Tuple

	#import smart.models as smart
	#from sdcclient import SdcClient

	# Initialize the SMART client
	# smart_client = smart.client

	# Initialize the SDC client
	#sdc_client = SdcClient()

	# Define a list of example patients
	EXAMPLE_PATIENTS = [
	{
	"name": "Alice Smith",
	"birthdate": "1980-01-01",
	"gender": "female",
	"address": {
	"line": ["123 Main St"],
	"city": "Anytown",
	"state": "NY",
	"postalCode": "12345",
	"country": "US"
	},
	"phone": "555-555-1212"
	},
	{
	"name": "Bob Johnson",
	"birthdate": "1975-05-05",
	"gender": "male",
	"address": {
	"line": ["456 Oak St"],
	"city": "Anytown",
	"state": "NY",
	"postalCode": "12345",
	"country": "US"
	},
	"phone": "555-555-1212"
	}
	]

	# Define a list of example social determinants of health
	EXAMPLE_SDH = [
	{
	"question": "Do you have reliable transportation?",
	"answer": "Yes"
	},
	{
	"question": "Do you have enough food to eat?",
	"answer": "No"
	},
	{
	"question": "Do you have stable housing?",
	"answer": "No"
	}
	]

	def get_patient(name: str) -> Tuple[Patient, str]:
	"""
	Returns a tuple containing the FHIR Patient resource and the SMART Patient model
	for the given patient name.
	"""
	# Get the example patient with the matching name
	example_patient = next((p for p in EXAMPLE_PATIENTS if p["name"] == name), None)
	if not example_patient:
	raise ValueError(f"No example patient found with name '{name}'")

	# Create a FHIR Patient resource
	patient = Patient()
	patient.name = [HumanName()]
	patient.name[0].given = [example_patient["name"].split()[0]]
	# patient.name[0].family = [example_patient["name"].split()[0]]

	# patient.birthDate = FHIRDate(example_patient["birthdate"])
	patient.gender = example_patient["gender"]
	#patient.address = [Address()]
	#patient.address.append(Address())

	#patient.address[0].line = example_patient["address"]["line"]
	#patient.address[0].city = example_patient["address"]["city"]
	#patient.address[0].state = example_patient["address"]["state"]
	#patient.address[0].postalCode = example_patient["address"]["postalCode"]
	#patient.address[0].country = example_patient["address"]["country"]
	#patient.telecom = [ContactPoint()]
	#patient.telecom[0].system = "phone"
	#patient.telecom[0].value = example_patient["phone"]

	# Create a SMART Patient model
	smart_patient = smart.Patient.read_from(patient.as_json())

	return patient, smart_patient

	def create_observation(patient_id: str, code: str, value: str, unit: str) -> Observation:
	"""
	Creates and returns a FHIR Observation resource with the given patient ID, code, value, and unit.
	"""
	observation = Observation()
	observation.subject = {"reference": f"Patient/{patient_id}"}
	observation.code = CodeableConcept()
	observation.code.coding = [Coding()]
	observation.code.coding[0].system = "http://loinc.org"
	observation.code.coding[0].code = code
	observation.valueQuantity = Quantity()
	observation.valueQuantity.value = float(value)
	observation.valueQuantity.unit = unit
	observation.status = "final"
	observation.effectiveDateTime = FHIRDate("2023-02-21")
	observation.meta = Meta()
	observation.meta.profile = ["http://hl7.org/fhir/StructureDefinition/vitalsigns"]

	return observation

	def create_assessment(patient_id: str, code: str, value: str) -> DiagnosticReport:
	"""
	Creates and returns a FHIR DiagnosticReport resource with the given patient ID, code, and value.
	"""
	report = DiagnosticReport()
	report.status = "final"
	report.subject = {"reference": f"Patient/{patient_id}"}
	report.code = CodeableConcept()
	report.code.coding = [Coding()]
	report.code.coding[0].system = "http://loinc.org"
	report.code.coding[0].code = code
	report.result = [Reference()]
	report.result[0].reference = f"Observation/{code}"
	report.result[0].display = value
	report.effectiveDateTime = FHIRDate("2023-02-21")

	return report


	def create_provider(name: str, organization_name: str) -> Tuple[Practitioner, PractitionerRole, Organization]:
	"""
	Creates and returns a tuple containing the FHIR Practitioner, PractitionerRole, and Organization resources for the given provider name and organization name.
	"""
	# Create a FHIR Practitioner resource
	practitioner = Practitioner()
	practitioner.name = [HumanName()]
	practitioner.name[0].text = name
	practitioner.identifier = [Identifier()]
	practitioner.identifier[0].system = "http://example.com/providers"
	practitioner.identifier[0].value = "12345"

	# Create a FHIR PractitionerRole resource
	practitioner_role = PractitionerRole()
	practitioner_role.practitioner = {"reference": f"Practitioner/{practitioner.id}"}
	practitioner_role.organization = {"reference": f"Organization/{organization_name}"}
	practitioner_role.code = [CodeableConcept()]
	practitioner_role.code[0].coding = [Coding()]
	practitioner_role.code[0].coding[0].system = "http://nucc.org/provider-taxonomy"
	practitioner_role.code[0].coding[0].code = "207Q00000X"
	practitioner_role.code[0].coding[0].display = "Family Medicine"
	practitioner_role.specialization = [CodeableConcept()]
	practitioner_role.specialization[0].coding = [Coding()]
	practitioner_role.specialization[0].coding[0].system = "http://snomed.info/sct"
	practitioner_role.specialization[0].coding[0].code = "123456"
	practitioner_role.specialization[0].coding[0].display = "Example Specialty"

	# Create a FHIR Organization resource
	organization = Organization()
	organization.name = organization_name

	return practitioner, practitioner_role, organization


	def create_fulfillment(patient_id: str, medication_name: str, quantity: int, dispense_date: str) -> MedicationStatement:
	"""
	Creates and returns a FHIR MedicationStatement resource representing a fulfillment of a prescription for the given patient ID, medication name, quantity, and dispense date.
	"""
	# Create a FHIR Medication resource
	medication = Medication()
	medication.code = CodeableConcept()
	medication.code.text = medication_name

	# Create a FHIR MedicationStatement resource
	fulfillment = MedicationStatement()
	fulfillment.status = "completed"
	fulfillment.medicationReference = {"reference": f"Medication/{medication.id}"}
	fulfillment.subject = {"reference": f"Patient/{patient_id}"}
	fulfillment.dosage = [Dosage()]
	fulfillment.dosage[0].route = CodeableConcept()
	fulfillment.dosage[0].route.coding = [Coding()]
	fulfillment.dosage[0].route.coding[0].system = "http://example.com/routes"
	fulfillment.dosage[0].route.coding[0].code = "123456"
	fulfillment.dosage[0].route.coding[0].display = "Example Route"
	fulfillment.dosage[0].quantity = Quantity()
	fulfillment.dosage[0].quantity.value = quantity
	fulfillment.dosage[0].quantity.unit = "pill"
	fulfillment.effectiveDateTime = FHIRDate(dispense_date)

	return fulfillment


	def create_note(patient_id: str, text: str) -> DocumentReference:
	"""
	Creates and returns a FHIR DocumentReference resource representing a note for the given patient ID and text.
	"""
	note = DocumentReference()
	note.status = "current"
	note.subject = {"reference": f"Patient/{patient_id}"}
	note.type = CodeableConcept()
	note.type.coding = [Coding()]
	note.type.coding[0].system = "http://loinc.org"
	note.type.coding[0].code = "11506-3"
	note.type.coding[0].display = "Consult note"
	note.content = [Attachment()]
	note.content[0].contentType = "text/plain"
	note.content[0].data = text.encode("utf-8")
	note.author = [Reference()]
	note.author[0].reference = f"Practitioner/example-provider"
	note.date = FHIRDate("2023-02-21")

	return note

	def create_social_determinant(question: str, answer: str) -> QuestionnaireResponse:
	"""
	Creates and returns a FHIR QuestionnaireResponse resource representing a social determinant of health with the given question and answer.
	"""
	response = SmartQuestionnaireResponse()
	response.questionnaire = "http://example.com/sdh-questionnaire"
	response.item = []
	item = QuestionnaireResponseItem()
	item.linkId = "1"
	item.text = question
	item.answer = []
	answer_item = QuestionnaireResponseItemAnswer()
	answer_item.valueString = answer
	item.answer.append(answer_item)
	response.item.append(item)

	return response

	def create_care_team(name: str, provider_names: List[str]) -> CareTeam:
	"""
	Creates and returns a FHIR CareTeam resource representing a care team with the given name and provider names.
	"""
	care_team = SmartCareTeam()
	care_team.status = "active"
	care_team.name = name
	care_team.participant = []
	for provider_name in provider_names:
	provider_ref = f"Practitioner/{provider_name}"
	care_team.participant.append(CareTeamParticipant({"member": {"reference": provider_ref}}))

	return care_team

	def create_activity_definition(title: str, description: str, category: str, code: str) -> ActivityDefinition:
	"""
	Creates and returns a FHIR ActivityDefinition resource representing an activity definition with the given title, description, category, and code.
	"""
	activity_definition = SmartActivityDefinition()
	activity_definition.status = "draft"
	activity_definition.kind = "procedure"
	activity_definition.title = title
	activity_definition.description = description
	activity_definition.category = CodeableConcept({"coding": [Coding({"system": "http://example.com/categories", "code": category, "display": f"Example {category}"})]})
	activity_definition.code = CodeableConcept({"coding": [Coding({"system": "http://example.com/codes", "code": code, "display": f"Example {code}"})]})

	return activity_definition

	# Streamlit app code
	import streamlit as st
	from fhir.resources import *

	st.set_page_config(page_title="FHIR Demo", page_icon=":heart:", layout="wide")

	st.title("FHIR Demo")

	st.sidebar.title("Navigation")
	navigation = st.sidebar.radio("Go to", ("Home", "Observations", "Assessments", "Rules", "Referrals", "Providers", "Programs", "Fulfillment", "Alerts", "Notes", "Social Determinants of Health"))

	st.sidebar.title("SMART App")
	smart = None
	if st.sidebar.button("Launch SMART App"):
	smart = SmartApp.launch()
	st.sidebar.write("SMART App launched")
	if st.sidebar.button("Close SMART App"):
	if smart:
	smart.close()
	st.sidebar.write("SMART App closed")
	else:
	st.sidebar.write("No SMART App to close")

	if navigation == "Home":
	st.write("Welcome to the FHIR Demo!")
	st.write("Use the sidebar to navigate to different FHIR resources.")
	st.write("Use the SMART App buttons in the sidebar to launch and close the app.")

	elif navigation == "Observations":
	st.write("# Observations")
	patient_name = st.selectbox("Select patient", [p["name"] for p in EXAMPLE_PATIENTS])
	patient, smart_patient = get_patient(patient_name)
	st.write("### Patient")
	st.write(f"Name: {patient.name[0].given[0]} {patient.name[0].family[0]}")
	st.write(f"Birthdate: {patient.birthDate.as_json()}")
	st.write(f"Gender: {patient.gender}")
	st.write(f"Address: {patient.address[0].line[0]}, {patient.address[0].city}, {patient.address[0].state} {patient.address[0].postalCode}, {patient.address[0].country}")
	st.write(f"Phone: {patient.telecom[0].value}")
	st.write("### Add Observation")
	code = st.selectbox("Select code", ["8302-2", "8462-4", "8867-4", "9279-1"])
	value = st.number_input("Value")
	unit = st.selectbox("Select unit", ["bpm", "mmHg", "mg/dL", "kg"])
	observation = create_observation(patient.id, code, value, unit)
	smart_client = smart.patient(smart_patient)
	response = smart_client.create(observation)
	st.write("Observation created:")
	st.write(response.as_json())

	elif navigation == "Assessments":
	st.write("# Assessments")
	patient_name = st.selectbox("Select patient", [p["name"] for p in EXAMPLE_PATIENTS])
	patient, smart_patient = get_patient(patient_name)
	st.write("### Patient")
	st.write(f"Name: {patient.name[0].given[0]} {patient.name[0].family[0]}")
	st.write(f"Birthdate: {patient.birthDate.as_json()}")
	st.write(f"Gender: {patient.gender}")
	st.write(f"Address: {patient.address[0].line[0]}, {patient.address[0].city}, {patient.address[0].state} {patient.address[0].postalCode}, {patient.address[0].country}")
	st.write(f"Phone: {patient.telecom[0].value}")
	st.write("### Add Assessment")
	code = st.selectbox("Select code", ["69646-2", "8150-9", "82810-3"])
	value = st.selectbox("Select value", ["Absent", "Mild", "Moderate", "Severe"])
	assessment = create_assessment(patient.id, code, value)
	smart_client = smart.patient(smart_patient)
	response = smart_client.create(assessment)
	st.write("Assessment created:")
	st.write(response.as_json())

	elif navigation == "Rules":
	st.write("# Rules")
	st.write("### Add Rule")
	code = st.selectbox("Select code", ["36405-2", "89053-6"])
	description = st.text_input("Description")
	rule = create_rule(code, description)
	response = smart.server.create(rule)
	st.write("Rule created:")
	st.write(response.as_json())

	elif navigation == "Referrals":
	st.write("# Referrals")
	patient_name = st.selectbox("Select patient", [p["name"] for p in EXAMPLE_PATIENTS])
	patient, smart_patient = get_patient(patient_name)
	st.write("### Patient")
	st.write(f"Name: {patient.name[0].given[0]} {patient.name[0].family[0]}")
	st.write(f"Birthdate: {patient.birthDate.as_json()}")
	st.write(f"Gender: {patient.gender}")
	st.write(f"Address: {patient.address[0].line[0]}, {patient.address[0].city}, {patient.address[0].state} {patient.address[0].postalCode}, {patient.address[0].country}")
	st.write(f"Phone: {patient.telecom[0].value}")
	st.write("### Add Referral")
	reason = st.text_input("Reason")
	specialty = st.text_input("Specialty")
	provider_name = st.selectbox("Select provider", [p["name"] for p in EXAMPLE_PROVIDERS])
	provider, practitioner_role, organization = create_provider(provider_name, "Example Healthcare")
	response1 = smart.server.create(provider)
	response2 = smart.server.create(practitioner_role)
	response3 = smart.server.create(organization)
	referral = create_referral_request(patient.id, reason, specialty, provider_name)
	response4 = smart.server.create(referral)
	st.write("Referral created:")
	st.write(response4.as_json())

	elif navigation == "Providers":
	st.write("# Providers")
	st.write("### Add Provider")
	name = st.text_input("Name")
	organization_name = st.text_input("Organization")
	provider, practitioner_role, organization = create_provider(name, organization_name)
	response1 = smart.server.create(provider)
	response2 = smart.server.create(practitioner_role)
	response3 = smart.server.create(organization)
	st.write("Provider created:")
	st.write(response1.as_json())

	elif navigation == "Programs":
	st.write("# Programs")
	st.write("### Add Program")
	name = st.text_input("Name")
	goal_description = st.text_input("Goal description")
	start_date = st.date_input("Start date")
	end_date = st.date_input("End date")
	program = create_program(name, goal_description, start_date.isoformat(), end_date.isoformat())
	response = smart.server.create(program)
	st.write("Program created:")
	st.write(response.as_json())

	elif navigation == "Fulfillment":
	st.write("# Fulfillment")
	patient_name = st.selectbox("Select patient", [p["name"] for p in EXAMPLE_PATIENTS])
	patient, smart_patient = get_patient(patient_name)
	st.write("### Patient")
	st.write(f"Name: {patient.name[0].given[0]} {patient.name[0].family[0]}")
	st.write(f"Birthdate: {patient.birthDate.as_json()}")
	st.write(f"Gender: {patient.gender}")
	st.write(f"Address: {patient.address[0].line[0]}, {patient.address[0].city}, {patient.address[0].state} {patient.address[0].postalCode}, {patient.address[0].country}")
	st.write(f"Phone: {patient.telecom[0].value}")
	st.write("### Add Fulfillment")
	medication_name = st.selectbox("Select medication", ["Aspirin", "Lisinopril", "Metformin"])
	quantity = st.number_input("Quantity")
	dispense_date = st.date_input("Dispense date")
	fulfillment = create_fulfillment(patient.id, medication_name, quantity, dispense_date.isoformat())
	smart_client = smart.patient(smart_patient)
	response = smart_client.create(fulfillment)
	st.write("Fulfillment created:")
	st.write(response.as_json())

	st.markdown("""


	\| Library \| Description \| PyPI URL \|
	\|---------\|-------------\|----------\|
	\| FHIR-Resources \| 🩺 A Python library for working with FHIR resources. It provides classes and methods for creating, manipulating, and serializing FHIR resources. \| https://pypi.org/project/fhir-resources/ \|
	\| SMART on FHIR Python Client \| 🔒 A Python library for accessing SMART on FHIR servers. It provides classes and methods for authenticating with SMART servers and accessing FHIR resources. \| https://pypi.org/project/smart-on-fhir/ \|
	\| PyFHIR \| 📦 A Python library for parsing and generating FHIR resources. It provides classes for representing FHIR resources and methods for serializing and deserializing them. \| https://pypi.org/project/pyfhir/ \|
	\| HAPI FHIR \| 💻 A Python library for working with FHIR servers. It provides classes and methods for querying FHIR servers and working with FHIR resources. \| https://pypi.org/project/hapi-fhir/ \|
	\| FHIR-Parser \| 📄 A Python library for parsing FHIR resources. It provides a parser class for deserializing FHIR resources. \| https://pypi.org/project/fhir-parser/ \|
	\| FHIR-FLAT \| 🧐 A Python library for working with FHIR resources. It provides classes for representing FHIR resources and methods for serializing and deserializing them. \| https://pypi.org/project/fhir-flat/ \|
	\| HL7apy \| 📩 A Python library for working with HL7 messages. It provides classes and methods for parsing and generating HL7 messages. \| https://pypi.org/project/hl7apy/ \|
	\| pyHl7 \| 📨 A Python library for parsing and generating HL7 messages. It provides classes for representing HL7 messages and methods for serializing and deserializing them. \| https://pypi.org/project/pyhl7/ \|
	\| FHIR-Utils \| 🔧 A Python library for working with FHIR resources. It provides utility functions for common FHIR tasks. \| https://pypi.org/project/fhir-utils/ \|

	""")

	import streamlit as st

	# Angular libraries
	st.header("Angular Libraries")
	st.write("Here are some popular Angular libraries:")

	st.markdown("- [ngx-charts](https://www.npmjs.com/package/@swimlane/ngx-charts)")
	st.markdown("- [angular-material](https://material.angular.io/)")

	# Node.JS libraries
	st.header("Node.JS Libraries")
	st.write("Here are some popular Node.JS libraries:")

	st.markdown("- [express](https://expressjs.com/)")
	st.markdown("- [axios](https://www.npmjs.com/package/axios)")

	# Docker libraries
	st.header("Docker Libraries")
	st.write("Here are some popular Docker libraries:")

	st.markdown("- [docker-py](https://pypi.org/project/docker/)")
	st.markdown("- [docker-compose](https://docs.docker.com/compose/)")

	# Kubernetes libraries
	st.header("Kubernetes Libraries")
	st.write("Here are some popular Kubernetes libraries:")

	st.markdown("- [kubernetes](https://pypi.org/project/kubernetes/)")
	st.markdown("- [kubeflow](https://www.kubeflow.org/)")

	# GraphQL libraries
	st.header("GraphQL Libraries")
	st.write("Here are some popular GraphQL libraries:")

	st.markdown("- [graphene](https://pypi.org/project/graphene/)")
	st.markdown("- [graphql-core](https://pypi.org/project/graphql-core/)")

	# PostgreSQL libraries
	st.header("PostgreSQL Libraries")
	st.write("Here are some popular PostgreSQL libraries:")

	st.markdown("- [psycopg2](https://pypi.org/project/psycopg2/)")
	st.markdown("- [sqlalchemy](https://pypi.org/project/SQLAlchemy/)")

	# Snowflake libraries
	st.header("Snowflake Libraries")
	st.write("Here are some popular Snowflake libraries:")

	st.markdown("- [snowflake-connector-python](https://pypi.org/project/snowflake-connector-python/)")
	st.markdown("- [snowflake-sqlalchemy](https://pypi.org/project/snowflake-sqlalchemy/)")

	# AI libraries
	st.header("AI Libraries")
	st.write("Here are some popular AI libraries:")

	st.markdown("- [tensorflow](https://pypi.org/project/tensorflow/)")
	st.markdown("- [scikit-learn](https://pypi.org/project/scikit-learn/)")