Initial commit

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.10
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN apt-get update && apt-get install -y \
+    ffmpeg \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+RUN --mount=type=secret,id=TELEGRAM_BOT_API,mode=0444,required=true
+RUN --mount=type=secret,id=TELEGRAM_USER,mode=0444,required=true
+
+COPY . .
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -0,0 +1,106 @@
+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]

requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.74.*
+requests==2.27.*
+uvicorn[standard]==0.17.*
+python-multipart
+python-telegram-bot

schema.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+
+
+class CarePlanItem(BaseModel):
+    name: str
+    description: str
+    quantity: float
+    units: str
+    frequency: str
+    start: str
+    adherence: float | None = None
+
+
+class CarePlan(BaseModel):
+    purpose: str
+    start: str
+    end: str
+    items: list[CarePlanItem]
+
+
+class PatientIssue(BaseModel):
+    title: str
+    reason: str
+    suggested_actions: list[str]
+
+
+class AdherenceStatus(BaseModel):
+    adherence: float

telegram_bot.py

@@ -0,0 +1,27 @@
+import asyncio
+import os
+import random
+
+import telegram
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+async def send_message_to_user(chat_id: str, message: str):
+    bot = telegram.Bot(token=os.getenv("TELEGRAM_BOT_TOKEN", ""))
+    await bot.send_message(chat_id=chat_id, text=message)
+
+
+checkup_messages = [
+    "Hey there, how are you feeling today?",
+    "Hello, how are you doing today?",
+    "Hi, how are you feeling today?",
+    "Hey, what's up?",
+    "Hi, how are you doing?",
+]
+
+
+if __name__ == "__main__":
+    msg = random.choice(checkup_messages)
+    asyncio.run(send_message_to_user(chat_id="1773171761", message=msg))