Coverage for tasks/cardinal_expdetthreshold.py: 33%
234 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-15 14:23 +0100
1"""
2camcops_server/tasks/cardinal_expdetthreshold.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 math
29import logging
30from typing import List, Optional, Tuple, Type
32from cardinal_pythonlib.maths_numpy import inv_logistic, logistic
33import cardinal_pythonlib.rnc_web as ws
34from matplotlib.figure import Figure
35import numpy as np
36from pendulum import DateTime as Pendulum
37from sqlalchemy.orm import Mapped, mapped_column
38from sqlalchemy.sql.sqltypes import Text, UnicodeText
40from camcops_server.cc_modules.cc_constants import (
41 CssClass,
42 MatplotlibConstants,
43 PlotDefaults,
44)
45from camcops_server.cc_modules.cc_db import (
46 ancillary_relationship,
47 GenericTabletRecordMixin,
48 TaskDescendant,
49)
50from camcops_server.cc_modules.cc_html import get_yes_no_none, tr_qa
51from camcops_server.cc_modules.cc_request import CamcopsRequest
52from camcops_server.cc_modules.cc_sqla_coltypes import (
53 mapped_camcops_column,
54 PendulumDateTimeAsIsoTextColType,
55)
56from camcops_server.cc_modules.cc_sqlalchemy import Base
57from camcops_server.cc_modules.cc_task import Task, TaskHasPatientMixin
58from camcops_server.cc_modules.cc_text import SS
60log = logging.getLogger(__name__)
63LOWER_MARKER = 0.25
64UPPER_MARKER = 0.75
65EQUATION_COMMENT = (
66 "logits: L(X) = intercept + slope * X; "
67 "probability: P = 1 / (1 + exp(-intercept - slope * X))"
68)
69MODALITY_AUDITORY = 0
70MODALITY_VISUAL = 1
71DP = 3
74# =============================================================================
75# CardinalExpDetThreshold
76# =============================================================================
79class CardinalExpDetThresholdTrial(
80 GenericTabletRecordMixin, TaskDescendant, Base
81):
82 __tablename__ = "cardinal_expdetthreshold_trials"
84 cardinal_expdetthreshold_id: Mapped[int] = mapped_column(
85 comment="FK to CardinalExpDetThreshold",
86 )
87 trial: Mapped[int] = mapped_column(comment="Trial number (0-based)")
89 # Results
90 trial_ignoring_catch_trials: Mapped[Optional[int]] = mapped_column(
91 comment="Trial number, ignoring catch trials (0-based)",
92 )
93 target_presented: Mapped[Optional[int]] = mapped_column(
94 comment="Target presented? (0 no, 1 yes)"
95 )
96 target_time: Mapped[Optional[Pendulum]] = mapped_column(
97 PendulumDateTimeAsIsoTextColType,
98 comment="Target presentation time (ISO-8601)",
99 )
100 intensity: Mapped[Optional[float]] = mapped_column(
101 comment="Target intensity (0.0-1.0)"
102 )
103 choice_time: Mapped[Optional[Pendulum]] = mapped_column(
104 PendulumDateTimeAsIsoTextColType,
105 comment="Time choice offered (ISO-8601)",
106 )
107 responded: Mapped[Optional[int]] = mapped_column(
108 comment="Responded? (0 no, 1 yes)"
109 )
110 response_time: Mapped[Optional[Pendulum]] = mapped_column(
111 PendulumDateTimeAsIsoTextColType,
112 comment="Time of response (ISO-8601)",
113 )
114 response_latency_ms: Mapped[Optional[int]] = mapped_column(
115 comment="Response latency (ms)"
116 )
117 yes: Mapped[Optional[int]] = mapped_column(
118 comment="Subject chose YES? (0 didn't, 1 did)"
119 )
120 no: Mapped[Optional[int]] = mapped_column(
121 comment="Subject chose NO? (0 didn't, 1 did)"
122 )
123 caught_out_reset: Mapped[Optional[int]] = mapped_column(
124 comment="Caught out on catch trial, thus reset? (0 no, 1 yes)",
125 )
126 trial_num_in_calculation_sequence: Mapped[Optional[int]] = mapped_column(
127 comment="Trial number as used for threshold calculation",
128 )
130 @classmethod
131 def get_html_table_header(cls) -> str:
132 return f"""
133 <table class="{CssClass.EXTRADETAIL}">
134 <tr>
135 <th>Trial# (0-based)</th>
136 <th>Trial# (ignoring catch trials) (0-based)</th>
137 <th>Target presented?</th>
138 <th>Target time</th>
139 <th>Intensity</th>
140 <th>Choice time</th>
141 <th>Responded?</th>
142 <th>Response time</th>
143 <th>Response latency (ms)</th>
144 <th>Yes?</th>
145 <th>No?</th>
146 <th>Caught out (and reset)?</th>
147 <th>Trial# in calculation sequence</th>
148 </tr>
149 """
151 def get_html_table_row(self) -> str:
152 return ("<tr>" + "<td>{}</td>" * 13 + "</th>").format(
153 self.trial,
154 self.trial_ignoring_catch_trials,
155 self.target_presented,
156 self.target_time,
157 ws.number_to_dp(self.intensity, DP),
158 self.choice_time,
159 self.responded,
160 self.response_time,
161 self.response_latency_ms,
162 self.yes,
163 self.no,
164 ws.webify(self.caught_out_reset),
165 ws.webify(self.trial_num_in_calculation_sequence),
166 )
168 # -------------------------------------------------------------------------
169 # TaskDescendant overrides
170 # -------------------------------------------------------------------------
172 @classmethod
173 def task_ancestor_class(cls) -> Optional[Type["Task"]]:
174 return CardinalExpDetThreshold
176 def task_ancestor(self) -> Optional["CardinalExpDetThreshold"]:
177 return CardinalExpDetThreshold.get_linked( # type: ignore[return-value] # noqa: E501
178 self.cardinal_expdetthreshold_id, self
179 )
182class CardinalExpDetThreshold(TaskHasPatientMixin, Task): # type: ignore[misc]
183 """
184 Server implementation of the Cardinal_ExpDetThreshold task.
185 """
187 __tablename__ = "cardinal_expdetthreshold"
188 shortname = "Cardinal_ExpDetThreshold"
189 use_landscape_for_pdf = True
191 # Config
192 modality: Mapped[Optional[int]] = mapped_column(
193 comment="Modality (0 auditory, 1 visual)"
194 )
195 target_number: Mapped[Optional[int]] = mapped_column(
196 comment="Target number (within available targets of that modality)",
197 )
198 background_filename: Mapped[Optional[str]] = mapped_camcops_column(
199 Text,
200 exempt_from_anonymisation=True,
201 comment="Filename of media used for background",
202 )
203 target_filename: Mapped[Optional[str]] = mapped_camcops_column(
204 "target_filename",
205 Text,
206 exempt_from_anonymisation=True,
207 comment="Filename of media used for target",
208 )
209 visual_target_duration_s: Mapped[Optional[float]] = mapped_column(
210 comment="Visual target duration (s)"
211 )
212 background_intensity: Mapped[Optional[float]] = mapped_column(
213 comment="Intensity of background (0.0-1.0)",
214 )
215 start_intensity_min: Mapped[Optional[float]] = mapped_column(
216 comment="Minimum starting intensity (0.0-1.0)",
217 )
218 start_intensity_max: Mapped[Optional[float]] = mapped_column(
219 comment="Maximum starting intensity (0.0-1.0)",
220 )
221 initial_large_intensity_step: Mapped[Optional[float]] = mapped_column(
222 comment="Initial, large, intensity step (0.0-1.0)",
223 )
224 main_small_intensity_step: Mapped[Optional[float]] = mapped_column(
225 comment="Main, small, intensity step (0.0-1.0)",
226 )
227 num_trials_in_main_sequence: Mapped[Optional[int]] = mapped_column(
228 comment="Number of trials required in main sequence",
229 )
230 p_catch_trial: Mapped[Optional[float]] = mapped_column(
231 comment="Probability of catch trial"
232 )
233 prompt: Mapped[Optional[str]] = mapped_camcops_column(
234 UnicodeText,
235 exempt_from_anonymisation=True,
236 comment="Prompt given to subject",
237 )
238 iti_s: Mapped[Optional[float]] = mapped_column(
239 comment="Intertrial interval (s)"
240 )
242 # Results
243 finished: Mapped[Optional[int]] = mapped_column(
244 comment="Subject finished successfully (0 no, 1 yes)",
245 )
246 intercept: Mapped[Optional[float]] = mapped_column(
247 comment=EQUATION_COMMENT
248 )
249 slope: Mapped[Optional[float]] = mapped_column(comment=EQUATION_COMMENT)
250 k: Mapped[Optional[float]] = mapped_column(
251 comment=EQUATION_COMMENT + "; k = slope"
252 )
253 theta: Mapped[Optional[float]] = mapped_column(
254 comment=EQUATION_COMMENT + "; theta = -intercept/k = -intercept/slope",
255 )
257 # Relationships
258 trials = ancillary_relationship( # type: ignore[assignment]
259 parent_class_name="CardinalExpDetThreshold",
260 ancillary_class_name="CardinalExpDetThresholdTrial",
261 ancillary_fk_to_parent_attr_name="cardinal_expdetthreshold_id",
262 ancillary_order_by_attr_name="trial",
263 ) # type: List[CardinalExpDetThresholdTrial]
265 @staticmethod
266 def longname(req: "CamcopsRequest") -> str:
267 _ = req.gettext
268 return _(
269 "Cardinal RN – Threshold determination for "
270 "Expectation–Detection task"
271 )
273 def is_complete(self) -> bool:
274 return bool(self.finished)
276 def _get_figures(
277 self, req: CamcopsRequest
278 ) -> Tuple[Figure, Optional[Figure]]:
279 """
280 Create and return figures. Returns ``trialfig, fitfig``.
281 """
282 trialarray = self.trials
284 # Constants
285 jitter_step = 0.02
286 dp_to_consider_same_for_jitter = 3
287 y_extra_space = 0.1
288 x_extra_space = 0.02
289 figsize = (
290 PlotDefaults.FULLWIDTH_PLOT_WIDTH / 2,
291 PlotDefaults.FULLWIDTH_PLOT_WIDTH / 2,
292 )
294 # Figure and axes
295 trialfig = req.create_figure(figsize=figsize)
296 trialax = trialfig.add_subplot(MatplotlibConstants.WHOLE_PANEL)
297 fitfig = None # type: Optional[Figure]
299 # Anything to do?
300 if not trialarray:
301 return trialfig, fitfig
303 # Data
304 notcalc_detected_x = []
305 notcalc_detected_y = []
306 notcalc_missed_x = []
307 notcalc_missed_y = []
308 calc_detected_x = []
309 calc_detected_y = []
310 calc_missed_x = []
311 calc_missed_y = []
312 catch_detected_x = []
313 catch_detected_y = []
314 catch_missed_x = []
315 catch_missed_y = []
316 all_x = []
317 all_y = []
318 for t in trialarray:
319 x = t.trial
320 y = t.intensity
321 all_x.append(x)
322 all_y.append(y)
323 if t.trial_num_in_calculation_sequence is not None:
324 if t.yes:
325 calc_detected_x.append(x)
326 calc_detected_y.append(y)
327 else:
328 calc_missed_x.append(x)
329 calc_missed_y.append(y)
330 elif t.target_presented:
331 if t.yes:
332 notcalc_detected_x.append(x)
333 notcalc_detected_y.append(y)
334 else:
335 notcalc_missed_x.append(x)
336 notcalc_missed_y.append(y)
337 else: # catch trial
338 if t.yes:
339 catch_detected_x.append(x)
340 catch_detected_y.append(y)
341 else:
342 catch_missed_x.append(x)
343 catch_missed_y.append(y)
345 # Create trialfig plots
346 trialax.plot(
347 all_x,
348 all_y,
349 marker=MatplotlibConstants.MARKER_NONE,
350 color=MatplotlibConstants.COLOUR_GREY_50,
351 linestyle=MatplotlibConstants.LINESTYLE_SOLID,
352 label=None,
353 )
354 trialax.plot(
355 notcalc_missed_x,
356 notcalc_missed_y,
357 marker=MatplotlibConstants.MARKER_CIRCLE,
358 color=MatplotlibConstants.COLOUR_BLACK,
359 linestyle=MatplotlibConstants.LINESTYLE_NONE,
360 label="miss",
361 )
362 trialax.plot(
363 notcalc_detected_x,
364 notcalc_detected_y,
365 marker=MatplotlibConstants.MARKER_PLUS,
366 color=MatplotlibConstants.COLOUR_BLACK,
367 linestyle=MatplotlibConstants.LINESTYLE_NONE,
368 label="hit",
369 )
370 trialax.plot(
371 calc_missed_x,
372 calc_missed_y,
373 marker=MatplotlibConstants.MARKER_CIRCLE,
374 color=MatplotlibConstants.COLOUR_RED,
375 linestyle=MatplotlibConstants.LINESTYLE_NONE,
376 label="miss, scored",
377 )
378 trialax.plot(
379 calc_detected_x,
380 calc_detected_y,
381 marker=MatplotlibConstants.MARKER_PLUS,
382 color=MatplotlibConstants.COLOUR_BLUE,
383 linestyle=MatplotlibConstants.LINESTYLE_NONE,
384 label="hit, scored",
385 )
386 trialax.plot(
387 catch_missed_x,
388 catch_missed_y,
389 marker=MatplotlibConstants.MARKER_CIRCLE,
390 color=MatplotlibConstants.COLOUR_GREEN,
391 linestyle=MatplotlibConstants.LINESTYLE_NONE,
392 label="CR",
393 )
394 trialax.plot(
395 catch_detected_x,
396 catch_detected_y,
397 marker=MatplotlibConstants.MARKER_STAR,
398 color=MatplotlibConstants.COLOUR_GREEN,
399 linestyle=MatplotlibConstants.LINESTYLE_NONE,
400 label="FA",
401 )
402 leg = trialax.legend(
403 numpoints=1,
404 fancybox=True, # for set_alpha (below)
405 loc="best", # bbox_to_anchor=(0.75, 1.05)
406 labelspacing=0,
407 handletextpad=0,
408 prop=req.fontprops,
409 )
410 leg.get_frame().set_alpha(0.5)
411 trialax.set_xlabel("Trial number (0-based)", fontdict=req.fontdict)
412 trialax.set_ylabel("Intensity", fontdict=req.fontdict)
413 trialax.set_ylim(0 - y_extra_space, 1 + y_extra_space)
414 trialax.set_xlim(-0.5, len(trialarray) - 0.5)
415 req.set_figure_font_sizes(trialax)
417 # Anything to do for fitfig?
418 if self.k is None or self.theta is None:
419 return trialfig, fitfig
421 # Create fitfig
422 fitfig = req.create_figure(figsize=figsize)
423 fitax = fitfig.add_subplot(MatplotlibConstants.WHOLE_PANEL)
424 detected_x = []
425 detected_x_approx = [] # type: ignore[var-annotated]
426 detected_y = []
427 missed_x = []
428 missed_x_approx = [] # type: ignore[var-annotated]
429 missed_y = []
430 all_x = []
431 for t in trialarray:
432 if t.trial_num_in_calculation_sequence is not None:
433 all_x.append(t.intensity)
434 approx_x = f"{t.intensity:.{dp_to_consider_same_for_jitter}f}"
435 if t.yes:
436 detected_y.append(
437 1 - detected_x_approx.count(approx_x) * jitter_step
438 )
439 detected_x.append(t.intensity)
440 detected_x_approx.append(approx_x)
441 else:
442 missed_y.append(
443 0 + missed_x_approx.count(approx_x) * jitter_step
444 )
445 missed_x.append(t.intensity)
446 missed_x_approx.append(approx_x)
448 # Again, anything to do for fitfig?
449 if not all_x:
450 return trialfig, fitfig
452 fit_x = np.arange(0.0 - x_extra_space, 1.0 + x_extra_space, 0.001)
453 fit_y = logistic(fit_x, self.k, self.theta)
454 fitax.plot(
455 fit_x,
456 fit_y,
457 color=MatplotlibConstants.COLOUR_GREEN,
458 linestyle=MatplotlibConstants.LINESTYLE_SOLID,
459 )
460 fitax.plot(
461 missed_x,
462 missed_y,
463 marker=MatplotlibConstants.MARKER_CIRCLE,
464 color=MatplotlibConstants.COLOUR_RED,
465 linestyle=MatplotlibConstants.LINESTYLE_NONE,
466 )
467 fitax.plot(
468 detected_x,
469 detected_y,
470 marker=MatplotlibConstants.MARKER_PLUS,
471 color=MatplotlibConstants.COLOUR_BLUE,
472 linestyle=MatplotlibConstants.LINESTYLE_NONE,
473 )
474 fitax.set_ylim(0 - y_extra_space, 1 + y_extra_space)
475 fitax.set_xlim(
476 np.amin(all_x) - x_extra_space, np.amax(all_x) + x_extra_space
477 )
478 marker_points = []
479 for y in (LOWER_MARKER, 0.5, UPPER_MARKER):
480 x = inv_logistic(y, self.k, self.theta) # type: ignore[assignment]
481 marker_points.append((x, y))
482 for p in marker_points:
483 fitax.plot(
484 [p[0], p[0]], # x
485 [-1, p[1]], # y
486 color=MatplotlibConstants.COLOUR_GREY_50,
487 linestyle=MatplotlibConstants.LINESTYLE_DOTTED,
488 )
489 fitax.plot(
490 [-1, p[0]], # x
491 [p[1], p[1]], # y
492 color=MatplotlibConstants.COLOUR_GREY_50,
493 linestyle=MatplotlibConstants.LINESTYLE_DOTTED,
494 )
495 fitax.set_xlabel("Intensity", fontdict=req.fontdict)
496 fitax.set_ylabel(
497 "Detected? (0=no, 1=yes; jittered)", fontdict=req.fontdict
498 )
499 req.set_figure_font_sizes(fitax)
501 # Done
502 return trialfig, fitfig
504 def get_trial_html(self, req: CamcopsRequest) -> str:
505 """
506 Note re plotting markers without lines:
508 .. code-block:: python
510 import matplotlib.pyplot as plt
512 fig, ax = plt.subplots()
513 ax.plot([1, 2], [1, 2], marker="+", color="r", linestyle="-")
514 ax.plot([1, 2], [2, 1], marker="o", color="b", linestyle="None")
515 fig.savefig("test.png")
516 # ... the "absent" line does NOT "cut" the red one.
518 Args:
519 req:
521 Returns:
523 """
524 trialarray = self.trials
525 html = CardinalExpDetThresholdTrial.get_html_table_header()
526 for t in trialarray:
527 html += t.get_html_table_row()
528 html += """</table>"""
530 # Don't add figures if we're incomplete
531 if not self.is_complete():
532 return html
534 # Add figures
535 trialfig, fitfig = self._get_figures(req)
537 html += f"""
538 <table class="{CssClass.NOBORDER}">
539 <tr>
540 <td class="{CssClass.NOBORDERPHOTO}">
541 {req.get_html_from_pyplot_figure(trialfig)}
542 </td>
543 <td class="{CssClass.NOBORDERPHOTO}">
544 {req.get_html_from_pyplot_figure(fitfig)}
545 </td>
546 </tr>
547 </table>
548 """
550 return html
552 def logistic_x_from_p(self, p: Optional[float]) -> Optional[float]:
553 try:
554 return (math.log(p / (1 - p)) - self.intercept) / self.slope
555 except (TypeError, ValueError):
556 return None
558 def get_task_html(self, req: CamcopsRequest) -> str:
559 if self.modality == MODALITY_AUDITORY:
560 modality = req.sstring(SS.AUDITORY)
561 elif self.modality == MODALITY_VISUAL:
562 modality = req.sstring(SS.VISUAL)
563 else:
564 modality = None
565 h = f"""
566 <div class="{CssClass.SUMMARY}">
567 <table class="{CssClass.SUMMARY}">
568 {self.get_is_complete_tr(req)}
569 </table>
570 </div>
571 <div class="{CssClass.EXPLANATION}">
572 The ExpDet-Threshold task measures visual and auditory
573 thresholds for stimuli on a noisy background, using a
574 single-interval up/down method. It is intended as a prequel to
575 the Expectation–Detection task.
576 </div>
577 <table class="{CssClass.TASKCONFIG}">
578 <tr>
579 <th width="50%">Configuration variable</th>
580 <th width="50%">Value</th>
581 </tr>
582 """
583 h += tr_qa("Modality", modality)
584 h += tr_qa("Target number", self.target_number)
585 h += tr_qa("Background filename", ws.webify(self.background_filename))
586 h += tr_qa("Background intensity", self.background_intensity)
587 h += tr_qa("Target filename", ws.webify(self.target_filename))
588 h += tr_qa(
589 "(For visual targets) Target duration (s)",
590 self.visual_target_duration_s,
591 )
592 h += tr_qa("Start intensity (minimum)", self.start_intensity_min)
593 h += tr_qa("Start intensity (maximum)", self.start_intensity_max)
594 h += tr_qa(
595 "Initial (large) intensity step", self.initial_large_intensity_step
596 )
597 h += tr_qa(
598 "Main (small) intensity step", self.main_small_intensity_step
599 )
600 h += tr_qa(
601 "Number of trials in main sequence",
602 self.num_trials_in_main_sequence,
603 )
604 h += tr_qa("Probability of a catch trial", self.p_catch_trial)
605 h += tr_qa("Prompt", self.prompt)
606 h += tr_qa("Intertrial interval (ITI) (s)", self.iti_s)
607 h += f"""
608 </table>
609 <table class="{CssClass.TASKDETAIL}">
610 <tr><th width="50%">Measure</th><th width="50%">Value</th></tr>
611 """
612 h += tr_qa("Finished?", get_yes_no_none(req, self.finished))
613 h += tr_qa("Logistic intercept", ws.number_to_dp(self.intercept, DP))
614 h += tr_qa("Logistic slope", ws.number_to_dp(self.slope, DP))
615 h += tr_qa("Logistic k (= slope)", ws.number_to_dp(self.k, DP))
616 h += tr_qa(
617 "Logistic theta (= –intercept/slope)",
618 ws.number_to_dp(self.theta, DP),
619 )
620 h += tr_qa(
621 f"Intensity for {100 * LOWER_MARKER}% detection",
622 ws.number_to_dp(self.logistic_x_from_p(LOWER_MARKER), DP),
623 )
624 h += tr_qa(
625 "Intensity for 50% detection", ws.number_to_dp(self.theta, DP)
626 )
627 h += tr_qa(
628 f"Intensity for {100 * UPPER_MARKER}% detection",
629 ws.number_to_dp(self.logistic_x_from_p(UPPER_MARKER), DP),
630 )
631 h += """
632 </table>
633 """
634 h += self.get_trial_html(req)
635 return h