common Package

common Package

common functions for the BOTMpy package

amplitude_histogram Module

amplitude histogram to track time-variant waveforms

class AmplitudeHistogram(ampl_range=(-2.0, 2.0, 0.1), ampl_noise=(-0.2, 0.2), nchan=1, bin_size=32000, bin_hist=3000)

Bases: object

amplitude histogram calculator

append_bin(bin)

append an AmplHistBin instance

Parameters:bin (ndarray like) – the amplHistBin to append

append_data_all(data, force=False)

append bin(s) calculated from a strip of data

with this method a histogram of the amplitude distribution of the passed data is generated as one observation and appended to the current amplitude histogram.

Parameters:

• data (ndarray) – the data to generate the bin(s) to append from
• force (bool) – if True, immediately start a new bin before calculation

append_data_peaks(data, force=False)

append bin(s) calculated from a strip of data

with this method the data is first queried for peaks. this should reduce the noise/smoothness of the histogram as observed from the amplitude distribution of the pure signal.

Parameters:

• data (ndarray) – the data to generate the bin(s) to append from
• force (bool) – if True, immediately start a new bin before calculation

force_new_bin()

force a new bin and finalize the current bin

get_channel_hist(channel=0, omit_noise=True)

yield the current amplitude histogram for the requested channel of the data

:type channel : int :param cahnnel: channel of the data :type omit_noise: bool :param omit_noise: if True, omit value in the noise band

get_hist(omit_noise=True)

yield tuple of amplitude histograms for all channels of the data

Parameters:omit_noise (bool) – if True, omit value in the noise band

plot_ampl_hist()

plot the amplitude histogram

static range(min_val, max_val, bin_size)

range function for the binning of the histogram

covariance_estimator Module

covariance estimator for timeseries data

class BaseTimeSeriesCovarianceEstimator(weight=0.05, cond=50, dtype=None)

Bases: object

covariance estimator base class

get_cmx(**kwargs)

get_cmx_loaded(**kwargs)

get_cond(**kwargs)

get_icmx(**kwargs)

get_icmx_loaded(**kwargs)

get_svd(**kwargs)

get_whitening_op(**kwargs)

is_cond_ok(**kwargs)

is_initialised()

reset()

reset the internal buffers to None

update(data, **kwargs)

update covariance matrix with epochs of :data:

Parameters:data (ndarray) – data vector [samples, channels] Raises:ValueError – data is not a ndarray with ndim == 2

class TimeSeriesCovE(tf_max=100, nc=4, weight=0.05, cond=50, with_default_chan_set=True, dtype=None)

Bases: botmpy.common.covariance_estimator.BaseTimeSeriesCovarianceEstimator

covariance estimator for timeseries data

Given strips of (multi-channeled) data, this covariance estimator is able to estimate the time-lagged (block-)covariance-matrix, which has a (block-)toeplitz structure. Estimates are built by taking (cross- and) auto-correlations of the (multi-channeled) data pieces, for all lags in the defined range. From this data a toeplitz matrix is build (for each combination of channels).

static cmx_init(cmx, nc, tf_max=None, dtype=None)

init from given covariance matrix

Parameters:

• cmx (ndarray) – covariance matrix in concatenated representation
• nc (int) – number of channels
• tf_max (int) –

  max length of template in samples, if None derive it from the passed matrix.

  Default=None

Return type:

TimeSeriesCovE

Returns:

TimeSeriesCovE with the passed matrix set

get_chan_set()

get_nc()

get_tf_max()

nc

new_chan_set(cs_new)

rm_chan_set(cs_rm)

set_tf_max(value)

tf_max

static white_noise_init(tf_max, nc, std=1.0, dtype=None)

init as white noise

Parameters:

• tf_max (int) – max length of template in samples
• nc (int) – number of channels
• std (float) –

  standard deviation of white noise process

  Default=1.0

• dtype (dtype resolvable) –

Return type:

TimeSeriesCovE

Returns:

TimeSeriesCovE with white noise init

class XcorrStore(tf=100, nc=4)

Bases: object

storage for cross-correlations

reset()

build_idx_set(ids)

builds the block index set for an upper triangular matrix

Parameters:ids (iterable) – set of channel ids

build_block_toeplitz_from_xcorrs(tf, chan_set, xcorrs, dtype=None)

builds a block toeplitz matrix from a set of channel xcorrs

Parameters:

• tf (int) – desired lag in samples
• chan_set (list) – list of channel ids to build the channel set from. the block covariance matrix will be build so that blocks are ordered from lower to higher channel id.
• xcorrs (XcorrStore) – XcorrStore object holding the xcorrs for various channel combinations
• dtype (dtype derivable) – will be passed to the constructor for the matrix returned. Default=None

funcs_filterutil Module

filter related utility functions

!! multichanneled crosscorrelation filtering implementations are outsourced to botmpy.common.mcfilter !!

xi_vs_f(xi, f, nc=4)

cross-correlation-tensor for a set of matched patterns and filters

The xcorr-tensor for a set of patterns (xi) and their matched filters (f) with a certain lag is returned as ndarray with dimensions [xi, f, tau]. All multichanneled vectors are presented in their concatenated form.

Parameters:

• xi (ndarray) – The patterns, one concatenated pattern per row.
• f (ndarray) – The filters, one concatenated filter per row.
• nc (int) –

  The channel count for the concatenated patterns and filters.

  Default=4

Returns:

ndarray - The tensor of cross-correlation for each pattern with each filter. Dimensions as [xi, f, xcorr].

kteo(data, k=1)

teager energy operator of range k [TEO]

The discrete teager energy operator (TEO) of window size k is defined as: M{S{Psi}[x(n)] = x^2(n) - x(n-k) x(n+k)}

Parameters:

• data (ndarray) – The signal to operate on. ndim=1
• k (int) – Parameter defining the window size for the TEO.

Returns:

ndarray - Array of same shape as the input signal, holding the kteo response.

Except:

If inconsistant dims or shapes.

mteo(data, kvalues=[1, 3, 5], condense=True)

multiresolution teager energy operator using given k-values [MTEO]

The multi-resolution teager energy operator (MTEO) applies TEO operators of varying k-values and returns the reduced maximum response TEO for each input sample.

To assure a constant noise power over all kteo channels, we convolve the individual kteo responses with a window: h_k(i) = hamming(4k+1) / sqrt(3sum(hamming(4k+1)^2) + sum(hamming(4k+1)) ^2), as suggested in Choi et al., 2006.

Parameters:

• data (ndarray) – The signal to operate on. ndim=1
• kvalues (list) – List of k-values to run the kteo for. If you want to give a single k-value, either use the kteo directly or put it in a list like [2].
• condense (bool) – if True, use max operator condensing onto one time series, else return a multichannel version with one channel per kvalue. Default=True

Returns:

ndarray- Array of same shape as the input signal, holding the response of the kteo which response was maximum after smoothing for each sample in the input signal.

funcs_general Module

general utility functions

sortrows(data)

sort matrix by rows

Parameters:data (ndarray) – ndarray that should be sorted by its rows Returns:ndarray - data sorted by its rows.

vec2ten(data, nchan=4)

converts from templates/spikes that are concatenated across the channels to tensors that have an extra dim for the channels

Parameters:

• data (ndarray) – input array [templates][vars * channels]
• nchan (int) – count of channels Default=4

Returns:

ndarray - data converted to tensor [templates][vars][channels]

ten2vec(data)

converts from templates/spikes that are not concatenated across the channels to vectors.

Parameters:data (ndarray) – input array [templates][vars][channels] Returns:ndarray- data converted to concatenated vectors [templates][channels * vars]

mcvec_to_conc(x)

returns the concatenated vector for a multichanneled vector

Parameters:x (ndarray) – multi-channeled vector in matrix form Returns:ndarray - multi-channeled vector in channel concatenated form

mcvec_from_conc(x, nc=4)

returns the multichanneled vector from a concatenated representation

Parameters:

• x (ndarray) – multi-channeled vector in channel concatenated form
• nc (int) – channel count

Returns:

ndarray - multi-channeled vector in matrix form

xcorr(a, b=None, lag=None, normalise=False, unbiased=False)

cross-correlation for one-dimensional input signals of equal size

If :b: is not given the auto-correlation of :a: will be computed.

Parameters:

• a (ndarray) – one-dimensional time series
• b (ndarray) – one-dimensional time series, if None :a: will be taken instead Default=None
• lag (int) – lag up to which the cross correlation will be calculated. If None all possible lags (2*a.size-1) will be computed. Default=None
• normalise (bool) – if True, normalise Default=True
• unbiased (bool) – if True and :normalise: is True, use a.size-|tau| to normalize instead of a.size Default=False

Returns:

ndarray - cross-correlate of :a: and :b: upt to lags :lag:

shifted_matrix_sub(data, sub, tau, pad_val=0.0)

Subtracts the multi-channeled vector (rows are channels) y from the vector x with a certain offset. x and y can due to the offset be only partly overlapping.

REM: from matlab

Parameters:

• data (ndarray) – data array to apply the subtractor to
• sub (ndarray) – subtractor array
• tau (int) – offset of :sub: w.r.t. start of :data:
• pad_val (float) – value to use for the padding Default=0.0

Returns:

ndarray - data minus sub at offset, len(data)

dict_sort_ndarrays(in_dict)

sort all arrays in a dictionary, works recursively

Parameters:in_dict (dict) – dict to work on Returns:

dict_list_to_ndarray(in_dict)

converts all lists in a dictionary to ndarray, works recursively

Parameters:in_dict (dict) – dict to workd on Returns:dict - all list are converted to ndarray

get_idx(idxset, append=False)

yields the first free index in a positive integer index set

Parameters:append (bool) – if True, returns max(:idxset:)+1, else find the first free index in :idxset: Returns:int - the first free index

funcs_spike Module

functions for spike sorting

threshold_detection(data, th, min_dist=1, mode='gt', find_max=True)

detect events by applying a threshold to the data

Parameters:

• data (ndarray) – the 2d-data to apply the threshold on. channels are in the second dimension (columns). Required
• th (ndarray or list) – list of threshold values, one value per channel in the data Required
• min_dist (int) – minimal distance two successive events have to be separated in samples, else the event is ignored. Default=1
• mode (str) – one of ‘gt’ for greater than or ‘lt’ for less than. will determine how the threshold is applied. Default=’gt’
• find_max (bool) – if True, will find the maximum for each event epoch, else will find the start for each event epoch. Default=True

Return type:

ndarray

Returns:

event samples

merge_epochs(*args, **kwargs)

for a set of epoch sets check if the combined set of epochs overlap and merge to one set with no overlapping epochs and no epochs of negative length.

Parameters:

• args – arbitrary count of epoch sets [[start, end]]
• min_dist – int - If present and greater than zero, this integer will be taken as the minimum distance in between epochs that is allowed. Should the gap in between two epochs smaller than min_dist, they are merged including the gap. This might reduce the segmentation of the data.

Returns:

ndarray - merged epoch set [[start, end]]

invert_epochs(epochs, end=None)

inverts epochs inverted

The first epoch will be mapped to [0, start] and the last will be mapped to [end of last epoch, :end:]. Epochs that accidentally become negative or zero-length will be omitted.

Parameters:

• epochs (ndarray) – epoch set to invert
• end (int) – If not None, it i taken for the end of the last epoch, else max(index-dtype) is taken instead. Default=None

Returns:

ndarray - inverted epoch set

epochs_from_binvec(binvec)

returns the discrete epochs where the :binvec: is true

Parameters:binvec (ndarray) – one-domensinal boolean ndarray. Returns:ndarray - epoch set where :binvec: is True [[start, end]]

epochs_from_spiketrain(st, cut, end=None, with_corrected_st=False)

yields epoch set, given a spiketrain and cut parameter

Parameters:

• st (ndarray) – spiketrains as 1d array
• cut (tuple) – 2-tuple of cutting parameters, (cut_left,cut_right) spike epochs will be generated by using cut_left and cut_right on the spike time. If an int is given, a symmetric cut tuple is assumed.
• end (int) – to determine potential problems with epochs overlapping data boundaries. If an event in the spiketrain is closer to 0 than :cut[0]: or closer to :end: than :cut[1]: the corresponding epoch will be omitted. If None, :end: will be set to max(INDEX_DTYPE) Default=None
• with_corrected_st (bool) – if True, return the corrected spiketrain by omitting spike events that cannot generate valid spike epochs given the passed cut settings. Default=False

Returns:

ndarray - epoch set of valid spike epochs, and if :with_corrected_st: is True additionally the corrected spike train

epochs_from_spiketrain_set(sts, cut, end=None)

yields epoch sets, given a spiketrain set and cut parameter

one set for each unit plus one for the noise epochs in a dict

Parameters:

• sts (dict) – dict with the spiketrains for each unit in the set. none of the units in the spiketrain set may have the key ‘noise’!
• cut (tuple) – 2-tuple of cutting parameters, (cut_left, cut_right) spike epochs will be generated by using cu_left and cut_right on the spike time. If an int is given, a symmetric cut tuple is assumed.
• end – to determine potential problems with epochs overlapping data boundaries. If an event in the spiketrain is closer to 0 than :cut[0]: or closer to :end: than :cut[1]: the corresponding epoch will be omitted. If None, :end: will be set to max(INDEX_DTYPE) Default=None

Returns:

dict - one epoch set per spike train plus the merged noise epoch set.

chunk_data(data, epochs=None, invert=False, min_len=0)

returns a generator of chunks from data given epochs

Parameters:

• data (ndarray) – signal data [[samples, channels]]
• epochs (ndarray) – epoch set, positive mask
• invert (bool) – invert epochs, negative mask instead of positive mask
• min_len (int) – epochs with fewer samples will be ignored and not returned

Returns:

generator - data chunks as per :epochs:

extract_spikes(data, epochs, mc=False)

extract spike waveforms of size tf from data

Parameters:

• data (ndarray) – signal data [[samples, channels]]
• epochs (ndarray) – spike epoch set [[start,end]]
• mc (bool) – if True, extract multichanneled spike waveforms as [n,tf,nc] else extract channel concatenated spike waveforms as [n, tf*nc] False as default for legacy compatibility Default=False

Returns:

ndarray - extracted spike data epochs

get_cut(tf, off=0)

cut 2-tuple (cut_left,cut_right) generating function

Used to generate epochs from events. Per default the epoch will be placed symmetrically around the event sample. :off: can be used to influence the placement. For odd tf values the extension of the cut_right part will be favored.

Parameters:

• tf (int) – length of the waveform in samples
• off (int) – offset for epoch start/end Default=0

snr_maha(waveforms, invC, mu=None)

SNR from Mahalanobis distance (generalised euclidean distance)

Definition of signal to noise ratio (SNR) as derived from the Mahalanobis distance. For C=eye this is equivalent to snr_power.

Parameters:

• waveforms (ndarray) – waveform data (signal), one per row
• invC (ndarray) – inverted noise covariance matrix (a block toeplitz matrix)
• mu (ndarray) – mean correction. Usually we assume zero-mean waveforms, so if this is None it will be ignored. Default=None

Returns:

ndarray - SNR per waveform

snr_peak(waveforms, noise_var)

SNR from instantaneous variance

Definition of signal to noise ratio (SNR) as the ratio between the peak of a waveforms and the noise standard deviation.

Parameters:

• waveforms (ndarray) – waveform data (signal), one per row
• noise_var (float) – instantaneous variance of the noise (noise)

Returns:

ndarray - SNR per waveform

snr_power(waveforms, noise_var)

SNR from signal energy

Definition of signal to noise ratio (SNR) using the waveforms energy as defined by Rutishauser et al (2006)

Parameters:

• waveforms (ndarray) – waveform data (signal), one per row
• noise_var (float) – instantaneous variance of the noise (noise)

Returns:

ndarray - SNR per waveform

overlaps(sts, window)

produces dict of boolean sequences indicating for all spikes in all spike trains in :sts: if it participates in an overlap event.

Parameters:

• sts (dict) – spike train set
• window (int) – overlap window size

Returns:

dict - boolean spike train set, dict - number of overlaps for each unit

matrix_ops Module

matrix operations

matrix_cond(mat)

yield the matrix condition number w.r.t. l2-norm (using svd)

Parameters:mat (ndarray) – input matrix Returns:float - condition number of :mat: or :inf: on error

diagonal_loading(mat, target_cond=50, overwrite_mat=False)

tries to condition the :mat: by imposing a spherical constraint on the covariance ellipsoid (adding alpha*eye)

solves: cond(mat + alpha*I) = target_cond for alpha

Note: this is a noop if the condition is already >= target_cond!

Parameters:

• mat (ndarray) – input matrix
• target_cond (float) – condition number to archive after loading
• overwrite_mat (bool) – if True, operate inplace and overwrite :mat:

Returns:

ndarray - matrix like :mat: conditioned s.t. cond = target_cond

coloured_loading(mat, target_cond=50, overwrite_mat=False)

tries to condition :mat: by inflating the badly conditioned subspace of :mat: using a spherical constraint.

Parameters:

• mat (ndarray) – input matrix
• target_cond (float) – condition number to archive after loading
• overwrite_mat (bool) – if True, operate inplace and overwrite :mat:

Returns:

ndarray - matrix like :mat: conditioned s.t. cond = target_cond

matrix_argmax(mat)

returns the indices (row,col) of the maximum value in :mat:

Parameters:mat (ndarray) – input matrix Returns:tuple - (row,col) of the maximum value in :mat:

matrix_argmin(mat)

returns the indices (row,col) of the minimum value in :mat:

Parameters:mat (ndarray) – input matrix Returns:tuple - (row,col) of the minimum value in :mat:

ringbuffer Module

matrix based ringbuffer implementation

class MxRingBuffer(capacity=64, dimension=1, dtype=None)

Bases: object

ringbuffer implementation based on pre-allocated ndarray

Ringbuffer behavior is archived by cycling though the buffer forward, wrapping around to the start upon reaching capacity.

append(datum)

append one datum at the end of the buffer, overwriting the oldest datum in the buffer if the capacity has been reached.

Parameters:datum (ndarray) – ndarray of shape :self.dimension:

capacity

clear()

clears the data and resets internals

dimension

extend(iterable)

append iterable at the end of the buffer using multiple append’s

Parameters:iterable (iterable) – iterable of objects to be stored in the ringbuffer

fill(datum)

fill all slots of the ringbuffer with the same datum.

Parameters:daaum – ndarray of shape :self.dimension:

flush()

return the buffer as a list and clear the RingBuffer

Convenience method. This returns self.tolist() and calls self.clear() afterwards.

Returns:list - the buffer as a python list of the objects stored

get_capacity()

get_dimension()

get_is_full()

is_full

mean(last=None)

yields the mean over the :last: entries

Parameters:last (int) – number entries from the back of the ringbuffer to include for mean calculation. If None, use all contents Default=None Returns:ndarray(self.dimension) - mean over the last entries, or the appropriate zero element if the ringbuffer is empty.

set_capacity(value)

tolist()

return the buffer as a list

Returns:list- the buffer as a python list

spike_alignment Module

spikes alignment functions

sinc_interp1d(x, s, r)

Interpolates x, sampled at times s Output y is sampled at times r

inspired from from Matlab: http://phaseportrait.blogspot.com/2008/06/sinc-interpolation-in-matlab.html

Parameters:

• x (ndarray) – input data time series
• s (ndarray) – input sampling time series (regular sample interval)
• r (ndarray) – output sampling time series

Return ndarray:

output data time series (regular sample interval)

get_tau_for_alignment(spikes, align_at)

return the per spike offset in samples (taus) of the maximum values to the desired alignment sample within the spike waveform.

Parameters:

• spikes (ndarray) – stacked mc spike waveforms [ns, tf, nc]
• align_at (int) – sample to align the maximum at

Returns:

ndarray - offset per spike

get_tau_align_min(spks, ali)

get_tau_align_max(spks, ali)

get_tau_align_energy(spks, ali)

get_aligned_spikes(data, spike_train, align_at=-1, tf=47, look_ahead=0, mc=True, kind='none', rsf=1.0, sample_back=True)

return the set of aligned spikes waveforms and the aligned spike train

Parameters:

• data (ndarray) – data with channels in the columns
• spike_train (ndarray or list) – spike train of events in data
• align_at (int) – align feature at this sample in the waveform
• tf (int) – temporal extend of the waveform in samples
• look_ahead (int) – samples to look beyond the cut window for finding the align feature
• mc (bool) – if True, return mc waveforms, else return concatenated waveforms. Default=True
• kind (str) –

  String giving the type of alignment to conduct. One of:

  • “max” - align on maximum of the waveform
  • “min” - align on minimum of the waveform
  • “energy” - align on peak of energy
  • “none” - no alignment

  Default=’none’

• rsf (float) – resampling factor (use integer values of powers of 2)
• sample_back (bool) – if True, resample spikes to original length after resampling

Return type:

ndarray, ndarray

Returns:

stacked spike events, spike train with events corrected for alignment

util Module

constants for the common package

class VERBOSE(level=None, has_print=False, has_plot=False)

Bases: object

verbosity manager

NONE = 0

PLOT = 16

PRINT = 1

get_has_plot(n=0)

get_has_print(n=0)

get_is_verbose()

has_plot

has_print

is_verbose

Subpackages

• datafile Package
  • datafile Package
  • atf Module
  • datafile Module
  • gdf Module
  • nas Module
  • wri Module
  • xpd Module
• mcfilter Package
  • mcfilter Package
  • mcfilter_py Module