Source code for NeuroTools.io

# -*- coding: utf-8 -*-
"""
NeuroTools.io
==================

A collection of functions to handle all the inputs/outputs of the NeuroTools.signals
file, used by the loaders.

Classes
-------

FileHandler        - abstract class which should be overriden, managing how a file will load/write
                     its data
StandardTextFile   - object used to manipulate text representation of NeuroTools objects (spikes or
                     analog signals)
StandardPickleFile - object used to manipulate pickle representation of NeuroTools objects (spikes or
                     analog signals)
NestFile           - object used to manipulate raw NEST file that would not have been saved by pyNN
                     (without headers)
DataHandler        - object to establish the interface between NeuroTools.signals and NeuroTools.io

All those objects can be used with NeuroTools.signals

    >> data = StandardTextFile("my_data.dat")
    >> spikes = load(data,'s')
"""


from NeuroTools import check_dependency

import os, logging, cPickle, numpy
DEFAULT_BUFFER_SIZE = -1



[docs]class FileHandler(object): """ Class to handle all the file read/write methods for the key objects of the signals class, i.e SpikeList and AnalogSignalList. Could be extented This is an abstract class that will be implemented for each format (txt, pickle, hdf5) The key methods of the class are: write(object) - Write an object to a file read_spikes(params) - Read a SpikeList file with some params read_analogs(type, params) - Read an AnalogSignalList of type `type` with some params Inputs: filename - the file name for reading/writing data If you want to implement your own file format, you just have to create an object that will inherit from this FileHandler class and implement the previous functions. See io.py for more details """ def __init__(self, filename): self.filename = filename def __str__(self): return "%s (%s)" % (self.__class__.__name__, self.filename)
[docs] def write(self, object): """ Write the object to the file. Examples: >> handler.write(SpikeListObject) >> handler.write(VmListObject) """ return _abstract_method(self)
[docs] def read_spikes(self, params): """ Read a SpikeList object from a file and return the SpikeList object, created from the File and from the additional params that may have been provided Examples: >> params = {'id_list' : range(100), 't_stop' : 1000} >> handler.read_spikes(params) SpikeList Object (with params taken into account) """ return _abstract_method(self)
[docs] def read_analogs(self, type, params): """ Read an AnalogSignalList object from a file and return the AnalogSignalList object of type `type`, created from the File and from the additional params that may have been provided `type` can be in ["vm", "current", "conductance"] Examples: >> params = {'id_list' : range(100), 't_stop' : 1000} >> handler.read_analogs("vm", params) VmList Object (with params taken into account) >> handler.read_analogs("current", params) CurrentList Object (with params taken into account) """ if not type in ["vm", "current", "conductance"]: raise Exception("The type %s is not available for the Analogs Signals" %type) return _abstract_method(self)
[docs]class StandardTextFile(FileHandler): def __init__(self, filename): FileHandler.__init__(self, filename) self.metadata = {} def __read_metadata(self): """ Read the informations that may be contained in the header of the NeuroTools object, if saved in a text file """ cmd = '' variable = None label = None f = open(self.filename, 'r') for line in f.readlines(): if line[0] == '#': if line[1:].strip().find('variable') != -1: variable = line[1:].strip().split(" = ") elif line[1:].strip().find('label') != -1: label = line[1:].strip().split(" = ") else: cmd += line[1:].strip() + ';' else: break f.close() exec cmd in None, self.metadata if not variable is None: self.metadata[variable[0]] = variable[1] if not variable is None: self.metadata[label[0]] = label[1] def __fill_metadata(self, object): """ Fill the metadata from those of a NeuroTools object before writing the object """ self.metadata['dimensions'] = str(object.dimensions) if len(object.id_list) > 0: self.metadata['first_id'] = numpy.min(object.id_list) self.metadata['last_id'] = numpy.max(object.id_list) if hasattr(object, "dt"): self.metadata['dt'] = object.dt def __check_params(self, params): """ Establish a control/completion/correction of the params to create an object by using comparison and data extracted from the metadata. """ if 'dt' in params: if params['dt'] is None and 'dt' in self.metadata: logging.debug("dt is infered from the file header") params['dt'] = self.metadata['dt'] if not ('id_list' in params) or (params['id_list'] is None): if ('first_id' in self.metadata) and ('last_id' in self.metadata): params['id_list'] = range(int(self.metadata['first_id']), int(self.metadata['last_id'])+1) logging.debug("id_list (%d...%d) is infered from the file header" % (int(self.metadata['first_id']), int(self.metadata['last_id'])+1)) else: raise Exception("id_list can not be infered while reading %s" %self.filename) elif isinstance(params['id_list'], int): # allows to just specify the number of neurons params['id_list'] = range(params['id_list']) elif not isinstance(params['id_list'], list): raise Exception("id_list should be an int or a list !") if not ('dims' in params) or (params['dims'] is None): if 'dimensions' in self.metadata: params['dims'] = self.metadata['dimensions'] else: params['dims'] = len(params['id_list']) return params
[docs] def get_data(self, sepchar = "\t", skipchar = "#"): """ Load data from a text file and returns an array of the data """ myfile = open(self.filename, "r", DEFAULT_BUFFER_SIZE) contents = myfile.readlines() myfile.close() data = [] header = True idx = 0 while header and idx < len(contents): if contents[idx][0] != skipchar: header = False break idx += 1 for i in xrange(idx, len(contents)): line = contents[i].strip().split(sepchar) id = [float(line[-1])] id += map(float, line[0:-1]) data.append(id) logging.debug("Loaded %d lines of data from %s" % (len(data), self)) data = numpy.array(data, numpy.float32) return data
[docs] def write(self, object): # can we write to the file more than once? In this case, should use seek, tell # to always put the header information at the top? # write header self.__fill_metadata(object) fileobj = open(self.filename, 'w', DEFAULT_BUFFER_SIZE) header_lines = ["# %s = %s" % item for item in self.metadata.items()] fileobj.write("\n".join(header_lines) + '\n') numpy.savetxt(fileobj, object.raw_data(), fmt = '%g', delimiter='\t') fileobj.close()
[docs] def read_spikes(self, params): self.__read_metadata() p = self.__check_params(params) from NeuroTools.signals import spikes data = self.get_data() result = spikes.SpikeList(data, p['id_list'], p['t_start'], p['t_stop'], p['dims']) del data return result
[docs] def read_analogs(self, type, params): self.__read_metadata() p = self.__check_params(params) from NeuroTools.signals import analogs if type == "vm": return analogs.VmList(self.get_data(), p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) elif type == "current": return analogs.CurrentList(self.get_data(), p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) elif type == "conductance": data = numpy.array(self.get_data()) if len(data[0,:]) > 2: g_exc = analogs.ConductanceList(data[:,[0,1]] , p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) g_inh = analogs.ConductanceList(data[:,[0,2]] , p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) return [g_exc, g_inh] else: return analogs.ConductanceList(data, p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims'])
[docs]class StandardPickleFile(FileHandler): # There's something kinda wrong with this right now... def __init__(self, filename): FileHandler.__init__(self, filename) self.metadata = {} def __fill_metadata(self, object): """ Fill the metadata from those of a NeuroTools object before writing the object """ self.metadata['dimensions'] = str(object.dimensions) self.metadata['first_id'] = numpy.min(object.id_list) self.metadata['last_id'] = numpy.max(object.id_list) if hasattr(object, 'dt'): self.metadata['dt'] = object.dt def __reformat(self, params, object): self.__fill_metadata(object) if 'id_list' in params and params['id_list'] != None: id_list = params['id_list'] if isinstance(id_list, int): # allows to just specify the number of neurons params['id_list'] = range(id_list) #if id_list != range(int(self.metadata['first_id']), int(self.metadata['last_id'])+1): # object = object.id_slice(params['id_list']) do_slice = False t_start = object.t_start t_stop = object.t_stop if 't_start' in params and params['t_start'] is not None and params['t_start'] != object.t_start: t_start = params['t_start'] do_slice = True if 't_stop' in params and params['t_stop'] is not None and params['t_stop'] != object.t_stop: t_stop = params['t_stop'] do_slice = True if do_slice: object = object.time_slice(t_start, t_stop) return object
[docs] def write(self, object): fileobj = file(self.filename,"w") return cPickle.dump(object, fileobj)
[docs] def read_spikes(self, params): fileobj = file(self.filename,"r") object = cPickle.load(fileobj) object = self.__reformat(params, object) return object
[docs] def read_analogs(self, type, params): return self.read_spikes(params)
[docs]class NestFile(FileHandler): def __init__(self, filename, padding=0, with_time=False, with_gid=True): self.filename = filename self.metadata = {} assert (padding >= 0) and (type(padding) == int), "ERROR ! padding should be a positive int" self.padding = padding self.with_time = with_time self.with_gid = with_gid self.standardtxtfile = StandardTextFile(filename)
[docs] def write(self, object): """ Write the object to the file. Examples: >> handler.write(SpikeListObject) >> handler.write(VmListObject) """ return self.standardtxtfile.write(object)
def __check_params(self, params): """ Establish a control/completion/correction of the params to create an object by using comparison and data extracted from the metadata. """ if 'dt' in params: if params['dt'] is None and 'dt' in self.metadata: logging.debug("dt is infered from the file header") params['dt'] = self.metadata['dt'] if params['id_list'] is None: print "WARNING: id_list will be infered based on active cells..." elif isinstance(params['id_list'], int): # allows to just specify the number of neurons params['id_list'] = range(params['id_list']) elif not isinstance(params['id_list'], list): raise Exception("id_list should be an int or a list !") if params['dims'] is None: if 'dimensions' in self.metadata: params['dims'] = self.metadata['dimensions'] else: raise Exception("dims can not be infered while reading %s" %self.filename) return params
[docs] def get_data(self, sepchar = "\t", skipchar = "#"): """ Load data from a text file and returns a list of data """ if HAVE_TABLEIO: data = TableIO.readTableAsArray(self.filename, skipchar) else: myfile = open(self.filename, "r", DEFAULT_BUFFER_SIZE) contents = myfile.readlines() myfile.close() data = [] header = True idx = 0 while header: if contents[idx][0] != skipchar: header = False break idx += 1 for i in xrange(idx, len(contents)): line = contents[i].strip().split(sepchar) id = [float(line[0])] id += map(float, line[1:]) data.append(id) return numpy.array(data)
def _fix_id_list(self, data, params): print "All gids are shifted by padding", self.padding data[:,0] = numpy.array(data[:,0], int) - self.padding if params['id_list'] is None: params['id_list'] = numpy.unique(data[:,0]) return data, params
[docs] def read_spikes(self, params): """ Read a SpikeList object from a file and return the SpikeList object, created from the File and from the additional params that may have been provided Examples: >> params = {'id_list' : range(100), 't_stop' : 1000} >> handler.read_spikes(params) SpikeList Object (with params taken into account) """ p = self.__check_params(params) from NeuroTools import signals data = self.get_data() data, p = self._fix_id_list(data, p) return signals.SpikeList(data, p['id_list'], p['t_start'], p['t_stop'], p['dims'])
[docs] def read_analogs(self, type, params): p = self.__check_params(params) data = self.get_data() data, p = self._fix_id_list(data, p) from NeuroTools.signals import analogs if type == "vm": return analogs.VmList(data, p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) elif type == "current": return analogs.CurrentList(data, p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) elif type == "conductance": if len(data[0,:]) > 2: g_exc = analogs.ConductanceList(data[:,[0,1]] , p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) g_inh = analogs.ConductanceList(data[:,[0,2]] , p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims']) return [g_exc, g_inh] else: return analogs.ConductanceList(data, p['id_list'], p['dt'], p['t_start'], p['t_stop'], p['dims'])
[docs]class PyNNNumpyBinaryFile(FileHandler): def __init__(self, filename): FileHandler.__init__(self, filename) self.fileobj = open(self.filename, 'r')
[docs] def read_spikes(self, params): from NeuroTools.signals import spikes contents = numpy.load(self.fileobj) spike_data = contents['data'][:, (1,0)] self.metadata = M = {} for k,v in contents['metadata']: M[k] = eval(v) id_list = range(M['first_id'], M['last_id']) t_stop = params['t_stop'] # really need to check the agreement between file metadata and # params for all metadata items return spikes.SpikeList(spike_data, id_list, t_start=0.0, t_stop=t_stop, dims=M['dimensions']) #def read_analogs(self, type, params): # contents = numpy.load(self.fileobj) # values, ids = contents['data'].T # need to check the shape first
[docs]class DataHandler(object): """ Class to establish the interface for loading/saving objects in NeuroTools Inputs: filename - the user file for reading/writing data. By default, if this is string, a StandardTextFile is created object - the object to be saved. Could be a SpikeList or an AnalogSignalList Examples: >> txtfile = StandardTextFile("results.dat") >> DataHandler(txtfile) >> picklefile = StandardPickelFile("results.dat") >> DataHandler(picklefile) """ def __init__(self, user_file, object = None): if type(user_file) == str: user_file = StandardTextFile(user_file) elif not isinstance(user_file, FileHandler): raise Exception ("The user_file object should be a string or herits from FileHandler !") self.user_file = user_file self.object = object
[docs] def load_spikes(self, **params): """ Function to load a SpikeList object from a file. The data type is automatically infered. Return a SpikeList object Inputs: params - a dictionnary with all the parameters used by the SpikeList constructor Examples: >> params = {'id_list' : range(100), 't_stop' : 1000} >> handler.load_spikes(params) SpikeList object See also SpikeList, load_analogs """ ### Here we should have an automatic selection of the correct manager ### acccording to the file format. ### For the moment, we try the pickle format, and if not working ### we assume this is a text file logging.debug("Loading spikes from %s, with parameters %s" % (self.user_file, params)) return self.user_file.read_spikes(params)
[docs] def load_analogs(self, type, **params): """ Read an AnalogSignalList object from a file and return the AnalogSignalList object of type `type`, created from the File and from the additional params that may have been provided `type` can be in ["vm", "current", "conductance"] Examples: >> params = {'id_list' : range(100), 't_stop' : 1000} >> handler.load_analogs("vm", params) VmList Object (with params taken into account) >> handler.load_analogs("current", params) CurrentList Object (with params taken into account) See also AnalogSignalList, load_spikes """ ### Here we should have an automatic selection of the correct manager ### acccording to the file format. ### For the moment, we try the pickle format, and if not working ### we assume this is a text file logging.debug("Loading analog signal of type '%s' from %s, with parameters %s" % (type, self.user_file, params)) return self.user_file.read_analogs(type, params)
[docs] def save(self): """ Save the object defined in self.object with the method os self.user_file Note that you can add your own format for I/O of such NeuroTools objects """ ### Here, you could add your own format if you have created the appropriate ### manager. ### The methods of the manager are quite simple: should just inherits from the FileHandler ### class and have read() / write() methods if self.object == None: raise Exception("No object has been defined to be saved !") else: self.user_file.write(self.object)