Source code for allensdk.brain_observatory.behavior.data_objects.projections
from PIL import Image as PILImage
from pynwb import NWBFile
from allensdk.core import DataObject
from allensdk.core import \
JsonReadableInterface, NwbReadableInterface, \
LimsReadableInterface
from allensdk.core import \
NwbWritableInterface
from allensdk.brain_observatory.behavior.image_api import ImageApi, Image
from allensdk.brain_observatory.nwb.nwb_utils import get_image, \
add_image_to_nwb
from allensdk.internal.api import PostgresQueryMixin
from allensdk.internal.core.lims_utilities import safe_system_path
[docs]
class Projections(DataObject, LimsReadableInterface, JsonReadableInterface,
NwbReadableInterface, NwbWritableInterface):
def __init__(self, max_projection: Image, avg_projection: Image):
super().__init__(name='projections', value=None, is_value_self=True)
self._max_projection = max_projection
self._avg_projection = avg_projection
@property
def max_projection(self) -> Image:
return self._max_projection
@property
def avg_projection(self) -> Image:
return self._avg_projection
[docs]
@classmethod
def from_lims(cls, ophys_experiment_id: int,
lims_db: PostgresQueryMixin) -> "Projections":
def _get_filepaths():
"""
Note
----
Legacy motion correction/segmentation stores images as well known
files attached to the segmentation run while the newer suite2p
pipeline produces them during motion correction and attached
to the ophys_experiment. In the code below, we first check
for the well known files attached to the experiment and then,
to support older datasets, we check for them on the segmentation
run if not all data products are found.
"""
query = """
SELECT
wkf.storage_directory || wkf.filename AS filepath,
wkft.name as wkfn
FROM ophys_experiments oe
JOIN well_known_files wkf ON wkf.attachable_id = oe.id
JOIN well_known_file_types wkft
ON wkft.id = wkf.well_known_file_type_id
WHERE wkf.attachable_type = 'OphysExperiment'
AND wkft.name IN ('OphysMaxIntImage',
'OphysAverageIntensityProjectionImage')
AND oe.id = {};
""".format(ophys_experiment_id)
res = lims_db.select(query=query)
# Check if the projections are attached to motion correction/the
# ophys_experiment. If not, this is an older experiment and
# need to load the projections from the segmentation.
if 'OphysMaxIntImage' not in res['wkfn'].to_list() \
or 'OphysAverageIntensityProjectionImage' \
not in res['wkfn'].to_list():
query = """
SELECT
wkf.storage_directory || wkf.filename AS filepath,
wkft.name as wkfn
FROM ophys_experiments oe
JOIN ophys_cell_segmentation_runs ocsr
ON ocsr.ophys_experiment_id = oe.id
JOIN well_known_files wkf ON wkf.attachable_id = ocsr.id
JOIN well_known_file_types wkft
ON wkft.id = wkf.well_known_file_type_id
WHERE ocsr.current = 't'
AND wkf.attachable_type = 'OphysCellSegmentationRun'
AND wkft.name IN ('OphysMaxIntImage',
'OphysAverageIntensityProjectionImage')
AND oe.id = {};
""".format(ophys_experiment_id)
res = lims_db.select(query=query)
res['filepath'] = res['filepath'].apply(safe_system_path)
return res
def _get_pixel_size():
query = """
SELECT sc.resolution
FROM ophys_experiments oe
JOIN scans sc ON sc.image_id=oe.ophys_primary_image_id
WHERE oe.id = {};
""".format(ophys_experiment_id)
return lims_db.fetchone(query, strict=True)
res = _get_filepaths()
pixel_size = _get_pixel_size()
max_projection_filepath = \
res[res['wkfn'] == 'OphysMaxIntImage'].iloc[0]['filepath']
max_projection = cls._from_filepath(filepath=max_projection_filepath,
pixel_size=pixel_size)
avg_projection_filepath = \
(res[res['wkfn'] == 'OphysAverageIntensityProjectionImage'].iloc[0]
['filepath'])
avg_projection = cls._from_filepath(filepath=avg_projection_filepath,
pixel_size=pixel_size)
return Projections(max_projection=max_projection,
avg_projection=avg_projection)
[docs]
@classmethod
def from_nwb(cls, nwbfile: NWBFile) -> "Projections":
max_projection = get_image(nwbfile=nwbfile, name='max_projection',
module='ophys')
avg_projection = get_image(nwbfile=nwbfile, name='average_image',
module='ophys')
return Projections(max_projection=max_projection,
avg_projection=avg_projection)
[docs]
def to_nwb(self, nwbfile: NWBFile) -> NWBFile:
add_image_to_nwb(nwbfile=nwbfile,
image_data=self._max_projection,
image_name='max_projection')
add_image_to_nwb(nwbfile=nwbfile,
image_data=self._avg_projection,
image_name='average_image')
return nwbfile
[docs]
@classmethod
def from_json(cls, dict_repr: dict) -> "Projections":
max_projection_filepath = dict_repr['max_projection_file']
avg_projection_filepath = \
dict_repr['average_intensity_projection_image_file']
pixel_size = dict_repr['surface_2p_pixel_size_um']
max_projection = cls._from_filepath(filepath=max_projection_filepath,
pixel_size=pixel_size)
avg_projection = cls._from_filepath(filepath=avg_projection_filepath,
pixel_size=pixel_size)
return Projections(max_projection=max_projection,
avg_projection=avg_projection)
@staticmethod
def _from_filepath(filepath: str, pixel_size: float) -> Image:
"""
:param filepath
path to image
:param pixel_size
pixel size in um
"""
img = PILImage.open(filepath)
img = ImageApi.serialize(img, [pixel_size / 1000.,
pixel_size / 1000.], 'mm')
img = ImageApi.deserialize(img=img)
return img