Dualmind-Assistant
129 lines
3.3 KiB
Python
129 lines
3.3 KiB
Python
from pathlib import Path
|
|
import json
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
|
|
|
|
# --- Domain models ---
|
|
|
|
|
|
class Answer(BaseModel):
|
|
question_code: str
|
|
value: Any
|
|
|
|
|
|
class SessionCreate(BaseModel):
|
|
language: str
|
|
answers: List[Answer]
|
|
|
|
|
|
class MtsPreparation(BaseModel):
|
|
session_id: str
|
|
proposed_presenting_flowchart: Optional[str] = None
|
|
red_flag_indicators: List[str] = []
|
|
suggested_priority_level: Optional[str] = None
|
|
|
|
|
|
class QuestionOption(BaseModel):
|
|
value: str
|
|
label: str
|
|
icon: str | None = None
|
|
|
|
|
|
class Question(BaseModel):
|
|
code: str
|
|
type: str
|
|
title: str
|
|
options: Optional[List[QuestionOption]] = None
|
|
min: Optional[int] = None
|
|
max: Optional[int] = None
|
|
|
|
|
|
class QuestionsResponse(BaseModel):
|
|
language: str
|
|
questions: List[Question]
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parents[2]
|
|
MTS_CONFIG_DIR = BASE_DIR / "mts-config"
|
|
|
|
|
|
def load_questions(language: str) -> QuestionsResponse:
|
|
"""Load questions configuration for the given language from mts-config.
|
|
|
|
This keeps the triage logic data-driven and allows adding further
|
|
languages by simply providing a new questions.<lang>.json file.
|
|
"""
|
|
|
|
file_path = MTS_CONFIG_DIR / f"questions.{language}.json"
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=404, detail=f"No question config for language '{language}'")
|
|
|
|
with file_path.open("r", encoding="utf-8") as f:
|
|
raw: Dict[str, Any] = json.load(f)
|
|
|
|
return QuestionsResponse(**raw)
|
|
|
|
|
|
app = FastAPI(title="Triage-Fragen API", version="0.2.0")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.get("/health")
|
|
async def health() -> Dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/questions/{language}", response_model=QuestionsResponse)
|
|
async def get_questions(language: str) -> QuestionsResponse:
|
|
"""Return the triage question configuration for the selected language."""
|
|
|
|
return load_questions(language)
|
|
|
|
|
|
@app.post("/sessions", response_model=MtsPreparation)
|
|
async def create_session(payload: SessionCreate) -> MtsPreparation:
|
|
"""Create a triage preparation proposal from the given answers.
|
|
|
|
For now this implements a very small ruleset as a walking skeleton
|
|
that we can refine step by step.
|
|
"""
|
|
|
|
session_id = "dummy-session-id"
|
|
|
|
flowchart: Optional[str] = None
|
|
red_flags: List[str] = []
|
|
priority: Optional[str] = None
|
|
|
|
for ans in payload.answers:
|
|
if ans.question_code == "chief_complaint" and ans.value == "chest_pain":
|
|
flowchart = "CHEST_PAIN"
|
|
if ans.question_code == "breathlessness" and ans.value is True:
|
|
red_flags.append("breathlessness")
|
|
if (
|
|
ans.question_code == "pain_intensity"
|
|
and isinstance(ans.value, int)
|
|
and ans.value >= 8
|
|
):
|
|
red_flags.append("severe_pain")
|
|
|
|
if "breathlessness" in red_flags:
|
|
priority = "RED_OR_ORANGE"
|
|
|
|
return MtsPreparation(
|
|
session_id=session_id,
|
|
proposed_presenting_flowchart=flowchart,
|
|
red_flag_indicators=red_flags,
|
|
suggested_priority_level=priority,
|
|
)
|