"""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)