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