91 lines
3.9 KiB
Python
91 lines
3.9 KiB
Python
"""Service Questionnaires cliniques (médical).
|
|
|
|
Périmètre CENTRAL (la passation on-device viendra plus tard, via une future spec RPC) :
|
|
- **authoring** par les médecins (QCM : questions → choix scorés) ; PHQ-9 fourni ;
|
|
- **assignation** à un/des patients avec une **fréquence** (1 fois / récurrent) ;
|
|
- **résultats** de passation → **dossier patient** (réponses chiffrées, score calculé).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from kazeia_central.core.store import Store
|
|
|
|
|
|
def list_questionnaires(store: Store) -> list[dict[str, Any]]:
|
|
return store.list_questionnaires()
|
|
|
|
|
|
def get_questionnaire(store: Store, qid: str) -> dict[str, Any] | None:
|
|
return store.get_questionnaire(qid)
|
|
|
|
|
|
def save_questionnaire(store: Store, qid: str, title: str, definition: dict, *,
|
|
now: int, actor: str | None = None) -> dict[str, Any]:
|
|
store.upsert_questionnaire(qid, title, definition, now=now)
|
|
store.audit("questionnaire_saved", actor=actor, target=qid,
|
|
detail={"questions": len(definition.get("questions", []))}, now=now)
|
|
return {"id": qid, "title": title}
|
|
|
|
|
|
def delete_questionnaire(store: Store, qid: str, *, now: int, actor: str | None = None) -> dict[str, Any]:
|
|
ok = store.delete_questionnaire(qid)
|
|
if ok:
|
|
store.audit("questionnaire_deleted", actor=actor, target=qid, now=now)
|
|
return {"deleted": ok}
|
|
|
|
|
|
def assign(store: Store, qid: str, serial: str, profile_id: str, frequency: dict, *,
|
|
now: int, actor: str | None = None) -> dict[str, Any]:
|
|
aid = store.assign_questionnaire(qid, serial, profile_id, frequency, now=now)
|
|
store.audit("questionnaire_assigned", actor=actor, target=f"{serial}/{profile_id}",
|
|
detail={"questionnaire": qid, "frequency": frequency}, now=now)
|
|
return {"assignment_id": aid, "questionnaire_id": qid}
|
|
|
|
|
|
def assignments(store: Store, *, serial: str | None = None, profile_id: str | None = None) -> list[dict[str, Any]]:
|
|
return store.assignments(serial=serial, profile_id=profile_id)
|
|
|
|
|
|
def deactivate(store: Store, aid: str, *, now: int, actor: str | None = None) -> dict[str, Any]:
|
|
store.deactivate_assignment(aid)
|
|
store.audit("questionnaire_unassigned", actor=actor, target=aid, now=now)
|
|
return {"assignment_id": aid, "active": False}
|
|
|
|
|
|
def score(definition: dict, answers: list[int | None]) -> dict[str, Any]:
|
|
"""Calcule le score (somme des choix sélectionnés) + l'interprétation (ranges)."""
|
|
scale = definition.get("scale") or []
|
|
qs = definition.get("questions") or []
|
|
total = 0
|
|
for i, q in enumerate(qs):
|
|
choices = q.get("choices") or scale
|
|
a = answers[i] if i < len(answers) else None
|
|
if a is not None and 0 <= a < len(choices):
|
|
total += int(choices[a].get("score", 0))
|
|
label = None
|
|
for rng in (definition.get("scoring") or {}).get("ranges", []):
|
|
if rng["min"] <= total <= rng["max"]:
|
|
label = rng["label"]
|
|
break
|
|
return {"total_score": total, "interpretation": label}
|
|
|
|
|
|
def record_result(store: Store, qid: str, serial: str, profile_id: str, answers: list, *,
|
|
now: int, actor: str | None = None) -> dict[str, Any]:
|
|
"""Enregistre une passation (réponses chiffrées + score) dans le dossier patient."""
|
|
q = store.get_questionnaire(qid)
|
|
if not q:
|
|
raise ValueError(f"questionnaire inconnu: {qid}")
|
|
scored = score(q["definition"], answers)
|
|
result = {"answers": answers, **scored}
|
|
rid = store.record_result(qid, serial, profile_id, result, now=now)
|
|
store.audit("questionnaire_result", actor=actor, target=f"{serial}/{profile_id}",
|
|
detail={"questionnaire": qid, "score": scored["total_score"]}, now=now)
|
|
return {"result_id": rid, **scored}
|
|
|
|
|
|
def patient_results(store: Store, serial: str, profile_id: str) -> list[dict[str, Any]]:
|
|
return store.patient_results(serial, profile_id)
|