Source code for allensdk.internal.api.queries.grid_data_api_prerelease
import os
import six
from allensdk.config.manifest import Manifest
from allensdk.api.cache import Cache, cacheable
from allensdk.api.queries.grid_data_api import GridDataApi
from allensdk.core import json_utilities
from ..api_prerelease import ApiPrerelease
from ...core import lims_utilities as lu
_STORAGE_DIRECTORY_QUERY = '''
select iser.id,
iser.storage_directory
from image_series as iser
where iser.storage_directory is not null
'''
@cacheable()
def _get_grid_storage_directories(grid_data_directory):
query_result = lu.query(_STORAGE_DIRECTORY_QUERY)
storage_directories = dict()
for row in query_result:
path = lu.safe_system_path(row[b'storage_directory'])
# NOTE: hacky, but grid directory contains files without having
# injection_density_*.nrrd, projection_density_*.nrrd, ...
grid_example = os.path.join(path, grid_data_directory, 'data_mask_100.nrrd')
if os.path.exists(grid_example):
storage_directories[str(row[b'id'])] = path
return storage_directories
[docs]class GridDataApiPrerelease(GridDataApi):
'''Client for retrieving prereleased mouse connectivity data from lims.
Parameters
----------
base_uri : string, optional
Does not affect pulling from lims.
file_name : string, optional
File name to save/read storage_directories dict. Passed to
GridDataApiPrerelease constructor.
'''
GRID_DATA_DIRECTORY = 'grid'
[docs] @classmethod
def from_file_name(cls, file_name, cache=True, **kwargs):
'''Alternative constructor using cache path file_name.
Parameters
----------
file_name : string
Path where storage_directories will be saved.
**kwargs
Keyword arguments to be supplied to __init__
Returns
-------
cls : instance of GridDataApiPrerelease
'''
if os.path.exists(file_name):
storage_directories = json_utilities.read(file_name)
else:
storage_directories = _get_grid_storage_directories(cls.GRID_DATA_DIRECTORY)
if cache:
Manifest.safe_make_parent_dirs(file_name)
json_utilities.write(file_name, storage_directories)
return cls(storage_directories, **kwargs)
def __init__(self, storage_directories, resolution=None, base_uri=None):
super(GridDataApiPrerelease, self).__init__(resolution=resolution,
base_uri=base_uri)
self.storage_directories = storage_directories
self.api = ApiPrerelease()
[docs] def download_projection_grid_data(self, path, experiment_id, file_name):
'''Copy data from path to file_name.
Parameters
----------
path : string
path to file in shared directory (copy source)
experiment_id : int
image series id.
file_name : string
path to file destination (copy target)
'''
try:
storage_path = self.storage_directories[str(experiment_id)]
except KeyError as e:
error = '''
experiment %s is not in the storage_directories dictionary
this can be a result of one or more of:
* an invalid experiment id
* a valid experiment id whose grid data has not yet been computed
try either removing the storage_directories_prerelease.json manifest
from you manifest directory, or passing an updated storage_directories
dict to the GridDataApiPrerelase constructor.
''' % experiment_id
self._file_download_log.error(error)
self.cleanup_truncated_file(path)
raise six.raise_from(ValueError(error), e)
storage_path = os.path.join(
storage_path, self.GRID_DATA_DIRECTORY, file_name)
self.api.retrieve_file_from_storage(storage_path, path)