Coverage for tasks/phq8.py: 48%

87 statements  

« prev     ^ index     » next       coverage.py v7.9.2, created at 2025-07-15 14:23 +0100

1""" 

2camcops_server/tasks/phq8.py 

3 

4=============================================================================== 

5 

6 Copyright (C) 2012, University of Cambridge, Department of Psychiatry. 

7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk). 

8 

9 This file is part of CamCOPS. 

10 

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. 

15 

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. 

20 

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/>. 

23 

24=============================================================================== 

25 

26""" 

27 

28import logging 

29from typing import Any, cast, Dict, List, Type 

30 

31from cardinal_pythonlib.stringfunc import strseq 

32from sqlalchemy.sql.sqltypes import Boolean, Integer 

33 

34from camcops_server.cc_modules.cc_constants import CssClass 

35from camcops_server.cc_modules.cc_ctvinfo import CtvInfo, CTV_INCOMPLETE 

36from camcops_server.cc_modules.cc_db import add_multiple_columns 

37from camcops_server.cc_modules.cc_fhir import ( 

38 FHIRAnsweredQuestion, 

39 FHIRAnswerType, 

40 FHIRQuestionType, 

41) 

42from camcops_server.cc_modules.cc_html import answer, get_yes_no, tr, tr_qa 

43from camcops_server.cc_modules.cc_request import CamcopsRequest 

44from camcops_server.cc_modules.cc_sqla_coltypes import ( 

45 SummaryCategoryColType, 

46) 

47from camcops_server.cc_modules.cc_summaryelement import SummaryElement 

48from camcops_server.cc_modules.cc_task import ( 

49 get_from_dict, 

50 Task, 

51 TaskHasPatientMixin, 

52) 

53from camcops_server.cc_modules.cc_text import SS 

54from camcops_server.cc_modules.cc_trackerhelpers import ( 

55 TrackerAxisTick, 

56 TrackerInfo, 

57 TrackerLabel, 

58) 

59 

60log = logging.getLogger(__name__) 

61 

62 

63# ============================================================================= 

64# PHQ-8 

65# ============================================================================= 

66 

67 

68class Phq8( # type: ignore[misc] 

69 TaskHasPatientMixin, 

70 Task, 

71): 

72 """ 

73 Server implementation of the Phq8 task. 

74 """ 

75 

76 __tablename__ = "phq8" 

77 shortname = "PHQ-8" 

78 provides_trackers = True 

79 

80 N_QUESTIONS = 8 

81 MAX_SCORE = 3 * N_QUESTIONS 

82 

83 @classmethod 

84 def extend_columns(cls: Type["Phq8"], **kwargs: Any) -> None: 

85 add_multiple_columns( 

86 cls, 

87 "q", 

88 1, 

89 cls.N_QUESTIONS, 

90 minimum=0, 

91 maximum=3, 

92 comment_fmt="Q{n} ({s}) (0 not at all - 3 nearly every day)", 

93 comment_strings=[ 

94 "anhedonia", 

95 "mood", 

96 "sleep", 

97 "energy", 

98 "appetite", 

99 "self-esteem/guilt", 

100 "concentration", 

101 "psychomotor", 

102 ], 

103 ) 

104 

105 QUESTIONS = strseq("q", 1, N_QUESTIONS) 

106 

107 @staticmethod 

108 def longname(req: "CamcopsRequest") -> str: 

109 _ = req.gettext 

110 return _("Patient Health Questionnaire 8-item depression scale") 

111 

112 def is_complete(self) -> bool: 

113 if self.any_fields_none(self.QUESTIONS): 

114 return False 

115 if not self.field_contents_valid(): 

116 return False 

117 return True 

118 

119 def get_trackers(self, req: CamcopsRequest) -> List[TrackerInfo]: 

120 return [ 

121 TrackerInfo( 

122 value=self.total_score(), 

123 plot_label="PHQ-8 total score (rating depressive symptoms)", 

124 axis_label=f"Score (out of {self.MAX_SCORE})", 

125 axis_min=-0.5, 

126 axis_max=self.MAX_SCORE + 0.5, 

127 axis_ticks=[ 

128 TrackerAxisTick(24, "24"), # maximum 

129 TrackerAxisTick(20, "20"), 

130 TrackerAxisTick(15, "15"), 

131 TrackerAxisTick(10, "10"), 

132 TrackerAxisTick(5, "5"), 

133 TrackerAxisTick(0, "0"), 

134 ], 

135 horizontal_lines=[19.5, 14.5, 9.5, 4.5], 

136 horizontal_labels=[ 

137 TrackerLabel(23, req.sstring(SS.SEVERE)), 

138 TrackerLabel(17, req.sstring(SS.MODERATELY_SEVERE)), 

139 TrackerLabel(12, req.sstring(SS.MODERATE)), 

140 TrackerLabel(7, req.sstring(SS.MILD)), 

141 TrackerLabel(2.25, req.sstring(SS.NONE)), 

142 ], 

143 ) 

144 ] 

145 

146 def get_clinical_text(self, req: CamcopsRequest) -> List[CtvInfo]: 

147 if not self.is_complete(): 

148 return CTV_INCOMPLETE 

149 return [ 

150 CtvInfo( 

151 content=( 

152 f"PHQ-8 total score " 

153 f"{self.total_score()}/{self.MAX_SCORE} " 

154 f"({self.severity(req)})" 

155 ) 

156 ) 

157 ] 

158 

159 def get_summaries(self, req: CamcopsRequest) -> List[SummaryElement]: 

160 return self.standard_task_summary_fields() + [ 

161 SummaryElement( 

162 name="total", 

163 coltype=Integer(), 

164 value=self.total_score(), 

165 comment=f"Total score (/{self.MAX_SCORE})", 

166 ), 

167 SummaryElement( 

168 name="n_core", 

169 coltype=Integer(), 

170 value=self.n_core(), 

171 comment="Number of core symptoms", 

172 ), 

173 SummaryElement( 

174 name="n_other", 

175 coltype=Integer(), 

176 value=self.n_other(), 

177 comment="Number of other symptoms", 

178 ), 

179 SummaryElement( 

180 name="n_total", 

181 coltype=Integer(), 

182 value=self.n_total(), 

183 comment="Total number of symptoms", 

184 ), 

185 SummaryElement( 

186 name="is_mds", 

187 coltype=Boolean(), 

188 value=self.is_mds(), 

189 comment="PHQ8 major depressive syndrome?", 

190 ), 

191 SummaryElement( 

192 name="is_ods", 

193 coltype=Boolean(), 

194 value=self.is_ods(), 

195 comment="PHQ8 other depressive syndrome?", 

196 ), 

197 SummaryElement( 

198 name="severity", 

199 coltype=SummaryCategoryColType, 

200 value=self.severity(req), 

201 comment="PHQ8 depression severity", 

202 ), 

203 ] 

204 

205 def total_score(self) -> int: 

206 return cast(int, self.sum_fields(self.QUESTIONS)) 

207 

208 def reaches_threshold(self, qnum: int) -> int: 

209 # Checks if a symptom scores >=2, meaning "more than half the days". 

210 # Kroenke et al. (2010); see Phq8::detail() in phq8.cpp 

211 threshold = 2 

212 value = getattr(self, "q" + str(qnum)) 

213 return 1 if value is not None and value >= threshold else 0 

214 

215 def n_core(self) -> int: 

216 # Questions 1 and 2 

217 return sum(self.reaches_threshold(qnum) for qnum in range(1, 2 + 1)) 

218 

219 def n_other(self) -> int: 

220 # Questions 3-8 

221 return sum(self.reaches_threshold(qnum) for qnum in range(3, 8 + 1)) 

222 

223 def n_total(self) -> int: 

224 return self.n_core() + self.n_other() 

225 

226 def is_mds(self) -> bool: 

227 # Kroenke et al. (2010); see Phq8::detail() in phq8.cpp 

228 return self.n_core() >= 1 and self.n_total() >= 5 

229 

230 def is_ods(self) -> bool: 

231 # Kroenke et al. (2010); see Phq8::detail() in phq8.cpp 

232 return self.n_core() >= 1 and 2 <= self.n_total() <= 4 

233 

234 def severity(self, req: CamcopsRequest) -> str: 

235 # Kroenke et al. (2010); see Phq8::severity() in phq8.cpp 

236 total = self.total_score() 

237 if total >= 20: 

238 return req.sstring(SS.SEVERE) 

239 elif total >= 15: 

240 return req.sstring(SS.MODERATELY_SEVERE) 

241 elif total >= 10: 

242 return req.sstring(SS.MODERATE) 

243 elif total >= 5: 

244 return req.sstring(SS.MILD) 

245 else: 

246 return req.sstring(SS.NONE) 

247 

248 def get_task_html(self, req: CamcopsRequest) -> str: 

249 answer_dict = { 

250 None: None, 

251 0: "0 — " + self.wxstring(req, "a0"), 

252 1: "1 — " + self.wxstring(req, "a1"), 

253 2: "2 — " + self.wxstring(req, "a2"), 

254 3: "3 — " + self.wxstring(req, "a3"), 

255 } 

256 q_a = "" 

257 for i in range(1, self.N_QUESTIONS + 1): 

258 nstr = str(i) 

259 q_a += tr_qa( 

260 self.wxstring(req, "q" + nstr), 

261 get_from_dict(answer_dict, getattr(self, "q" + nstr)), 

262 ) 

263 

264 h = """ 

265 <div class="{CssClass.SUMMARY}"> 

266 <table class="{CssClass.SUMMARY}"> 

267 {tr_is_complete} 

268 {total_score} 

269 {depression_severity} 

270 {n_symptoms} 

271 {mds} 

272 {ods} 

273 </table> 

274 </div> 

275 <div class="{CssClass.EXPLANATION}"> 

276 Ratings are over the last 2 weeks. 

277 </div> 

278 <table class="{CssClass.TASKDETAIL}"> 

279 <tr> 

280 <th width="60%">Question</th> 

281 <th width="40%">Answer</th> 

282 </tr> 

283 {q_a} 

284 </table> 

285 <div class="{CssClass.FOOTNOTES}"> 

286 [1] Sum for questions 1–8. 

287 [2] Total score ≥20 severe, ≥15 moderately severe, 

288 ≥10 moderate, ≥5 mild, &lt;5 none.<sup>[7]</sup> 

289 [3] Number of questions 1–2 rated ≥2.<sup>[7]</sup> 

290 [4] Number of questions 3–8 rated ≥2.<sup>[7]</sup> 

291 [5] ≥1 core symptom and ≥5 total symptoms.<sup>[7]</sup> 

292 [6] ≥1 core symptom and 2–4 total symptoms.<sup>[7]</sup> 

293 [7] Kroenke et al. (2010), PMID 18752852. 

294 </div> 

295 """.format( 

296 CssClass=CssClass, 

297 tr_is_complete=self.get_is_complete_tr(req), 

298 total_score=tr( 

299 req.sstring(SS.TOTAL_SCORE) + " <sup>[1]</sup>", 

300 answer(self.total_score()) + f" / {self.MAX_SCORE}", 

301 ), 

302 depression_severity=tr_qa( 

303 self.wxstring(req, "depression_severity") + " <sup>[2]</sup>", 

304 self.severity(req), 

305 ), 

306 n_symptoms=tr( 

307 "Number of symptoms: core <sup>[3]</sup>, other " 

308 "<sup>[4]</sup>, total", 

309 answer(self.n_core()) 

310 + "/2, " 

311 + answer(self.n_other()) 

312 + "/6, " 

313 + answer(self.n_total()) 

314 + "/8", 

315 ), 

316 mds=tr_qa( 

317 self.wxstring(req, "mds") + " <sup>[5]</sup>", 

318 get_yes_no(req, self.is_mds()), 

319 ), 

320 ods=tr_qa( 

321 self.wxstring(req, "ods") + " <sup>[6]</sup>", 

322 get_yes_no(req, self.is_ods()), 

323 ), 

324 q_a=q_a, 

325 ) 

326 return h 

327 

328 # No SNOMED CT codes for the PHQ-8 (just the PHQ-9), 2022-11-30. 

329 

330 def get_fhir_questionnaire( 

331 self, req: "CamcopsRequest" 

332 ) -> List[FHIRAnsweredQuestion]: 

333 items = [] # type: List[FHIRAnsweredQuestion] 

334 

335 answer_options = {} # type: Dict[int, str] 

336 for index in range(4): 

337 answer_options[index] = self.wxstring(req, f"a{index}") 

338 for q_field in self.QUESTIONS: 

339 items.append( 

340 FHIRAnsweredQuestion( 

341 qname=q_field, 

342 qtext=self.xstring(req, q_field), 

343 qtype=FHIRQuestionType.CHOICE, 

344 answer_type=FHIRAnswerType.INTEGER, 

345 answer=getattr(self, q_field), 

346 answer_options=answer_options, 

347 ) 

348 ) 

349 

350 return items