Hide keyboard shortcuts

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 

2 

3""" 

4camcops_server/tasks/phq9.py 

5 

6=============================================================================== 

7 

8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com). 

9 

10 This file is part of CamCOPS. 

11 

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. 

16 

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. 

21 

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

24 

25=============================================================================== 

26 

27""" 

28 

29from typing import Any, Dict, List, Tuple, Type, TYPE_CHECKING 

30 

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 

39 

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) 

63 

64if TYPE_CHECKING: 

65 from camcops_server.cc_modules.cc_exportrecipient import ExportRecipient 

66 

67 

68# ============================================================================= 

69# PHQ-9 

70# ============================================================================= 

71 

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) 

95 

96 

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 

105 

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 ) 

112 

113 N_MAIN_QUESTIONS = 9 

114 MAX_SCORE_MAIN = 3 * N_MAIN_QUESTIONS 

115 MAIN_QUESTIONS = strseq("q", 1, N_MAIN_QUESTIONS) 

116 

117 @staticmethod 

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

119 _ = req.gettext 

120 return _("Patient Health Questionnaire-9") 

121 

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 

130 

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

161 

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

169 

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 ] 

201 

202 def total_score(self) -> int: 

203 return self.sum_fields(self.MAIN_QUESTIONS) 

204 

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 

208 

209 def n_core(self) -> int: 

210 return (self.one_if_q_ge(1, 2) + 

211 self.one_if_q_ge(2, 2)) 

212 

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 

222 

223 def n_total(self) -> int: 

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

225 

226 def is_mds(self) -> bool: 

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

228 

229 def is_ods(self) -> bool: 

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

231 

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) 

244 

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

267 

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, &lt;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 

330 

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 

349 

350 def get_fhir_questionnaire_items( 

351 self, 

352 req: "CamcopsRequest", 

353 recipient: "ExportRecipient") -> List[QuestionnaireItem]: 

354 items = [] 

355 

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

362 

363 items.append(QuestionnaireItem(jsondict={ 

364 "linkId": "q10", 

365 "text": "10. " + self.wxstring(req, "finalq"), 

366 "type": "choice", 

367 }).as_json()) 

368 

369 return items 

370 

371 def get_fhir_questionnaire_response_items( 

372 self, 

373 req: "CamcopsRequest", 

374 recipient: "ExportRecipient") -> List[QuestionnaireResponseItem]: 

375 

376 items = [] 

377 

378 for q_field in self.MAIN_QUESTIONS: 

379 answer = QuestionnaireResponseItemAnswer(jsondict={ 

380 "valueInteger": getattr(self, q_field) 

381 }) 

382 

383 items.append(QuestionnaireResponseItem(jsondict={ 

384 "linkId": q_field, 

385 "text": self.wxstring(req, q_field), 

386 "answer": [answer.as_json()], 

387 }).as_json()) 

388 

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

397 

398 return items