Source code for allensdk.model.biophysical.runner

# 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 ..biophys_sim.config import Config
from .utils import create_utils
from allensdk.core.nwb_data_set import NwbDataSet
import allensdk.ephys.extract_cell_features as extract_cell_features
from shutil import copy
import numpy
import logging
import time
import os
import multiprocessing as mp
from functools import partial

_runner_log = logging.getLogger('allensdk.model.biophysical.runner')

_lock = None

def _init_lock(lock):
    global _lock
    _lock = lock

[docs]def run(description, sweeps=None, procs=6): '''Main function for simulating sweeps in a biophysical experiment. Parameters ---------- description : Config All information needed to run the experiment. procs : int number of sweeps to simulate simultaneously. sweeps : list list of experiment sweep numbers to simulate. If None, simulate all sweeps. ''' prepare_nwb_output(description.manifest.get_path('stimulus_path'), description.manifest.get_path('output_path')) if procs == 1: run_sync(description, sweeps) return if sweeps is None: stimulus_path = description.manifest.get_path('stimulus_path') run_params = description.data['runs'][0] sweeps = run_params['sweeps'] lock = mp.Lock() pool = mp.Pool(procs, initializer=_init_lock, initargs=(lock,)) pool.map(partial(run_sync, description), [[sweep] for sweep in sweeps]) pool.close() pool.join()
[docs]def run_sync(description, sweeps=None): '''Single-process main function for simulating sweeps in a biophysical experiment. Parameters ---------- description : Config All information needed to run the experiment. sweeps : list list of experiment sweep numbers to simulate. If None, simulate all sweeps. ''' # configure NEURON utils = create_utils(description) h = utils.h # configure model manifest = description.manifest morphology_path = description.manifest.get_path('MORPHOLOGY') utils.generate_morphology(morphology_path.encode('ascii', 'ignore')) utils.load_cell_parameters() # configure stimulus and recording stimulus_path = description.manifest.get_path('stimulus_path') run_params = description.data['runs'][0] if sweeps is None: sweeps = run_params['sweeps'] sweeps_by_type = run_params['sweeps_by_type'] output_path = manifest.get_path("output_path") # run sweeps for sweep in sweeps: _runner_log.info("Loading sweep: %d" % (sweep)) utils.setup_iclamp(stimulus_path, sweep=sweep) _runner_log.info("Simulating sweep: %d" % (sweep)) vec = utils.record_values() tstart = time.time() h.finitialize() h.run() tstop = time.time() _runner_log.info("Time: %f" % (tstop - tstart)) # write to an NWB File _runner_log.info("Writing sweep: %d" % (sweep)) recorded_data = utils.get_recorded_data(vec) if _lock is not None: _lock.acquire() save_nwb(output_path, recorded_data["v"], sweep, sweeps_by_type) if _lock is not None: _lock.release()
[docs]def prepare_nwb_output(nwb_stimulus_path, nwb_result_path): '''Copy the stimulus file, zero out the recorded voltages and spike times. Parameters ---------- nwb_stimulus_path : string NWB file name nwb_result_path : string NWB file name ''' output_dir = os.path.dirname(nwb_result_path) if not os.path.exists(output_dir): os.makedirs(output_dir) copy(nwb_stimulus_path, nwb_result_path) data_set = NwbDataSet(nwb_result_path) data_set.fill_sweep_responses(0.0, extend_experiment=True) for sweep in data_set.get_sweep_numbers(): data_set.set_spike_times(sweep, [])
[docs]def save_nwb(output_path, v, sweep, sweeps_by_type): '''Save a single voltage output result into an existing sweep in a NWB file. This is intended to overwrite a recorded trace with a simulated voltage. Parameters ---------- output_path : string file name of a pre-existing NWB file. v : numpy array voltage sweep : integer which entry to overwrite in the file. ''' output = NwbDataSet(output_path) output.set_sweep(sweep, None, v) sweep_by_type = {t: [sweep] for t, ss in sweeps_by_type.items() if sweep in ss} sweep_features = extract_cell_features.extract_sweep_features(output, sweep_by_type) try: spikes = sweep_features[sweep]['spikes'] spike_times = [s['threshold_t'] for s in spikes] output.set_spike_times(sweep, spike_times) except Exception as e: logging.info("sweep %d has no sweep features. %s" % (sweep, e.args))
[docs]def load_description(manifest_json_path): '''Read configuration file. Parameters ---------- manifest_json_path : string File containing the experiment configuration. Returns ------- Config Object with all information needed to run the experiment. ''' description = Config().load(manifest_json_path) # fix nonstandard description sections fix_sections = ['passive', 'axon_morph,', 'conditions', 'fitting'] description.fix_unary_sections(fix_sections) return description
if '__main__' == __name__: import sys description = load_description(sys.argv[-1]) run(description)