Coverage for ibllib/plots/snapshot.py: 56%
161 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-08 17:16 +0100
1import logging
2import requests
3import traceback
4import json
5import abc
6import numpy as np
8from one.api import ONE
9from ibllib.pipes import tasks
10from one.alf.exceptions import ALFObjectNotFound
11from neuropixel import trace_header, TIP_SIZE_UM
13from ibllib import __version__ as ibllib_version
14from ibllib.pipes.ephys_alignment import EphysAlignment
15from ibllib.pipes.histology import interpolate_along_track
16from iblatlas.atlas import AllenAtlas
18_logger = logging.getLogger(__name__)
21class ReportSnapshot(tasks.Task):
23 def __init__(self, session_path, object_id, content_type='session', **kwargs):
24 self.object_id = object_id 1ab
25 self.content_type = content_type 1ab
26 self.images = [] 1ab
27 super(ReportSnapshot, self).__init__(session_path, **kwargs) 1ab
29 def _run(self, overwrite=False):
30 # Can be used to generate the image if desired
31 pass
33 def register_images(self, widths=None, function=None, extra_dict=None):
34 report_tag = '## report ##' 1ab
35 snapshot = Snapshot(one=self.one, object_id=self.object_id, content_type=self.content_type) 1ab
36 jsons = [] 1ab
37 texts = [] 1ab
38 for f in self.outputs: 1ab
39 json_dict = dict(tag=report_tag, version=ibllib_version, 1ab
40 function=(function or str(self.__class__).split("'")[1]), name=f.stem)
41 if extra_dict is not None: 1ab
42 assert isinstance(extra_dict, dict)
43 json_dict.update(extra_dict)
44 jsons.append(json_dict) 1ab
45 texts.append(f"{f.stem}") 1ab
46 return snapshot.register_images(self.outputs, jsons=jsons, texts=texts, widths=widths) 1ab
49class ReportSnapshotProbe(ReportSnapshot):
50 signature = {
51 'input_files': [], # see setUp method for declaration of inputs
52 'output_files': [] # see setUp method for declaration of inputs
53 }
55 def __init__(self, pid, session_path=None, one=None, brain_regions=None, brain_atlas=None, **kwargs):
56 """
57 :param pid: probe insertion UUID from Alyx
58 :param one: one instance
59 :param brain_regions: (optional) iblatlas.regions.BrainRegion object
60 :param brain_atlas: (optional) iblatlas.atlas.AllenAtlas object
61 :param kwargs:
62 """
63 assert one
64 self.one = one
65 self.brain_atlas = brain_atlas
66 self.brain_regions = brain_regions
67 if self.brain_atlas and not self.brain_regions:
68 self.brain_regions = self.brain_atlas.regions
69 self.content_type = 'probeinsertion'
70 self.pid = pid
71 self.eid, self.pname = self.one.pid2eid(self.pid)
72 self.session_path = session_path or self.one.eid2path(self.eid)
73 self.output_directory = self.session_path.joinpath('snapshot', self.pname)
74 self.output_directory.mkdir(exist_ok=True, parents=True)
75 self.histology_status = None
76 self.get_probe_signature()
77 super(ReportSnapshotProbe, self).__init__(self.session_path, object_id=pid, content_type=self.content_type, one=self.one,
78 **kwargs)
80 @property
81 def pid_label(self):
82 """returns a probe insertion stub to label titles, for example: 'SWC_054_2020-10-05_001_probe01'"""
83 return '_'.join(list(self.session_path.parts[-3:]) + [self.pname])
85 @abc.abstractmethod
86 def get_probe_signature(self):
87 # method that gets input and output signatures from the probe name. The format is a dictionary as follows:
88 # return {'input_files': input_signature, 'output_files': output_signature}
89 pass
91 def get_histology_status(self):
92 """
93 Finds at which point in histology pipeline the probe insertion is
94 :return:
95 """
97 self.hist_lookup = {'Resolved': 3,
98 'Aligned': 2,
99 'Traced': 1,
100 None: 0} # is this bad practice?
102 self.ins = self.one.alyx.rest('insertions', 'list', id=self.pid)[0]
103 traced = self.ins.get('json', {}).get('extended_qc', {}).get('tracing_exists', False)
104 aligned = self.ins.get('json', {}).get('extended_qc', {}).get('alignment_count', 0)
105 resolved = self.ins.get('json', {}).get('extended_qc', {}).get('alignment_resolved', False)
107 if resolved:
108 return 'Resolved'
109 elif aligned > 0:
110 return 'Aligned'
111 elif traced:
112 return 'Traced'
113 else:
114 return None
116 def get_channels(self, alf_object, collection):
117 electrodes = {}
119 try:
120 electrodes = self.one.load_object(self.eid, alf_object, collection=collection)
121 electrodes['axial_um'] = electrodes['localCoordinates'][:, 1]
122 except ALFObjectNotFound:
123 _logger.warning(f'{alf_object} does not yet exist')
125 if self.hist_lookup[self.histology_status] == 3:
126 try:
127 electrodes['atlas_id'] = electrodes['brainLocationIds_ccf_2017']
128 electrodes['mlapdv'] = electrodes['mlapdv'] / 1e6
129 except KeyError:
130 _logger.warning('Insertion resolved but brainLocationIds_ccf_2017 attribute do not exist')
132 if self.hist_lookup[self.histology_status] > 0 and 'atlas_id' not in electrodes.keys():
133 if not self.brain_atlas:
134 self.brain_atlas = AllenAtlas()
135 self.brain_regions = self.brain_regions or self.brain_atlas.regions
136 if 'localCoordinates' not in electrodes.keys():
137 geometry = trace_header(version=1)
138 electrodes['localCoordinates'] = np.c_[geometry['x'], geometry['y']]
139 electrodes['axial_um'] = electrodes['localCoordinates'][:, 1]
141 depths = electrodes['localCoordinates'][:, 1]
142 xyz = np.array(self.ins['json']['xyz_picks']) / 1e6
144 if self.hist_lookup[self.histology_status] >= 2:
145 traj = self.one.alyx.rest('trajectories', 'list', provenance='Ephys aligned histology track',
146 probe_insertion=self.pid)[0]
147 align_key = self.ins['json']['extended_qc']['alignment_stored']
148 feature = traj['json'][align_key][0]
149 track = traj['json'][align_key][1]
150 ephysalign = EphysAlignment(xyz, depths, track_prev=track,
151 feature_prev=feature,
152 brain_atlas=self.brain_atlas, speedy=True)
153 electrodes['mlapdv'] = ephysalign.get_channel_locations(feature, track)
154 electrodes['atlas_id'] = self.brain_atlas.regions.get(self.brain_atlas.get_labels(electrodes['mlapdv']))['id']
156 if self.hist_lookup[self.histology_status] == 1:
157 xyz = xyz[np.argsort(xyz[:, 2]), :]
158 electrodes['mlapdv'] = interpolate_along_track(xyz, (depths + TIP_SIZE_UM) / 1e6)
159 electrodes['atlas_id'] = self.brain_atlas.regions.get(self.brain_atlas.get_labels(electrodes['mlapdv']))['id']
161 return electrodes
163 def register_images(self, widths=None, function=None):
164 super(ReportSnapshotProbe, self).register_images(widths=widths, function=function,
165 extra_dict={'channels': self.histology_status})
168class Snapshot:
169 """
170 A class to register images in form of Notes, linked to an object on Alyx.
172 :param object_id: The id of the object the image should be linked to
173 :param content_type: Which type of object to link to, e.g. 'session', 'probeinsertion', 'subject',
174 default is 'session'
175 :param one: An ONE instance, if None is given it will be instantiated.
176 """
178 def __init__(self, object_id, content_type='session', one=None):
179 self.one = one or ONE() 1hgfdeab
180 self.object_id = object_id 1hgfdeab
181 self.content_type = content_type 1hgfdeab
182 self.images = [] 1hgfdeab
184 def plot(self):
185 """
186 Placeholder method to be overriden by child object
187 :return:
188 """
189 pass
191 def generate_image(self, plt_func, plt_kwargs):
192 """
193 Takes a plotting function and adds the output to the Snapshot.images list for registration
195 :param plt_func: A plotting function that returns the path to an image.
196 :param plt_kwargs: Dictionary with keyword arguments for the plotting function
197 """
198 img_path = plt_func(**plt_kwargs) 1g
199 if isinstance(img_path, list): 1g
200 self.images.extend(img_path)
201 else:
202 self.images.append(img_path) 1g
203 return img_path 1g
205 def register_image(self, image_file, text='', json_field=None, width=None):
206 """
207 Registers an image as a Note, attached to the object specified by Snapshot.object_id
209 :param image_file: Path to the image to to registered
210 :param text: str, text to describe the image, defaults ot empty string
211 :param json_field: dict, to be added to the json field of the Note
212 :param width: width to scale the image to, defaults to None (scale to UPLOADED_IMAGE_WIDTH in alyx.settings.py),
213 other options are 'orig' (don't change size) or any integer (scale to width=int, aspect ratios won't be changed)
215 :returns: dict, note as registered in database
216 """
217 # the protocol is not compatible with byte streaming and json, so serialize the json object here
218 # Make sure that user is logged in, if not, try to log in
219 assert self.one.alyx.is_logged_in, 'No Alyx user is logged in, try running one.alyx.authenticate() first' 1fdeab
220 note = { 1fdeab
221 'user': self.one.alyx.user, 'content_type': self.content_type, 'object_id': self.object_id,
222 'text': text, 'width': width, 'json': json.dumps(json_field)}
223 _logger.info(f'Registering image to {self.content_type} with id {self.object_id}') 1fdeab
224 # to make sure an eventual note gets deleted with the image call the delete REST endpoint first
225 current_note = self.one.alyx.rest('notes', 'list', 1fdeab
226 django=f"object_id,{self.object_id},text,{text},json__name,{text}",
227 no_cache=True)
228 if len(current_note) == 1: 1fdeab
229 self.one.alyx.rest('notes', 'delete', id=current_note[0]['id']) 1ab
230 # Open image for upload
231 fig_open = open(image_file, 'rb') 1fdeab
232 # Catch error that results from object_id - content_type mismatch
233 try: 1fdeab
234 note_db = self.one.alyx.rest('notes', 'create', data=note, files={'image': fig_open}) 1fdeab
235 return note_db 1fdeab
236 except requests.HTTPError as e: 1e
237 if 'matching query does not exist' in str(e): 1e
238 _logger.error(f'The object_id {self.object_id} does not match an object of type {self.content_type}') 1e
239 _logger.debug(traceback.format_exc()) 1e
240 else:
241 raise e
242 finally:
243 fig_open.close() 1fdeab
245 def register_images(self, image_list=None, texts=None, widths=None, jsons=None):
246 """
247 Registers a list of images as Notes, attached to the object specified by Snapshot.object_id.
248 The images can be passed as image_list. If None are passed, will try to register the images in Snapshot.images.
250 :param image_list: List of paths to the images to to registered. If None, will try to register any images in
251 Snapshot.images
252 :param texts: List of text to describe the images. If len(texts)==1, the same text will be used for all images
253 :param widths: List of width to scale the figure to (see Snapshot.register_image). If len(widths)==1,
254 the same width will be used for all images
255 :param jsons: List of dictionaries to populate the json field of the note in Alyx. If len(jsons)==1,
256 the same dict will be used for all images
257 :returns: list of dicts, notes as registered in database
258 """
259 if not image_list or len(image_list) == 0: 1dab
260 if len(self.images) == 0: 1d
261 _logger.warning( 1d
262 "No figures were passed to register_figures, and self.figures is empty. No figures to register")
263 return 1d
264 else:
265 image_list = self.images 1d
266 widths = widths or [None] 1dab
267 texts = texts or [''] 1dab
268 jsons = jsons or [None] 1dab
270 if len(texts) == 1: 1dab
271 texts = len(image_list) * texts 1d
272 if len(widths) == 1: 1dab
273 widths = len(image_list) * widths 1dab
274 if len(jsons) == 1: 1dab
275 jsons = len(image_list) * jsons 1d
276 note_dbs = [] 1dab
277 for figure, text, width, json_field in zip(image_list, texts, widths, jsons): 1dab
278 note_dbs.append(self.register_image(figure, text=text, width=width, json_field=json_field)) 1dab
279 return note_dbs 1dab