Spaces:
Runtime error
Runtime error
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() | |
async def echo(msg: str): | |
return msg | |
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) | |
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] | |
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] | |
) | |
) | |
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] | |
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] | |
async def log_message_response(msg: dict = Body(...)): | |
print(msg) | |
user_id = JSONPath("$..from.id").parse(msg) | |
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) | |
action = JSONPath("$..queryResult.action").parse(msg) | |
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) | |
async def reset_db(): | |
db.reset() | |
return Response(status_code=status.HTTP_200_OK) | |