Coverage for ibllib/qc/critical_reasons.py: 88%
230 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-08 17:16 +0100
1"""
2Prompt experimenter for reason for marking session/insertion as CRITICAL
3Choices are listed in the global variables. Multiple reasons can be selected.
4Places info in Alyx session note in a format that is machine retrievable (text->json)
5"""
6import abc
7import logging
8import json
9import warnings
10from datetime import datetime
11from one.api import OneAlyx
12from one.webclient import AlyxClient
14_logger = logging.getLogger('ibllib')
17def main_gui(uuid, reasons_selected, one=None, alyx=None):
18 """
19 Main function to call to input a reason for marking an insertion as CRITICAL from the alignment GUI.
21 It wil create note text, after deleting any similar notes existing already.
23 Parameters
24 ----------
25 uuid : uuid.UUID, str
26 An insertion ID.
27 reasons_selected : list of str
28 A subset of REASONS_INS_CRIT_GUI.
29 one : one.api.OneAlyx
30 (DEPRECATED) An instance of ONE. NB: Pass in an instance of AlyxClient instead.
31 alyx : one.webclient.AlyxClient
32 An AlyxClient instance.
33 """
34 # hit the database to check if uuid is insertion uuid
35 if alyx is None and one is not None: 1g
36 # Deprecate ONE in future because instantiating it takes longer and is unnecessary
37 warnings.warn('In future please pass in an AlyxClient instance (i.e. `one.alyx`)', FutureWarning) 1g
38 alyx = one if isinstance(one, AlyxClient) else one.alyx 1g
39 ins_list = alyx.rest('insertions', 'list', id=uuid, no_cache=True) 1g
40 if len(ins_list) != 1: 1g
41 raise ValueError(f'N={len(ins_list)} insertion found, expected N=1. Check uuid provided.')
43 note = CriticalInsertionNote(uuid, alyx) 1g
45 # assert that reasons are all within REASONS_INS_CRIT_GUI
46 for item_str in reasons_selected: 1g
47 assert item_str in note.descriptions_gui 1g
49 note.selected_reasons = reasons_selected 1g
50 note.other_reason = [] 1g
51 note._upload_note(overwrite=True) 1g
54def main(uuid, one=None, alyx=None):
55 """
56 Main function to call to input a reason for marking a session/insertion as CRITICAL programmatically.
58 It will:
59 - ask reasons for selection of critical status
60 - check if 'other' reason has been selected, inquire why (free text)
61 - create note text, checking whether similar notes exist already
62 - upload note to Alyx if none exist previously or if overwrite is chosen Q&A are prompted via the Python terminal
64 Parameters
65 ----------
66 uuid : uuid.UUID, str
67 An experiment UUID or an insertion UUID.
68 one : one.api.OneAlyx
69 (DEPRECATED) An instance of ONE. NB: Pass in an instance of AlyxClient instead.
70 alyx : one.webclient.AlyxClient
71 An AlyxClient instance.
73 Examples
74 --------
75 Retrieve Alyx note to test
77 >>> alyx = AlyxClient(base_url='https://dev.alyx.internationalbrainlab.org')
78 >>> uuid = '2ffd3ed5-477e-4153-9af7-7fdad3c6946b'
79 >>> main(uuid=uuid, alyx=alyx)
81 Get notes with pattern
83 >>> notes = alyx.rest('notes', 'list',
84 ... django=f'text__icontains,{STR_NOTES_STATIC},'
85 ... f'object_id,{uuid}')
86 >>> test_json_read = json.loads(notes[0]['text'])
88 """
89 if alyx is None and one is not None: 1cef
90 # Deprecate ONE in future because instantiating it takes longer and is unnecessary
91 warnings.warn('In future please pass in an AlyxClient instance (i.e. `one.alyx`)', FutureWarning) 1cef
92 alyx = one if isinstance(one, AlyxClient) else one.alyx 1cef
93 if not alyx: 1cef
94 alyx = AlyxClient()
95 # ask reasons for selection of critical status
97 # hit the database to know if uuid is insertion or session uuid
98 sess_list = alyx.get('/sessions?&django=pk,' + uuid, clobber=True) 1cef
99 ins_list = alyx.get('/insertions?&django=pk,' + uuid, clobber=True) 1cef
101 if len(sess_list) > 0 and len(ins_list) == 0: # session 1cef
102 note = CriticalSessionNote(uuid, alyx) 1cf
103 elif len(ins_list) > 0 and len(sess_list) == 0: # insertion 1e
104 note = CriticalInsertionNote(uuid, alyx) 1e
105 else:
106 raise ValueError(f'Inadequate number of session (n={len(sess_list)}) '
107 f'or insertion (n={len(ins_list)}) found for uuid {uuid}.'
108 f'The query output should be of length 1.')
110 note.upload_note() 1cef
113class Note(abc.ABC):
114 descriptions = []
116 @property
117 def default_descriptions(self):
118 return self.descriptions + ['Other'] 1bdcf
120 @property
121 def extra_prompt(self):
122 return ''
124 @property
125 def note_title(self):
126 return ''
128 @property
129 def n_description(self):
130 return len(self.default_descriptions) 1bdcef
132 def __init__(self, uuid, alyx, content_type=None):
133 """
134 Base class for attaching notes to an alyx endpoint. Do not use this class directly but use parent classes that inherit
135 this base class
137 Parameters
138 ----------
139 uuid : uuid.UUID, str
140 A UUID of a session, insertion, or other Alyx model to attach note to.
141 alyx : one.webclient.AlyxClient
142 An AlyxClient instance.
143 content_type : str
144 The Alyx model name of the UUID.
145 """
146 self.uuid = uuid 1hbdgcef
147 if isinstance(alyx, OneAlyx): 1hbdgcef
148 # Deprecate ONE in future because instantiating it takes longer and is unnecessary
149 warnings.warn('In future please pass in an AlyxClient instance (i.e. `one.alyx`)', FutureWarning) 1hbd
150 alyx = alyx.alyx 1hbd
151 self.alyx = alyx 1hbdgcef
152 self.selected_reasons = [] 1hbdgcef
153 self.other_reason = [] 1hbdgcef
154 if content_type is not None: 1hbdgcef
155 self.content_type = content_type 1hbdgcef
156 else:
157 self.content_type = self.get_content_type()
159 def get_content_type(self):
160 """
161 Infer the content_type from the uuid. Only checks to see if uuid is a session or insertion.
162 If not recognised will raise an error and the content_type must be specified on note
163 initialisation e.g. Note(uuid, alyx, content_type='subject')
165 Returns
166 -------
167 str
168 The Alyx model name, inferred from the UUID.
169 """
171 # see if it as session or an insertion
172 if self.alyx.rest('sessions', 'list', id=self.uuid):
173 content_type = 'session'
174 elif self.alyx.rest('insertions', 'list', id=self.uuid):
175 content_type = 'probeinsertion'
176 else:
177 raise ValueError(f'Content type cannot be recognised from {self.uuid}. '
178 'Specify on initialistion e.g Note(uuid, alyx, content_type="subject"')
179 return content_type
181 def describe(self):
182 """
183 Print list of default reasons that can be chosen from
184 :return:
185 """
186 for i, d in enumerate(self.descriptions):
187 print(f'{i}. {d} \n')
189 def numbered_descriptions(self):
190 """
191 Return list of numbered default reasons
192 :return:
193 """
194 return [f'{i}) {d}' for i, d in enumerate(self.default_descriptions)] 1bdcef
196 def upload_note(self, nums=None, other_reason=None, **kwargs):
197 """
198 Upload note to Alyx.
200 If no values for nums and other_reason are specified, user will receive a prompt in command
201 line asking them to choose from default list of reasons to add to note as well as option
202 for free text. To upload without receiving prompt a value for either `nums` or
203 `other_reason` must be given.
205 Parameters
206 ----------
207 nums : str
208 string of numbers matching those in default descriptions, e.g, '1,3'. Options can be
209 seen using note.describe().
210 other_reason : str
211 Other comment or reason(s) to add to note.
213 """
215 if nums is None and other_reason is None: 1bdcef
216 self.selected_reasons, self.other_reason = self.reasons_prompt() 1bdcef
217 else:
218 self.selected_reasons = self._map_num_to_description(nums) 1b
219 self.other_reason = other_reason or [] 1b
221 self._upload_note(**kwargs) 1bdcef
223 def _upload_note(self, **kwargs):
224 existing_note, notes = self._check_existing_note() 1bdgcef
225 if existing_note: 1bdgcef
226 self.update_existing_note(notes, **kwargs) 1bc
227 else:
228 text = self.format_note(**kwargs) 1bdgcef
229 self._create_note(text) 1bdgcef
230 _logger.info('The selected reasons were saved on Alyx.') 1bdgcef
232 def _create_note(self, text):
234 data = {'user': self.alyx.user, 1bdgcef
235 'content_type': self.content_type,
236 'object_id': self.uuid,
237 'text': f'{text}'}
238 self.alyx.rest('notes', 'create', data=data) 1bdgcef
240 def _update_note(self, note_id, text):
242 data = {'user': self.alyx.user, 1b
243 'content_type': self.content_type,
244 'object_id': self.uuid,
245 'text': f'{text}'}
246 self.alyx.rest('notes', 'partial_update', id=note_id, data=data) 1b
248 def _delete_note(self, note_id):
249 self.alyx.rest('notes', 'delete', id=note_id) 1c
251 def _delete_notes(self, notes):
252 for note in notes: 1c
253 self._delete_note(note['id']) 1c
255 def _check_existing_note(self):
256 notes = self.alyx.rest('notes', 'list', django=f'text__icontains,{self.note_title},object_id,{self.uuid}', no_cache=True) 1bdgcef
257 if len(notes) == 0: 1bdgcef
258 return False, None 1bdgcef
259 else:
260 return True, notes 1bc
262 def _map_num_to_description(self, nums):
264 if nums is None: 1bdcef
265 return []
267 string_list = nums.split(',') 1bdcef
268 int_list = list(map(int, string_list)) 1bdcef
270 if max(int_list) >= self.n_description or min(int_list) < 0: 1bdcef
271 raise ValueError(f'Chosen values out of range, must be between 0 and {self.n_description - 1}')
273 return [self.default_descriptions[n] for n in int_list] 1bdcef
275 def reasons_prompt(self):
276 """
277 Prompt for user to enter reasons
278 :return:
279 """
281 prompt = f'{self.extra_prompt} ' \ 1bdcef
282 f'\n {self.numbered_descriptions()} \n ' \
283 f'and enter the corresponding numbers separated by commas, e.g. 1,3 -> enter: '
285 ans = input(prompt).strip().lower() 1bdcef
287 try: 1bdcef
288 selected_reasons = self._map_num_to_description(ans) 1bdcef
289 print(f'You selected reason(s): {selected_reasons}') 1bdcef
290 if 'Other' in selected_reasons: 1bdcef
291 other_reasons = self.other_reason_prompt() 1bdcef
292 return selected_reasons, other_reasons 1bdcef
293 else:
294 return selected_reasons, []
296 except ValueError:
297 print(f'{ans} is invalid, please try again...')
298 return self.reasons_prompt()
300 def other_reason_prompt(self):
301 """
302 Prompt for user to enter other reasons
303 :return:
304 """
306 prompt = 'Explain why you selected "other" (free text): ' 1bdcef
307 ans = input(prompt).strip().lower() 1bdcef
308 return ans 1bdcef
310 @abc.abstractmethod
311 def format_note(self, **kwargs):
312 """
313 Method to format text field of note according to type of note wanting to be uploaded
314 :param kwargs:
315 :return:
316 """
318 @abc.abstractmethod
319 def update_existing_note(self, note, **kwargs):
320 """
321 Method to specify behavior in the case of a note with the same title already existing
322 :param note:
323 :param kwargs:
324 :return:
325 """
328class CriticalNote(Note):
329 """
330 Class for uploading a critical note to a session or insertion. Do not use directly but use CriticalSessionNote or
331 CriticalInsertionNote instead
332 """
334 def format_note(self, **kwargs):
335 note_text = { 1gcef
336 "title": self.note_title,
337 "reasons_selected": self.selected_reasons,
338 "reason_for_other": self.other_reason
339 }
340 return json.dumps(note_text) 1gcef
342 def update_existing_note(self, notes, **kwargs):
344 overwrite = kwargs.get('overwrite', None) 1c
345 if overwrite is None: 1c
346 overwrite = self.delete_note_prompt(notes) 1c
348 if overwrite: 1c
349 self._delete_notes(notes) 1c
350 text = self.format_note() 1c
351 self._create_note(text) 1c
352 _logger.info('The selected reasons were saved on Alyx; old notes were deleted') 1c
353 else:
354 _logger.info('The selected reasons were NOT saved on Alyx; old notes remain.')
356 def delete_note_prompt(self, notes):
358 prompt = f'You are about to delete {len(notes)} existing notes; ' \ 1c
359 f'do you want to proceed? y/n: '
361 ans = input(prompt).strip().lower() 1c
363 if ans not in ['y', 'n']: 1c
364 print(f'{ans} is invalid, please try again...')
365 return self.delete_note_prompt(notes)
366 else:
367 return True if ans == 'y' else False 1c
370class CriticalInsertionNote(CriticalNote):
371 """
372 Class for uploading a critical note to an insertion.
374 Examples
375 --------
376 >>> note = CriticalInsertionNote(pid, AlyxClient())
378 Print list of default reasons
380 >>> note.describe()
382 To receive a command line prompt to fill in note
384 >>> note.upload_note()
386 To upload note automatically without prompt
388 >>> note.upload_note(nums='1,4', other_reason='lots of bad channels')
389 """
391 descriptions_gui = [
392 'Noise and artifact',
393 'Drift',
394 'Poor neural yield',
395 'Brain Damage',
396 'Other'
397 ]
399 descriptions = [
400 'Histological images missing',
401 'Track not visible on imaging data'
402 ]
404 @property
405 def default_descriptions(self):
406 return self.descriptions + self.descriptions_gui 1e
408 @property
409 def extra_prompt(self):
410 return 'Select from this list the reason(s) why you are marking the insertion as CRITICAL:' 1e
412 @property
413 def note_title(self):
414 return '=== EXPERIMENTER REASON(S) FOR MARKING THE INSERTION AS CRITICAL ===' 1ge
416 def __init__(self, uuid, alyx):
417 super(CriticalInsertionNote, self).__init__(uuid, alyx, content_type='probeinsertion') 1ge
420class CriticalSessionNote(CriticalNote):
421 """
422 Class for uploading a critical note to a session.
424 Example
425 -------
426 >>> note = CriticalInsertionNote(uuid, AlyxClient)
428 Print list of default reasons
430 >>> note.describe()
432 To receive a command line prompt to fill in note
434 >>> note.upload_note()
436 To upload note automatically without prompt
438 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
439 """
441 descriptions = [
442 'within experiment system crash',
443 'synching impossible',
444 'dud or mock session',
445 'essential dataset missing',
446 ]
448 @property
449 def extra_prompt(self):
450 return 'Select from this list the reason(s) why you are marking the session as CRITICAL:' 1cf
452 @property
453 def note_title(self):
454 return '=== EXPERIMENTER REASON(S) FOR MARKING THE SESSION AS CRITICAL ===' 1cf
456 def __init__(self, uuid, alyx):
457 super(CriticalSessionNote, self).__init__(uuid, alyx, content_type='session') 1cf
460class SignOffNote(Note):
461 """
462 Class for signing off a session and optionally adding a related explanation note.
463 Do not use directly but use classes that inherit from this class e.g TaskSignOffNote, RawEphysSignOffNote
464 """
466 @property
467 def extra_prompt(self):
468 return 'Select from this list the reason(s) that describe issues with this session:' 1bd
470 @property
471 def note_title(self):
472 return f'=== SIGN-OFF NOTE FOR {self.sign_off_key} ===' 1bd
474 def __init__(self, uuid, alyx, sign_off_key):
475 self.sign_off_key = sign_off_key 1hbd
476 super(SignOffNote, self).__init__(uuid, alyx, content_type='session') 1hbd
477 self.datetime_key = self.get_datetime_key() 1hbd
478 self.session = self.alyx.rest('sessions', 'read', id=self.uuid, no_cache=True) 1hbd
480 def upload_note(self, nums=None, other_reason=None, **kwargs):
481 super(SignOffNote, self).upload_note(nums=nums, other_reason=other_reason, **kwargs) 1bd
482 self.sign_off() 1bd
484 def sign_off(self):
486 json = self.session['json'] 1hbd
487 sign_off_checklist = json.get('sign_off_checklist', None) 1hbd
488 if sign_off_checklist is None: 1hbd
489 sign_off_checklist = {self.sign_off_key: {'date': self.datetime_key.split('_')[0],
490 'user': self.datetime_key.split('_')[1]}}
491 else:
492 sign_off_checklist[self.sign_off_key] = {'date': self.datetime_key.split('_')[0], 1hbd
493 'user': self.datetime_key.split('_')[1]}
495 json['sign_off_checklist'] = sign_off_checklist 1hbd
497 self.alyx.json_field_update("sessions", self.uuid, 'json', data=json) 1hbd
499 def format_note(self, **kwargs):
501 note_text = { 1bd
502 "title": self.note_title,
503 f'{self.datetime_key}': {"reasons_selected": self.selected_reasons,
504 "reason_for_other": self.other_reason}
505 }
507 return json.dumps(note_text) 1bd
509 def format_existing_note(self, orignal_note):
511 extra_note = {f'{self.datetime_key}': {"reasons_selected": self.selected_reasons, 1b
512 "reason_for_other": self.other_reason}
513 }
515 orignal_note.update(extra_note) 1b
517 return json.dumps(orignal_note) 1b
519 def update_existing_note(self, notes):
520 if len(notes) != 1: 1b
521 raise ValueError(f'{len(notes)} with same title found, only expect at most 1. Clean up before proceeding')
522 else:
523 original_note = json.loads(notes[0]['text']) 1b
524 text = self.format_existing_note(original_note) 1b
525 self._update_note(notes[0]['id'], text) 1b
527 def get_datetime_key(self):
528 if not self.alyx.is_logged_in: 1hbd
529 self.alyx.authenticate()
530 assert self.alyx.is_logged_in, 'you must be logged in to the AlyxClient'
531 user = self.alyx.user 1hbd
532 date = datetime.now().date().isoformat() 1hbd
533 return date + '_' + user 1hbd
536class TaskSignOffNote(SignOffNote):
538 """
539 Class for signing off a task part of a session and optionally adding a related explanation note.
541 Examples
542 --------
543 >>> note = TaskSignOffNote(eid, AlyxClient(), '_ephysChoiceWorld_00')
545 To sign off session without any note
547 >>> note.sign_off()
549 Print list of default reasons
551 >>> note.describe()
553 To upload note and sign off with prompt
555 >>> note.upload_note()
557 To upload note automatically without prompt
559 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
560 """
562 descriptions = [
563 'raw trial data does not exist',
564 'wheel data corrupt',
565 'task data could not be synced',
566 'stimulus timings unreliable'
567 ]
570class PassiveSignOffNote(SignOffNote):
572 """
573 Class for signing off a passive part of a session and optionally adding a related explanation note.
575 Examples
576 --------
577 >>> note = PassiveSignOffNote(eid, AlyxClient(), '_passiveChoiceWorld_00')
579 To sign off session without any note
581 >>> note.sign_off()
583 Print list of default reasons
585 >>> note.describe()
587 To upload note and sign off with prompt
589 >>> note.upload_note()
591 To upload note automatically without prompt
593 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
594 """
596 descriptions = [
597 'Raw passive data doesn’t exist (no. of spacers = 0)',
598 'Incorrect number or spacers (i.e passive cutoff midway)',
599 'RFmap file doesn’t exist',
600 'Gabor patches couldn’t be extracted',
601 'Trial playback couldn’t be extracted',
602 ]
605class VideoSignOffNote(SignOffNote):
607 """
608 Class for signing off a video part of a session and optionally adding a related explanation note.
610 Examples
611 --------
612 >>> note = VideoSignOffNote(eid, AlyxClient(), '_camera_left')
614 To sign off session without any note
616 >>> note.sign_off()
618 Print list of default reasons
620 >>> note.describe()
622 To upload note and sign off with prompt
624 >>> note.upload_note()
626 To upload note automatically without prompt
628 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
629 """
631 descriptions = [
632 'The video timestamps are not the same length as the video file (either empty or slightly longer/shorter)',
633 'The rotary encoder trace doesn’t not appear synced with the video',
634 'The QC fails because the GPIO file is missing or empty',
635 'The frame rate in the video header is wrong (the video plays too slow or fast)',
636 'The resolution is not what is defined in the experiment description file',
637 'The DLC QC fails because something is obscuring the pupil',
638 ]
641class RawEphysSignOffNote(SignOffNote):
643 """
644 Class for signing off a raw ephys part of a session and optionally adding a related explanation note.
646 Examples
647 --------
648 >>> note = RawEphysSignOffNote(uuid, AlyxClient(), '_neuropixel_raw_probe00')
650 To sign off session without any note
652 >>> note.sign_off()
654 Print list of default reasons
656 >>> note.describe()
658 To upload note and sign off with prompt
660 >>> note.upload_note()
662 To upload note automatically without prompt
664 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
665 """
667 descriptions = [
668 'Data has striping',
669 'Horizontal band',
670 'Discontunuity',
671 ]
674class SpikeSortingSignOffNote(SignOffNote):
676 """
677 Class for signing off a spike sorting part of a session and optionally adding a related explanation note.
679 Examples
680 --------
681 >>> note = SpikeSortingSignOffNote(uuid, AlyxClient(), '_neuropixel_spike_sorting_probe00')
683 To sign off session without any note
685 >>> note.sign_off()
687 Print list of default reasons
689 >>> note.describe()
691 To upload note and sign off with prompt
693 >>> note.upload_note()
695 To upload note automatically without prompt
697 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
698 """
700 descriptions = [
701 'Spikesorting could not be run',
702 'Poor quality spikesorting',
703 ]
706class AlignmentSignOffNote(SignOffNote):
708 """
709 Class for signing off a alignment part of a session and optionally adding a related explanation note.
711 Examples
712 --------
713 >>> note = AlignmentSignOffNote(uuid, AlyxClient(), '_neuropixel_alignment_probe00')
715 To sign off session without any note
717 >>> note.sign_off()
719 Print list of default reasons
721 >>> note.describe()
723 To upload note and sign off with prompt
725 >>> note.upload_note()
727 To upload note automatically without prompt
729 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
730 """
732 descriptions = []