Spaces:
Runtime error
Runtime error
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() | |
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_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] | |