# Copyright 2016 Allen Institute for Brain Science
# This file is part of Allen SDK.
#
# Allen SDK is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# Allen SDK is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Allen SDK. If not, see <http://www.gnu.org/licenses/>.
import os
import logging
import allensdk.internal.core.lims_utilities as lims_utilities
from allensdk.config.manifest_builder import ManifestBuilder
from allensdk.config.manifest import Manifest
import json
import traceback
[docs]class OptimizeConfigReader(object):
_log = logging.getLogger('allensdk.internal.api.queries.lims.optimize_config_reader')
STIMULUS_CONTENT_TYPE = None
MORPHOLOGY_TYPE_ID = 303941301
MOD_FILE_TYPE_ID = 292178729
NEURONAL_MODEL_PARAMETERS = 329230374 # fit.json file.
def __init__(self):
self.lims_path = None
self.lims_data = None
self.lims_update_data = None
[docs] def read_lims_message(self, message, lims_path):
self.lims_path = lims_path
self.lims_data = message[0]
self.lims_update_data = dict(self.lims_data)
[docs] def read_lims_file(self, lims_path):
self.lims_path = lims_path
self.read_json(lims_path)
self.lims_update_data = dict(self.lims_data)
[docs] def read_json(self, path):
self.lims_path = os.path.realpath(path)
with open(self.lims_path) as f:
json_string = f.read()
self.read_json_string(json_string)
return self.lims_data
[docs] def read_json_string(self, json_string):
self.lims_data = json.loads(json_string)
self.lims_update_data = dict(self.lims_data)
[docs] def write_file(self, path):
with open(path, 'wb') as f:
f.write(json.dumps(self.lims_update_data, indent=2))
[docs] def stimulus_file_entries(self):
''' read the well known file path from the lims result
corresponding to the stimulus file
:return: well_known_file entries
:rtype: array of dicts
'''
neuronal_model = self.lims_data
specimen = neuronal_model['specimen']
roi_result = specimen['ephys_roi_result']
well_known_files = roi_result['well_known_files']
stimulus_file_entries = []
for well_known_file in well_known_files:
try:
file_type_id = well_known_file['well_known_file_type_id']
if file_type_id == lims_utilities.NWB_FILE_TYPE_ID:
stimulus_file_entries.append(well_known_file)
except:
OptimizeConfigReader._log.warn('skipping well known file record with no well known file type.')
return stimulus_file_entries
[docs] def lims_working_directory(self):
''' While this is the same directory as the optimize
directory, it can be mocked out for testing if the
optimize directory is write only.
'''
return self.neuronal_model_optimize_dir()
[docs] def output_directory(self):
return os.path.join(self.lims_working_directory(), 'work')
[docs] def stimulus_path(self):
''' Get the path to the stimulus file from the lims result.
:return: path to stimulus file
:rtype: string
'''
file_entries = self.stimulus_file_entries()
if len(file_entries) > 1:
OptimizeConfigReader._log.warning('More than one stimulus file found.')
file_entry = file_entries[0]
stimulus_path = os.path.join(file_entry['storage_directory'],
file_entry['filename'])
return stimulus_path
[docs] def neuronal_model_optimize_dir(self):
''' read the directory path where
output goes from the lims optimization config json
Parameters
----------
Returns
-------
string:
directory path
'''
return self.lims_data['storage_directory']
[docs] def morphology_file_entries(self):
''' read the well known file paths
from the lims result corresponding to the morphology
Returns
-------
arrary of dicts:
well known file entries
'''
neuronal_model = self.lims_data
specimen = neuronal_model['specimen']
reconstructions = specimen['neuron_reconstructions']
morphology_file_entries = []
for reconstruction in reconstructions:
superseded = reconstruction['superseded']
manual = reconstruction['manual']
if manual == True and superseded == False:
well_known_files = reconstruction['well_known_files']
for well_known_file in well_known_files:
file_type_id = well_known_file['well_known_file_type_id']
if file_type_id == OptimizeConfigReader.MORPHOLOGY_TYPE_ID:
morphology_file_entries.append(well_known_file)
return morphology_file_entries
[docs] def morphology_path(self):
''' Get the path to the morphology file from the lims result.
:return: path to morphology file
:rtype: string
'''
file_entries = self.morphology_file_entries()
if len(file_entries) > 1:
OptimizeConfigReader._log.warning('More than one morphology file found.')
file_entry = file_entries[0]
morphology_path = os.path.join(file_entry['storage_directory'],
file_entry['filename'])
return morphology_path
[docs] def sweep_entries(self):
''' read the sweep entries
from the lims result corresponding to the stimulus
:return: stimulus sweep entries
:rtype: array of dicts
'''
neuronal_model = self.lims_data
specimen = neuronal_model['specimen']
sweeps = specimen['ephys_sweeps']
return sweeps
[docs] def sweep_numbers(self):
''' Get the stimulus sweep numbers from the lims result
:return: list of sweep numbers
:rtype: array of ints
'''
sweep_entries = self.sweep_entries()
if not sweep_entries or len(sweep_entries) < 1:
OptimizeConfigReader._log.warning('No sweeps found.')
sweeps = [sweep_entry['sweep_number'] \
for sweep_entry in sweep_entries \
if sweep_entry['workflow_state'] == 'auto_passed' or \
sweep_entry['workflow_state'] == 'manual_passed' ]
return list(set(sweeps))
[docs] def mod_file_entries(self):
''' read the NERUON .mod file entries
from the lims result corresponding to the NeuronModel
:return: well known file entries
:rtype: array of dicts
'''
neuronal_model = self.lims_data
model_template = neuronal_model['neuronal_model_template']
well_known_files = model_template['well_known_files']
mod_file_entries = []
for well_known_file in well_known_files:
file_type_id = well_known_file['well_known_file_type_id']
if file_type_id == OptimizeConfigReader.MOD_FILE_TYPE_ID:
mod_file_entries.append(well_known_file)
return mod_file_entries
[docs] def mod_file_paths(self):
''' Get the paths to the mod files from the lims result.
:return: paths to mod files
:rtype: array of strings
'''
file_entries = self.mod_file_entries()
if not file_entries or len(file_entries) < 1:
OptimizeConfigReader._log.warning('No mod files found.')
mod_file_paths = []
for file_entry in file_entries:
mod_path = os.path.join(file_entry['storage_directory'],
file_entry['filename'])
OptimizeConfigReader._log.info(mod_path)
mod_file_paths.append(mod_path)
return mod_file_paths
[docs] def update_well_known_file(self,
path,
well_known_file_type_id=None):
if well_known_file_type_id == None:
well_known_file_type_id = lims_utilities.MODEL_PARAMETERS_FILE_TYPE_ID
well_known_files = self.lims_data['well_known_files']
def get_model_parameter_file_id(f):
if ('well_known_file_type_id' in f and
f['well_known_file_type_id'] == well_known_file_type_id):
return f['id']
else:
return None
def not_fit_param(f):
if ('well_known_file_type_id' in f and
f['well_known_file_type_id'] == well_known_file_type_id):
return False
else:
return True
try:
existing_file_id = \
next(wkf_id for wkf_id in
(get_model_parameter_file_id(f2)
for f2 in well_known_files) if wkf_id)
# existing parameter file found
self.lims_update_data['well_known_files'] = [f for f in well_known_files if not_fit_param(f)]
(dirname, filename) = os.path.split(os.path.abspath(path))
self.lims_update_data['well_known_files'] += [{
'id': existing_file_id,
'filename': filename,
'storage_directory': dirname,
'well_known_file_type_id': well_known_file_type_id
}]
except StopIteration:
# no parameter files found
(dirname, filename) = os.path.split(os.path.abspath(path))
self.lims_update_data['well_known_files'] += [{
'content_type': 'application/json',
'filename': filename,
'storage_directory': dirname,
'well_known_file_type_id': well_known_file_type_id
}]
[docs] def build_manifest(self, manifest_path=None):
b = ManifestBuilder()
b.add_path('BASEDIR', os.path.realpath(os.curdir))
b.add_path('WORKDIR',
self.output_directory())
b.add_path('MORPHOLOGY',
self.morphology_path(),
typename='file')
b.add_path('MODFILE_DIR', 'modfiles')
for modfile in self.mod_file_entries():
b.add_path('MOD_FILE_%s' % (os.path.splitext(modfile['filename'])[0]),
os.path.join(modfile['storage_directory'],
modfile['filename']),
typename='file',
format='MODFILE')
b.add_path('stimulus_path',
self.stimulus_path(),
typename='file',
format='NWB')
b.add_path('manifest',
os.path.join(os.path.realpath(os.curdir),
manifest_path),
typename='file')
b.add_path('output',
os.path.basename(self.stimulus_path()),
typename='file',
parent_key='WORKDIR',
format='NWB')
b.add_path('neuronal_model_data',
self.lims_path,
typename='file')
b.add_path('upfile',
'upbase.dat',
typename='file',
parent_key='WORKDIR')
b.add_path('downfile',
'downbase.dat',
typename='file',
parent_key='WORKDIR')
b.add_path('passive_fit_data',
'passive_fit_data.json',
typename='file',
parent_key='WORKDIR')
b.add_path('stage_1_jobs',
'stage_1_jobs.json',
typename='file',
parent_key='WORKDIR')
b.add_path('fit_1_file',
'fit_1_data.json',
typename='file',
parent_key='WORKDIR')
b.add_path('fit_2_file',
'fit_2_data.json',
typename='file',
parent_key='WORKDIR')
b.add_path('fit_3_file',
'fit_3_data.json',
typename='file',
parent_key='WORKDIR')
b.add_path('fit_type_path',
typename='file',
spec='%s',
parent_key='WORKDIR')
b.add_path('target_path',
typename='file',
spec='target.json',
parent_key='WORKDIR')
b.add_path('fit_config_json',
typename='file',
spec='%s/config.json',
parent_key='WORKDIR')
b.add_path('final_hof_fit',
typename='file',
spec='%s/s%d/final_hof_fit.txt',
parent_key='WORKDIR')
b.add_path('final_hof',
typename='file',
spec='%s/s%d/final_hof.txt',
parent_key='WORKDIR')
b.add_path('output_fit_file',
typename='file',
spec='fit_%s_%s.json')
b.add_section('biophys', {"biophys": [
{"model_file": [ manifest_path ] }]})
b.add_section('stimulus_conf',
{"runs": [{"sweeps": self.sweep_numbers(),
"specimen_id": self.lims_data['specimen_id']
}]})
b.add_section('hoc_conf',
{"neuron" : [{"hoc": [ "stdgui.hoc", "import3d.hoc", "cell.hoc" ]
}]})
return b
[docs] def to_manifest(self, manifest_path=None):
b = self.build_manifest(manifest_path)
m = Manifest(config=b.path_info)
if manifest_path != None:
b.write_json_file(manifest_path, overwrite=True)
return m