Source code for ibllib.qc.critical_reasons

"""
Prompt experimenter for reason for marking session/insertion as CRITICAL
Choices are listed in the global variables. Multiple reasons can be selected.
Places info in Alyx session note in a format that is machine retrievable (text->json)
"""
import abc
import logging
import json
import warnings
from datetime import datetime
from one.api import OneAlyx
from one.webclient import AlyxClient

_logger = logging.getLogger('ibllib')


[docs] def main_gui(uuid, reasons_selected, one=None, alyx=None): """ Main function to call to input a reason for marking an insertion as CRITICAL from the alignment GUI. It wil create note text, after deleting any similar notes existing already. Parameters ---------- uuid : uuid.UUID, str An insertion ID. reasons_selected : list of str A subset of REASONS_INS_CRIT_GUI. one : one.api.OneAlyx (DEPRECATED) An instance of ONE. NB: Pass in an instance of AlyxClient instead. alyx : one.webclient.AlyxClient An AlyxClient instance. """ # hit the database to check if uuid is insertion uuid if alyx is None and one is not None: # Deprecate ONE in future because instantiating it takes longer and is unnecessary warnings.warn('In future please pass in an AlyxClient instance (i.e. `one.alyx`)', FutureWarning) alyx = one if isinstance(one, AlyxClient) else one.alyx ins_list = alyx.rest('insertions', 'list', id=uuid, no_cache=True) if len(ins_list) != 1: raise ValueError(f'N={len(ins_list)} insertion found, expected N=1. Check uuid provided.') note = CriticalInsertionNote(uuid, alyx) # assert that reasons are all within REASONS_INS_CRIT_GUI for item_str in reasons_selected: assert item_str in note.descriptions_gui note.selected_reasons = reasons_selected note.other_reason = [] note._upload_note(overwrite=True)
[docs] def main(uuid, one=None, alyx=None): """ Main function to call to input a reason for marking a session/insertion as CRITICAL programmatically. It will: - ask reasons for selection of critical status - check if 'other' reason has been selected, inquire why (free text) - create note text, checking whether similar notes exist already - upload note to Alyx if none exist previously or if overwrite is chosen Q&A are prompted via the Python terminal Parameters ---------- uuid : uuid.UUID, str An experiment UUID or an insertion UUID. one : one.api.OneAlyx (DEPRECATED) An instance of ONE. NB: Pass in an instance of AlyxClient instead. alyx : one.webclient.AlyxClient An AlyxClient instance. Examples -------- Retrieve Alyx note to test >>> alyx = AlyxClient(base_url='https://dev.alyx.internationalbrainlab.org') >>> uuid = '2ffd3ed5-477e-4153-9af7-7fdad3c6946b' >>> main(uuid=uuid, alyx=alyx) Get notes with pattern >>> notes = alyx.rest('notes', 'list', ... django=f'text__icontains,{STR_NOTES_STATIC},' ... f'object_id,{uuid}') >>> test_json_read = json.loads(notes[0]['text']) """ if alyx is None and one is not None: # Deprecate ONE in future because instantiating it takes longer and is unnecessary warnings.warn('In future please pass in an AlyxClient instance (i.e. `one.alyx`)', FutureWarning) alyx = one if isinstance(one, AlyxClient) else one.alyx if not alyx: alyx = AlyxClient() # ask reasons for selection of critical status # hit the database to know if uuid is insertion or session uuid sess_list = alyx.get('/sessions?&django=pk,' + uuid, clobber=True) ins_list = alyx.get('/insertions?&django=pk,' + uuid, clobber=True) if len(sess_list) > 0 and len(ins_list) == 0: # session note = CriticalSessionNote(uuid, alyx) elif len(ins_list) > 0 and len(sess_list) == 0: # insertion note = CriticalInsertionNote(uuid, alyx) else: raise ValueError(f'Inadequate number of session (n={len(sess_list)}) ' f'or insertion (n={len(ins_list)}) found for uuid {uuid}.' f'The query output should be of length 1.') note.upload_note()
[docs] class Note(abc.ABC): descriptions = [] @property def default_descriptions(self): return self.descriptions + ['Other'] @property def extra_prompt(self): return '' @property def note_title(self): return '' @property def n_description(self): return len(self.default_descriptions) def __init__(self, uuid, alyx, content_type=None): """ Base class for attaching notes to an alyx endpoint. Do not use this class directly but use parent classes that inherit this base class Parameters ---------- uuid : uuid.UUID, str A UUID of a session, insertion, or other Alyx model to attach note to. alyx : one.webclient.AlyxClient An AlyxClient instance. content_type : str The Alyx model name of the UUID. """ self.uuid = uuid if isinstance(alyx, OneAlyx): # Deprecate ONE in future because instantiating it takes longer and is unnecessary warnings.warn('In future please pass in an AlyxClient instance (i.e. `one.alyx`)', FutureWarning) alyx = alyx.alyx self.alyx = alyx self.selected_reasons = [] self.other_reason = [] if content_type is not None: self.content_type = content_type else: self.content_type = self.get_content_type()
[docs] def get_content_type(self): """ Infer the content_type from the uuid. Only checks to see if uuid is a session or insertion. If not recognised will raise an error and the content_type must be specified on note initialisation e.g. Note(uuid, alyx, content_type='subject') Returns ------- str The Alyx model name, inferred from the UUID. """ # see if it as session or an insertion if self.alyx.rest('sessions', 'list', id=self.uuid): content_type = 'session' elif self.alyx.rest('insertions', 'list', id=self.uuid): content_type = 'probeinsertion' else: raise ValueError(f'Content type cannot be recognised from {self.uuid}. ' 'Specify on initialistion e.g Note(uuid, alyx, content_type="subject"') return content_type
[docs] def describe(self): """ Print list of default reasons that can be chosen from :return: """ for i, d in enumerate(self.descriptions): print(f'{i}. {d} \n')
[docs] def numbered_descriptions(self): """ Return list of numbered default reasons :return: """ return [f'{i}) {d}' for i, d in enumerate(self.default_descriptions)]
[docs] def upload_note(self, nums=None, other_reason=None, **kwargs): """ Upload note to Alyx. If no values for nums and other_reason are specified, user will receive a prompt in command line asking them to choose from default list of reasons to add to note as well as option for free text. To upload without receiving prompt a value for either `nums` or `other_reason` must be given. Parameters ---------- nums : str string of numbers matching those in default descriptions, e.g, '1,3'. Options can be seen using note.describe(). other_reason : str Other comment or reason(s) to add to note. """ if nums is None and other_reason is None: self.selected_reasons, self.other_reason = self.reasons_prompt() else: self.selected_reasons = self._map_num_to_description(nums) self.other_reason = other_reason or [] self._upload_note(**kwargs)
def _upload_note(self, **kwargs): existing_note, notes = self._check_existing_note() if existing_note: self.update_existing_note(notes, **kwargs) else: text = self.format_note(**kwargs) self._create_note(text) _logger.info('The selected reasons were saved on Alyx.') def _create_note(self, text): data = {'user': self.alyx.user, 'content_type': self.content_type, 'object_id': self.uuid, 'text': f'{text}'} self.alyx.rest('notes', 'create', data=data) def _update_note(self, note_id, text): data = {'user': self.alyx.user, 'content_type': self.content_type, 'object_id': self.uuid, 'text': f'{text}'} self.alyx.rest('notes', 'partial_update', id=note_id, data=data) def _delete_note(self, note_id): self.alyx.rest('notes', 'delete', id=note_id) def _delete_notes(self, notes): for note in notes: self._delete_note(note['id']) def _check_existing_note(self): notes = self.alyx.rest('notes', 'list', django=f'text__icontains,{self.note_title},object_id,{self.uuid}', no_cache=True) if len(notes) == 0: return False, None else: return True, notes def _map_num_to_description(self, nums): if nums is None: return [] string_list = nums.split(',') int_list = list(map(int, string_list)) if max(int_list) >= self.n_description or min(int_list) < 0: raise ValueError(f'Chosen values out of range, must be between 0 and {self.n_description - 1}') return [self.default_descriptions[n] for n in int_list]
[docs] def reasons_prompt(self): """ Prompt for user to enter reasons :return: """ prompt = f'{self.extra_prompt} ' \ f'\n {self.numbered_descriptions()} \n ' \ f'and enter the corresponding numbers separated by commas, e.g. 1,3 -> enter: ' ans = input(prompt).strip().lower() try: selected_reasons = self._map_num_to_description(ans) print(f'You selected reason(s): {selected_reasons}') if 'Other' in selected_reasons: other_reasons = self.other_reason_prompt() return selected_reasons, other_reasons else: return selected_reasons, [] except ValueError: print(f'{ans} is invalid, please try again...') return self.reasons_prompt()
[docs] def other_reason_prompt(self): """ Prompt for user to enter other reasons :return: """ prompt = 'Explain why you selected "other" (free text): ' ans = input(prompt).strip().lower() return ans
[docs] @abc.abstractmethod def format_note(self, **kwargs): """ Method to format text field of note according to type of note wanting to be uploaded :param kwargs: :return: """
[docs] @abc.abstractmethod def update_existing_note(self, note, **kwargs): """ Method to specify behavior in the case of a note with the same title already existing :param note: :param kwargs: :return: """
[docs] class CriticalNote(Note): """ Class for uploading a critical note to a session or insertion. Do not use directly but use CriticalSessionNote or CriticalInsertionNote instead """
[docs] def format_note(self, **kwargs): note_text = { "title": self.note_title, "reasons_selected": self.selected_reasons, "reason_for_other": self.other_reason } return json.dumps(note_text)
[docs] def update_existing_note(self, notes, **kwargs): overwrite = kwargs.get('overwrite', None) if overwrite is None: overwrite = self.delete_note_prompt(notes) if overwrite: self._delete_notes(notes) text = self.format_note() self._create_note(text) _logger.info('The selected reasons were saved on Alyx; old notes were deleted') else: _logger.info('The selected reasons were NOT saved on Alyx; old notes remain.')
[docs] def delete_note_prompt(self, notes): prompt = f'You are about to delete {len(notes)} existing notes; ' \ f'do you want to proceed? y/n: ' ans = input(prompt).strip().lower() if ans not in ['y', 'n']: print(f'{ans} is invalid, please try again...') return self.delete_note_prompt(notes) else: return True if ans == 'y' else False
[docs] class CriticalInsertionNote(CriticalNote): """ Class for uploading a critical note to an insertion. Examples -------- >>> note = CriticalInsertionNote(pid, AlyxClient()) Print list of default reasons >>> note.describe() To receive a command line prompt to fill in note >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='lots of bad channels') """ descriptions_gui = [ 'Noise and artifact', 'Drift', 'Poor neural yield', 'Brain Damage', 'Other' ] descriptions = [ 'Histological images missing', 'Track not visible on imaging data' ] @property def default_descriptions(self): return self.descriptions + self.descriptions_gui @property def extra_prompt(self): return 'Select from this list the reason(s) why you are marking the insertion as CRITICAL:' @property def note_title(self): return '=== EXPERIMENTER REASON(S) FOR MARKING THE INSERTION AS CRITICAL ===' def __init__(self, uuid, alyx): super(CriticalInsertionNote, self).__init__(uuid, alyx, content_type='probeinsertion')
[docs] class CriticalSessionNote(CriticalNote): """ Class for uploading a critical note to a session. Example ------- >>> note = CriticalInsertionNote(uuid, AlyxClient) Print list of default reasons >>> note.describe() To receive a command line prompt to fill in note >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = [ 'within experiment system crash', 'synching impossible', 'dud or mock session', 'essential dataset missing', ] @property def extra_prompt(self): return 'Select from this list the reason(s) why you are marking the session as CRITICAL:' @property def note_title(self): return '=== EXPERIMENTER REASON(S) FOR MARKING THE SESSION AS CRITICAL ===' def __init__(self, uuid, alyx): super(CriticalSessionNote, self).__init__(uuid, alyx, content_type='session')
[docs] class SignOffNote(Note): """ Class for signing off a session and optionally adding a related explanation note. Do not use directly but use classes that inherit from this class e.g TaskSignOffNote, RawEphysSignOffNote """ @property def extra_prompt(self): return 'Select from this list the reason(s) that describe issues with this session:' @property def note_title(self): return f'=== SIGN-OFF NOTE FOR {self.sign_off_key} ===' def __init__(self, uuid, alyx, sign_off_key): self.sign_off_key = sign_off_key super(SignOffNote, self).__init__(uuid, alyx, content_type='session') self.datetime_key = self.get_datetime_key() self.session = self.alyx.rest('sessions', 'read', id=self.uuid, no_cache=True)
[docs] def upload_note(self, nums=None, other_reason=None, **kwargs): super(SignOffNote, self).upload_note(nums=nums, other_reason=other_reason, **kwargs) self.sign_off()
[docs] def sign_off(self): json = self.session['json'] sign_off_checklist = json.get('sign_off_checklist', None) if sign_off_checklist is None: sign_off_checklist = {self.sign_off_key: {'date': self.datetime_key.split('_')[0], 'user': self.datetime_key.split('_')[1]}} else: sign_off_checklist[self.sign_off_key] = {'date': self.datetime_key.split('_')[0], 'user': self.datetime_key.split('_')[1]} json['sign_off_checklist'] = sign_off_checklist self.alyx.json_field_update("sessions", self.uuid, 'json', data=json)
[docs] def format_note(self, **kwargs): note_text = { "title": self.note_title, f'{self.datetime_key}': {"reasons_selected": self.selected_reasons, "reason_for_other": self.other_reason} } return json.dumps(note_text)
[docs] def format_existing_note(self, orignal_note): extra_note = {f'{self.datetime_key}': {"reasons_selected": self.selected_reasons, "reason_for_other": self.other_reason} } orignal_note.update(extra_note) return json.dumps(orignal_note)
[docs] def update_existing_note(self, notes): if len(notes) != 1: raise ValueError(f'{len(notes)} with same title found, only expect at most 1. Clean up before proceeding') else: original_note = json.loads(notes[0]['text']) text = self.format_existing_note(original_note) self._update_note(notes[0]['id'], text)
[docs] def get_datetime_key(self): if not self.alyx.is_logged_in: self.alyx.authenticate() assert self.alyx.is_logged_in, 'you must be logged in to the AlyxClient' user = self.alyx.user date = datetime.now().date().isoformat() return date + '_' + user
[docs] class TaskSignOffNote(SignOffNote): """ Class for signing off a task part of a session and optionally adding a related explanation note. Examples -------- >>> note = TaskSignOffNote(eid, AlyxClient(), '_ephysChoiceWorld_00') To sign off session without any note >>> note.sign_off() Print list of default reasons >>> note.describe() To upload note and sign off with prompt >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = [ 'raw trial data does not exist', 'wheel data corrupt', 'task data could not be synced', 'stimulus timings unreliable' ]
[docs] class PassiveSignOffNote(SignOffNote): """ Class for signing off a passive part of a session and optionally adding a related explanation note. Examples -------- >>> note = PassiveSignOffNote(eid, AlyxClient(), '_passiveChoiceWorld_00') To sign off session without any note >>> note.sign_off() Print list of default reasons >>> note.describe() To upload note and sign off with prompt >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = [ 'Raw passive data doesn’t exist (no. of spacers = 0)', 'Incorrect number or spacers (i.e passive cutoff midway)', 'RFmap file doesn’t exist', 'Gabor patches couldn’t be extracted', 'Trial playback couldn’t be extracted', ]
[docs] class VideoSignOffNote(SignOffNote): """ Class for signing off a video part of a session and optionally adding a related explanation note. Examples -------- >>> note = VideoSignOffNote(eid, AlyxClient(), '_camera_left') To sign off session without any note >>> note.sign_off() Print list of default reasons >>> note.describe() To upload note and sign off with prompt >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = [ 'The video timestamps are not the same length as the video file (either empty or slightly longer/shorter)', 'The rotary encoder trace doesn’t not appear synced with the video', 'The QC fails because the GPIO file is missing or empty', 'The frame rate in the video header is wrong (the video plays too slow or fast)', 'The resolution is not what is defined in the experiment description file', 'The DLC QC fails because something is obscuring the pupil', ]
[docs] class RawEphysSignOffNote(SignOffNote): """ Class for signing off a raw ephys part of a session and optionally adding a related explanation note. Examples -------- >>> note = RawEphysSignOffNote(uuid, AlyxClient(), '_neuropixel_raw_probe00') To sign off session without any note >>> note.sign_off() Print list of default reasons >>> note.describe() To upload note and sign off with prompt >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = [ 'Data has striping', 'Horizontal band', 'Discontunuity', ]
[docs] class SpikeSortingSignOffNote(SignOffNote): """ Class for signing off a spike sorting part of a session and optionally adding a related explanation note. Examples -------- >>> note = SpikeSortingSignOffNote(uuid, AlyxClient(), '_neuropixel_spike_sorting_probe00') To sign off session without any note >>> note.sign_off() Print list of default reasons >>> note.describe() To upload note and sign off with prompt >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = [ 'Spikesorting could not be run', 'Poor quality spikesorting', ]
[docs] class AlignmentSignOffNote(SignOffNote): """ Class for signing off a alignment part of a session and optionally adding a related explanation note. Examples -------- >>> note = AlignmentSignOffNote(uuid, AlyxClient(), '_neuropixel_alignment_probe00') To sign off session without any note >>> note.sign_off() Print list of default reasons >>> note.describe() To upload note and sign off with prompt >>> note.upload_note() To upload note automatically without prompt >>> note.upload_note(nums='1,4', other_reason='session with no ephys recording') """ descriptions = []