Coverage for tasks/phq9.py : 57%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
3"""
4camcops_server/tasks/phq9.py
6===============================================================================
8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com).
10 This file is part of CamCOPS.
12 CamCOPS is free software: you can redistribute it and/or modify
13 it under the terms of the GNU General Public License as published by
14 the Free Software Foundation, either version 3 of the License, or
15 (at your option) any later version.
17 CamCOPS is distributed in the hope that it will be useful,
18 but WITHOUT ANY WARRANTY; without even the implied warranty of
19 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20 GNU General Public License for more details.
22 You should have received a copy of the GNU General Public License
23 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
25===============================================================================
27"""
29from typing import Any, Dict, List, Tuple, Type, TYPE_CHECKING
31from cardinal_pythonlib.stringfunc import strseq
32from fhirclient.models.questionnaire import QuestionnaireItem
33from fhirclient.models.questionnaireresponse import (
34 QuestionnaireResponseItem,
35 QuestionnaireResponseItemAnswer,
36)
37from sqlalchemy.ext.declarative import DeclarativeMeta
38from sqlalchemy.sql.sqltypes import Boolean, Integer
40from camcops_server.cc_modules.cc_constants import CssClass
41from camcops_server.cc_modules.cc_ctvinfo import CtvInfo, CTV_INCOMPLETE
42from camcops_server.cc_modules.cc_db import add_multiple_columns
43from camcops_server.cc_modules.cc_html import answer, get_yes_no, tr, tr_qa
44from camcops_server.cc_modules.cc_request import CamcopsRequest
45from camcops_server.cc_modules.cc_snomed import SnomedExpression, SnomedLookup
46from camcops_server.cc_modules.cc_sqla_coltypes import (
47 CamcopsColumn,
48 SummaryCategoryColType,
49 ZERO_TO_THREE_CHECKER,
50)
51from camcops_server.cc_modules.cc_summaryelement import SummaryElement
52from camcops_server.cc_modules.cc_task import (
53 get_from_dict,
54 Task,
55 TaskHasPatientMixin,
56)
57from camcops_server.cc_modules.cc_text import SS
58from camcops_server.cc_modules.cc_trackerhelpers import (
59 TrackerAxisTick,
60 TrackerInfo,
61 TrackerLabel,
62)
64if TYPE_CHECKING:
65 from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient
68# =============================================================================
69# PHQ-9
70# =============================================================================
72class Phq9Metaclass(DeclarativeMeta):
73 # noinspection PyInitNewSignature
74 def __init__(cls: Type['Phq9'],
75 name: str,
76 bases: Tuple[Type, ...],
77 classdict: Dict[str, Any]) -> None:
78 add_multiple_columns(
79 cls, "q", 1, cls.N_MAIN_QUESTIONS,
80 minimum=0, maximum=3,
81 comment_fmt="Q{n} ({s}) (0 not at all - 3 nearly every day)",
82 comment_strings=[
83 "anhedonia",
84 "mood",
85 "sleep",
86 "energy",
87 "appetite",
88 "self-esteem/guilt",
89 "concentration",
90 "psychomotor",
91 "death/self-harm",
92 ]
93 )
94 super().__init__(name, bases, classdict)
97class Phq9(TaskHasPatientMixin, Task,
98 metaclass=Phq9Metaclass):
99 """
100 Server implementation of the PHQ9 task.
101 """
102 __tablename__ = "phq9"
103 shortname = "PHQ-9"
104 provides_trackers = True
106 q10 = CamcopsColumn(
107 "q10", Integer,
108 permitted_value_checker=ZERO_TO_THREE_CHECKER,
109 comment="Q10 (difficulty in activities) (0 not difficult at "
110 "all - 3 extremely difficult)"
111 )
113 N_MAIN_QUESTIONS = 9
114 MAX_SCORE_MAIN = 3 * N_MAIN_QUESTIONS
115 MAIN_QUESTIONS = strseq("q", 1, N_MAIN_QUESTIONS)
117 @staticmethod
118 def longname(req: "CamcopsRequest") -> str:
119 _ = req.gettext
120 return _("Patient Health Questionnaire-9")
122 def is_complete(self) -> bool:
123 if self.any_fields_none(self.MAIN_QUESTIONS):
124 return False
125 if self.total_score() > 0 and self.q10 is None:
126 return False
127 if not self.field_contents_valid():
128 return False
129 return True
131 def get_trackers(self, req: CamcopsRequest) -> List[TrackerInfo]:
132 return [TrackerInfo(
133 value=self.total_score(),
134 plot_label="PHQ-9 total score (rating depressive symptoms)",
135 axis_label=f"Score for Q1-9 (out of {self.MAX_SCORE_MAIN})",
136 axis_min=-0.5,
137 axis_max=self.MAX_SCORE_MAIN + 0.5,
138 axis_ticks=[
139 TrackerAxisTick(27, "27"),
140 TrackerAxisTick(25, "25"),
141 TrackerAxisTick(20, "20"),
142 TrackerAxisTick(15, "15"),
143 TrackerAxisTick(10, "10"),
144 TrackerAxisTick(5, "5"),
145 TrackerAxisTick(0, "0"),
146 ],
147 horizontal_lines=[
148 19.5,
149 14.5,
150 9.5,
151 4.5
152 ],
153 horizontal_labels=[
154 TrackerLabel(23, req.sstring(SS.SEVERE)),
155 TrackerLabel(17, req.sstring(SS.MODERATELY_SEVERE)),
156 TrackerLabel(12, req.sstring(SS.MODERATE)),
157 TrackerLabel(7, req.sstring(SS.MILD)),
158 TrackerLabel(2.25, req.sstring(SS.NONE)),
159 ]
160 )]
162 def get_clinical_text(self, req: CamcopsRequest) -> List[CtvInfo]:
163 if not self.is_complete():
164 return CTV_INCOMPLETE
165 return [CtvInfo(content=(
166 f"PHQ-9 total score {self.total_score()}/{self.MAX_SCORE_MAIN} "
167 f"({self.severity(req)})"
168 ))]
170 def get_summaries(self, req: CamcopsRequest) -> List[SummaryElement]:
171 return self.standard_task_summary_fields() + [
172 SummaryElement(
173 name="total", coltype=Integer(),
174 value=self.total_score(),
175 comment=f"Total score (/{self.MAX_SCORE_MAIN})"),
176 SummaryElement(
177 name="n_core", coltype=Integer(),
178 value=self.n_core(),
179 comment="Number of core symptoms"),
180 SummaryElement(
181 name="n_other", coltype=Integer(),
182 value=self.n_other(),
183 comment="Number of other symptoms"),
184 SummaryElement(
185 name="n_total", coltype=Integer(),
186 value=self.n_total(),
187 comment="Total number of symptoms"),
188 SummaryElement(
189 name="is_mds", coltype=Boolean(),
190 value=self.is_mds(),
191 comment="PHQ9 major depressive syndrome?"),
192 SummaryElement(
193 name="is_ods", coltype=Boolean(),
194 value=self.is_ods(),
195 comment="PHQ9 other depressive syndrome?"),
196 SummaryElement(
197 name="severity", coltype=SummaryCategoryColType,
198 value=self.severity(req),
199 comment="PHQ9 depression severity"),
200 ]
202 def total_score(self) -> int:
203 return self.sum_fields(self.MAIN_QUESTIONS)
205 def one_if_q_ge(self, qnum: int, threshold: int) -> int:
206 value = getattr(self, "q" + str(qnum))
207 return 1 if value is not None and value >= threshold else 0
209 def n_core(self) -> int:
210 return (self.one_if_q_ge(1, 2) +
211 self.one_if_q_ge(2, 2))
213 def n_other(self) -> int:
214 return (self.one_if_q_ge(3, 2) +
215 self.one_if_q_ge(4, 2) +
216 self.one_if_q_ge(5, 2) +
217 self.one_if_q_ge(6, 2) +
218 self.one_if_q_ge(7, 2) +
219 self.one_if_q_ge(8, 2) +
220 self.one_if_q_ge(9, 1)) # suicidality
221 # suicidality counted whenever present
223 def n_total(self) -> int:
224 return self.n_core() + self.n_other()
226 def is_mds(self) -> bool:
227 return self.n_core() >= 1 and self.n_total() >= 5
229 def is_ods(self) -> bool:
230 return self.n_core() >= 1 and 2 <= self.n_total() <= 4
232 def severity(self, req: CamcopsRequest) -> str:
233 total = self.total_score()
234 if total >= 20:
235 return req.sstring(SS.SEVERE)
236 elif total >= 15:
237 return req.sstring(SS.MODERATELY_SEVERE)
238 elif total >= 10:
239 return req.sstring(SS.MODERATE)
240 elif total >= 5:
241 return req.sstring(SS.MILD)
242 else:
243 return req.sstring(SS.NONE)
245 def get_task_html(self, req: CamcopsRequest) -> str:
246 main_dict = {
247 None: None,
248 0: "0 — " + self.wxstring(req, "a0"),
249 1: "1 — " + self.wxstring(req, "a1"),
250 2: "2 — " + self.wxstring(req, "a2"),
251 3: "3 — " + self.wxstring(req, "a3")
252 }
253 q10_dict = {
254 None: None,
255 0: "0 — " + self.wxstring(req, "fa0"),
256 1: "1 — " + self.wxstring(req, "fa1"),
257 2: "2 — " + self.wxstring(req, "fa2"),
258 3: "3 — " + self.wxstring(req, "fa3")
259 }
260 q_a = ""
261 for i in range(1, self.N_MAIN_QUESTIONS + 1):
262 nstr = str(i)
263 q_a += tr_qa(self.wxstring(req, "q" + nstr),
264 get_from_dict(main_dict, getattr(self, "q" + nstr)))
265 q_a += tr_qa("10. " + self.wxstring(req, "finalq"),
266 get_from_dict(q10_dict, self.q10))
268 h = """
269 <div class="{CssClass.SUMMARY}">
270 <table class="{CssClass.SUMMARY}">
271 {tr_is_complete}
272 {total_score}
273 {depression_severity}
274 {n_symptoms}
275 {mds}
276 {ods}
277 </table>
278 </div>
279 <div class="{CssClass.EXPLANATION}">
280 Ratings are over the last 2 weeks.
281 </div>
282 <table class="{CssClass.TASKDETAIL}">
283 <tr>
284 <th width="60%">Question</th>
285 <th width="40%">Answer</th>
286 </tr>
287 {q_a}
288 </table>
289 <div class="{CssClass.FOOTNOTES}">
290 [1] Sum for questions 1–9.
291 [2] Total score ≥20 severe, ≥15 moderately severe,
292 ≥10 moderate, ≥5 mild, <5 none.
293 [3] Number of questions 1–2 rated ≥2.
294 [4] Number of questions 3–8 rated ≥2, or question 9
295 rated ≥1.
296 [5] ≥1 core symptom and ≥5 total symptoms (as per
297 DSM-IV-TR page 356).
298 [6] ≥1 core symptom and 2–4 total symptoms (as per
299 DSM-IV-TR page 775).
300 </div>
301 """.format(
302 CssClass=CssClass,
303 tr_is_complete=self.get_is_complete_tr(req),
304 total_score=tr(
305 req.sstring(SS.TOTAL_SCORE) + " <sup>[1]</sup>",
306 answer(self.total_score()) + f" / {self.MAX_SCORE_MAIN}"
307 ),
308 depression_severity=tr_qa(
309 self.wxstring(req, "depression_severity") + " <sup>[2]</sup>",
310 self.severity(req)
311 ),
312 n_symptoms=tr(
313 "Number of symptoms: core <sup>[3]</sup>, other "
314 "<sup>[4]</sup>, total",
315 answer(self.n_core()) + "/2, " +
316 answer(self.n_other()) + "/7, " +
317 answer(self.n_total()) + "/9"
318 ),
319 mds=tr_qa(
320 self.wxstring(req, "mds") + " <sup>[5]</sup>",
321 get_yes_no(req, self.is_mds())
322 ),
323 ods=tr_qa(
324 self.wxstring(req, "ods") + " <sup>[6]</sup>",
325 get_yes_no(req, self.is_ods())
326 ),
327 q_a=q_a,
328 )
329 return h
331 def get_snomed_codes(self, req: CamcopsRequest) -> List[SnomedExpression]:
332 procedure = req.snomed(SnomedLookup.PHQ9_PROCEDURE_DEPRESSION_SCREENING) # noqa
333 codes = [SnomedExpression(procedure)]
334 if self.is_complete():
335 scale = req.snomed(SnomedLookup.PHQ9_SCALE)
336 score = req.snomed(SnomedLookup.PHQ9_SCORE)
337 screen_negative = req.snomed(SnomedLookup.PHQ9_FINDING_NEGATIVE_SCREENING_FOR_DEPRESSION) # noqa
338 screen_positive = req.snomed(SnomedLookup.PHQ9_FINDING_POSITIVE_SCREENING_FOR_DEPRESSION) # noqa
339 if self.is_mds() or self.is_ods():
340 # Threshold debatable, but if you have "other depressive
341 # syndrome", it seems wrong to say you've screened negative for
342 # depression.
343 procedure_result = screen_positive
344 else:
345 procedure_result = screen_negative
346 codes.append(SnomedExpression(scale, {score: self.total_score()}))
347 codes.append(SnomedExpression(procedure_result))
348 return codes
350 def get_fhir_questionnaire_items(
351 self,
352 req: "CamcopsRequest",
353 recipient: "ExportRecipient") -> List[QuestionnaireItem]:
354 items = []
356 for q_field in self.MAIN_QUESTIONS:
357 items.append(QuestionnaireItem(jsondict={
358 "linkId": q_field,
359 "text": self.wxstring(req, q_field),
360 "type": "choice",
361 }).as_json())
363 items.append(QuestionnaireItem(jsondict={
364 "linkId": "q10",
365 "text": "10. " + self.wxstring(req, "finalq"),
366 "type": "choice",
367 }).as_json())
369 return items
371 def get_fhir_questionnaire_response_items(
372 self,
373 req: "CamcopsRequest",
374 recipient: "ExportRecipient") -> List[QuestionnaireResponseItem]:
376 items = []
378 for q_field in self.MAIN_QUESTIONS:
379 answer = QuestionnaireResponseItemAnswer(jsondict={
380 "valueInteger": getattr(self, q_field)
381 })
383 items.append(QuestionnaireResponseItem(jsondict={
384 "linkId": q_field,
385 "text": self.wxstring(req, q_field),
386 "answer": [answer.as_json()],
387 }).as_json())
389 answer = QuestionnaireResponseItemAnswer(jsondict={
390 "valueInteger": self.q10
391 })
392 items.append(QuestionnaireResponseItem(jsondict={
393 "linkId": "q10",
394 "text": "10. " + self.wxstring(req, "finalq"),
395 "answer": [answer.as_json()],
396 }).as_json())
398 return items