Single-threaded pipeline for data prep that can be started in a pool.

def pipeline_dataprep(d, segment):
    """ Single-threaded pipeline for data prep that can be started in a pool.
    """

    logger.debug('dataprep starting for segment %d' % segment)

    # dataprep reads for a single segment, so d['segment'] defined here
    d['segment'] = segment

    # set up numpy arrays, as expected by dataprep functions
    data_read = numpyview(data_read_mem, 'complex64', datashape(d), raw=False); data = numpyview(data_mem, 'complex64', datashape(d), raw=False)
    u_read = numpyview(u_read_mem, 'float32', d['nbl'], raw=False); u = numpyview(u_mem, 'float32', d['nbl'], raw=False)
    v_read = numpyview(v_read_mem, 'float32', d['nbl'], raw=False); v = numpyview(v_mem, 'float32', d['nbl'], raw=False)
    w_read = numpyview(w_read_mem, 'float32', d['nbl'], raw=False); w = numpyview(w_mem, 'float32', d['nbl'], raw=False)

    ####    ####    ####    ####
    # 1) Read data
    ####    ####    ####    ####

    with data_read_mem.get_lock():
        if d['dataformat'] == 'ms':   # CASA-based read
            segread = pm.readsegment(d, segment)
            data_read[:] = segread[0]
            (u_read[:], v_read[:], w_read[:]) = (segread[1][d['readints']/2], segread[2][d['readints']/2], segread[3][d['readints']/2])  # mid int good enough for segment. could extend this to save per chunk
            del segread
        elif d['dataformat'] == 'sdm':
            data_read[:] = ps.read_bdf_segment(d, segment)
            (u_read[:], v_read[:], w_read[:]) = ps.get_uvw_segment(d, segment)

        ####    ####    ####    ####
        # 2) Prepare data
        ####    ####    ####    ####

        # calibrate data
        if os.path.exists(d['gainfile']):
            try:
                radec = (); spwind = []; calname = ''  # set defaults
                if '.GN' in d['gainfile']: # if telcal file
                    if d.has_key('calname'):
                        calname = d['calname']

                    sols = pc.telcal_sol(d['gainfile'])   # parse gainfile
                else:   # if CASA table
                    if d.has_key('calradec'):
                        radec = d['calradec']  # optionally defined cal location

                    spwind = d['spw']
                    sols = pc.casa_sol(d['gainfile'], flagants=d['flagantsol'])   # parse gainfile
                    sols.parsebp(d['bpfile'])   # parse bpfile

                # if gainfile parsed ok, choose best solution for data
                sols.set_selection(d['segmenttimes'][segment].mean(), d['freq']*1e9, rtlib.calc_blarr(d), calname=calname, pols=d['pols'], radec=radec, spwind=spwind)
                sols.apply(data_read)
            except:
                logger.warning('Could not parse or apply gainfile %s.' % d['gainfile'])
                raise
        else:
            logger.warn('Calibration file not found. Proceeding with no calibration applied.')

        # flag data
        if len(d['flaglist']):
            logger.info('Flagging with flaglist: %s' % d['flaglist'])
            dataflag(d, data_read)
        else:
            logger.warn('No real-time flagging.')

        # mean t vis subtration
        if d['timesub'] == 'mean':
            logger.info('Subtracting mean visibility in time...')
            rtlib.meantsub(data_read, [0, d['nbl']])
        else:
            logger.warn('No mean time subtraction.')

        # save noise pickle
        if d['savenoise']:
            noisepickle(d, data_read, u_read, v_read, w_read, chunk=200)

        # phase to new location if l1,m1 set and nonzero value
        try:
            if any([d['l1'], d['m1']]):
                logger.info('Rephasing data to (l, m)=(%.4f, %.4f).' % (d['l1'], d['m1']))
                rtlib.phaseshift_threaded(data_read, d, d['l1'], d['m1'], u_read, v_read)
                d['l0'] = d['l1']
                d['m0'] = d['m1']
            else:
                logger.debug('Not rephasing.')
        except KeyError:
            pass

        if d['mock']: # could be list or int
            # assume that std of vis in the middle of the segment is
            # characteristic of noise throughout the segment
            falsecands = {}
            datamid = n.ma.masked_equal(data_read[d['readints']/2].real, 0, copy=True)
            madstd = 1.4826 * n.ma.median(n.abs(datamid - n.ma.median(datamid)))/n.sqrt(d['npol']*d['nbl']*d['nchan'])
            std = datamid.std()/n.sqrt(d['npol']*d['nbl']*d['nchan'])
            logger.debug('Noise per vis in central int: madstd {}, std {}'.format(madstd, std))
            dt = 1 # pulse width in integrations

            if isinstance(d['mock'], int):
                for i in n.random.randint(d['datadelay'][-1], d['readints'], d['mock']):  # add nmock transients at random ints
                    (loff, moff, A, DM) = make_transient(madstd, max(d['dmarr']), Amin=1.2*d['sigma_image1'])
                    candid =  (int(segment), int(i), DM, int(dt), int(0))
                    falsecands[candid] = [A/madstd, A, loff, moff]
            elif isinstance(d['mock'], list):
                for mock in d['mock']:
                    try:
                        (i, DM, loff, moff, SNR) = mock
                        candid =  (int(segment), int(i), DM, int(dt), int(0))
                        falsecands[candid] = [SNR, SNR*madstd, loff, moff]
                    except:
                        logger.warn('Could not parse mock parameters: {}'.format(mock))
            else:
                logger.warn('Not a recognized type for mock.')

            for candid in falsecands:
                (segment, i, DM, dt, beamnum) = candid
                (SNR, A, loff, moff) = falsecands[candid]
                logger.info('Adding mock transient at int %d, DM %.1f, (l, m) = (%f, %f) at est SNR %.1f' % (i, DM, loff, moff, SNR))
                add_transient(d, data_read, u_read, v_read, w_read, loff, moff, i, A, DM, dt)
            
            if d['savecands']:
                savecands(d, falsecands, domock=True)

        with data_mem.get_lock():
            data[:] = data_read[:]
            u[:] = u_read[:]; v[:] = v_read[:]; w[:] = w_read[:]
    logger.debug('All data unlocked for segment %d' % segment)

    # d now has segment keyword defined
    return d