Coverage for tasks/cardinal_expectationdetection.py : 31%

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/cardinal_expectationdetection.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"""
29import logging
30from typing import Any, Dict, List, Optional, Sequence, Tuple, Type
32from cardinal_pythonlib.logs import BraceStyleAdapter
33from matplotlib.axes import Axes
34import numpy
35import scipy.stats # http://docs.scipy.org/doc/scipy/reference/stats.html
36from sqlalchemy.sql.schema import Column, ForeignKey
37from sqlalchemy.sql.sqltypes import Float, Integer
39from camcops_server.cc_modules.cc_constants import (
40 CssClass,
41 MatplotlibConstants,
42 PlotDefaults,
43)
44from camcops_server.cc_modules.cc_db import (
45 ancillary_relationship,
46 GenericTabletRecordMixin,
47 TaskDescendant,
48)
49from camcops_server.cc_modules.cc_html import (
50 answer,
51 div,
52 get_yes_no_none,
53 identity,
54 italic,
55 td,
56 tr,
57 tr_qa,
58)
59from camcops_server.cc_modules.cc_request import CamcopsRequest
60from camcops_server.cc_modules.cc_sqla_coltypes import (
61 PendulumDateTimeAsIsoTextColType,
62)
63from camcops_server.cc_modules.cc_sqlalchemy import Base
64from camcops_server.cc_modules.cc_summaryelement import (
65 ExtraSummaryTable,
66 SummaryElement,
67)
68from camcops_server.cc_modules.cc_task import Task, TaskHasPatientMixin
70log = BraceStyleAdapter(logging.getLogger(__name__))
73CONVERT_0_P_TO = 0.001 # for Z-transformed ROC plot
74CONVERT_1_P_TO = 0.999 # for Z-transformed ROC plot
76NRATINGS = 5 # numbered 0-4 in the database
77# -- to match DETECTION_OPTIONS.length in the original task
78N_CUES = 8 # to match magic number in original task
80ERROR_RATING_OUT_OF_RANGE = f"""
81 <div class="{CssClass.ERROR}">Can't draw figure: rating out of range</div>
82"""
83WARNING_INSUFFICIENT_DATA = f"""
84 <div class="{CssClass.WARNING}">Insufficient data</div>
85"""
86WARNING_RATING_MISSING = f"""
87 <div class="{CssClass.WARNING}">One or more ratings are missing</div>
88"""
89PLAIN_ROC_TITLE = "ROC"
90Z_ROC_TITLE = (
91 f"ROC in Z coordinates (0/1 first mapped to "
92 f"{CONVERT_0_P_TO}/{CONVERT_1_P_TO})"
93)
95AUDITORY = 0
96VISUAL = 1
99def a(x: Any) -> str:
100 """Answer formatting for this task."""
101 return answer(x, formatter_answer=identity, default="")
104# =============================================================================
105# Cardinal_ExpectationDetection
106# =============================================================================
108class ExpDetTrial(GenericTabletRecordMixin, TaskDescendant, Base):
109 __tablename__ = "cardinal_expdet_trials"
111 cardinal_expdet_id = Column(
112 "cardinal_expdet_id", Integer,
113 nullable=False,
114 comment="FK to cardinal_expdet"
115 )
116 trial = Column(
117 "trial", Integer,
118 nullable=False,
119 comment="Trial number (0-based)"
120 )
122 # Config determines these (via an autogeneration process):
123 block = Column(
124 "block", Integer,
125 comment="Block number (0-based)"
126 )
127 group_num = Column(
128 "group_num", Integer,
129 comment="Group number (0-based)"
130 )
131 cue = Column(
132 "cue", Integer,
133 comment="Cue number (0-based)"
134 )
135 raw_cue_number = Column(
136 "raw_cue_number", Integer,
137 comment="Raw cue number (following counterbalancing) (0-based)"
138 )
139 target_modality = Column(
140 "target_modality", Integer,
141 comment="Target modality (0 auditory, 1 visual)"
142 )
143 target_number = Column(
144 "target_number", Integer,
145 comment="Target number (0-based)"
146 )
147 target_present = Column(
148 "target_present", Integer,
149 comment="Target present? (0 no, 1 yes)"
150 )
151 iti_length_s = Column(
152 "iti_length_s", Float,
153 comment="Intertrial interval (s)"
154 )
156 # Task determines these (on the fly):
157 pause_given_before_trial = Column(
158 "pause_given_before_trial", Integer,
159 comment="Pause given before trial? (0 no, 1 yes)"
160 )
161 pause_start_time = Column(
162 "pause_start_time", PendulumDateTimeAsIsoTextColType,
163 comment="Pause start time (ISO-8601)"
164 )
165 pause_end_time = Column(
166 "pause_end_time", PendulumDateTimeAsIsoTextColType,
167 comment="Pause end time (ISO-8601)"
168 )
169 trial_start_time = Column(
170 "trial_start_time", PendulumDateTimeAsIsoTextColType,
171 comment="Trial start time (ISO-8601)"
172 )
173 cue_start_time = Column(
174 "cue_start_time", PendulumDateTimeAsIsoTextColType,
175 comment="Cue start time (ISO-8601)"
176 )
177 target_start_time = Column(
178 "target_start_time", PendulumDateTimeAsIsoTextColType,
179 comment="Target start time (ISO-8601)"
180 )
181 detection_start_time = Column(
182 "detection_start_time", PendulumDateTimeAsIsoTextColType,
183 comment="Detection response start time (ISO-8601)"
184 )
185 iti_start_time = Column(
186 "iti_start_time", PendulumDateTimeAsIsoTextColType,
187 comment="Intertrial interval start time (ISO-8601)"
188 )
189 iti_end_time = Column(
190 "iti_end_time", PendulumDateTimeAsIsoTextColType,
191 comment="Intertrial interval end time (ISO-8601)"
192 )
193 trial_end_time = Column(
194 "trial_end_time", PendulumDateTimeAsIsoTextColType,
195 comment="Trial end time (ISO-8601)"
196 )
198 # Subject decides these:
199 responded = Column(
200 "responded", Integer,
201 comment="Responded? (0 no, 1 yes)"
202 )
203 response_time = Column(
204 "response_time", PendulumDateTimeAsIsoTextColType,
205 comment="Response time (ISO-8601)"
206 )
207 response_latency_ms = Column(
208 "response_latency_ms", Integer,
209 comment="Response latency (ms)"
210 )
211 rating = Column(
212 "rating", Integer,
213 comment="Rating (0 definitely not - 4 definitely)"
214 )
215 correct = Column(
216 "correct", Integer,
217 comment="Correct side of the middle rating? (0 no, 1 yes)"
218 )
219 points = Column(
220 "points", Integer,
221 comment="Points earned this trial"
222 )
223 cumulative_points = Column(
224 "cumulative_points", Integer,
225 comment="Cumulative points earned"
226 )
228 @classmethod
229 def get_html_table_header(cls) -> str:
230 return f"""
231 <table class="{CssClass.EXTRADETAIL}">
232 <tr>
233 <th>Trial# (0-based)</th>
234 <th>Block# (0-based)</th>
235 <th>Group# (0-based)</th>
236 <th>Cue</th>
237 <th>Raw cue</th>
238 <th>Target modality</th>
239 <th>Target#</th>
240 <th>Target present?</th>
241 <th>ITI (s)</th>
242 <th>Pause before trial?</th>
243 </tr>
244 <tr class="{CssClass.EXTRADETAIL2}">
245 <th>...</th>
246 <th>Pause start@</th>
247 <th>Pause end@</th>
248 <th>Trial start@</th>
249 <th>Cue@</th>
250 <th>Target@</th>
251 <th>Detection start@</th>
252 <th>ITI start@</th>
253 <th>ITI end@</th>
254 <th>Trial end@</th>
255 </tr>
256 <tr class="{CssClass.EXTRADETAIL2}">
257 <th>...</th>
258 <th>Responded?</th>
259 <th>Responded@</th>
260 <th>Response latency (ms)</th>
261 <th>Rating</th>
262 <th>Correct?</th>
263 <th>Points</th>
264 <th>Cumulative points</th>
265 </tr>
266 """
268 # ratings: 0, 1 absent -- 2 don't know -- 3, 4 present
269 def judged_present(self) -> Optional[bool]:
270 if not self.responded:
271 return None
272 elif self.rating >= 3:
273 return True
274 else:
275 return False
277 def judged_absent(self) -> Optional[bool]:
278 if not self.responded:
279 return None
280 elif self.rating <= 1:
281 return True
282 else:
283 return False
285 def didnt_know(self) -> Optional[bool]:
286 if not self.responded:
287 return None
288 return self.rating == 2
290 def get_html_table_row(self) -> str:
291 return tr(
292 a(self.trial),
293 a(self.block),
294 a(self.group_num),
295 a(self.cue),
296 a(self.raw_cue_number),
297 a(self.target_modality),
298 a(self.target_number),
299 a(self.target_present),
300 a(self.iti_length_s),
301 a(self.pause_given_before_trial),
302 ) + tr(
303 "...",
304 a(self.pause_start_time),
305 a(self.pause_end_time),
306 a(self.trial_start_time),
307 a(self.cue_start_time),
308 a(self.target_start_time),
309 a(self.detection_start_time),
310 a(self.iti_start_time),
311 a(self.iti_end_time),
312 a(self.trial_end_time),
313 tr_class=CssClass.EXTRADETAIL2
314 ) + tr(
315 "...",
316 a(self.responded),
317 a(self.response_time),
318 a(self.response_latency_ms),
319 a(self.rating),
320 a(self.correct),
321 a(self.points),
322 a(self.cumulative_points),
323 tr_class=CssClass.EXTRADETAIL2
324 )
326 # -------------------------------------------------------------------------
327 # TaskDescendant overrides
328 # -------------------------------------------------------------------------
330 @classmethod
331 def task_ancestor_class(cls) -> Optional[Type["Task"]]:
332 return CardinalExpectationDetection
334 def task_ancestor(self) -> Optional["CardinalExpectationDetection"]:
335 return CardinalExpectationDetection.get_linked(
336 self.cardinal_expdet_id, self)
339class ExpDetTrialGroupSpec(GenericTabletRecordMixin, TaskDescendant, Base):
340 __tablename__ = "cardinal_expdet_trialgroupspec"
342 cardinal_expdet_id = Column(
343 "cardinal_expdet_id", Integer,
344 nullable=False,
345 comment="FK to cardinal_expdet"
346 )
347 group_num = Column(
348 "group_num", Integer,
349 nullable=False,
350 comment="Group number (0-based)"
351 )
353 # Group spec
354 cue = Column(
355 "cue", Integer,
356 comment="Cue number (0-based)"
357 )
358 target_modality = Column(
359 "target_modality", Integer,
360 comment="Target modality (0 auditory, 1 visual)"
361 )
362 target_number = Column(
363 "target_number", Integer,
364 comment="Target number (0-based)"
365 )
366 n_target = Column(
367 "n_target", Integer,
368 comment="Number of trials with target present"
369 )
370 n_no_target = Column(
371 "n_no_target", Integer,
372 comment="Number of trials with target absent"
373 )
375 DP = 3
377 @classmethod
378 def get_html_table_header(cls) -> str:
379 return f"""
380 <table class="{CssClass.EXTRADETAIL}">
381 <tr>
382 <th>Group# (0-based)</th>
383 <th>Cue (0-based)</th>
384 <th>Target modality (0 auditory, 1 visual)</th>
385 <th>Target# (0-based)</th>
386 <th># target trials</th>
387 <th># no-target trials</th>
388 </tr>
389 """
391 def get_html_table_row(self) -> str:
392 return tr(
393 a(self.group_num),
394 a(self.cue),
395 a(self.target_modality),
396 a(self.target_number),
397 a(self.n_target),
398 a(self.n_no_target),
399 )
401 # -------------------------------------------------------------------------
402 # TaskDescendant overrides
403 # -------------------------------------------------------------------------
405 @classmethod
406 def task_ancestor_class(cls) -> Optional[Type["Task"]]:
407 return CardinalExpectationDetection
409 def task_ancestor(self) -> Optional["CardinalExpectationDetection"]:
410 return CardinalExpectationDetection.get_linked(
411 self.cardinal_expdet_id, self)
414class CardinalExpectationDetection(TaskHasPatientMixin, Task):
415 """
416 Server implementation of the Cardinal_ExpDet task.
417 """
418 __tablename__ = "cardinal_expdet"
419 shortname = "Cardinal_ExpDet"
420 use_landscape_for_pdf = True
422 # Config
423 num_blocks = Column(
424 "num_blocks", Integer,
425 comment="Number of blocks"
426 )
427 stimulus_counterbalancing = Column(
428 "stimulus_counterbalancing", Integer,
429 comment="Stimulus counterbalancing condition"
430 )
431 is_detection_response_on_right = Column(
432 "is_detection_response_on_right", Integer,
433 comment='Is the "detection" response on the right? (0 no, 1 yes)'
434 )
435 pause_every_n_trials = Column(
436 "pause_every_n_trials", Integer,
437 comment="Pause every n trials"
438 )
439 # ... cue
440 cue_duration_s = Column(
441 "cue_duration_s", Float,
442 comment="Cue duration (s)"
443 )
444 visual_cue_intensity = Column(
445 "visual_cue_intensity", Float,
446 comment="Visual cue intensity (0.0-1.0)"
447 )
448 auditory_cue_intensity = Column(
449 "auditory_cue_intensity", Float,
450 comment="Auditory cue intensity (0.0-1.0)"
451 )
452 # ... ISI
453 isi_duration_s = Column(
454 "isi_duration_s", Float,
455 comment="Interstimulus interval (s)"
456 )
457 # .. target
458 visual_target_duration_s = Column(
459 "visual_target_duration_s", Float,
460 comment="Visual target duration (s)"
461 )
462 visual_background_intensity = Column(
463 "visual_background_intensity", Float,
464 comment="Visual background intensity (0.0-1.0)"
465 )
466 visual_target_0_intensity = Column(
467 "visual_target_0_intensity", Float,
468 comment="Visual target 0 intensity (0.0-1.0)"
469 )
470 visual_target_1_intensity = Column(
471 "visual_target_1_intensity", Float,
472 comment="Visual target 1 intensity (0.0-1.0)"
473 )
474 auditory_background_intensity = Column(
475 "auditory_background_intensity", Float,
476 comment="Auditory background intensity (0.0-1.0)"
477 )
478 auditory_target_0_intensity = Column(
479 "auditory_target_0_intensity", Float,
480 comment="Auditory target 0 intensity (0.0-1.0)"
481 )
482 auditory_target_1_intensity = Column(
483 "auditory_target_1_intensity", Float,
484 comment="Auditory target 1 intensity (0.0-1.0)"
485 )
486 # ... ITI
487 iti_min_s = Column(
488 "iti_min_s", Float,
489 comment="Intertrial interval minimum (s)"
490 )
491 iti_max_s = Column(
492 "iti_max_s", Float,
493 comment="Intertrial interval maximum (s)"
494 )
496 # Results
497 aborted = Column(
498 "aborted", Integer,
499 comment="Was the task aborted? (0 no, 1 yes)"
500 )
501 finished = Column(
502 "finished", Integer,
503 comment="Was the task finished? (0 no, 1 yes)"
504 )
505 last_trial_completed = Column(
506 "last_trial_completed", Integer,
507 comment="Number of last trial completed"
508 )
510 # Relationships
511 trials = ancillary_relationship(
512 parent_class_name="CardinalExpectationDetection",
513 ancillary_class_name="ExpDetTrial",
514 ancillary_fk_to_parent_attr_name="cardinal_expdet_id",
515 ancillary_order_by_attr_name="trial"
516 ) # type: List[ExpDetTrial]
517 groupspecs = ancillary_relationship(
518 parent_class_name="CardinalExpectationDetection",
519 ancillary_class_name="ExpDetTrialGroupSpec",
520 ancillary_fk_to_parent_attr_name="cardinal_expdet_id",
521 ancillary_order_by_attr_name="group_num"
522 ) # type: List[ExpDetTrialGroupSpec]
524 @staticmethod
525 def longname(req: "CamcopsRequest") -> str:
526 _ = req.gettext
527 return _("Cardinal RN – Expectation–Detection task")
529 def is_complete(self) -> bool:
530 return bool(self.finished)
532 def get_summaries(self, req: CamcopsRequest) -> List[SummaryElement]:
533 return self.standard_task_summary_fields() + [
534 SummaryElement(name="final_score",
535 coltype=Integer(),
536 value=self.get_final_score()),
537 SummaryElement(name="overall_p_detect_present",
538 coltype=Float(),
539 value=self.get_overall_p_detect_present()),
540 SummaryElement(name="overall_p_detect_absent",
541 coltype=Float(),
542 value=self.get_overall_p_detect_absent()),
543 SummaryElement(name="overall_c",
544 coltype=Float(),
545 value=self.get_overall_c()),
546 SummaryElement(name="overall_d",
547 coltype=Float(),
548 value=self.get_overall_d()),
549 ]
551 def get_final_score(self) -> Optional[int]:
552 trialarray = self.trials
553 if not trialarray:
554 return None
555 return trialarray[-1].cumulative_points
557 def get_group_html(self) -> str:
558 grouparray = self.groupspecs
559 html = ExpDetTrialGroupSpec.get_html_table_header()
560 for g in grouparray:
561 html += g.get_html_table_row()
562 html += """</table>"""
563 return html
565 @staticmethod
566 def get_c_dprime(h: Optional[float],
567 fa: Optional[float],
568 two_alternative_forced_choice: bool = False) \
569 -> Tuple[Optional[float], Optional[float]]:
570 if h is None or fa is None:
571 return None, None
572 # In this task, we're only presenting a single alternative.
573 if fa == 0:
574 fa = CONVERT_0_P_TO
575 if fa == 1:
576 fa = CONVERT_1_P_TO
577 if h == 0:
578 h = CONVERT_0_P_TO
579 if h == 1:
580 h = CONVERT_1_P_TO
581 z_fa = scipy.stats.norm.ppf(fa)
582 z_h = scipy.stats.norm.ppf(h)
583 if two_alternative_forced_choice:
584 dprime = (1.0 / numpy.sqrt(2)) * (z_h - z_fa)
585 else:
586 dprime = z_h - z_fa
587 c = -0.5 * (z_h + z_fa)
588 # c is zero when FA rate and miss rate (1 - H) are equal
589 # c is negative when FA > miss
590 # c is positive when miss > FA
591 return c, dprime
593 @staticmethod
594 def get_sdt_values(count_stimulus: Sequence[int],
595 count_nostimulus: Sequence[int]) -> Dict:
596 # Probabilities and cumulative probabilities
597 sum_count_stimulus = numpy.sum(count_stimulus)
598 sum_count_nostimulus = numpy.sum(count_nostimulus)
599 if sum_count_stimulus == 0 or sum_count_nostimulus == 0:
600 fa = []
601 h = []
602 z_fa = []
603 z_h = []
604 else:
605 p_stimulus = count_stimulus / sum_count_stimulus
606 p_nostimulus = count_nostimulus / sum_count_nostimulus
607 # ... may produce a RuntimeWarning in case of division by zero
608 cump_stimulus = numpy.cumsum(p_stimulus) # hit rates
609 cump_nostimulus = numpy.cumsum(p_nostimulus) # false alarm rates
610 # We're interested in all pairs except the last:
611 fa = cump_stimulus[:-1]
612 h = cump_nostimulus[:-1]
613 # WHICH WAY ROUND YOU ASSIGN THESE DETERMINES THE ROC'S APPEARANCE.
614 # However, it's arbitrary, in the sense that the left/right
615 # assignment of the ratings is arbitrary. To make the ROC look
616 # conventional (top left), assign this way round, so that "fa"
617 # starts low and grows, and "h" starts high and falls. Hmm...
618 fa[fa == 0] = CONVERT_0_P_TO
619 fa[fa == 1] = CONVERT_1_P_TO
620 h[h == 0] = CONVERT_0_P_TO
621 h[h == 1] = CONVERT_1_P_TO
622 z_fa = scipy.stats.norm.ppf(fa)
623 z_h = scipy.stats.norm.ppf(h)
625 # log.debug("p_stimulus: " + str(p_stimulus))
626 # log.debug("p_nostimulus: " + str(p_nostimulus))
627 # log.debug("cump_stimulus: " + str(cump_stimulus))
628 # log.debug("cump_nostimulus: " + str(cump_nostimulus))
630 # log.debug("h: " + str(h))
631 # log.debug("fa: " + str(fa))
632 # log.debug("z_h: " + str(z_h))
633 # log.debug("z_fa: " + str(z_fa))
634 return {
635 "fa": fa,
636 "h": h,
637 "z_fa": z_fa,
638 "z_h": z_h,
639 }
641 def plot_roc(self,
642 req: CamcopsRequest,
643 ax: Axes,
644 count_stimulus: Sequence[int],
645 count_nostimulus: Sequence[int],
646 show_x_label: bool,
647 show_y_label: bool,
648 plainroc: bool,
649 subtitle: str) -> None:
650 extraspace = 0.05
651 sdtval = self.get_sdt_values(count_stimulus, count_nostimulus)
653 # Calculate d' for all pairs but the last
654 if plainroc:
655 x = sdtval["fa"]
656 y = sdtval["h"]
657 xlabel = "FA"
658 ylabel = "H"
659 ax.set_xlim(0 - extraspace, 1 + extraspace)
660 ax.set_ylim(0 - extraspace, 1 + extraspace)
661 else:
662 x = sdtval["z_fa"]
663 y = sdtval["z_h"]
664 xlabel = "Z(FA)"
665 ylabel = "Z(H)"
666 # Plot
667 ax.plot(
668 x, y,
669 marker=MatplotlibConstants.MARKER_PLUS,
670 color=MatplotlibConstants.COLOUR_BLUE,
671 linestyle=MatplotlibConstants.LINESTYLE_SOLID
672 )
673 ax.set_xlabel(xlabel if show_x_label else "", fontdict=req.fontdict)
674 ax.set_ylabel(ylabel if show_y_label else "", fontdict=req.fontdict)
675 ax.set_title(subtitle, fontdict=req.fontdict)
676 req.set_figure_font_sizes(ax)
678 @staticmethod
679 def get_roc_info(trialarray: List[ExpDetTrial],
680 blocks: List[int],
681 groups: Optional[List[int]]) -> Dict:
682 # Collect counts (Macmillan & Creelman p61)
683 total_n = 0
684 count_stimulus = numpy.zeros(NRATINGS)
685 count_nostimulus = numpy.zeros(NRATINGS)
686 rating_missing = False
687 rating_out_of_range = False
688 for t in trialarray:
689 if t.rating is None:
690 rating_missing = True
691 continue
692 if t.rating < 0 or t.rating >= NRATINGS:
693 rating_out_of_range = True
694 break
695 if groups and t.group_num not in groups:
696 continue
697 if blocks and t.block not in blocks:
698 continue
699 total_n += 1
700 if t.target_present:
701 count_stimulus[t.rating] += 1
702 else:
703 count_nostimulus[t.rating] += 1
704 return {
705 "total_n": total_n,
706 "count_stimulus": count_stimulus,
707 "count_nostimulus": count_nostimulus,
708 "rating_missing": rating_missing,
709 "rating_out_of_range": rating_out_of_range,
710 }
712 def get_roc_figure_by_group(self,
713 req: CamcopsRequest,
714 trialarray: List[ExpDetTrial],
715 grouparray: List[ExpDetTrialGroupSpec],
716 plainroc: bool) -> str:
717 if not trialarray or not grouparray:
718 return WARNING_INSUFFICIENT_DATA
719 figsize = (
720 PlotDefaults.FULLWIDTH_PLOT_WIDTH * 2,
721 PlotDefaults.FULLWIDTH_PLOT_WIDTH
722 )
723 html = ""
724 fig = req.create_figure(figsize=figsize)
725 warned = False
726 for groupnum in range(len(grouparray)):
727 ax = fig.add_subplot(2, 4, groupnum+1)
728 # ... rows, cols, plotnum (in reading order from 1)
729 rocinfo = self.get_roc_info(trialarray, [], [groupnum])
730 if rocinfo["rating_out_of_range"]:
731 return ERROR_RATING_OUT_OF_RANGE
732 if rocinfo["rating_missing"] and not warned:
733 html += WARNING_RATING_MISSING
734 warned = True
735 show_x_label = (groupnum > 3)
736 show_y_label = (groupnum % 4 == 0)
737 subtitle = f"Group {groupnum} (n = {rocinfo['total_n']})"
738 self.plot_roc(
739 req,
740 ax,
741 rocinfo["count_stimulus"],
742 rocinfo["count_nostimulus"],
743 show_x_label,
744 show_y_label,
745 plainroc,
746 subtitle
747 )
748 title = PLAIN_ROC_TITLE if plainroc else Z_ROC_TITLE
749 fontprops = req.fontprops
750 fontprops.set_weight("bold")
751 fig.suptitle(title, fontproperties=fontprops)
752 html += req.get_html_from_pyplot_figure(fig)
753 return html
755 def get_roc_figure_firsthalf_lasthalf(self,
756 req: CamcopsRequest,
757 trialarray: List[ExpDetTrial],
758 plainroc: bool) -> str:
759 if not trialarray or not self.num_blocks:
760 return WARNING_INSUFFICIENT_DATA
761 figsize = (
762 PlotDefaults.FULLWIDTH_PLOT_WIDTH,
763 PlotDefaults.FULLWIDTH_PLOT_WIDTH / 2
764 )
765 html = ""
766 fig = req.create_figure(figsize=figsize)
767 warned = False
768 for half in range(2):
769 ax = fig.add_subplot(1, 2, half+1)
770 # ... rows, cols, plotnum (in reading order from 1)
771 blocks = list(range(half * self.num_blocks // 2,
772 self.num_blocks // (2 - half)))
773 rocinfo = self.get_roc_info(trialarray, blocks, None)
774 if rocinfo["rating_out_of_range"]:
775 return ERROR_RATING_OUT_OF_RANGE
776 if rocinfo["rating_missing"] and not warned:
777 html += WARNING_RATING_MISSING
778 warned = True
779 show_x_label = True
780 show_y_label = (half == 0)
781 subtitle = "First half" if half == 0 else "Second half"
782 self.plot_roc(
783 req,
784 ax,
785 rocinfo["count_stimulus"],
786 rocinfo["count_nostimulus"],
787 show_x_label,
788 show_y_label,
789 plainroc,
790 subtitle
791 )
792 title = PLAIN_ROC_TITLE if plainroc else Z_ROC_TITLE
793 fontprops = req.fontprops
794 fontprops.set_weight("bold")
795 fig.suptitle(title, fontproperties=fontprops)
796 html += req.get_html_from_pyplot_figure(fig)
797 return html
799 def get_trial_html(self) -> str:
800 trialarray = self.trials
801 html = ExpDetTrial.get_html_table_header()
802 for t in trialarray:
803 html += t.get_html_table_row()
804 html += """</table>"""
805 return html
807 def get_task_html(self, req: CamcopsRequest) -> str:
808 grouparray = self.groupspecs
809 trialarray = self.trials
810 # THIS IS A NON-EDITABLE TASK, so we *ignore* the problem
811 # of matching to no-longer-current records.
812 # (See PhotoSequence.py for a task that does it properly.)
814 # Provide HTML
815 # HTML
816 h = f"""
817 <div class="{CssClass.SUMMARY}">
818 <table class="{CssClass.SUMMARY}">
819 {self.get_is_complete_tr(req)}
820 </table>
821 </div>
822 <div class="{CssClass.EXPLANATION}">
823 Putative assay of propensity to hallucinations.
824 </div>
825 <table class="{CssClass.TASKCONFIG}">
826 <tr>
827 <th width="50%">Configuration variable</th>
828 <th width="50%">Value</th>
829 </tr>
830 """
831 h += tr_qa("Number of blocks", self.num_blocks)
832 h += tr_qa("Stimulus counterbalancing", self.stimulus_counterbalancing)
833 h += tr_qa("“Detection” response on right?",
834 self.is_detection_response_on_right)
835 h += tr_qa("Pause every <i>n</i> trials (0 = no pauses)",
836 self.pause_every_n_trials)
837 h += tr_qa("Cue duration (s)", self.cue_duration_s)
838 h += tr_qa("Visual cue intensity (0–1)", self.visual_cue_intensity)
839 h += tr_qa("Auditory cue intensity (0–1)",
840 self.auditory_cue_intensity)
841 h += tr_qa("ISI duration (s)", self.isi_duration_s)
842 h += tr_qa("Visual target duration (s)", self.visual_target_duration_s)
843 h += tr_qa("Visual background intensity",
844 self.visual_background_intensity)
845 h += tr_qa("Visual target 0 (circle) intensity",
846 self.visual_target_0_intensity)
847 h += tr_qa("Visual target 1 (“sun”) intensity",
848 self.visual_target_1_intensity)
849 h += tr_qa("Auditory background intensity",
850 self.auditory_background_intensity)
851 h += tr_qa("Auditory target 0 (tone) intensity",
852 self.auditory_target_0_intensity)
853 h += tr_qa("Auditory target 1 (“moon”) intensity",
854 self.auditory_target_1_intensity)
855 h += tr_qa("ITI minimum (s)", self.iti_min_s)
856 h += tr_qa("ITI maximum (s)", self.iti_max_s)
857 h += f"""
858 </table>
859 <table class="{CssClass.TASKDETAIL}">
860 <tr><th width="50%">Measure</th><th width="50%">Value</th></tr>
861 """
862 h += tr_qa("Aborted?", get_yes_no_none(req, self.aborted))
863 h += tr_qa("Finished?", get_yes_no_none(req, self.finished))
864 h += tr_qa("Last trial completed", self.last_trial_completed)
865 h += (
866 """
867 </table>
868 <div>
869 Trial group specifications (one block is a full set of
870 all these trials):
871 </div>
872 """ +
873 self.get_group_html() +
874 """
875 <div>
876 Detection probabilities by block and group (c > 0 when
877 miss rate > false alarm rate; c < 0 when false alarm
878 rate > miss rate):
879 </div>
880 """ +
881 self.get_html_correct_by_group_and_block(trialarray) +
882 "<div>Detection probabilities by block:</div>" +
883 self.get_html_correct_by_block(trialarray) +
884 "<div>Detection probabilities by group:</div>" +
885 self.get_html_correct_by_group(trialarray) +
886 """
887 <div>
888 Detection probabilities by half and high/low association
889 probability:
890 </div>
891 """ +
892 self.get_html_correct_by_half_and_probability(trialarray,
893 grouparray) +
894 """
895 <div>
896 Detection probabilities by block and high/low association
897 probability:
898 </div>
899 """ +
900 self.get_html_correct_by_block_and_probability(trialarray,
901 grouparray) +
902 """
903 <div>
904 Receiver operating characteristic (ROC) curves by group:
905 </div>
906 """ +
907 self.get_roc_figure_by_group(req, trialarray, grouparray, True) +
908 self.get_roc_figure_by_group(req, trialarray, grouparray, False) +
909 "<div>First-half/last-half ROCs:</div>" +
910 self.get_roc_figure_firsthalf_lasthalf(req, trialarray, True) +
911 "<div>Trial-by-trial results:</div>" +
912 self.get_trial_html()
913 )
914 return h
916 def get_html_correct_by_group_and_block(
917 self,
918 trialarray: List[ExpDetTrial]) -> str:
919 if not trialarray:
920 return div(italic("No trials"))
921 html = f"""
922 <table class="{CssClass.EXTRADETAIL}">
923 <tr>
924 <th>Block</th>
925 """
926 for g in range(N_CUES):
927 # Have spaces around | to allow web browsers to word-wrap
928 html += f"""
929 <th>Group {g} P(detected | present)</th>
930 <th>Group {g} P(detected | absent)</th>
931 <th>Group {g} c</th>
932 <th>Group {g} d'</th>
933 """
934 html += """
935 </th>
936 </tr>
937 """
938 for b in range(self.num_blocks):
939 html += "<tr>" + td(str(b))
940 for g in range(N_CUES):
941 (p_detected_given_present,
942 p_detected_given_absent,
943 c,
944 dprime,
945 n_trials) = self.get_p_detected(trialarray, [b], [g])
946 html += td(a(p_detected_given_present))
947 html += td(a(p_detected_given_absent))
948 html += td(a(c))
949 html += td(a(dprime))
950 html += "</tr>\n"
951 html += """
952 </table>
953 """
954 return html
956 def get_html_correct_by_half_and_probability(
957 self,
958 trialarray: List[ExpDetTrial],
959 grouparray: List[ExpDetTrialGroupSpec]) -> str:
960 if (not trialarray) or (not grouparray):
961 return div(italic("No trials or no groups"))
962 n_target_highprob = max([x.n_target for x in grouparray])
963 n_target_lowprob = min([x.n_target for x in grouparray])
964 groups_highprob = [x.group_num
965 for x in grouparray
966 if x.n_target == n_target_highprob]
967 groups_lowprob = [x.group_num
968 for x in grouparray
969 if x.n_target == n_target_lowprob]
970 html = f"""
971 <div><i>
972 High probability groups (cues):
973 {", ".join([str(x) for x in groups_highprob])}.\n
974 Low probability groups (cues):
975 {", ".join([str(x) for x in groups_lowprob])}.\n
976 </i></div>
977 <table class="{CssClass.EXTRADETAIL}">
978 <tr>
979 <th>Half (0 first, 1 second)</th>
980 <th>Target probability given stimulus (0 low, 1 high)</th>
981 <th>P(detected | present)</th>
982 <th>P(detected | absent)</th>
983 <th>c</th>
984 <th>d'</th>
985 </tr>
986 """
987 for half in [0, 1]:
988 for prob in [0, 1]:
989 blocks = list(range(half * self.num_blocks // 2,
990 self.num_blocks // (2 - half)))
991 groups = groups_lowprob if prob == 0 else groups_highprob
992 (p_detected_given_present,
993 p_detected_given_absent,
994 c,
995 dprime,
996 n_trials) = self.get_p_detected(trialarray, blocks, groups)
997 html += tr(
998 half,
999 a(prob),
1000 a(p_detected_given_present),
1001 a(p_detected_given_absent),
1002 a(c),
1003 a(dprime),
1004 )
1005 html += """
1006 </table>
1007 """
1008 return html
1010 def get_html_correct_by_block_and_probability(
1011 self,
1012 trialarray: List[ExpDetTrial],
1013 grouparray: List[ExpDetTrialGroupSpec]) -> str:
1014 if (not trialarray) or (not grouparray):
1015 return div(italic("No trials or no groups"))
1016 n_target_highprob = max([x.n_target for x in grouparray])
1017 n_target_lowprob = min([x.n_target for x in grouparray])
1018 groups_highprob = [x.group_num
1019 for x in grouparray
1020 if x.n_target == n_target_highprob]
1021 groups_lowprob = [x.group_num
1022 for x in grouparray
1023 if x.n_target == n_target_lowprob]
1024 html = f"""
1025 <div><i>
1026 High probability groups (cues):
1027 {", ".join([str(x) for x in groups_highprob])}.\n
1028 Low probability groups (cues):
1029 {", ".join([str(x) for x in groups_lowprob])}.\n
1030 </i></div>
1031 <table class="{CssClass.EXTRADETAIL}">
1032 <tr>
1033 <th>Block (0-based)</th>
1034 <th>Target probability given stimulus (0 low, 1 high)</th>
1035 <th>P(detected | present)</th>
1036 <th>P(detected | absent)</th>
1037 <th>c</th>
1038 <th>d'</th>
1039 </tr>
1040 """
1041 for b in range(self.num_blocks):
1042 for prob in [0, 1]:
1043 groups = groups_lowprob if prob == 0 else groups_highprob
1044 (p_detected_given_present,
1045 p_detected_given_absent,
1046 c,
1047 dprime,
1048 n_trials) = self.get_p_detected(trialarray, [b], groups)
1049 html += tr(
1050 b,
1051 prob,
1052 a(p_detected_given_present),
1053 a(p_detected_given_absent),
1054 a(c),
1055 a(dprime),
1056 )
1057 html += """
1058 </table>
1059 """
1060 return html
1062 def get_html_correct_by_group(self, trialarray: List[ExpDetTrial]) -> str:
1063 if not trialarray:
1064 return div(italic("No trials"))
1065 html = f"""
1066 <table class="{CssClass.EXTRADETAIL}">
1067 <tr>
1068 <th>Group</th>
1069 <th>P(detected | present)</th>
1070 <th>P(detected | absent)</th>
1071 <th>c</th>
1072 <th>d'</th>
1073 </tr>
1074 """
1075 for g in range(N_CUES):
1076 (p_detected_given_present,
1077 p_detected_given_absent,
1078 c,
1079 dprime,
1080 n_trials) = self.get_p_detected(trialarray, None, [g])
1081 html += tr(
1082 g,
1083 a(p_detected_given_present),
1084 a(p_detected_given_absent),
1085 a(c),
1086 a(dprime),
1087 )
1088 html += """
1089 </table>
1090 """
1091 return html
1093 def get_html_correct_by_block(self, trialarray: List[ExpDetTrial]) -> str:
1094 if not trialarray:
1095 return div(italic("No trials"))
1096 html = f"""
1097 <table class="{CssClass.EXTRADETAIL}">
1098 <tr>
1099 <th>Block</th>
1100 <th>P(detected | present)</th>
1101 <th>P(detected | absent)</th>
1102 <th>c</th>
1103 <th>d'</th>
1104 </tr>
1105 """
1106 for b in range(self.num_blocks):
1107 (p_detected_given_present,
1108 p_detected_given_absent,
1109 c,
1110 dprime,
1111 n_trials) = self.get_p_detected(trialarray, [b], None)
1112 html += tr(
1113 b,
1114 a(p_detected_given_present),
1115 a(p_detected_given_absent),
1116 a(c),
1117 a(dprime),
1118 )
1119 html += """
1120 </table>
1121 """
1122 return html
1124 def get_p_detected(self,
1125 trialarray: List[ExpDetTrial],
1126 blocks: Optional[List[int]],
1127 groups: Optional[List[int]]) -> \
1128 Tuple[Optional[float], Optional[float],
1129 Optional[float], Optional[float],
1130 int]:
1131 n_present = 0
1132 n_absent = 0
1133 n_detected_given_present = 0
1134 n_detected_given_absent = 0
1135 n_trials = 0
1136 for t in trialarray:
1137 if (not t.responded or
1138 (blocks is not None and t.block not in blocks) or
1139 (groups is not None and t.group_num not in groups)):
1140 continue
1141 if t.target_present:
1142 n_present += 1
1143 if t.judged_present():
1144 n_detected_given_present += 1
1145 else:
1146 n_absent += 1
1147 if t.judged_present():
1148 n_detected_given_absent += 1
1149 n_trials += 1
1150 p_detected_given_present = (
1151 float(n_detected_given_present) / float(n_present)
1152 ) if n_present > 0 else None
1153 p_detected_given_absent = (
1154 float(n_detected_given_absent) / float(n_absent)
1155 ) if n_absent > 0 else None
1156 (c, dprime) = self.get_c_dprime(p_detected_given_present,
1157 p_detected_given_absent)
1158 # hits: p_detected_given_present
1159 # false alarms: p_detected_given_absent
1160 return (p_detected_given_present, p_detected_given_absent, c, dprime,
1161 n_trials)
1163 def get_extra_summary_tables(self, req: CamcopsRequest) \
1164 -> List[ExtraSummaryTable]:
1165 grouparray = self.groupspecs
1166 trialarray = self.trials
1167 trialarray_auditory = [x for x in trialarray
1168 if x.target_modality == AUDITORY]
1169 blockprob_values = [] # type: List[Dict[str, Any]]
1170 halfprob_values = [] # type: List[Dict[str, Any]]
1172 if grouparray and trialarray:
1173 n_target_highprob = max([x.n_target for x in grouparray])
1174 n_target_lowprob = min([x.n_target for x in grouparray])
1175 groups_highprob = [x.group_num for x in grouparray
1176 if x.n_target == n_target_highprob]
1177 groups_lowprob = [x.group_num for x in grouparray
1178 if x.n_target == n_target_lowprob]
1179 for block in range(self.num_blocks):
1180 for target_probability_low_high in (0, 1):
1181 groups = (
1182 groups_lowprob if target_probability_low_high == 0
1183 else groups_highprob
1184 )
1185 (p_detected_given_present,
1186 p_detected_given_absent,
1187 c, dprime,
1188 n_trials) = self.get_p_detected(
1189 trialarray, [block], groups)
1190 (auditory_p_detected_given_present,
1191 auditory_p_detected_given_absent,
1192 auditory_c, auditory_dprime,
1193 auditory_n_trials) = self.get_p_detected(
1194 trialarray_auditory, [block], groups)
1195 blockprob_values.append(dict(
1196 cardinal_expdet_pk=self._pk, # tablename_pk
1197 n_blocks_overall=self.num_blocks,
1198 block=block,
1199 target_probability_low_high=target_probability_low_high, # noqa
1200 n_trials=n_trials,
1201 p_detect_present=p_detected_given_present,
1202 p_detect_absent=p_detected_given_absent,
1203 c=c,
1204 d=dprime,
1205 auditory_n_trials=auditory_n_trials,
1206 auditory_p_detect_present=auditory_p_detected_given_present, # noqa
1207 auditory_p_detect_absent=auditory_p_detected_given_absent, # noqa
1208 auditory_c=auditory_c,
1209 auditory_d=auditory_dprime,
1210 ))
1212 # Now another one...
1213 for half in range(2):
1214 blocks = list(range(half * self.num_blocks // 2,
1215 self.num_blocks // (2 - half)))
1216 for target_probability_low_high in (0, 1):
1217 groups = (
1218 groups_lowprob if target_probability_low_high == 0
1219 else groups_highprob
1220 )
1221 (p_detected_given_present,
1222 p_detected_given_absent,
1223 c,
1224 dprime,
1225 n_trials) = self.get_p_detected(
1226 trialarray, blocks, groups)
1227 (auditory_p_detected_given_present,
1228 auditory_p_detected_given_absent,
1229 auditory_c, auditory_dprime,
1230 auditory_n_trials) = self.get_p_detected(
1231 trialarray_auditory, blocks, groups)
1232 halfprob_values.append(dict(
1233 cardinal_expdet_pk=self._pk, # tablename_pk
1234 half=half,
1235 target_probability_low_high=target_probability_low_high, # noqa
1236 n_trials=n_trials,
1237 p_detect_present=p_detected_given_present,
1238 p_detect_absent=p_detected_given_absent,
1239 c=c,
1240 d=dprime,
1241 auditory_n_trials=auditory_n_trials,
1242 auditory_p_detect_present=auditory_p_detected_given_present, # noqa
1243 auditory_p_detect_absent=auditory_p_detected_given_absent, # noqa
1244 auditory_c=auditory_c,
1245 auditory_d=auditory_dprime,
1246 ))
1248 blockprob_table = ExtraSummaryTable(
1249 tablename="cardinal_expdet_blockprobs",
1250 xmlname="blockprobs",
1251 columns=self.get_blockprob_columns(),
1252 rows=blockprob_values,
1253 task=self
1254 )
1255 halfprob_table = ExtraSummaryTable(
1256 tablename="cardinal_expdet_halfprobs",
1257 xmlname="halfprobs",
1258 columns=self.get_halfprob_columns(),
1259 rows=halfprob_values,
1260 task=self
1261 )
1262 return [blockprob_table, halfprob_table]
1264 @staticmethod
1265 def get_blockprob_columns() -> List[Column]:
1266 # Must be a function, not a constant, because as SQLAlchemy builds the
1267 # tables, it assigns the Table object to each Column. Therefore, a
1268 # constant list works for the first request, but fails on subsequent
1269 # requests with e.g. "sqlalchemy.exc.ArgumentError: Column object 'id'
1270 # already assigned to Table 'cardinal_expdet_blockprobs'"
1271 return [
1272 Column("id", Integer, primary_key=True, autoincrement=True,
1273 comment="Arbitrary PK"),
1274 Column("cardinal_expdet_pk", Integer,
1275 ForeignKey("cardinal_expdet._pk"),
1276 nullable=False,
1277 comment="FK to the source table's _pk field"),
1278 Column("n_blocks_overall", Integer,
1279 comment="Number of blocks (OVERALL)"),
1280 Column("block", Integer,
1281 comment="Block number"),
1282 Column("target_probability_low_high", Integer,
1283 comment="Target probability given stimulus "
1284 "(0 low, 1 high)"),
1285 Column("n_trials", Integer,
1286 comment="Number of trials in this condition"),
1287 Column("p_detect_present", Float,
1288 comment="P(detect | present)"),
1289 Column("p_detect_absent", Float,
1290 comment="P(detect | absent)"),
1291 Column("c", Float,
1292 comment="c (bias; c > 0 when miss rate > false alarm rate; "
1293 "c < 0 when false alarm rate > miss rate)"),
1294 Column("d", Float,
1295 comment="d' (discriminability)"),
1296 Column("auditory_n_trials", Integer,
1297 comment="Number of auditory trials in this condition"),
1298 Column("auditory_p_detect_present", Float,
1299 comment="AUDITORY P(detect | present)"),
1300 Column("auditory_p_detect_absent", Float,
1301 comment="AUDITORY P(detect | absent)"),
1302 Column("auditory_c", Float,
1303 comment="AUDITORY c"),
1304 Column("auditory_d", Float,
1305 comment="AUDITORY d'"),
1306 ]
1308 @staticmethod
1309 def get_halfprob_columns() -> List[Column]:
1310 return [
1311 Column("id", Integer, primary_key=True, autoincrement=True,
1312 comment="Arbitrary PK"),
1313 Column("cardinal_expdet_pk", Integer,
1314 ForeignKey("cardinal_expdet._pk"),
1315 nullable=False,
1316 comment="FK to the source table's _pk field"),
1317 Column("half", Integer,
1318 comment="Half number"),
1319 Column("target_probability_low_high", Integer,
1320 comment="Target probability given stimulus "
1321 "(0 low, 1 high)"),
1322 Column("n_trials", Integer,
1323 comment="Number of trials in this condition"),
1324 Column("p_detect_present", Float,
1325 comment="P(detect | present)"),
1326 Column("p_detect_absent", Float,
1327 comment="P(detect | absent)"),
1328 Column("c", Float,
1329 comment="c (bias; c > 0 when miss rate > false alarm rate; "
1330 "c < 0 when false alarm rate > miss rate)"),
1331 Column("d", Float,
1332 comment="d' (discriminability)"),
1333 Column("auditory_n_trials", Integer,
1334 comment="Number of auditory trials in this condition"),
1335 Column("auditory_p_detect_present", Float,
1336 comment="AUDITORY P(detect | present)"),
1337 Column("auditory_p_detect_absent", Float,
1338 comment="AUDITORY P(detect | absent)"),
1339 Column("auditory_c", Float,
1340 comment="AUDITORY c"),
1341 Column("auditory_d", Float,
1342 comment="AUDITORY d'"),
1343 ]
1345 def get_overall_p_detect_present(self) -> Optional[float]:
1346 trialarray = self.trials
1347 (p_detected_given_present,
1348 p_detected_given_absent,
1349 c,
1350 dprime,
1351 n_trials) = self.get_p_detected(trialarray, None, None)
1352 return p_detected_given_present
1354 def get_overall_p_detect_absent(self) -> Optional[float]:
1355 trialarray = self.trials
1356 (p_detected_given_present,
1357 p_detected_given_absent,
1358 c,
1359 dprime,
1360 n_trials) = self.get_p_detected(trialarray, None, None)
1361 return p_detected_given_absent
1363 def get_overall_c(self) -> Optional[float]:
1364 trialarray = self.trials
1365 (p_detected_given_present,
1366 p_detected_given_absent,
1367 c,
1368 dprime,
1369 n_trials) = self.get_p_detected(trialarray, None, None)
1370 return c
1372 def get_overall_d(self) -> Optional[float]:
1373 trialarray = self.trials
1374 (p_detected_given_present,
1375 p_detected_given_absent,
1376 c,
1377 dprime,
1378 n_trials) = self.get_p_detected(trialarray, None, None)
1379 return dprime