Take a zipped timeseries and peaks found in it
    and integrate it to return peaks. Used to allow
    multiprocessing support.
437815
def _integrate_mpwrap(ts_and_pks, integrate, fopts):
    
    ts, tpks = ts_and_pks
    pks = integrate(ts, tpks, **fopts)
    # for p in pks:
    #     p.info['mz'] = str(ts.name)
    return pks
1058041
def peak_interval(self, name, alpha=_alpha, npoints=_npoints, **kwargs):
        
        data = self.get(name, **kwargs)
        return peak_interval(data,alpha,npoints)
800199
def zero_pad(ts, n_zeros):
    
    
    zeros_shape = ts.shape[:-1] + (n_zeros,)
    zzs = np.zeros(zeros_shape)
    # Concatenate along the time-dimension:
    new_data = np.concatenate((zzs, ts.data, zzs), axis=-1)

    return nts.TimeSeries(new_data, sampling_rate=ts.sampling_rate)
646765
def trend(self, order=LINEAR):
        ''''''
        coefficients = self.trend_coefficients(order)
        x = self.timestamps
        trend_y = LazyImport.numpy().polyval(coefficients, x)
        return TimeSeries(zip(x, trend_y))
468198
def merge(tup):
    
    if not all(tuple(ts.shape[1:] == tup[0].shape[1:] for ts in tup[1:])):
        raise ValueError('Timeseries to merge must have compatible shapes')
    indices = np.vstack(tuple(ts.tspan for ts in tup)).argsort()
    return np.vstack((tup))[indices]
1297146
def _convert_results(self, data, **kwargs):
        

        if kwargs.get("multiprocessing", False):
            m = mp.Manager()
            output = m.Queue()
            pdb.set_trace()
            # processes = [mp.Process(target=convert_row_main,
            #                         args=(row, output,))
            #              for row in data]
            # # Run processes
            # for p in processes:
            #     p.start()

            # # Exit the completed processes
            # for p in processes:
            #     p.join()
            # # Get process results from the output queue
            # return [output.get() for p in processes]

            pool = mp.Pool(processes=pool_size)
            for i, row in enumerate(data):
                for key, val in row.items():
                    try:
                        pool.apply(convert_row_main, args=(val, i, key, output,))
                    except:
                        pass #
            # run = [pool.apply(convert_row_main, args=(row, i, output))
            #        for i, row in enumerate(data)]
            for item in output:
                pdb.set_trace()
            return output
            # with multiprocessing.Pool(processes=pool_size) as pool:
            #     results = [convert_row_main, (row,))
            #                for row in data]
            #     converted = [r.get() for r in results]
            # return converted #pool_outputs
        else:
            return [{key:pyrdf(value) for key, value in row.items()}
                    for row in data]
900705
def get_results(self, stream, time_interval):
        
        return [StreamInstance(t, self.data[stream.stream_id][t])
                for t in sorted(self.data[stream.stream_id]) if t in time_interval]
1270672
def _postprocess_for_cut(fac, bins, retbins, x_is_series,
                         series_index, name, dtype):
    
    if x_is_series:
        fac = Series(fac, index=series_index, name=name)

    if not retbins:
        return fac

    bins = _convert_bin_to_datelike_type(bins, dtype)

    return fac, bins
1191665
def _format_return_timestamps(self, return_timestamps=None):
        
        if return_timestamps is None:
            # Build based upon model file Start, Stop times and Saveper
            # Vensim's standard is to expect that the data set includes the `final time`,
            # so we have to add an extra period to make sure we get that value in what
            # numpy's `arange` gives us.
            return_timestamps_array = np.arange(
                self.components.initial_time(),
                self.components.final_time() + self.components.saveper(),
                self.components.saveper(), dtype=np.float64
            )
        elif inspect.isclass(range) and isinstance(return_timestamps, range):
            return_timestamps_array = np.array(return_timestamps, ndmin=1)
        elif isinstance(return_timestamps, (list, int, float, np.ndarray)):
            return_timestamps_array = np.array(return_timestamps, ndmin=1)
        elif isinstance(return_timestamps, _pd.Series):
            return_timestamps_array = return_timestamps.as_matrix()
        else:
            raise TypeError('`return_timestamps` expects a list, array, pandas Series, '
                            'or numeric value')
        return return_timestamps_array
872446
def query_timeseries(self, query, from_time=None, to_time=None, by_post=False):
    
    return timeseries.query_timeseries(self, query, from_time, to_time, by_post=by_post)
16810
def _join_factory(cls, gap, pad):
    
    if issubclass(cls, dict):
        def _join(data):
            out = cls()
            data = list(data)
            while data:
                tsd = data.pop(0)
                out.append(tsd, gap=gap, pad=pad)
                del tsd
            return out
    else:
        from .. import TimeSeriesBaseList

        def _join(arrays):
            list_ = TimeSeriesBaseList(*arrays)
            return list_.join(pad=pad, gap=gap)
    return _join
934933
def trimmed(self, n_peaks):
        
        result = self.copy()
        result.trim(n_peaks)
        return result
453164
def run(items, run_parallel):
    
    to_process = []
    extras = []
    for batch, cur_items in _group_by_batches(items).items():
        if _ready_for_het_analysis(cur_items):
            to_process.append((batch, cur_items))
        else:
            for data in cur_items:
                extras.append([data])
    processed = run_parallel("heterogeneity_estimate", ([xs, b, xs[0]["config"]] for b, xs in to_process))
    return _group_by_sample_and_batch(extras + processed)
999055
def spectrogram(timeseries, method_func, **kwargs):
    
    from ...spectrogram import Spectrogram

    # get params
    sampling = timeseries.sample_rate.to('Hz').value
    nproc = kwargs.pop('nproc', 1)
    nfft = kwargs.pop('nfft')
    noverlap = kwargs.pop('noverlap')
    nstride = nfft - noverlap

    # sanity check parameters
    if noverlap >= nfft:
        raise ValueError("overlap must be less than fftlength")

    # set up single process Spectrogram method
    def _psd(series):
        """Calculate a single PSD for a spectrogram
        """
        return method_func(series, nfft=nfft, **kwargs)[1]

    # define chunks
    chunks = []
    x = 0
    while x + nfft <= timeseries.size:
        y = min(timeseries.size, x + nfft)
        chunks.append((x, y))
        x += nstride

    tschunks = (timeseries.value[i:j] for i, j in chunks)

    # calculate PSDs with multiprocessing
    psds = mp_utils.multiprocess_with_queues(nproc, _psd, tschunks)

    # convert PSDs to array with spacing for averages
    numtimes = 1 + int((timeseries.size - nstride) / nstride)
    numfreqs = int(nfft / 2 + 1)
    data = numpy.zeros((numtimes, numfreqs), dtype=timeseries.dtype)
    data[:len(psds)] = psds

    # create output spectrogram
    unit = fft_utils.scale_timeseries_unit(
        timeseries.unit, scaling=kwargs.get('scaling', 'density'))
    out = Spectrogram(numpy.empty((numtimes, numfreqs),
                                  dtype=timeseries.dtype),
                      copy=False, dt=nstride * timeseries.dt, t0=timeseries.t0,
                      f0=0, df=sampling/nfft, unit=unit,
                      name=timeseries.name, channel=timeseries.channel)

    # normalize over-dense grid
    density = nfft // nstride
    weights = get_window('triangle', density)
    for i in range(numtimes):
        # get indices of overlapping columns
        x = max(0, i+1-density)
        y = min(i+1, numtimes-density+1)
        if x == 0:
            wgt = weights[-y:]
        elif y == numtimes - density + 1:
            wgt = weights[:y-x]
        else:
            wgt = weights
        # calculate weighted average
        out.value[i, :] = numpy.average(data[x:y], axis=0, weights=wgt)

    return out
868790
def _compute(self, arrays, dates, assets, mask):
        
        return super(NumExprFilter, self)._compute(
            arrays,
            dates,
            assets,
            mask,
        ) & mask
159296
def compare_csv_timeseries_files(file1, file2, header=True):
    
    return compare_csv_decimal_files(file1, file2, header, True)
964822
def runMultiprocessNoiseExperiment(resultName, repeat, numWorkers,
                                   appendResults, **kwargs):
  
  experiments = [{}]
  for key, values in kwargs.items():
    if type(values) is list:
      newExperiments = []
      for experiment in experiments:
        for val in values:
          newExperiment = copy(experiment)
          newExperiment[key] = val
          newExperiments.append(newExperiment)
      experiments = newExperiments
    else:
      [a.__setitem__(key, values) for a in experiments]

  newExperiments = []
  for experiment in experiments:
    for _ in xrange(repeat):
      newExperiments.append(copy(experiment))
  experiments = newExperiments

  return runExperiments(experiments, resultName, numWorkers, appendResults)