# Allen Institute Software License - This software license is the 2-clause BSD
# license plus a third clause that prohibits redistribution for commercial
# purposes without further permission.
#
# Copyright 2017. Allen Institute. All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# 3. Redistributions for commercial purposes are not permitted without the
# Allen Institute's written permission.
# For purposes of this license, commercial purposes is the incorporation of the
# Allen Institute's software into anything for which you will charge fees or
# other compensation. Contact terms@alleninstitute.org for commercial licensing
# opportunities.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
from allensdk.api.queries.rma_template import RmaTemplate
from allensdk.api.cache import cacheable
import os
import simplejson as json
from collections import OrderedDict
from allensdk.config.manifest import Manifest
[docs]class BiophysicalApi(RmaTemplate):
_NWB_file_type = 'NWBDownload'
_SWC_file_type = '3DNeuronReconstruction'
_MOD_file_type = 'BiophysicalModelDescription'
_FIT_file_type = 'NeuronalModelParameters'
_MARKER_file_type = '3DNeuronMarker'
BIOPHYSICAL_MODEL_TYPE_IDS = (491455321, 32923071,)
rma_templates = \
{"model_queries": [
{'name': 'models_by_specimen',
'description': 'see name',
'model': 'NeuronalModel',
'num_rows': 'all',
'count': False,
'criteria': '[neuronal_model_template_id$in{{biophysical_model_types}}],[specimen_id$in{{specimen_ids}}]',
'criteria_params': ['specimen_ids', 'biophysical_model_types']
}]}
def __init__(self, base_uri=None):
super(BiophysicalApi, self).__init__(base_uri, query_manifest=BiophysicalApi.rma_templates)
self.cache_stimulus = True
self.ids = {}
self.sweeps = []
self.manifest = {}
self.model_type = None
[docs] @cacheable()
def get_neuronal_models(self, specimen_ids, num_rows='all', count=False, model_type_ids=None, **kwargs):
'''Fetch all of the biophysically detailed model records associated with
a particular specimen_id
Parameters
----------
specimen_ids : list
One or more integer ids identifying specimen records.
num_rows : int, optional
how many records to retrieve. Default is 'all'.
count : bool, optional
If True, return a count of the lines found by the query. Default is False.
model_type_ids : list, optional
One or more integer ids identifying categories of neuronal model. Defaults
to all-active and perisomatic biophysical_models.
Returns
-------
List of dict
Each element is a biophysical model record, containing a unique integer
id, the id of the associated specimen, and the id of the model type to
which this model belongs.
'''
if model_type_ids is None:
model_type_ids = self.BIOPHYSICAL_MODEL_TYPE_IDS
return self.template_query('model_queries', 'models_by_specimen',
specimen_ids=specimen_ids,
biophysical_model_types=list(model_type_ids),
num_rows=num_rows, count=count)
[docs] def build_rma(self, neuronal_model_id, fmt='json'):
'''Construct a query to find all files related to a neuronal model.
Parameters
----------
neuronal_model_id : integer or string representation
key of experiment to retrieve.
fmt : string, optional
json (default) or xml
Returns
-------
string
RMA query url.
'''
include_associations = ''.join([
'neuronal_model_template(well_known_files(well_known_file_type)),',
'specimen(ephys_result(well_known_files(well_known_file_type)),',
'neuron_reconstructions(well_known_files(well_known_file_type)),',
'ephys_sweeps),',
'well_known_files(well_known_file_type)'])
criteria_associations = ''.join([
("[id$eq%d]," % (neuronal_model_id)),
include_associations])
return ''.join([self.rma_endpoint,
'/query.',
fmt,
'?q=',
'model::NeuronalModel,',
'rma::criteria,',
criteria_associations,
',rma::include,',
include_associations])
[docs] def read_json(self, json_parsed_data):
'''Get the list of well_known_file ids from a response body
containing nested sample,microarray_slides,well_known_files.
Parameters
----------
json_parsed_data : dict
Response from the Allen Institute Api RMA.
Returns
-------
list of strings
Well known file ids.
'''
self.ids = {
'stimulus': {},
'morphology': {},
'marker': {},
'modfiles': {},
'fit': {}
}
self.sweeps = []
if 'msg' in json_parsed_data:
for neuronal_model in json_parsed_data['msg']:
if 'well_known_files' in neuronal_model:
for well_known_file in neuronal_model['well_known_files']:
if ('id' in well_known_file and
'path' in well_known_file and
self.is_well_known_file_type(well_known_file,
BiophysicalApi._FIT_file_type)):
self.ids['fit'][str(well_known_file['id'])] = \
os.path.split(well_known_file['path'])[1]
if 'neuronal_model_template' in neuronal_model:
neuronal_model_template = neuronal_model[
'neuronal_model_template']
self.model_type = neuronal_model_template['name']
if 'well_known_files' in neuronal_model_template:
for well_known_file in neuronal_model_template['well_known_files']:
if ('id' in well_known_file and
'path' in well_known_file and
self.is_well_known_file_type(well_known_file,
BiophysicalApi._MOD_file_type)):
self.ids['modfiles'][str(well_known_file['id'])] = \
os.path.join('modfiles',
os.path.split(well_known_file['path'])[1])
if 'specimen' in neuronal_model:
specimen = neuronal_model['specimen']
if 'neuron_reconstructions' in specimen:
for neuron_reconstruction in specimen['neuron_reconstructions']:
if 'well_known_files' in neuron_reconstruction:
for well_known_file in neuron_reconstruction['well_known_files']:
if ('id' in well_known_file and 'path' in well_known_file):
if self.is_well_known_file_type(well_known_file, BiophysicalApi._SWC_file_type):
self.ids['morphology'][str(well_known_file['id'])] = \
os.path.split(
well_known_file['path'])[1]
elif self.is_well_known_file_type(well_known_file, BiophysicalApi._MARKER_file_type):
self.ids['marker'][str(well_known_file['id'])] = \
os.path.split(
well_known_file['path'])[1]
if 'ephys_result' in specimen:
ephys_result = specimen['ephys_result']
if 'well_known_files' in ephys_result:
for well_known_file in ephys_result['well_known_files']:
if ('id' in well_known_file and
'path' in well_known_file and
self.is_well_known_file_type(well_known_file, BiophysicalApi._NWB_file_type)):
self.ids['stimulus'][str(well_known_file['id'])] = \
"%d.nwb" % (ephys_result['id'])
self.sweeps = [sweep['sweep_number']
for sweep in specimen['ephys_sweeps']
if sweep['stimulus_name'] != 'Test']
return self.ids
[docs] def is_well_known_file_type(self, wkf, name):
'''Check if a structure has the expected name.
Parameters
----------
wkf : dict
A well-known-file structure with nested type information.
name : string
The expected type name
See Also
--------
read_json: where this helper function is used.
'''
try:
return wkf['well_known_file_type']['name'] == name
except:
return False
[docs] def get_well_known_file_ids(self, neuronal_model_id):
'''Query the current RMA endpoint with a neuronal_model id
to get the corresponding well known file ids.
Returns
-------
list
A list of well known file id strings.
'''
rma_builder_fn = self.build_rma
json_traversal_fn = self.read_json
return self.do_query(rma_builder_fn, json_traversal_fn, neuronal_model_id)
[docs] def create_manifest(self,
fit_path='',
model_type='',
stimulus_filename='',
swc_morphology_path='',
marker_path='',
sweeps=[]):
'''Generate a json configuration file with parameters for a
a biophysical experiment.
Parameters
----------
fit_path : string
filename of a json configuration file with cell parameters.
stimulus_filename : string
path to an NWB file with input currents.
swc_morphology_path : string
file in SWC format.
sweeps : array of integers
which sweeps in the stimulus file are to be used.
'''
self.manifest = OrderedDict()
self.manifest['biophys'] = [{
'model_file': ['manifest.json', fit_path],
'model_type': model_type
}]
self.manifest['runs'] = [{
'sweeps': sweeps
}]
self.manifest['neuron'] = [{
'hoc': ['stdgui.hoc', 'import3d.hoc']
}]
self.manifest['manifest'] = [
{
'type': 'dir',
'spec': '.',
'key': 'BASEDIR'
},
{
'type': 'dir',
'spec': 'work',
'key': 'WORKDIR',
'parent': 'BASEDIR'
},
{
'type': 'file',
'spec': swc_morphology_path,
'key': 'MORPHOLOGY'
},
{
'type': 'file',
'spec': marker_path,
'key': 'MARKER'
},
{
'type': 'dir',
'spec': 'modfiles',
'key': 'MODFILE_DIR'
},
{
'type': 'file',
'format': 'NWB',
'spec': stimulus_filename,
'key': 'stimulus_path'
},
{
'parent_key': 'WORKDIR',
'type': 'file',
'format': 'NWB',
'spec': stimulus_filename,
'key': 'output_path'
}
]
[docs] def cache_data(self,
neuronal_model_id,
working_directory=None):
'''Take a an experiment id, query the Api RMA to get well-known-files
download the files, and store them in the working directory.
Parameters
----------
neuronal_model_id : int or string representation
found in the neuronal_model table in the api
working_directory : string
Absolute path name where the downloaded well-known files will be stored.
'''
if working_directory is None:
working_directory = self.default_working_directory
well_known_file_id_dict = self.get_well_known_file_ids(
neuronal_model_id)
if not well_known_file_id_dict or \
(not any(list(well_known_file_id_dict.values()))):
raise(Exception("No data found for neuronal model id %d" %
(neuronal_model_id)))
Manifest.safe_mkdir(working_directory)
work_dir = os.path.join(working_directory, 'work')
Manifest.safe_mkdir(work_dir)
modfile_dir = os.path.join(working_directory, 'modfiles')
Manifest.safe_mkdir(modfile_dir)
for key, id_dict in well_known_file_id_dict.items():
if (not self.cache_stimulus) and (key == 'stimulus'):
continue
for well_known_id, filename in id_dict.items():
well_known_file_url = self.construct_well_known_file_download_url(
well_known_id)
cached_file_path = os.path.join(working_directory, filename)
self.retrieve_file_over_http(
well_known_file_url, cached_file_path)
fit_path = list(self.ids['fit'].values())[0]
stimulus_filename = list(self.ids['stimulus'].values())[0]
swc_morphology_path = list(self.ids['morphology'].values())[0]
marker_path = \
list(self.ids['marker'].values())[0] if 'marker' in self.ids else ''
sweeps = sorted(self.sweeps)
self.create_manifest(fit_path,
self.model_type,
stimulus_filename,
swc_morphology_path,
marker_path,
sweeps)
manifest_path = os.path.join(working_directory, 'manifest.json')
with open(manifest_path, 'w') as f:
json.dump(self.manifest, f, indent=2)