Coverage for tasks/phq9.py: 41%
111 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 14:23 +0100
1"""
2camcops_server/tasks/phq9.py
4===============================================================================
6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CamCOPS.
11 CamCOPS is free software: you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation, either version 3 of the License, or
14 (at your option) any later version.
16 CamCOPS is distributed in the hope that it will be useful,
17 but WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 GNU General Public License for more details.
21 You should have received a copy of the GNU General Public License
22 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26"""
28import logging
29from typing import Any, cast, Dict, List, Optional, Type
31from cardinal_pythonlib.stringfunc import strseq
32from sqlalchemy.orm import Mapped
33from sqlalchemy.sql.sqltypes import Boolean, Integer
35from camcops_server.cc_modules.cc_constants import CssClass
36from camcops_server.cc_modules.cc_ctvinfo import CtvInfo, CTV_INCOMPLETE
37from camcops_server.cc_modules.cc_db import add_multiple_columns
38from camcops_server.cc_modules.cc_fhir import (
39 FHIRAnsweredQuestion,
40 FHIRAnswerType,
41 FHIRQuestionType,
42)
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 mapped_camcops_column,
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)
64log = logging.getLogger(__name__)
67# =============================================================================
68# PHQ-9
69# =============================================================================
72class Phq9( # type: ignore[misc]
73 TaskHasPatientMixin,
74 Task,
75):
76 """
77 Server implementation of the PHQ9 task.
78 """
80 __tablename__ = "phq9"
81 shortname = "PHQ-9"
82 provides_trackers = True
84 @classmethod
85 def extend_columns(cls: Type["Phq9"], **kwargs: Any) -> None:
86 add_multiple_columns(
87 cls,
88 "q",
89 1,
90 cls.N_MAIN_QUESTIONS,
91 minimum=0,
92 maximum=3,
93 comment_fmt="Q{n} ({s}) (0 not at all - 3 nearly every day)",
94 comment_strings=[
95 "anhedonia",
96 "mood",
97 "sleep",
98 "energy",
99 "appetite",
100 "self-esteem/guilt",
101 "concentration",
102 "psychomotor",
103 "death/self-harm",
104 ],
105 )
107 q10: Mapped[Optional[int]] = mapped_camcops_column(
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 [
133 TrackerInfo(
134 value=self.total_score(),
135 plot_label="PHQ-9 total score (rating depressive symptoms)",
136 axis_label=f"Score for Q1-9 (out of {self.MAX_SCORE_MAIN})",
137 axis_min=-0.5,
138 axis_max=self.MAX_SCORE_MAIN + 0.5,
139 axis_ticks=[
140 TrackerAxisTick(27, "27"),
141 TrackerAxisTick(25, "25"),
142 TrackerAxisTick(20, "20"),
143 TrackerAxisTick(15, "15"),
144 TrackerAxisTick(10, "10"),
145 TrackerAxisTick(5, "5"),
146 TrackerAxisTick(0, "0"),
147 ],
148 horizontal_lines=[19.5, 14.5, 9.5, 4.5],
149 horizontal_labels=[
150 TrackerLabel(23, req.sstring(SS.SEVERE)),
151 TrackerLabel(17, req.sstring(SS.MODERATELY_SEVERE)),
152 TrackerLabel(12, req.sstring(SS.MODERATE)),
153 TrackerLabel(7, req.sstring(SS.MILD)),
154 TrackerLabel(2.25, req.sstring(SS.NONE)),
155 ],
156 )
157 ]
159 def get_clinical_text(self, req: CamcopsRequest) -> List[CtvInfo]:
160 if not self.is_complete():
161 return CTV_INCOMPLETE
162 return [
163 CtvInfo(
164 content=(
165 f"PHQ-9 total score "
166 f"{self.total_score()}/{self.MAX_SCORE_MAIN} "
167 f"({self.severity(req)})"
168 )
169 )
170 ]
172 def get_summaries(self, req: CamcopsRequest) -> List[SummaryElement]:
173 return self.standard_task_summary_fields() + [
174 SummaryElement(
175 name="total",
176 coltype=Integer(),
177 value=self.total_score(),
178 comment=f"Total score (/{self.MAX_SCORE_MAIN})",
179 ),
180 SummaryElement(
181 name="n_core",
182 coltype=Integer(),
183 value=self.n_core(),
184 comment="Number of core symptoms",
185 ),
186 SummaryElement(
187 name="n_other",
188 coltype=Integer(),
189 value=self.n_other(),
190 comment="Number of other symptoms",
191 ),
192 SummaryElement(
193 name="n_total",
194 coltype=Integer(),
195 value=self.n_total(),
196 comment="Total number of symptoms",
197 ),
198 SummaryElement(
199 name="is_mds",
200 coltype=Boolean(),
201 value=self.is_mds(),
202 comment="PHQ9 major depressive syndrome?",
203 ),
204 SummaryElement(
205 name="is_ods",
206 coltype=Boolean(),
207 value=self.is_ods(),
208 comment="PHQ9 other depressive syndrome?",
209 ),
210 SummaryElement(
211 name="severity",
212 coltype=SummaryCategoryColType,
213 value=self.severity(req),
214 comment="PHQ9 depression severity",
215 ),
216 ]
218 def total_score(self) -> int:
219 return cast(int, self.sum_fields(self.MAIN_QUESTIONS))
221 def one_if_q_ge(self, qnum: int, threshold: int) -> int:
222 value = getattr(self, "q" + str(qnum))
223 return 1 if value is not None and value >= threshold else 0
225 def n_core(self) -> int:
226 return self.one_if_q_ge(1, 2) + self.one_if_q_ge(2, 2)
228 def n_other(self) -> int:
229 return (
230 self.one_if_q_ge(3, 2)
231 + self.one_if_q_ge(4, 2)
232 + self.one_if_q_ge(5, 2)
233 + self.one_if_q_ge(6, 2)
234 + self.one_if_q_ge(7, 2)
235 + self.one_if_q_ge(8, 2)
236 + self.one_if_q_ge(9, 1)
237 ) # suicidality
238 # suicidality counted whenever present
240 def n_total(self) -> int:
241 return self.n_core() + self.n_other()
243 def is_mds(self) -> bool:
244 return self.n_core() >= 1 and self.n_total() >= 5
246 def is_ods(self) -> bool:
247 return self.n_core() >= 1 and 2 <= self.n_total() <= 4
249 def severity(self, req: CamcopsRequest) -> str:
250 total = self.total_score()
251 if total >= 20:
252 return req.sstring(SS.SEVERE)
253 elif total >= 15:
254 return req.sstring(SS.MODERATELY_SEVERE)
255 elif total >= 10:
256 return req.sstring(SS.MODERATE)
257 elif total >= 5:
258 return req.sstring(SS.MILD)
259 else:
260 return req.sstring(SS.NONE)
262 def get_task_html(self, req: CamcopsRequest) -> str:
263 main_dict = {
264 None: None,
265 0: "0 — " + self.wxstring(req, "a0"),
266 1: "1 — " + self.wxstring(req, "a1"),
267 2: "2 — " + self.wxstring(req, "a2"),
268 3: "3 — " + self.wxstring(req, "a3"),
269 }
270 q10_dict = {
271 None: None,
272 0: "0 — " + self.wxstring(req, "fa0"),
273 1: "1 — " + self.wxstring(req, "fa1"),
274 2: "2 — " + self.wxstring(req, "fa2"),
275 3: "3 — " + self.wxstring(req, "fa3"),
276 }
277 q_a = ""
278 for i in range(1, self.N_MAIN_QUESTIONS + 1):
279 nstr = str(i)
280 q_a += tr_qa(
281 self.wxstring(req, "q" + nstr),
282 get_from_dict(main_dict, getattr(self, "q" + nstr)),
283 )
284 q_a += tr_qa(
285 "10. " + self.wxstring(req, "finalq"),
286 get_from_dict(q10_dict, self.q10),
287 )
289 h = """
290 <div class="{CssClass.SUMMARY}">
291 <table class="{CssClass.SUMMARY}">
292 {tr_is_complete}
293 {total_score}
294 {depression_severity}
295 {n_symptoms}
296 {mds}
297 {ods}
298 </table>
299 </div>
300 <div class="{CssClass.EXPLANATION}">
301 Ratings are over the last 2 weeks.
302 </div>
303 <table class="{CssClass.TASKDETAIL}">
304 <tr>
305 <th width="60%">Question</th>
306 <th width="40%">Answer</th>
307 </tr>
308 {q_a}
309 </table>
310 <div class="{CssClass.FOOTNOTES}">
311 [1] Sum for questions 1–9.
312 [2] Total score ≥20 severe, ≥15 moderately severe,
313 ≥10 moderate, ≥5 mild, <5 none.
314 [3] Number of questions 1–2 rated ≥2.
315 [4] Number of questions 3–8 rated ≥2, or question 9
316 rated ≥1.
317 [5] ≥1 core symptom and ≥5 total symptoms (as per
318 DSM-IV-TR page 356).
319 [6] ≥1 core symptom and 2–4 total symptoms (as per
320 DSM-IV-TR page 775).
321 </div>
322 """.format(
323 CssClass=CssClass,
324 tr_is_complete=self.get_is_complete_tr(req),
325 total_score=tr(
326 req.sstring(SS.TOTAL_SCORE) + " <sup>[1]</sup>",
327 answer(self.total_score()) + f" / {self.MAX_SCORE_MAIN}",
328 ),
329 depression_severity=tr_qa(
330 self.wxstring(req, "depression_severity") + " <sup>[2]</sup>",
331 self.severity(req),
332 ),
333 n_symptoms=tr(
334 "Number of symptoms: core <sup>[3]</sup>, other "
335 "<sup>[4]</sup>, total",
336 answer(self.n_core())
337 + "/2, "
338 + answer(self.n_other())
339 + "/7, "
340 + answer(self.n_total())
341 + "/9",
342 ),
343 mds=tr_qa(
344 self.wxstring(req, "mds") + " <sup>[5]</sup>",
345 get_yes_no(req, self.is_mds()),
346 ),
347 ods=tr_qa(
348 self.wxstring(req, "ods") + " <sup>[6]</sup>",
349 get_yes_no(req, self.is_ods()),
350 ),
351 q_a=q_a,
352 )
353 return h
355 def get_snomed_codes(self, req: CamcopsRequest) -> List[SnomedExpression]:
356 procedure = req.snomed(
357 SnomedLookup.PHQ9_PROCEDURE_DEPRESSION_SCREENING
358 )
359 codes = [SnomedExpression(procedure)]
360 if self.is_complete():
361 scale = req.snomed(SnomedLookup.PHQ9_SCALE)
362 score = req.snomed(SnomedLookup.PHQ9_SCORE)
363 screen_negative = req.snomed(
364 SnomedLookup.PHQ9_FINDING_NEGATIVE_SCREENING_FOR_DEPRESSION
365 )
366 screen_positive = req.snomed(
367 SnomedLookup.PHQ9_FINDING_POSITIVE_SCREENING_FOR_DEPRESSION
368 )
369 if self.is_mds() or self.is_ods():
370 # Threshold debatable, but if you have "other depressive
371 # syndrome", it seems wrong to say you've screened negative for
372 # depression.
373 procedure_result = screen_positive
374 else:
375 procedure_result = screen_negative
376 codes.append(SnomedExpression(scale, {score: self.total_score()}))
377 codes.append(SnomedExpression(procedure_result))
378 return codes
380 def get_fhir_questionnaire(
381 self, req: "CamcopsRequest"
382 ) -> List[FHIRAnsweredQuestion]:
383 items = [] # type: List[FHIRAnsweredQuestion]
385 main_options = {} # type: Dict[int, str]
386 for index in range(4):
387 main_options[index] = self.wxstring(req, f"a{index}")
388 for q_field in self.MAIN_QUESTIONS:
389 items.append(
390 FHIRAnsweredQuestion(
391 qname=q_field,
392 qtext=self.xstring(req, q_field),
393 qtype=FHIRQuestionType.CHOICE,
394 answer_type=FHIRAnswerType.INTEGER,
395 answer=getattr(self, q_field),
396 answer_options=main_options,
397 )
398 )
400 q10_options = {}
401 for index in range(4):
402 q10_options[index] = self.wxstring(req, f"fa{index}")
403 items.append(
404 FHIRAnsweredQuestion(
405 qname="q10",
406 qtext="10. " + self.xstring(req, "finalq"),
407 qtype=FHIRQuestionType.CHOICE,
408 answer_type=FHIRAnswerType.INTEGER,
409 answer=self.q10,
410 answer_options=q10_options,
411 )
412 )
414 return items