Coverage for ibllib/qc/critical_reasons.py: 88%
223 statements
« prev ^ index » next coverage.py v7.7.0, created at 2025-03-17 15:25 +0000
1"""
2Methods for adding QC sign-off notes to Alyx.
4Includes a GUI to prompt experimenter for reason for marking session/insertion as CRITICAL.
5Choices are listed in the global variables. Multiple reasons can be selected.
6Places info in Alyx session note in a format that is machine retrievable (text->json).
7"""
8import abc
9import logging
10import json
11from datetime import datetime
12from one.webclient import AlyxClient
13from one.alf.spec import is_uuid
15_logger = logging.getLogger('ibllib')
18def main_gui(uuid, reasons_selected, alyx=None):
19 """
20 Main function to call to input a reason for marking an insertion as CRITICAL from the alignment GUI.
22 It wil create note text, after deleting any similar notes existing already.
24 Parameters
25 ----------
26 uuid : uuid.UUID, str
27 An insertion ID.
28 reasons_selected : list of str
29 A subset of REASONS_INS_CRIT_GUI.
30 alyx : one.webclient.AlyxClient
31 An AlyxClient instance.
32 """
33 # hit the database to check if uuid is insertion uuid
34 alyx = alyx or AlyxClient() 1g
35 ins_list = alyx.rest('insertions', 'list', id=uuid, no_cache=True) 1g
36 if len(ins_list) != 1: 1g
37 raise ValueError(f'N={len(ins_list)} insertion found, expected N=1. Check uuid provided.')
39 note = CriticalInsertionNote(uuid, alyx) 1g
41 # assert that reasons are all within REASONS_INS_CRIT_GUI
42 for item_str in reasons_selected: 1g
43 assert item_str in note.descriptions_gui 1g
45 note.selected_reasons = reasons_selected 1g
46 note.other_reason = [] 1g
47 note._upload_note(overwrite=True) 1g
50def main(uuid, alyx=None):
51 """
52 Main function to call to input a reason for marking a session/insertion as CRITICAL programmatically.
54 It will:
55 - ask reasons for selection of critical status
56 - check if 'other' reason has been selected, inquire why (free text)
57 - create note text, checking whether similar notes exist already
58 - upload note to Alyx if none exist previously or if overwrite is chosen Q&A are prompted via the Python terminal
60 Parameters
61 ----------
62 uuid : uuid.UUID, str
63 An experiment UUID or an insertion UUID.
64 alyx : one.webclient.AlyxClient
65 An AlyxClient instance.
67 Examples
68 --------
69 Retrieve Alyx note to test
71 >>> alyx = AlyxClient(base_url='https://dev.alyx.internationalbrainlab.org')
72 >>> uuid = '2ffd3ed5-477e-4153-9af7-7fdad3c6946b'
73 >>> main(uuid=uuid, alyx=alyx)
75 Get notes with pattern
77 >>> notes = alyx.rest('notes', 'list',
78 ... django=f'text__icontains,{STR_NOTES_STATIC},'
79 ... f'object_id,{uuid}')
80 >>> test_json_read = json.loads(notes[0]['text'])
82 """
83 alyx = alyx or AlyxClient() 1cef
84 # ask reasons for selection of critical status
86 # hit the database to know if uuid is insertion or session uuid
87 sess_list = alyx.get('/sessions?&django=pk,' + str(uuid), clobber=True) 1cef
88 ins_list = alyx.get('/insertions?&django=pk,' + str(uuid), clobber=True) 1cef
90 if len(sess_list) > 0 and len(ins_list) == 0: # session 1cef
91 note = CriticalSessionNote(uuid, alyx) 1cf
92 elif len(ins_list) > 0 and len(sess_list) == 0: # insertion 1e
93 note = CriticalInsertionNote(uuid, alyx) 1e
94 else:
95 raise ValueError(f'Inadequate number of session (n={len(sess_list)}) '
96 f'or insertion (n={len(ins_list)}) found for uuid {uuid}.'
97 f'The query output should be of length 1.')
99 note.upload_note() 1cef
102class Note(abc.ABC):
103 descriptions = []
105 @property
106 def default_descriptions(self):
107 return self.descriptions + ['Other'] 1bdcf
109 @property
110 def extra_prompt(self):
111 return ''
113 @property
114 def note_title(self):
115 return ''
117 @property
118 def n_description(self):
119 return len(self.default_descriptions) 1bdcef
121 def __init__(self, uuid, alyx, content_type=None):
122 """
123 Base class for attaching notes to an alyx endpoint. Do not use this class directly but use parent classes that inherit
124 this base class
126 Parameters
127 ----------
128 uuid : uuid.UUID, str
129 A UUID of a session, insertion, or other Alyx model to attach note to.
130 alyx : one.webclient.AlyxClient
131 An AlyxClient instance.
132 content_type : str
133 The Alyx model name of the UUID.
134 """
135 if not is_uuid(uuid, versions=(4,)): 1hbdgcef
136 raise ValueError('Expected `uuid` to be a UUIDv4 object')
137 self.uuid = uuid 1hbdgcef
138 self.alyx = alyx 1hbdgcef
139 self.selected_reasons = [] 1hbdgcef
140 self.other_reason = [] 1hbdgcef
141 if content_type is not None: 1hbdgcef
142 self.content_type = content_type 1hbdgcef
143 else:
144 self.content_type = self.get_content_type()
146 def get_content_type(self):
147 """
148 Infer the content_type from the uuid. Only checks to see if uuid is a session or insertion.
149 If not recognised will raise an error and the content_type must be specified on note
150 initialisation e.g. Note(uuid, alyx, content_type='subject')
152 Returns
153 -------
154 str
155 The Alyx model name, inferred from the UUID.
156 """
158 # see if it as session or an insertion
159 if self.alyx.rest('sessions', 'list', id=self.uuid):
160 content_type = 'session'
161 elif self.alyx.rest('insertions', 'list', id=self.uuid):
162 content_type = 'probeinsertion'
163 else:
164 raise ValueError(f'Content type cannot be recognised from {self.uuid}. '
165 'Specify on initialistion e.g Note(uuid, alyx, content_type="subject"')
166 return content_type
168 def describe(self):
169 """
170 Print list of default reasons that can be chosen from
171 :return:
172 """
173 for i, d in enumerate(self.descriptions):
174 print(f'{i}. {d} \n')
176 def numbered_descriptions(self):
177 """
178 Return list of numbered default reasons
179 :return:
180 """
181 return [f'{i}) {d}' for i, d in enumerate(self.default_descriptions)] 1bdcef
183 def upload_note(self, nums=None, other_reason=None, **kwargs):
184 """
185 Upload note to Alyx.
187 If no values for nums and other_reason are specified, user will receive a prompt in command
188 line asking them to choose from default list of reasons to add to note as well as option
189 for free text. To upload without receiving prompt a value for either `nums` or
190 `other_reason` must be given.
192 Parameters
193 ----------
194 nums : str
195 string of numbers matching those in default descriptions, e.g, '1,3'. Options can be
196 seen using note.describe().
197 other_reason : str
198 Other comment or reason(s) to add to note.
200 """
202 if nums is None and other_reason is None: 1bdcef
203 self.selected_reasons, self.other_reason = self.reasons_prompt() 1bdcef
204 else:
205 self.selected_reasons = self._map_num_to_description(nums) 1b
206 self.other_reason = other_reason or [] 1b
208 self._upload_note(**kwargs) 1bdcef
210 def _upload_note(self, **kwargs):
211 existing_note, notes = self._check_existing_note() 1bdgcef
212 if existing_note: 1bdgcef
213 self.update_existing_note(notes, **kwargs) 1bc
214 else:
215 text = self.format_note(**kwargs) 1bdgcef
216 self._create_note(text) 1bdgcef
217 _logger.info('The selected reasons were saved on Alyx.') 1bdgcef
219 def _create_note(self, text):
221 data = {'user': self.alyx.user, 1bdgcef
222 'content_type': self.content_type,
223 'object_id': self.uuid,
224 'text': f'{text}'}
225 self.alyx.rest('notes', 'create', data=data) 1bdgcef
227 def _update_note(self, note_id, text):
229 data = {'user': self.alyx.user, 1b
230 'content_type': self.content_type,
231 'object_id': self.uuid,
232 'text': f'{text}'}
233 self.alyx.rest('notes', 'partial_update', id=note_id, data=data) 1b
235 def _delete_note(self, note_id):
236 self.alyx.rest('notes', 'delete', id=note_id) 1c
238 def _delete_notes(self, notes):
239 for note in notes: 1c
240 self._delete_note(note['id']) 1c
242 def _check_existing_note(self):
243 query = f'text__icontains,{self.note_title},object_id,{str(self.uuid)}' 1bdgcef
244 notes = self.alyx.rest('notes', 'list', django=query, no_cache=True) 1bdgcef
245 if len(notes) == 0: 1bdgcef
246 return False, None 1bdgcef
247 else:
248 return True, notes 1bc
250 def _map_num_to_description(self, nums):
252 if nums is None: 1bdcef
253 return []
255 string_list = nums.split(',') 1bdcef
256 int_list = list(map(int, string_list)) 1bdcef
258 if max(int_list) >= self.n_description or min(int_list) < 0: 1bdcef
259 raise ValueError(f'Chosen values out of range, must be between 0 and {self.n_description - 1}')
261 return [self.default_descriptions[n] for n in int_list] 1bdcef
263 def reasons_prompt(self):
264 """
265 Prompt for user to enter reasons
266 :return:
267 """
269 prompt = f'{self.extra_prompt} ' \ 1bdcef
270 f'\n {self.numbered_descriptions()} \n ' \
271 f'and enter the corresponding numbers separated by commas, e.g. 1,3 -> enter: '
273 ans = input(prompt).strip().lower() 1bdcef
275 try: 1bdcef
276 selected_reasons = self._map_num_to_description(ans) 1bdcef
277 print(f'You selected reason(s): {selected_reasons}') 1bdcef
278 if 'Other' in selected_reasons: 1bdcef
279 other_reasons = self.other_reason_prompt() 1bdcef
280 return selected_reasons, other_reasons 1bdcef
281 else:
282 return selected_reasons, []
284 except ValueError:
285 print(f'{ans} is invalid, please try again...')
286 return self.reasons_prompt()
288 def other_reason_prompt(self):
289 """
290 Prompt for user to enter other reasons
291 :return:
292 """
294 prompt = 'Explain why you selected "other" (free text): ' 1bdcef
295 ans = input(prompt).strip().lower() 1bdcef
296 return ans 1bdcef
298 @abc.abstractmethod
299 def format_note(self, **kwargs):
300 """
301 Method to format text field of note according to type of note wanting to be uploaded
302 :param kwargs:
303 :return:
304 """
306 @abc.abstractmethod
307 def update_existing_note(self, note, **kwargs):
308 """
309 Method to specify behavior in the case of a note with the same title already existing
310 :param note:
311 :param kwargs:
312 :return:
313 """
316class CriticalNote(Note):
317 """
318 Class for uploading a critical note to a session or insertion. Do not use directly but use CriticalSessionNote or
319 CriticalInsertionNote instead
320 """
322 def format_note(self, **kwargs):
323 note_text = { 1gcef
324 "title": self.note_title,
325 "reasons_selected": self.selected_reasons,
326 "reason_for_other": self.other_reason
327 }
328 return json.dumps(note_text) 1gcef
330 def update_existing_note(self, notes, **kwargs):
332 overwrite = kwargs.get('overwrite', None) 1c
333 if overwrite is None: 1c
334 overwrite = self.delete_note_prompt(notes) 1c
336 if overwrite: 1c
337 self._delete_notes(notes) 1c
338 text = self.format_note() 1c
339 self._create_note(text) 1c
340 _logger.info('The selected reasons were saved on Alyx; old notes were deleted') 1c
341 else:
342 _logger.info('The selected reasons were NOT saved on Alyx; old notes remain.')
344 def delete_note_prompt(self, notes):
346 prompt = f'You are about to delete {len(notes)} existing notes; ' \ 1c
347 f'do you want to proceed? y/n: '
349 ans = input(prompt).strip().lower() 1c
351 if ans not in ['y', 'n']: 1c
352 print(f'{ans} is invalid, please try again...')
353 return self.delete_note_prompt(notes)
354 else:
355 return True if ans == 'y' else False 1c
358class CriticalInsertionNote(CriticalNote):
359 """
360 Class for uploading a critical note to an insertion.
362 Examples
363 --------
364 >>> note = CriticalInsertionNote(pid, AlyxClient())
366 Print list of default reasons
368 >>> note.describe()
370 To receive a command line prompt to fill in note
372 >>> note.upload_note()
374 To upload note automatically without prompt
376 >>> note.upload_note(nums='1,4', other_reason='lots of bad channels')
377 """
379 descriptions_gui = [
380 'Noise and artifact',
381 'Drift',
382 'Poor neural yield',
383 'Brain Damage',
384 'Other'
385 ]
387 descriptions = [
388 'Histological images missing',
389 'Track not visible on imaging data'
390 ]
392 @property
393 def default_descriptions(self):
394 return self.descriptions + self.descriptions_gui 1e
396 @property
397 def extra_prompt(self):
398 return 'Select from this list the reason(s) why you are marking the insertion as CRITICAL:' 1e
400 @property
401 def note_title(self):
402 return '=== EXPERIMENTER REASON(S) FOR MARKING THE INSERTION AS CRITICAL ===' 1ge
404 def __init__(self, uuid, alyx):
405 super(CriticalInsertionNote, self).__init__(uuid, alyx, content_type='probeinsertion') 1ge
408class CriticalSessionNote(CriticalNote):
409 """
410 Class for uploading a critical note to a session.
412 Example
413 -------
414 >>> note = CriticalInsertionNote(uuid, AlyxClient)
416 Print list of default reasons
418 >>> note.describe()
420 To receive a command line prompt to fill in note
422 >>> note.upload_note()
424 To upload note automatically without prompt
426 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
427 """
429 descriptions = [
430 'within experiment system crash',
431 'synching impossible',
432 'dud or mock session',
433 'essential dataset missing',
434 ]
436 @property
437 def extra_prompt(self):
438 return 'Select from this list the reason(s) why you are marking the session as CRITICAL:' 1cf
440 @property
441 def note_title(self):
442 return '=== EXPERIMENTER REASON(S) FOR MARKING THE SESSION AS CRITICAL ===' 1cf
444 def __init__(self, uuid, alyx):
445 super(CriticalSessionNote, self).__init__(uuid, alyx, content_type='session') 1cf
448class SignOffNote(Note):
449 """
450 Class for signing off a session and optionally adding a related explanation note.
451 Do not use directly but use classes that inherit from this class e.g TaskSignOffNote, RawEphysSignOffNote
452 """
454 @property
455 def extra_prompt(self):
456 return 'Select from this list the reason(s) that describe issues with this session:' 1bd
458 @property
459 def note_title(self):
460 return f'=== SIGN-OFF NOTE FOR {self.sign_off_key} ===' 1bd
462 def __init__(self, uuid, alyx, sign_off_key):
463 self.sign_off_key = sign_off_key 1hbd
464 super(SignOffNote, self).__init__(uuid, alyx, content_type='session') 1hbd
465 self.datetime_key = self.get_datetime_key() 1hbd
466 self.session = self.alyx.rest('sessions', 'read', id=self.uuid, no_cache=True) 1hbd
468 def upload_note(self, nums=None, other_reason=None, **kwargs):
469 super(SignOffNote, self).upload_note(nums=nums, other_reason=other_reason, **kwargs) 1bd
470 self.sign_off() 1bd
472 def sign_off(self):
474 json = self.session['json'] 1hbd
475 sign_off_checklist = json.get('sign_off_checklist', None) 1hbd
476 if sign_off_checklist is None: 1hbd
477 sign_off_checklist = {self.sign_off_key: {'date': self.datetime_key.split('_')[0],
478 'user': self.datetime_key.split('_')[1]}}
479 else:
480 sign_off_checklist[self.sign_off_key] = {'date': self.datetime_key.split('_')[0], 1hbd
481 'user': self.datetime_key.split('_')[1]}
483 json['sign_off_checklist'] = sign_off_checklist 1hbd
485 self.alyx.json_field_update("sessions", self.uuid, 'json', data=json) 1hbd
487 def format_note(self, **kwargs):
489 note_text = { 1bd
490 "title": self.note_title,
491 f'{self.datetime_key}': {"reasons_selected": self.selected_reasons,
492 "reason_for_other": self.other_reason}
493 }
495 return json.dumps(note_text) 1bd
497 def format_existing_note(self, orignal_note):
499 extra_note = {f'{self.datetime_key}': {"reasons_selected": self.selected_reasons, 1b
500 "reason_for_other": self.other_reason}
501 }
503 orignal_note.update(extra_note) 1b
505 return json.dumps(orignal_note) 1b
507 def update_existing_note(self, notes):
508 if len(notes) != 1: 1b
509 raise ValueError(f'{len(notes)} with same title found, only expect at most 1. Clean up before proceeding')
510 else:
511 original_note = json.loads(notes[0]['text']) 1b
512 text = self.format_existing_note(original_note) 1b
513 self._update_note(notes[0]['id'], text) 1b
515 def get_datetime_key(self):
516 if not self.alyx.is_logged_in: 1hbd
517 self.alyx.authenticate()
518 assert self.alyx.is_logged_in, 'you must be logged in to the AlyxClient'
519 user = self.alyx.user 1hbd
520 date = datetime.now().date().isoformat() 1hbd
521 return date + '_' + user 1hbd
524class TaskSignOffNote(SignOffNote):
526 """
527 Class for signing off a task part of a session and optionally adding a related explanation note.
529 Examples
530 --------
531 >>> note = TaskSignOffNote(eid, AlyxClient(), '_ephysChoiceWorld_00')
533 To sign off session without any note
535 >>> note.sign_off()
537 Print list of default reasons
539 >>> note.describe()
541 To upload note and sign off with prompt
543 >>> note.upload_note()
545 To upload note automatically without prompt
547 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
548 """
550 descriptions = [
551 'raw trial data does not exist',
552 'wheel data corrupt',
553 'task data could not be synced',
554 'stimulus timings unreliable'
555 ]
558class PassiveSignOffNote(SignOffNote):
560 """
561 Class for signing off a passive part of a session and optionally adding a related explanation note.
563 Examples
564 --------
565 >>> note = PassiveSignOffNote(eid, AlyxClient(), '_passiveChoiceWorld_00')
567 To sign off session without any note
569 >>> note.sign_off()
571 Print list of default reasons
573 >>> note.describe()
575 To upload note and sign off with prompt
577 >>> note.upload_note()
579 To upload note automatically without prompt
581 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
582 """
584 descriptions = [
585 'Raw passive data doesn’t exist (no. of spacers = 0)',
586 'Incorrect number or spacers (i.e passive cutoff midway)',
587 'RFmap file doesn’t exist',
588 'Gabor patches couldn’t be extracted',
589 'Trial playback couldn’t be extracted',
590 ]
593class VideoSignOffNote(SignOffNote):
595 """
596 Class for signing off a video part of a session and optionally adding a related explanation note.
598 Examples
599 --------
600 >>> note = VideoSignOffNote(eid, AlyxClient(), '_camera_left')
602 To sign off session without any note
604 >>> note.sign_off()
606 Print list of default reasons
608 >>> note.describe()
610 To upload note and sign off with prompt
612 >>> note.upload_note()
614 To upload note automatically without prompt
616 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
617 """
619 descriptions = [
620 'The video timestamps are not the same length as the video file (either empty or slightly longer/shorter)',
621 'The rotary encoder trace doesn’t not appear synced with the video',
622 'The QC fails because the GPIO file is missing or empty',
623 'The frame rate in the video header is wrong (the video plays too slow or fast)',
624 'The resolution is not what is defined in the experiment description file',
625 'The DLC QC fails because something is obscuring the pupil',
626 ]
629class RawEphysSignOffNote(SignOffNote):
631 """
632 Class for signing off a raw ephys part of a session and optionally adding a related explanation note.
634 Examples
635 --------
636 >>> note = RawEphysSignOffNote(uuid, AlyxClient(), '_neuropixel_raw_probe00')
638 To sign off session without any note
640 >>> note.sign_off()
642 Print list of default reasons
644 >>> note.describe()
646 To upload note and sign off with prompt
648 >>> note.upload_note()
650 To upload note automatically without prompt
652 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
653 """
655 descriptions = [
656 'Data has striping',
657 'Horizontal band',
658 'Discontunuity',
659 ]
662class SpikeSortingSignOffNote(SignOffNote):
664 """
665 Class for signing off a spike sorting part of a session and optionally adding a related explanation note.
667 Examples
668 --------
669 >>> note = SpikeSortingSignOffNote(uuid, AlyxClient(), '_neuropixel_spike_sorting_probe00')
671 To sign off session without any note
673 >>> note.sign_off()
675 Print list of default reasons
677 >>> note.describe()
679 To upload note and sign off with prompt
681 >>> note.upload_note()
683 To upload note automatically without prompt
685 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
686 """
688 descriptions = [
689 'Spikesorting could not be run',
690 'Poor quality spikesorting',
691 ]
694class AlignmentSignOffNote(SignOffNote):
696 """
697 Class for signing off a alignment part of a session and optionally adding a related explanation note.
699 Examples
700 --------
701 >>> note = AlignmentSignOffNote(uuid, AlyxClient(), '_neuropixel_alignment_probe00')
703 To sign off session without any note
705 >>> note.sign_off()
707 Print list of default reasons
709 >>> note.describe()
711 To upload note and sign off with prompt
713 >>> note.upload_note()
715 To upload note automatically without prompt
717 >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording')
718 """
720 descriptions = []