Spaces:

natexcvi
/

med-hackaathon

Runtime error

med-hackaathon / main.py

natexcvi

Fix json path parsing

e51d953 unverified over 1 year ago

No virus

5.82 kB

	import os
	import random
	import statistics
	from typing import Any, Dict, Optional

	from fastapi import BackgroundTasks, Body, FastAPI, Response, UploadFile, status
	from jsonpath import JSONPath

	from schema import *
	from telegram_bot import checkup_messages, send_message_to_user

	app = FastAPI()
	tasks: Dict[str, Optional[Dict]] = {}


	class DB:
	def __init__(self):
	self.care_plans = {
	"123": CarePlan(
	purpose="Diabetes",
	start="2021-01-01",
	end="2021-12-31",
	items=[
	CarePlanItem(
	name="Glyburide",
	description="Glyburide to treat diabetes",
	quantity=2.5,
	units="mg",
	frequency="daily",
	start="2021-01-01",
	adherence=0.9,
	),
	CarePlanItem(
	name="Metformin",
	description="Metformin to treat diabetes",
	quantity=1.0,
	units="tablet",
	frequency="daily",
	start="2021-01-01",
	adherence=0.8,
	),
	],
	)
	}
	self.issues = {
	"123": [
	PatientIssue(
	title="Patient does not engage in physical activity",
	reason="Reports feeling tired and fatigued",
	suggested_actions=[
	"Discuss reason with physician",
	"Suggest lighter activities",
	],
	),
	]
	}
	self.adherence_history = {
	"123": AdherenceHistory(
	history=[
	AdherenceHistoryRecord(date="2021-01-01", adherence=0.9),
	AdherenceHistoryRecord(date="2021-02-01", adherence=0.3),
	AdherenceHistoryRecord(date="2021-03-01", adherence=0.5),
	AdherenceHistoryRecord(date="2021-04-01", adherence=0.79),
	AdherenceHistoryRecord(date="2021-05-01", adherence=0.7),
	AdherenceHistoryRecord(date="2021-06-01", adherence=0.8),
	AdherenceHistoryRecord(date="2021-07-01", adherence=0.75),
	AdherenceHistoryRecord(date="2021-08-01", adherence=0.78),
	AdherenceHistoryRecord(date="2021-09-01", adherence=0.80),
	AdherenceHistoryRecord(date="2021-10-01", adherence=0.85),
	]
	)
	}
	self.conversation_state = {
	"1773171761": {
	"current_question": 0,
	"patient_id": "123",
	"action": "",
	}
	}

	def reset(self):
	self.__init__()


	db = DB()


	@app.post("/echo")
	async def echo(msg: str):
	return msg


	@app.post("/trigger-medication-check")
	async def trigger_medication_check(background_tasks: BackgroundTasks):
	background_tasks.add_task(
	send_message_to_user,
	chat_id=os.getenv("TELEGRAM_USER_ID", ""),
	message=random.choice(checkup_messages),
	)
	return Response(status_code=status.HTTP_202_ACCEPTED)


	@app.get("/{patient_id}/care_plan", response_model=CarePlan)
	async def get_patient_care_plan(patient_id: str) -> CarePlan \| Response:
	if patient_id not in db.care_plans:
	return Response(status_code=status.HTTP_404_NOT_FOUND)
	return db.care_plans[patient_id]


	@app.get("/{patient_id}/adherence_status", response_model=AdherenceStatus)
	async def get_patient_adherence_status(patient_id: str) -> AdherenceStatus \| Response:
	if patient_id not in db.care_plans:
	return Response(status_code=status.HTTP_404_NOT_FOUND)
	return AdherenceStatus(
	adherence=statistics.mean(
	[item.adherence for item in db.care_plans[patient_id].items]
	)
	)


	@app.get("/{patient_id}/adherence_history", response_model=AdherenceHistory)
	async def get_patient_adherence_history(patient_id: str) -> Response:
	if patient_id not in db.adherence_history:
	return Response(status_code=status.HTTP_404_NOT_FOUND)
	return db.adherence_history[patient_id]


	@app.get("/{patient_id}/issues", response_model=list[PatientIssue])
	async def get_patient_issues(patient_id: str) -> list[PatientIssue] \| Response:
	if patient_id not in db.issues:
	return Response(status_code=status.HTTP_404_NOT_FOUND)
	return db.issues[patient_id]


	@app.post("/log-message-response")
	async def log_message_response(msg: dict = Body(...)):
	print(msg)
	user_id = JSONPath("$..from.id").parse(msg)[0]
	if user_id not in db.conversation_state:
	return Response(status_code=status.HTTP_404_NOT_FOUND)
	state = db.conversation_state[user_id]
	patient_id = state["patient_id"]
	response = JSONPath("$..queryResult.queryText").parse(msg)[0]
	action = JSONPath("$..queryResult.action").parse(msg)[0]
	state["action"] = action
	state["current_question"] += 1
	if state["action"] == "good-flow.good-flow-no.good-flow-no-no-2":
	db.care_plans[patient_id].items[0].adherence -= 0.02
	db.issues[patient_id].append(
	PatientIssue(
	title="Patient is not taking glyburide",
	reason=response,
	suggested_actions=[
	"Discuss side effects with physician",
	"Suggest taking glyburide with food",
	],
	)
	)
	return Response(status_code=status.HTTP_202_ACCEPTED)


	@app.delete("/reset-db")
	async def reset_db():
	db.reset()
	return Response(status_code=status.HTTP_200_OK)