7961675356254463

Build Type

continuous-integration/UCL

Committed by olivier

Commit Message

add test

Pull Request Pull Request #557: Chained protocols

Run Details

718 of 718 new or added lines in 27 files covered. (100.0%)

12554 of 18072 relevant lines covered (69.47%)

0.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.94

/ibllib/qc/critical_reasons.py

"""
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)
"""
# Author: Gaelle
import abc
import logging
import json
from requests.exceptions import HTTPError
from datetime import datetime
from one.api import ONE

_logger = logging.getLogger("ibllib")


def main_gui(uuid, reasons_selected, one=None):
    """
    Main function to call to input a reason for marking an insertion as
    CRITICAL from the alignment GUI. It will:
    - create note text, after deleting any similar notes existing already

    :param: uuid: insertion id
    :param: reasons_selected: list of str, str are picked within REASONS_INS_CRIT_GUI
    """
    # hit the database to check if uuid is insertion uuid
    ins_list = one.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, one)

    # 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)


def main(uuid, one=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.

    Example:
    # Retrieve Alyx note to test
    one = ONE(base_url='https://dev.alyx.internationalbrainlab.org')
    uuid = '2ffd3ed5-477e-4153-9af7-7fdad3c6946b'
    main(uuid=uuid, one=one)

    # Get notes with pattern
    notes = one.alyx.rest('notes', 'list',
                          django=f'text__icontains,{STR_NOTES_STATIC},'
                                 f'object_id,{uuid}')
    test_json_read = json.loads(notes[0]['text'])

    :param uuid: session/insertion uuid
    :param one: default: None -> ONE()
    :return:
    """
    if one is None:
        one = ONE()
    # ask reasons for selection of critical status

    # hit the database to know if uuid is insertion or session uuid
    sess_list = one.alyx.get('/sessions?&django=pk,' + uuid, clobber=True)
    ins_list = one.alyx.get('/insertions?&django=pk,' + uuid, clobber=True)

    if len(sess_list) > 0 and len(ins_list) == 0:  # session
        note = CriticalSessionNote(uuid, one)
    elif len(ins_list) > 0 and len(sess_list) == 0:  # insertion
        note = CriticalInsertionNote(uuid, one)
    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()


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, one, 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

        :param uuid: uuid of session/ insertion or other model to attach note to
        :param one: ONE instance
        :param content_type: alyx endpoint of uuid
        """
        self.uuid = uuid
        self.one = one
        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()

    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, one, content_type='subject')
        :return:
        """

        # see if it as session or an insertion
        if self.one.uuid2path(self.uuid):
            content_type = 'session'
        else:
            try:
                _ = self.one.pid2uuid(self.uuid)
                content_type = 'probeinsertion'
            except HTTPError:
                raise ValueError('Content type cannot be recognised from {uuid}. Specify on '
                                 'initialistion e.g Note(uuid, one, content_type="subject"')

        return content_type

    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')

    def numbered_descriptions(self):
        """
        Return list of numbered default reasons
        :return:
        """
        return [f'{i}) {d}' for i, d in enumerate(self.default_descriptions)]

    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

        :param nums: string of numbers matching those in default descrptions, e.g, '1,3'. Options can be see using note.describe()
        :param other_reason: other comment or reasons to add to note (string)
        :param kwargs:
        :return:
        """

        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.one.alyx.user,
                'content_type': self.content_type,
                'object_id': self.uuid,
                'text': f'{text}'}
        self.one.alyx.rest('notes', 'create', data=data)

    def _update_note(self, note_id, text):

        data = {'user': self.one.alyx.user,
                'content_type': self.content_type,
                'object_id': self.uuid,
                'text': f'{text}'}
        self.one.alyx.rest('notes', 'partial_update', id=note_id, data=data)

    def _delete_note(self, note_id):
        self.one.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.one.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]

    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()

    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

    @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:
        """

    @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:
        """


class CriticalNote(Note):
    """
    Class for uploading a critical note to a session or insertion. Do not use directly but use CriticalSessionNote or
    CriticalInsertionNote instead
    """

    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)

    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.')

    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


class CriticalInsertionNote(CriticalNote):
    """
    Class for uploading a critical note to an insertion.

    Example
    -------
    note = CriticalInsertionNote(pid, one)
    # 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, one):
        super(CriticalInsertionNote, self).__init__(uuid, one, content_type='probeinsertion')


class CriticalSessionNote(CriticalNote):
    """
    Class for uploading a critical note to a session.

    Example
    -------
    note = CriticalInsertionNote(uuid, one)
    # 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, one):
        super(CriticalSessionNote, self).__init__(uuid, one, content_type='session')


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, one, sign_off_key):
        self.sign_off_key = sign_off_key
        super(SignOffNote, self).__init__(uuid, one, content_type='session')
        self.datetime_key = self.get_datetime_key()
        self.session = one.alyx.rest('sessions', 'read', id=self.uuid, no_cache=True)

    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()

    def sign_off(self):

        json = self.session['json']
        json['sign_off_checklist'][self.sign_off_key] = {'date': self.datetime_key.split('_')[0],
                                                         'user': self.datetime_key.split('_')[1]}

        self.one.alyx.json_field_update("sessions", self.uuid, 'json', data=json)

    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)

    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)

    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)

    def get_datetime_key(self):
        user = self.one.alyx.user
        date = datetime.now().date().isoformat()
        return date + '_' + user


class TaskSignOffNote(SignOffNote):

    """
    Class for signing off a task part of a session and optionally adding a related explanation note.

    Example
    -------
    note = TaskSignOffNote(uuid, one, '_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',
    ]


class PassiveSignOffNote(SignOffNote):

    """
    Class for signing off a passive part of a session and optionally adding a related explanation note.

    Example
    -------
    note = PassiveSignOffNote(uuid, one, '_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',
    ]


class VideoSignOffNote(SignOffNote):

    """
    Class for signing off a video part of a session and optionally adding a related explanation note.

    Example
    -------
    note = VideoSignOffNote(uuid, one, '_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',
    ]


class RawEphysSignOffNote(SignOffNote):

    """
    Class for signing off a raw ephys part of a session and optionally adding a related explanation note.

    Example
    -------
    note = RawEphysSignOffNote(uuid, one, '_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',
    ]


class SpikeSortingSignOffNote(SignOffNote):

    """
    Class for signing off a spikesorting part of a session and optionally adding a related explanation note.

    Example
    -------
    note = SpikeSortingSignOffNote(uuid, one, '_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',
    ]


class AlignmentSignOffNote(SignOffNote):

    """
    Class for signing off a alignment part of a session and optionally adding a related explanation note.

    Example
    -------
    note = AlignmentSignOffNote(uuid, one, '_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 = []

1	"""	1✔
2	Prompt experimenter for reason for marking session/insertion as CRITICAL
3	Choices are listed in the global variables. Multiple reasons can be selected.
4	Places info in Alyx session note in a format that is machine retrievable (text->json)
5	"""
6	# Author: Gaelle
7	import abc	1✔
8	import logging	1✔
9	import json	1✔
10	from requests.exceptions import HTTPError	1✔
11	from datetime import datetime	1✔
12	from one.api import ONE	1✔
13
14	_logger = logging.getLogger("ibllib")	1✔
15
16
17	def main_gui(uuid, reasons_selected, one=None):	1✔
18	"""
19	Main function to call to input a reason for marking an insertion as
20	CRITICAL from the alignment GUI. It will:
21	- create note text, after deleting any similar notes existing already
22
23	:param: uuid: insertion id
24	:param: reasons_selected: list of str, str are picked within REASONS_INS_CRIT_GUI
25	"""
26	# hit the database to check if uuid is insertion uuid
27	ins_list = one.alyx.rest('insertions', 'list', id=uuid, no_cache=True)	1✔
28	if len(ins_list) != 1:	1✔
29	raise ValueError(f'N={len(ins_list)} insertion found, expected N=1. Check uuid provided.')	×
30
31	note = CriticalInsertionNote(uuid, one)	1✔
32
33	# assert that reasons are all within REASONS_INS_CRIT_GUI
34	for item_str in reasons_selected:	1✔
35	assert item_str in note.descriptions_gui	1✔
36
37	note.selected_reasons = reasons_selected	1✔
38	note.other_reason = []	1✔
39	note._upload_note(overwrite=True)	1✔
40
41
42	def main(uuid, one=None):	1✔
43	"""
44	Main function to call to input a reason for marking a session/insertion
45	as CRITICAL programmatically. It will:
46	- ask reasons for selection of critical status
47	- check if 'other' reason has been selected, inquire why (free text)
48	- create note text, checking whether similar notes exist already
49	- upload note to Alyx if none exist previously or if overwrite is chosen
50	Q&A are prompted via the Python terminal.
51
52	Example:
53	# Retrieve Alyx note to test
54	one = ONE(base_url='https://dev.alyx.internationalbrainlab.org')
55	uuid = '2ffd3ed5-477e-4153-9af7-7fdad3c6946b'
56	main(uuid=uuid, one=one)
57
58	# Get notes with pattern
59	notes = one.alyx.rest('notes', 'list',
60	django=f'text__icontains,{STR_NOTES_STATIC},'
61	f'object_id,{uuid}')
62	test_json_read = json.loads(notes[0]['text'])
63
64	:param uuid: session/insertion uuid
65	:param one: default: None -> ONE()
66	:return:
67	"""
68	if one is None:	1✔
69	one = ONE()	×
70	# ask reasons for selection of critical status
71
72	# hit the database to know if uuid is insertion or session uuid
73	sess_list = one.alyx.get('/sessions?&django=pk,' + uuid, clobber=True)	1✔
74	ins_list = one.alyx.get('/insertions?&django=pk,' + uuid, clobber=True)	1✔
75
76	if len(sess_list) > 0 and len(ins_list) == 0: # session	1✔
77	note = CriticalSessionNote(uuid, one)	1✔
78	elif len(ins_list) > 0 and len(sess_list) == 0: # insertion	1✔
79	note = CriticalInsertionNote(uuid, one)	1✔
80	else:
81	raise ValueError(f'Inadequate number of session (n={len(sess_list)}) '	×
82	f'or insertion (n={len(ins_list)}) found for uuid {uuid}.'
83	f'The query output should be of length 1.')
84
85	note.upload_note()	1✔
86
87
88	class Note(abc.ABC):	1✔
89	descriptions = []	1✔
90
91	@property	1✔
92	def default_descriptions(self):	1✔
93	return self.descriptions + ['Other']	1✔
94
95	@property	1✔
96	def extra_prompt(self):	1✔
97	return ''	×
98
99	@property	1✔
100	def note_title(self):	1✔
101	return ''	×
102
103	@property	1✔
104	def n_description(self):	1✔
105	return len(self.default_descriptions)	1✔
106
107	def __init__(self, uuid, one, content_type=None):	1✔
108	"""
109	Base class for attaching notes to an alyx endpoint. Do not use this class directly but use parent classes that inherit
110	this base class
111
112	:param uuid: uuid of session/ insertion or other model to attach note to
113	:param one: ONE instance
114	:param content_type: alyx endpoint of uuid
115	"""
116	self.uuid = uuid	1✔
117	self.one = one	1✔
118	self.selected_reasons = []	1✔
119	self.other_reason = []	1✔
120	if content_type is not None:	1✔
121	self.content_type = content_type	1✔
122	else:
123	self.content_type = self.get_content_type()	×
124
125	def get_content_type(self):	1✔
126	"""
127	Infer the content_type from the uuid. Only checks to see if uuid is a session or insertion. If not recognised will raise
128	an error and the content_type must be specified on note initialisation e.g Note(uuid, one, content_type='subject')
129	:return:
130	"""
131
132	# see if it as session or an insertion
133	if self.one.uuid2path(self.uuid):	×
134	content_type = 'session'	×
135	else:
136	try:	×
137	_ = self.one.pid2uuid(self.uuid)	×
138	content_type = 'probeinsertion'	×
139	except HTTPError:	×
140	raise ValueError('Content type cannot be recognised from {uuid}. Specify on '	×
141	'initialistion e.g Note(uuid, one, content_type="subject"')
142
143	return content_type	×
144
145	def describe(self):	1✔
146	"""
147	Print list of default reasons that can be chosen from
148	:return:
149	"""
150	for i, d in enumerate(self.descriptions):	×
151	print(f'{i}. {d} \n')	×
152
153	def numbered_descriptions(self):	1✔
154	"""
155	Return list of numbered default reasons
156	:return:
157	"""
158	return [f'{i}) {d}' for i, d in enumerate(self.default_descriptions)]	1✔
159
160	def upload_note(self, nums=None, other_reason=None, **kwargs):	1✔
161	"""
162	Upload note to alyx. If no values for nums and other_reason are specified, user will receive a prompt in command line
163	asking them to choose from default list of reasons to add to note as well as option for free text. To upload without
164	receiving prompt a value for either nums or other_reason must be given
165
166	:param nums: string of numbers matching those in default descrptions, e.g, '1,3'. Options can be see using note.describe()
167	:param other_reason: other comment or reasons to add to note (string)
168	:param kwargs:
169	:return:
170	"""
171
172	if nums is None and other_reason is None:	1✔
173	self.selected_reasons, self.other_reason = self.reasons_prompt()	1✔
174	else:
175	self.selected_reasons = self._map_num_to_description(nums)	1✔
176	self.other_reason = other_reason or []	1✔
177
178	self._upload_note(**kwargs)	1✔
179
180	def _upload_note(self, **kwargs):	1✔
181	existing_note, notes = self._check_existing_note()	1✔
182	if existing_note:	1✔
183	self.update_existing_note(notes, **kwargs)	1✔
184	else:
185	text = self.format_note(**kwargs)	1✔
186	self._create_note(text)	1✔
187	_logger.info('The selected reasons were saved on Alyx.')	1✔
188
189	def _create_note(self, text):	1✔
190
191	data = {'user': self.one.alyx.user,	1✔
192	'content_type': self.content_type,
193	'object_id': self.uuid,
194	'text': f'{text}'}
195	self.one.alyx.rest('notes', 'create', data=data)	1✔
196
197	def _update_note(self, note_id, text):	1✔
198
199	data = {'user': self.one.alyx.user,	1✔
200	'content_type': self.content_type,
201	'object_id': self.uuid,
202	'text': f'{text}'}
203	self.one.alyx.rest('notes', 'partial_update', id=note_id, data=data)	1✔
204
205	def _delete_note(self, note_id):	1✔
206	self.one.alyx.rest('notes', 'delete', id=note_id)	1✔
207
208	def _delete_notes(self, notes):	1✔
209	for note in notes:	1✔
210	self._delete_note(note['id'])	1✔
211
212	def _check_existing_note(self):	1✔
213	notes = self.one.alyx.rest('notes', 'list', django=f'text__icontains,{self.note_title},object_id,{self.uuid}',	1✔
214	no_cache=True)
215	if len(notes) == 0:	1✔
216	return False, None	1✔
217	else:
218	return True, notes	1✔
219
220	def _map_num_to_description(self, nums):	1✔
221
222	if nums is None:	1✔
223	return []	×
224
225	string_list = nums.split(',')	1✔
226	int_list = list(map(int, string_list))	1✔
227
228	if max(int_list) >= self.n_description or min(int_list) < 0:	1✔
229	raise ValueError(f'Chosen values out of range, must be between 0 and {self.n_description - 1}')	×
230
231	return [self.default_descriptions[n] for n in int_list]	1✔
232
233	def reasons_prompt(self):	1✔
234	"""
235	Prompt for user to enter reasons
236	:return:
237	"""
238
239	prompt = f'{self.extra_prompt} ' \	1✔
240	f'\n {self.numbered_descriptions()} \n ' \
241	f'and enter the corresponding numbers separated by commas, e.g. 1,3 -> enter: '
242
243	ans = input(prompt).strip().lower()	1✔
244
245	try:	1✔
246	selected_reasons = self._map_num_to_description(ans)	1✔
247	print(f'You selected reason(s): {selected_reasons}')	1✔
248	if 'Other' in selected_reasons:	1✔
249	other_reasons = self.other_reason_prompt()	1✔
250	return selected_reasons, other_reasons	1✔
251	else:
252	return selected_reasons, []	1✔
253
254	except ValueError:	×
255	print(f'{ans} is invalid, please try again...')	×
256	return self.reasons_prompt()	×
257
258	def other_reason_prompt(self):	1✔
259	"""
260	Prompt for user to enter other reasons
261	:return:
262	"""
263
264	prompt = 'Explain why you selected "other" (free text): '	1✔
265	ans = input(prompt).strip().lower()	1✔
266	return ans	1✔
267
268	@abc.abstractmethod	1✔
269	def format_note(self, **kwargs):	1✔
270	"""
271	Method to format text field of note according to type of note wanting to be uploaded
272	:param kwargs:
273	:return:
274	"""
275
276	@abc.abstractmethod	1✔
277	def update_existing_note(self, note, **kwargs):	1✔
278	"""
279	Method to specify behavior in the case of a note with the same title already existing
280	:param note:
281	:param kwargs:
282	:return:
283	"""
284
285
286	class CriticalNote(Note):	1✔
287	"""	1✔
288	Class for uploading a critical note to a session or insertion. Do not use directly but use CriticalSessionNote or
289	CriticalInsertionNote instead
290	"""
291
292	def format_note(self, **kwargs):	1✔
293	note_text = {	1✔
294	"title": self.note_title,
295	"reasons_selected": self.selected_reasons,
296	"reason_for_other": self.other_reason
297	}
298	return json.dumps(note_text)	1✔
299
300	def update_existing_note(self, notes, **kwargs):	1✔
301
302	overwrite = kwargs.get('overwrite', None)	1✔
303	if overwrite is None:	1✔
304	overwrite = self.delete_note_prompt(notes)	1✔
305
306	if overwrite:	1✔
307	self._delete_notes(notes)	1✔
308	text = self.format_note()	1✔
309	self._create_note(text)	1✔
310	_logger.info('The selected reasons were saved on Alyx; old notes were deleted')	1✔
311	else:
312	_logger.info('The selected reasons were NOT saved on Alyx; old notes remain.')	×
313
314	def delete_note_prompt(self, notes):	1✔
315
316	prompt = f'You are about to delete {len(notes)} existing notes; ' \	1✔
317	f'do you want to proceed? y/n: '
318
319	ans = input(prompt).strip().lower()	1✔
320
321	if ans not in ['y', 'n']:	1✔
322	print(f'{ans} is invalid, please try again...')	×
323	return self.delete_note_prompt(notes)	×
324	else:
325	return True if ans == 'y' else False	1✔
326
327
328	class CriticalInsertionNote(CriticalNote):	1✔
329	"""	1✔
330	Class for uploading a critical note to an insertion.
331
332	Example
333	-------
334	note = CriticalInsertionNote(pid, one)
335	# print list of default reasons
336	note.describe()
337	# to receive a command line prompt to fill in note
338	note.upload_note()
339	# to upload note automatically without prompt
340	note.upload_note(nums='1,4', other_reason='lots of bad channels')
341	"""
342
343	descriptions_gui = [	1✔
344	'Noise and artifact',
345	'Drift',
346	'Poor neural yield',
347	'Brain Damage'
348	'Other'
349	]
350
351	descriptions = [	1✔
352	'Histological images missing',
353	'Track not visible on imaging data'
354	]
355
356	@property	1✔
357	def default_descriptions(self):	1✔
358	return self.descriptions + self.descriptions_gui	1✔
359
360	@property	1✔
361	def extra_prompt(self):	1✔
362	return 'Select from this list the reason(s) why you are marking the insertion as CRITICAL:'	1✔
363
364	@property	1✔
365	def note_title(self):	1✔
366	return '=== EXPERIMENTER REASON(S) FOR MARKING THE INSERTION AS CRITICAL ==='	1✔
367
368	def __init__(self, uuid, one):	1✔
369	super(CriticalInsertionNote, self).__init__(uuid, one, content_type='probeinsertion')	1✔
370
371
372	class CriticalSessionNote(CriticalNote):	1✔
373	"""	1✔
374	Class for uploading a critical note to a session.
375
376	Example
377	-------
378	note = CriticalInsertionNote(uuid, one)
379	# print list of default reasons
380	note.describe()
381	# to receive a command line prompt to fill in note
382	note.upload_note()
383	# to upload note automatically without prompt
384	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
385	"""
386
387	descriptions = [	1✔
388	'within experiment system crash',
389	'synching impossible',
390	'dud or mock session',
391	'essential dataset missing',
392	]
393
394	@property	1✔
395	def extra_prompt(self):	1✔
396	return 'Select from this list the reason(s) why you are marking the session as CRITICAL:'	1✔
397
398	@property	1✔
399	def note_title(self):	1✔
400	return '=== EXPERIMENTER REASON(S) FOR MARKING THE SESSION AS CRITICAL ==='	1✔
401
402	def __init__(self, uuid, one):	1✔
403	super(CriticalSessionNote, self).__init__(uuid, one, content_type='session')	1✔
404
405
406	class SignOffNote(Note):	1✔
407	"""	1✔
408	Class for signing off a session and optionally adding a related explanation note.
409	Do not use directly but use classes that inherit from this class e.g TaskSignOffNote, RawEphysSignOffNote
410	"""
411
412	@property	1✔
413	def extra_prompt(self):	1✔
414	return 'Select from this list the reason(s) that describe issues with this session:'	1✔
415
416	@property	1✔
417	def note_title(self):	1✔
418	return f'=== SIGN-OFF NOTE FOR {self.sign_off_key} ==='	1✔
419
420	def __init__(self, uuid, one, sign_off_key):	1✔
421	self.sign_off_key = sign_off_key	1✔
422	super(SignOffNote, self).__init__(uuid, one, content_type='session')	1✔
423	self.datetime_key = self.get_datetime_key()	1✔
424	self.session = one.alyx.rest('sessions', 'read', id=self.uuid, no_cache=True)	1✔
425
426	def upload_note(self, nums=None, other_reason=None, **kwargs):	1✔
427	super(SignOffNote, self).upload_note(nums=nums, other_reason=other_reason, **kwargs)	1✔
428	self.sign_off()	1✔
429
430	def sign_off(self):	1✔
431
432	json = self.session['json']	1✔
433	json['sign_off_checklist'][self.sign_off_key] = {'date': self.datetime_key.split('_')[0],	1✔
434	'user': self.datetime_key.split('_')[1]}
435
436	self.one.alyx.json_field_update("sessions", self.uuid, 'json', data=json)	1✔
437
438	def format_note(self, **kwargs):	1✔
439
440	note_text = {	1✔
441	"title": self.note_title,
442	f'{self.datetime_key}': {"reasons_selected": self.selected_reasons,
443	"reason_for_other": self.other_reason}
444	}
445
446	return json.dumps(note_text)	1✔
447
448	def format_existing_note(self, orignal_note):	1✔
449
450	extra_note = {f'{self.datetime_key}': {"reasons_selected": self.selected_reasons,	1✔
451	"reason_for_other": self.other_reason}
452	}
453
454	orignal_note.update(extra_note)	1✔
455
456	return json.dumps(orignal_note)	1✔
457
458	def update_existing_note(self, notes):	1✔
459	if len(notes) != 1:	1✔
460	raise ValueError(f'{len(notes)} with same title found, only expect at most 1. Clean up before proceeding')	×
461	else:
462	original_note = json.loads(notes[0]['text'])	1✔
463	text = self.format_existing_note(original_note)	1✔
464	self._update_note(notes[0]['id'], text)	1✔
465
466	def get_datetime_key(self):	1✔
467	user = self.one.alyx.user	1✔
468	date = datetime.now().date().isoformat()	1✔
469	return date + '_' + user	1✔
470
471
472	class TaskSignOffNote(SignOffNote):	1✔
473
474	"""	1✔
475	Class for signing off a task part of a session and optionally adding a related explanation note.
476
477	Example
478	-------
479	note = TaskSignOffNote(uuid, one, '_ephysChoiceWorld_00')
480	# to sign off session without any note
481	note.sign_off()
482	# print list of default reasons
483	note.describe()
484	# to upload note and sign off with prompt
485	note.upload_note()
486	# to upload note automatically without prompt
487	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
488	"""
489
490	descriptions = [	1✔
491	'raw trial data does not exist',
492	'wheel data corrupt',
493	'task data could not be synced',
494	]
495
496
497	class PassiveSignOffNote(SignOffNote):	1✔
498
499	"""	1✔
500	Class for signing off a passive part of a session and optionally adding a related explanation note.
501
502	Example
503	-------
504	note = PassiveSignOffNote(uuid, one, '_passiveChoiceWorld_00')
505	# to sign off session without any note
506	note.sign_off()
507	# print list of default reasons
508	note.describe()
509	# to upload note and sign off with prompt
510	note.upload_note()
511	# to upload note automatically without prompt
512	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
513	"""
514
515	descriptions = [	1✔
516	'Raw passive data doesn’t exist (no. of spacers = 0)',
517	'Incorrect number or spacers (i.e passive cutoff midway)',
518	'RFmap file doesn’t exist',
519	'Gabor patches couldn’t be extracted',
520	'Trial playback couldn’t be extracted',
521	]
522
523
524	class VideoSignOffNote(SignOffNote):	1✔
525
526	"""	1✔
527	Class for signing off a video part of a session and optionally adding a related explanation note.
528
529	Example
530	-------
531	note = VideoSignOffNote(uuid, one, '_camera_left')
532	# to sign off session without any note
533	note.sign_off()
534	# print list of default reasons
535	note.describe()
536	# to upload note and sign off with prompt
537	note.upload_note()
538	# to upload note automatically without prompt
539	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
540	"""
541
542	descriptions = [	1✔
543	'The video timestamps are not the same length as the video file (either empty or slightly longer/shorter)',
544	'The rotary encoder trace doesn’t not appear synced with the video',
545	'The QC fails because the GPIO file is missing or empty',
546	'The frame rate in the video header is wrong (the video plays too slow or fast)',
547	'The resolution is not what is defined in the experiment description file',
548	'The DLC QC fails because something is obscuring the pupil',
549	]
550
551
552	class RawEphysSignOffNote(SignOffNote):	1✔
553
554	"""	1✔
555	Class for signing off a raw ephys part of a session and optionally adding a related explanation note.
556
557	Example
558	-------
559	note = RawEphysSignOffNote(uuid, one, '_neuropixel_raw_probe00')
560	# to sign off session without any note
561	note.sign_off()
562	# print list of default reasons
563	note.describe()
564	# to upload note and sign off with prompt
565	note.upload_note()
566	# to upload note automatically without prompt
567	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
568	"""
569
570	descriptions = [	1✔
571	'Data has striping',
572	'Horizontal band',
573	'Discontunuity',
574	]
575
576
577	class SpikeSortingSignOffNote(SignOffNote):	1✔
578
579	"""	1✔
580	Class for signing off a spikesorting part of a session and optionally adding a related explanation note.
581
582	Example
583	-------
584	note = SpikeSortingSignOffNote(uuid, one, '_neuropixel_spike_sorting_probe00')
585	# to sign off session without any note
586	note.sign_off()
587	# print list of default reasons
588	note.describe()
589	# to upload note and sign off with prompt
590	note.upload_note()
591	# to upload note automatically without prompt
592	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
593	"""
594
595	descriptions = [	1✔
596	'Spikesorting could not be run',
597	'Poor quality spikesorting',
598	]
599
600
601	class AlignmentSignOffNote(SignOffNote):	1✔
602
603	"""	1✔
604	Class for signing off a alignment part of a session and optionally adding a related explanation note.
605
606	Example
607	-------
608	note = AlignmentSignOffNote(uuid, one, '_neuropixel_alignment_probe00')
609	# to sign off session without any note
610	note.sign_off()
611	# print list of default reasons
612	note.describe()
613	# to upload note and sign off with prompt
614	note.upload_note()
615	# to upload note automatically without prompt
616	note.upload_note(nums='1,4', other_reason='session with no ephys recording')
617	"""
618
619	descriptions = []	1✔

int-brain-lab / ibllib / 7961675356254463

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous