Spaces:
Runtime error
Runtime error
File size: 3,628 Bytes
1150b5c 0e1a6eb 1150b5c 0e1a6eb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
import os
import random
import statistics
from typing import Dict, Optional
from fastapi import BackgroundTasks, Body, FastAPI, Response, UploadFile, status
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="Insulin",
description="Insulin to treat diabetes",
quantity=1.0,
units="ml",
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",
],
),
PatientIssue(
title="Patient is not taking metformin",
reason="Reports feeling nauseous",
suggested_actions=[
"Discuss reason with physician",
"Suggest taking metformin with food",
],
),
]
}
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}/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)
return Response(status_code=status.HTTP_202_ACCEPTED)
|