signature:
def collate_files ( data , sname , tmp1s , tmp2s ) :
    """Collate temp fastq files in tmp - dir into 1 gzipped sample ."""

implementation:
# # out handle
out1 = os . path . join ( data . dirs . fastqs , "{}_R1_.fastq.gz" . format ( sname ) )
out = io . BufferedWriter ( gzip . open ( out1 , 'w' ) )
# # build cmd
cmd1 = [ 'cat' ]
for tmpfile in tmp1s :
    cmd1 += [ tmpfile ]
# # compression function
proc = sps . Popen ( [ 'which' , 'pigz' ] , stderr = sps . PIPE , stdout = sps . PIPE ) . communicate ( )
if proc [ 0 ] . strip ( ) :
    compress = [ "pigz" ]
else :
    compress = [ "gzip" ]
# # call cmd
proc1 = sps . Popen ( cmd1 , stderr = sps . PIPE , stdout = sps . PIPE )
proc2 = sps . Popen ( compress , stdin = proc1 . stdout , stderr = sps . PIPE , stdout = out )
err = proc2 . communicate ( )
if proc2 . returncode :
    raise IPyradWarningExit ( "error in collate_files R1 %s" , err )
proc1 . stdout . close ( )
out . close ( )
# # then cleanup
for tmpfile in tmp1s :
    os . remove ( tmpfile )
if 'pair' in data . paramsdict [ "datatype" ] : # # out handle
    out2 = os . path . join ( data . dirs . fastqs , "{}_R2_.fastq.gz" . format ( sname ) )
    out = io . BufferedWriter ( gzip . open ( out2 , 'w' ) )
    # # build cmd
    cmd1 = [ 'cat' ]
    for tmpfile in tmp2s :
        cmd1 += [ tmpfile ]
    # # call cmd
    proc1 = sps . Popen ( cmd1 , stderr = sps . PIPE , stdout = sps . PIPE )
    proc2 = sps . Popen ( compress , stdin = proc1 . stdout , stderr = sps . PIPE , stdout = out )
    err = proc2 . communicate ( )
    if proc2 . returncode :
        raise IPyradWarningExit ( "error in collate_files R2 %s" , err )
    proc1 . stdout . close ( )
    out . close ( )
    # # then cleanup
    for tmpfile in tmp2s :
        os . remove ( tmpfile )


signature:
def prechecks2 ( data , force ) :
    """A new simplified version of prechecks func before demux
 Checks before starting analysis .
 1 ) Is there data in raw _ fastq _ path
 2 ) Is there a barcode file
 3 ) Is there a workdir and fastqdir
 4 ) remove old fastq / tmp _ sample _ R * _ dirs /
 5 ) return file names as pairs ( r1 , r2 ) or fakepairs ( r1 , 1)
 6 ) get ambiguous cutter resolutions
 7 ) get optim size"""

implementation:
# # check for data using glob for fuzzy matching
if not glob . glob ( data . paramsdict [ "raw_fastq_path" ] ) :
    raise IPyradWarningExit ( NO_RAWS . format ( data . paramsdict [ "raw_fastq_path" ] ) )
# # find longest barcode
try : # # Handle 3rad multi - barcodes . Gets len of the first one .
# # Should be harmless for single barcode data
    barlens = [ len ( i . split ( "+" ) [ 0 ] ) for i in data . barcodes . values ( ) ]
    if len ( set ( barlens ) ) == 1 :
        longbar = ( barlens [ 0 ] , 'same' )
    else :
        longbar = ( max ( barlens ) , 'diff' )
    # # For 3rad we need to add the length info for barcodes _ R2
    if "3rad" in data . paramsdict [ "datatype" ] :
        barlens = [ len ( i . split ( "+" ) [ 1 ] ) for i in data . barcodes . values ( ) ]
        longbar = ( longbar [ 0 ] , longbar [ 1 ] , max ( barlens ) )
except ValueError :
    raise IPyradWarningExit ( NO_BARS . format ( data . paramsdict [ "barcodes_path" ] ) )
# # setup dirs : [ workdir ] and a [ workdir / name _ fastqs ]
opj = os . path . join
# # create project dir
pdir = os . path . realpath ( data . paramsdict [ "project_dir" ] )
if not os . path . exists ( pdir ) :
    os . mkdir ( pdir )
# # create fastq dir
data . dirs . fastqs = opj ( pdir , data . name + "_fastqs" )
if os . path . exists ( data . dirs . fastqs ) and force :
    print ( OVERWRITING_FASTQS . format ( ** { "spacer" : data . _spacer } ) )
    shutil . rmtree ( data . dirs . fastqs )
if not os . path . exists ( data . dirs . fastqs ) :
    os . mkdir ( data . dirs . fastqs )
# # insure no leftover tmp files from a previous run ( there shouldn ' t be )
oldtmps = glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*_R1_" ) )
oldtmps += glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*_R2_" ) )
for oldtmp in oldtmps :
    os . remove ( oldtmp )
# # gather raw sequence filenames ( people want this to be flexible . . . )
if 'pair' in data . paramsdict [ "datatype" ] :
    raws = combinefiles ( data . paramsdict [ "raw_fastq_path" ] )
else :
    raws = zip ( glob . glob ( data . paramsdict [ "raw_fastq_path" ] ) , iter ( int , 1 ) )
# # returns a list of both resolutions of cut site 1
# # ( TGCAG , ) = = > [ TGCAG , ]
# # ( TWGC , ) = = > [ TAGC , TTGC ]
# # ( TWGC , AATT ) = = > [ TAGC , TTGC ]
cutters = [ ambigcutters ( i ) for i in data . paramsdict [ "restriction_overhang" ] ]
print ( cutters )
assert cutters , "Must enter a `restriction_overhang` for demultiplexing."
# # get matchdict
matchdict = inverse_barcodes ( data )
# # return all
return raws , longbar , cutters , matchdict


signature:
def inverse_barcodes ( data ) :
    """Build full inverse barcodes dictionary"""

implementation:
matchdict = { }
bases = set ( "CATGN" )
poss = set ( )
# # do perfect matches
for sname , barc in data . barcodes . items ( ) : # # remove - technical - replicate - N if present
    if "-technical-replicate-" in sname :
        sname = sname . rsplit ( "-technical-replicate" , 1 ) [ 0 ]
    matchdict [ barc ] = sname
    poss . add ( barc )
    if data . paramsdict [ "max_barcode_mismatch" ] > 0 : # # get 1 - base diffs
        for idx1 , base in enumerate ( barc ) :
            diffs = bases . difference ( base )
            for diff in diffs :
                lbar = list ( barc )
                lbar [ idx1 ] = diff
                tbar1 = "" . join ( lbar )
                if tbar1 not in poss :
                    matchdict [ tbar1 ] = sname
                    poss . add ( tbar1 )
                else :
                    if matchdict . get ( tbar1 ) != sname :
                        print ( """\
        Note: barcodes {}:{} and {}:{} are within {} base change of each other
            Ambiguous barcodes that match to both samples will arbitrarily
            be assigned to the first sample. If you do not like this idea
            then lower the value of max_barcode_mismatch and rerun step 1\n""" . format ( sname , barc , matchdict [ tbar1 ] , data . barcodes [ matchdict [ tbar1 ] ] , data . paramsdict [ "max_barcode_mismatch" ] ) )
            # # if allowing two base difference things get big
            # # for each modified bar , allow one modification to other bases
            if data . paramsdict [ "max_barcode_mismatch" ] > 1 :
                for idx2 , _ in enumerate ( tbar1 ) : # # skip the base that is already modified
                    if idx2 != idx1 :
                        for diff in bases . difference ( tbar1 [ idx2 ] ) :
                            ltbar = list ( tbar1 )
                            ltbar [ idx2 ] = diff
                            tbar2 = "" . join ( ltbar )
                            if tbar2 not in poss :
                                matchdict [ tbar2 ] = sname
                                poss . add ( tbar2 )
                            else :
                                if matchdict . get ( tbar2 ) != sname :
                                    print ( """\
        Note: barcodes {}:{} and {}:{} are within {} base change of each other\
             Ambiguous barcodes that match to both samples will arbitrarily
             be assigned to the first sample. If you do not like this idea
             then lower the value of max_barcode_mismatch and rerun step 1\n""" . format ( sname , barc , matchdict [ tbar2 ] , data . barcodes [ matchdict [ tbar2 ] ] , data . paramsdict [ "max_barcode_mismatch" ] ) )
return matchdict


signature:
def estimate_optim ( data , testfile , ipyclient ) :
    """Estimate a reasonable optim value by grabbing a chunk of sequences ,
 decompressing and counting them , to estimate the full file size ."""

implementation:
# # count the len of one file and assume all others are similar len
insize = os . path . getsize ( testfile )
tmp_file_name = os . path . join ( data . paramsdict [ "project_dir" ] , "tmp-step1-count.fq" )
if testfile . endswith ( ".gz" ) :
    infile = gzip . open ( testfile )
    outfile = gzip . open ( tmp_file_name , 'wb' , compresslevel = 5 )
else :
    infile = open ( testfile )
    outfile = open ( tmp_file_name , 'w' )
# # We ' ll take the average of the size of a file based on the
# # first 10000 reads to approximate number of reads in the main file
outfile . write ( "" . join ( itertools . islice ( infile , 40000 ) ) )
outfile . close ( )
infile . close ( )
# # Get the size of the tmp file
tmp_size = os . path . getsize ( tmp_file_name )
# # divide by the tmp file size and multiply by 10000 to approximate
# # the size of the input . fq files
inputreads = int ( insize / tmp_size ) * 10000
os . remove ( tmp_file_name )
return inputreads


signature:
def run2 ( data , ipyclient , force ) :
    """One input file ( or pair ) is run on two processors , one for reading
 and decompressing the data , and the other for demuxing it ."""

implementation:
# # get file handles , name - lens , cutters , and matchdict
raws , longbar , cutters , matchdict = prechecks2 ( data , force )
# # wrap funcs to ensure we can kill tmpfiles
kbd = 0
try : # # if splitting files , split files into smaller chunks for demuxing
    chunkfiles = splitfiles ( data , raws , ipyclient )
    # # send chunks to be demux ' d
    statdicts = demux2 ( data , chunkfiles , cutters , longbar , matchdict , ipyclient )
    # # concat tmp files
    concat_chunks ( data , ipyclient )
    # # build stats from dictionaries
    perfile , fsamplehits , fbarhits , fmisses , fdbars = statdicts
    make_stats ( data , perfile , fsamplehits , fbarhits , fmisses , fdbars )
except KeyboardInterrupt :
    print ( "\n  ...interrupted, just a second while we ensure proper cleanup" )
    kbd = 1
# # cleanup
finally : # # cleaning up the tmpdir is safe from ipyclient
    tmpdir = os . path . join ( data . paramsdict [ "project_dir" ] , "tmp-chunks-" + data . name )
    if os . path . exists ( tmpdir ) :
        shutil . rmtree ( tmpdir )
    if kbd :
        raise KeyboardInterrupt ( "s1" )
    else :
        _cleanup_and_die ( data )


signature:
def _cleanup_and_die ( data ) :
    """cleanup func for step 1"""

implementation:
tmpfiles = glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*_R*.fastq" ) )
tmpfiles += glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*.p" ) )
for tmpf in tmpfiles :
    os . remove ( tmpf )


signature:
def run3 ( data , ipyclient , force ) :
    """One input file ( or pair ) is run on two processors , one for reading
 and decompressing the data , and the other for demuxing it ."""

implementation:
start = time . time ( )
# # get file handles , name - lens , cutters , and matchdict ,
# # and remove any existing files if a previous run failed .
raws , longbar , cutters , matchdict = prechecks2 ( data , force )
# # wrap funcs to ensure we can kill tmpfiles
kbd = 0
try : # # send chunks to be demux ' d , nothing is parallelized yet .
    lbview = ipyclient . load_balanced_view ( )
    args = ( data , raws , cutters , longbar , matchdict )
    async = lbview . apply ( demux3 , * args )
    # # track progress
    while 1 : # # how many of this func have finished so far
        elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
        printstr = ' writing/compressing   | {} | s1 |'
        progressbar ( len ( ready ) , sum ( ready ) , printstr , spacer = spacer )
        time . sleep ( 0.1 )
        if async . ready ( ) :
            print ( "" )
            break
    if async . successful ( ) :
        statdicts = async . get ( )
    else :
        raise IPyradWarningExit ( async . get ( ) )
    # # build stats from dictionaries
    perfile , fsamplehits , fbarhits , fmisses , fdbars = statdicts
    make_stats ( data , perfile , fsamplehits , fbarhits , fmisses , fdbars )
except KeyboardInterrupt :
    print ( "\n  ...interrupted, just a second while we ensure proper cleanup" )
    kbd = 1
# # cleanup
finally : # # cleaning up the tmpdir is safe from ipyclient
    tmpdir = os . path . join ( data . paramsdict [ "project_dir" ] , "tmp-chunks-" + data . name )
    if os . path . exists ( tmpdir ) :
        shutil . rmtree ( tmpdir )
    tmpfiles = glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*_R*.fastq" ) )
    tmpfiles += glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*.p" ) )
    for tmpf in tmpfiles :
        if os . path . exists ( tmpf ) :
            os . remove ( tmpf )
    if kbd :
        raise


signature:
def splitfiles ( data , raws , ipyclient ) :
    """sends raws to be chunked"""

implementation:
# # create a tmpdir for chunked _ files and a chunk optimizer
tmpdir = os . path . join ( data . paramsdict [ "project_dir" ] , "tmp-chunks-" + data . name )
if os . path . exists ( tmpdir ) :
    shutil . rmtree ( tmpdir )
os . makedirs ( tmpdir )
# # chunk into 8M reads
totalreads = estimate_optim ( data , raws [ 0 ] [ 0 ] , ipyclient )
optim = int ( 8e6 )
njobs = int ( totalreads / ( optim / 4. ) ) * len ( raws )
# # if more files than cpus : no chunking
nosplit = 0
if ( len ( raws ) > len ( ipyclient ) ) or ( totalreads < optim ) :
    nosplit = 1
# # send slices N at a time . The dict chunkfiles stores a tuple of rawpairs
# # dictionary to store asyncresults for sorting jobs
start = time . time ( )
chunkfiles = { }
for fidx , tups in enumerate ( raws ) :
    handle = os . path . splitext ( os . path . basename ( tups [ 0 ] ) ) [ 0 ]
    # # if number of lines is > 20M then just submit it
    if nosplit :
        chunkfiles [ handle ] = [ tups ]
    else : # # chunk the file using zcat _ make _ temps
        chunklist = zcat_make_temps ( data , tups , fidx , tmpdir , optim , njobs , start )
        chunkfiles [ handle ] = chunklist
if not nosplit :
    print ( "" )
return chunkfiles


signature:
def concat_chunks ( data , ipyclient ) :
    """Concatenate chunks . If multiple chunk files match to the same sample name
 but with different barcodes ( i . e . , they are technical replicates ) then this
 will assign all the files to the same sample name file ."""

implementation:
# # collate files progress bar
start = time . time ( )
printstr = ' writing/compressing   | {} | s1 |'
lbview = ipyclient . load_balanced_view ( )
elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
progressbar ( 10 , 0 , printstr . format ( elapsed ) , spacer = data . _spacer )
# # get all the files
ftmps = glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*.fastq" ) )
# # a dict to assign tmp files to names / reads
r1dict = { }
r2dict = { }
for sname in data . barcodes :
    if "-technical-replicate-" in sname :
        sname = sname . rsplit ( "-technical-replicate" , 1 ) [ 0 ]
    r1dict [ sname ] = [ ]
    r2dict [ sname ] = [ ]
# # assign to name keys
for ftmp in ftmps :
    base , orient , _ = ftmp . rsplit ( "_" , 2 )
    sname = base . rsplit ( "/" , 1 ) [ - 1 ] . split ( "tmp_" , 1 ) [ 1 ]
    if orient == "R1" :
        r1dict [ sname ] . append ( ftmp )
    else :
        r2dict [ sname ] . append ( ftmp )
# # concatenate files
snames = [ ]
for sname in data . barcodes :
    if "-technical-replicate-" in sname :
        sname = sname . rsplit ( "-technical-replicate" , 1 ) [ 0 ]
    snames . append ( sname )
writers = [ ]
for sname in set ( snames ) :
    tmp1s = sorted ( r1dict [ sname ] )
    tmp2s = sorted ( r2dict [ sname ] )
    writers . append ( lbview . apply ( collate_files , * [ data , sname , tmp1s , tmp2s ] ) )
total = len ( writers )
while 1 :
    ready = [ i . ready ( ) for i in writers ]
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( total , sum ( ready ) , printstr . format ( elapsed ) , spacer = data . _spacer )
    time . sleep ( 0.1 )
    if all ( ready ) :
        print ( "" )
        break


signature:
def demux2 ( data , chunkfiles , cutters , longbar , matchdict , ipyclient ) :
    """Submit chunks to be sorted by the barmatch ( ) function then
 calls putstats ( ) ."""

implementation:
# # parallel stuff , limit to 1/4 of available cores for RAM limits .
start = time . time ( )
printstr = ' sorting reads         | {} | s1 |'
lbview = ipyclient . load_balanced_view ( targets = ipyclient . ids [ : : 4 ] )
# # store statcounters and async results in dicts
perfile = { }
filesort = { }
total = 0
done = 0
# # chunkfiles is a dict with { handle : chunkslist , . . . } . The func barmatch
# # writes results to samplename files with PID number , and also writes a
# # pickle for chunk specific results with fidx suffix , which it returns .
for handle , rawtuplist in chunkfiles . items ( ) : # # get args for job
    for fidx , rawtuple in enumerate ( rawtuplist ) : # handle = os . path . splitext ( os . path . basename ( rawtuple [ 0 ] ) ) [ 0]
        args = ( data , rawtuple , cutters , longbar , matchdict , fidx )
        # # submit the job
        async = lbview . apply ( barmatch , * args )
        filesort [ total ] = ( handle , async )
        total += 1
        # # get ready to receive stats : ' total ' , ' cutfound ' , ' matched '
        perfile [ handle ] = np . zeros ( 3 , dtype = np . int )
# # stats for each sample
fdbars = { }
fsamplehits = Counter ( )
fbarhits = Counter ( )
fmisses = Counter ( )
# # a tuple to hold my dictionaries
statdicts = perfile , fsamplehits , fbarhits , fmisses , fdbars
# # wait for jobs to finish
while 1 :
    fin = [ i for i , j in filesort . items ( ) if j [ 1 ] . ready ( ) ]
    # fin = [ i for i in jobs if i [ 1 ] . ready ( ) ]
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( total , done , printstr . format ( elapsed ) , spacer = data . _spacer )
    time . sleep ( 0.1 )
    # # should we break ?
    if total == done :
        print ( "" )
        break
    # # cleanup
    for key in fin :
        tup = filesort [ key ]
        if tup [ 1 ] . successful ( ) :
            pfile = tup [ 1 ] . result ( )
            handle = tup [ 0 ]
            if pfile : # # check if this needs to return data
                putstats ( pfile , handle , statdicts )
                # # purge to conserve memory
                del filesort [ key ]
                done += 1
return statdicts


signature:
def demux ( data , chunkfiles , cutters , longbar , matchdict , ipyclient ) :
    """submit chunks to be sorted"""

implementation:
# # parallel stuff
start = time . time ( )
printstr = ' sorting reads         | {} | s1 |'
lbview = ipyclient . load_balanced_view ( )
# # store statcounters and async results in dicts
perfile = { }
filesort = { }
for handle , rawtuplist in chunkfiles . items ( ) : # # get args for job
    for fidx , rawtuple in enumerate ( rawtuplist ) : # handle = os . path . splitext ( os . path . basename ( rawtuple [ 0 ] ) ) [ 0]
        args = ( data , rawtuple , cutters , longbar , matchdict , fidx )
        # # submit the job
        filesort [ handle ] = lbview . apply ( barmatch , * args )
        # # get ready to receive stats : ' total ' , ' cutfound ' , ' matched '
        perfile [ handle ] = np . zeros ( 3 , dtype = np . int )
# # stats for each sample
fdbars = { }
fsamplehits = Counter ( )
fbarhits = Counter ( )
fmisses = Counter ( )
# # a tuple to hold my dictionaries
statdicts = perfile , fsamplehits , fbarhits , fmisses , fdbars
try :
    kbd = 0
    total = len ( chunkfiles )
    done = 0
    # # wait for jobs to finish
    while 1 :
        fin = [ i for i , j in filesort . items ( ) if j . ready ( ) ]
        elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
        progressbar ( total , done , printstr . format ( elapsed ) , spacer = data . _spacer )
        time . sleep ( 0.1 )
        # # should we break ?
        if total == done :
            print ( "" )
            break
        # # cleanup
        for job in fin :
            if filesort [ job ] . successful ( ) :
                pfile = filesort [ job ] . result ( )
                # if result :
                if pfile : # # check if this needs to return data
                    putstats ( pfile , handle , statdicts )
                    # # purge to conserve memory
                    del filesort [ job ]
                    done += 1
    # # keep tacking progreess during writing stage
    start = time . time ( )
    printstr = ' writing/compressing   | {} | s1 |'
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( 10 , 0 , printstr . format ( elapsed ) , spacer = data . _spacer )
except KeyboardInterrupt : # # wait to cleanup
    kbd = 1
    raise
# # only proceed here if barmatch jobs were not interrupted
else : # # collate files and do progress bar
    ftmps = glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*.fastq" ) )
    # # a dict to assign tmp files to names / reads
    r1dict = { }
    r2dict = { }
    for sname in data . barcodes :
        r1dict [ sname ] = [ ]
        r2dict [ sname ] = [ ]
    # # assign to name keys
    for ftmp in ftmps : # # split names
        base , orient , _ = ftmp . rsplit ( "_" , 2 )
        sname = base . rsplit ( "/" , 1 ) [ - 1 ] . split ( "tmp_" , 1 ) [ 1 ]
        # # put into dicts
        if orient == "R1" :
            r1dict [ sname ] . append ( ftmp )
        else :
            r2dict [ sname ] . append ( ftmp )
    # # concatenate files
    total = len ( data . barcodes )
    done = 0
    # # store asyncs of collate jobs
    writers = [ ]
    for sname in data . barcodes :
        tmp1s = sorted ( r1dict [ sname ] )
        tmp2s = sorted ( r2dict [ sname ] )
        writers . append ( lbview . apply ( collate_files , * [ data , sname , tmp1s , tmp2s ] ) )
    # # track progress of collate jobs
    while 1 :
        ready = [ i . ready ( ) for i in writers ]
        elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
        progressbar ( total , sum ( ready ) , printstr . format ( elapsed ) , spacer = data . _spacer )
        time . sleep ( 0.1 )
        if all ( ready ) :
            print ( "" )
            break
finally : # # clean up junk files
    tmpfiles = glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*_R*.fastq" ) )
    tmpfiles += glob . glob ( os . path . join ( data . dirs . fastqs , "tmp_*.p" ) )
    for tmpf in tmpfiles :
        os . remove ( tmpf )
    if kbd :
        raise KeyboardInterrupt ( )
    else : # # build stats from dictionaries
        perfile , fsamplehits , fbarhits , fmisses , fdbars = statdicts
        make_stats ( data , perfile , fsamplehits , fbarhits , fmisses , fdbars )


signature:
def putstats ( pfile , handle , statdicts ) :
    """puts stats from pickles into a dictionary"""

implementation:
# # load in stats
with open ( pfile , 'r' ) as infile :
    filestats , samplestats = pickle . load ( infile )
# # get dicts from statdicts tuple
perfile , fsamplehits , fbarhits , fmisses , fdbars = statdicts
# # pull new stats
# handle = os . path . splitext ( os . path . basename ( handle ) ) [ 0]
perfile [ handle ] += filestats
# # update sample stats
samplehits , barhits , misses , dbars = samplestats
fsamplehits . update ( samplehits )
fbarhits . update ( barhits )
fmisses . update ( misses )
fdbars . update ( dbars )
# # repack the tuple and return
statdicts = perfile , fsamplehits , fbarhits , fmisses , fdbars
return statdicts


signature:
def zcat_make_temps ( data , raws , num , tmpdir , optim , njobs , start ) :
    """Call bash command ' cat ' and ' split ' to split large files . The goal
 is to create N splitfiles where N is a multiple of the number of processors
 so that each processor can work on a file in parallel ."""

implementation:
printstr = ' chunking large files  | {} | s1 |'
# # split args
tmpdir = os . path . realpath ( tmpdir )
LOGGER . info ( "zcat is using optim = %s" , optim )
# # read it , is it gzipped ?
catcmd = [ "cat" ]
if raws [ 0 ] . endswith ( ".gz" ) :
    catcmd = [ "gunzip" , "-c" ]
# # get reading commands for r1s , r2s
cmd1 = catcmd + [ raws [ 0 ] ]
cmd2 = catcmd + [ raws [ 1 ] ]
# # second command splits and writes with name prefix
cmd3 = [ "split" , "-a" , "4" , "-l" , str ( int ( optim ) ) , "-" , os . path . join ( tmpdir , "chunk1_" + str ( num ) + "_" ) ]
cmd4 = [ "split" , "-a" , "4" , "-l" , str ( int ( optim ) ) , "-" , os . path . join ( tmpdir , "chunk2_" + str ( num ) + "_" ) ]
# # # run splitter
proc1 = sps . Popen ( cmd1 , stderr = sps . STDOUT , stdout = sps . PIPE )
proc3 = sps . Popen ( cmd3 , stderr = sps . STDOUT , stdout = sps . PIPE , stdin = proc1 . stdout )
# # wrap the actual call so we can kill it if anything goes awry
while 1 :
    try :
        if not isinstance ( proc3 . poll ( ) , int ) :
            elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
            done = len ( glob . glob ( os . path . join ( tmpdir , 'chunk1_*' ) ) )
            progressbar ( njobs , min ( njobs , done ) , printstr . format ( elapsed ) , spacer = data . _spacer )
            time . sleep ( 0.1 )
        else :
            res = proc3 . communicate ( ) [ 0 ]
            proc1 . stdout . close ( )
            break
    except KeyboardInterrupt :
        proc1 . kill ( )
        proc3 . kill ( )
        raise KeyboardInterrupt ( )
if proc3 . returncode :
    raise IPyradWarningExit ( " error in %s: %s" , cmd3 , res )
# # grab output handles
chunks1 = glob . glob ( os . path . join ( tmpdir , "chunk1_" + str ( num ) + "_*" ) )
chunks1 . sort ( )
if "pair" in data . paramsdict [ "datatype" ] :
    proc2 = sps . Popen ( cmd2 , stderr = sps . STDOUT , stdout = sps . PIPE )
    proc4 = sps . Popen ( cmd4 , stderr = sps . STDOUT , stdout = sps . PIPE , stdin = proc2 . stdout )
    # # wrap the actual call so we can kill it if anything goes awry
    while 1 :
        try :
            if not isinstance ( proc4 . poll ( ) , int ) :
                elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
                done = len ( glob . glob ( os . path . join ( tmpdir , 'chunk1_*' ) ) )
                progressbar ( njobs , min ( njobs , done ) , printstr . format ( elapsed ) , data . _spacer )
                time . sleep ( 0.1 )
            else :
                res = proc4 . communicate ( ) [ 0 ]
                proc2 . stdout . close ( )
                break
        except KeyboardInterrupt :
            proc2 . kill ( )
            proc4 . kill ( )
            raise KeyboardInterrupt ( )
    if proc4 . returncode :
        raise IPyradWarningExit ( " error in %s: %s" , cmd4 , res )
    # # grab output handles
    chunks2 = glob . glob ( os . path . join ( tmpdir , "chunk2_" + str ( num ) + "_*" ) )
    chunks2 . sort ( )
else :
    chunks2 = [ 0 ] * len ( chunks1 )
assert len ( chunks1 ) == len ( chunks2 ) , "R1 and R2 files are not the same length."
# # ensure full progress bar b / c estimates njobs could be off
progressbar ( 10 , 10 , printstr . format ( elapsed ) , spacer = data . _spacer )
return zip ( chunks1 , chunks2 )


signature:
def _plotshare ( share , names , ** kwargs ) :
    """make toyplot matrix fig"""

implementation:
# # set the colormap
colormap = toyplot . color . LinearMap ( toyplot . color . brewer . palette ( "Spectral" ) , domain_min = share . min ( ) , domain_max = share . max ( ) )
# # set up canvas
if not kwargs . get ( 'width' ) :
    width = 900
else :
    width = kwargs [ 'width' ]
canvas = toyplot . Canvas ( width = width , height = width * 0.77778 )
# # order the dta
table = canvas . matrix ( ( share , colormap ) , bounds = ( 50 , canvas . height - 100 , 50 , canvas . height - 100 ) , step = 5 , tshow = False , lshow = False )
# # put a box around the table
table . body . grid . vlines [ ... , [ 0 , - 1 ] ] = 'single'
table . body . grid . hlines [ [ 0 , - 1 ] , ... ] = 'single'
# # make hover info on grid
for i , j in itertools . product ( range ( len ( share ) ) , repeat = 2 ) :
    table . body . cell ( i , j ) . title = "%s, %s : %s" % ( names [ i ] , names [ j ] , int ( share [ i , j ] ) )
# # create barplot
axes = canvas . cartesian ( bounds = ( 665 , 800 , 90 , 560 ) )
# # make a hover for barplot
zf = zip ( names [ : : - 1 ] , share . diagonal ( ) [ : : - 1 ] )
barfloater = [ "%s: %s" % ( i , int ( j ) ) for i , j in zf ]
# # plot bars
axes . bars ( share . diagonal ( ) [ : : - 1 ] , along = 'y' , title = barfloater )
# # hide spine , move labels to the left ,
# # use taxon names , rotate angle , align
axes . y . spine . show = False
axes . y . ticks . labels . offset = 0
axes . y . ticks . locator = toyplot . locator . Explicit ( range ( len ( names ) ) , labels = names [ : : - 1 ] )
axes . y . ticks . labels . angle = - 90
axes . y . ticks . labels . style = { "baseline-shift" : 0 , "text-anchor" : "end" , "font-size" : "8px" }
# # rotate xlabels , align with ticks , change to thousands , move up on canvas
# # show ticks , and hide popup coordinates
axes . x . ticks . labels . angle = 90
axes . x . ticks . labels . offset = 20
axes . x . ticks . locator = toyplot . locator . Explicit ( range ( 0 , int ( share . max ( ) ) , int ( share . max ( ) / 10 ) ) , [ "{}" . format ( i ) for i in range ( 0 , int ( share . max ( ) ) , int ( share . max ( ) / 10 ) ) ] )
axes . x . ticks . labels . style = { "baseline-shift" : 0 , "text-anchor" : "end" , "-toyplot-anchor-shift" : "15px" }
axes . x . ticks . show = True
# # add labels
label_style = { "font-size" : "16px" , "font-weight" : "bold" }
canvas . text ( 300 , 60 , "Matrix of shared RAD loci" , style = label_style )
canvas . text ( 700 , 60 , "N RAD loci per sample" , style = label_style )
return canvas , axes


signature:
def _getarray ( loci , tree ) :
    """parse the loci file list and return presence / absence matrix
 ordered by the tips on the tree"""

implementation:
# # order tips
tree . ladderize ( )
# # get tip names
snames = tree . get_leaf_names ( )
# # make an empty matrix
lxs = np . zeros ( ( len ( snames ) , len ( loci ) ) , dtype = np . int )
# # fill the matrix
for loc in xrange ( len ( loci ) ) :
    for seq in loci [ loc ] . split ( "\n" ) [ : - 1 ] :
        lxs [ snames . index ( seq . split ( ) [ 0 ] ) , loc ] += 1
return lxs , snames


signature:
def _countmatrix ( lxs ) :
    """fill a matrix with pairwise data sharing"""

implementation:
# # an empty matrix
share = np . zeros ( ( lxs . shape [ 0 ] , lxs . shape [ 0 ] ) )
# # fill above
names = range ( lxs . shape [ 0 ] )
for row in lxs :
    for samp1 , samp2 in itertools . combinations ( names , 2 ) :
        shared = lxs [ samp1 , lxs [ samp2 ] > 0 ] . sum ( )
        share [ samp1 , samp2 ] = shared
# # mirror below
# # share [ ]
# # fill diagonal with total sample coverage
for row in xrange ( len ( names ) ) :
    share [ row , row ] = lxs [ row ] . sum ( )
return share


signature:
def paramname ( param = "" ) :
    """Get the param name from the dict index value ."""

implementation:
try :
    name = pinfo [ str ( param ) ] [ 0 ] . strip ( ) . split ( " " ) [ 1 ]
except ( KeyError , ValueError ) as err : # # TODO : paramsinfo get description by param string not working .
# # It would be cool to have an assembly object bcz then you could
# # just do this :
# # print ( pinfo [ data . paramsinfo . keys ( ) . index ( param ) ] )
    print ( "\tKey name/number not recognized - " . format ( param ) , err )
    raise
return name


signature:
def paraminfo ( param = "" , short = False ) :
    """Returns detailed information for the numbered parameter .
 Further information is available in the tutorial .
 Unlike params ( ) this function doesn ' t deal well with *
 It only takes one parameter at a time and returns the desc"""

implementation:
# # If the short flag is set return the short description , otherwise
# # return the long .
if short :
    desc = 1
else :
    desc = 0
try :
    description = pinfo [ str ( param ) ] [ desc ]
except ( KeyError , ValueError ) as err : # # TODO : paramsinfo get description by param string not working .
# # It would be cool to have an assembly object bcz then you could
# # just do this :
# # print ( pinfo [ data . paramsinfo . keys ( ) . index ( param ) ] )
    print ( "\tKey name/number not recognized - " . format ( param ) , err )
    raise
return description


signature:
def paramsinfo ( param = "" , short = False ) :
    """This is the human readable version of the paramsinfo ( ) function .
 You give it a param and it prints to stdout ."""

implementation:
if short :
    desc = 1
else :
    desc = 0
if param == "*" :
    for key in pinfo :
        print ( pinfo [ str ( key ) ] [ desc ] )
elif param :
    try :
        print ( pinfo [ str ( param ) ] [ desc ] )
    except ( KeyError , ValueError ) as err : # # TODO : paramsinfo get description by param string not working .
    # # It would be cool to have an assembly object bcz then you could
    # # just do this :
    # # print ( pinfo [ data . paramsinfo . keys ( ) . index ( param ) ] )
        print ( "\tKey name/number not recognized" , err )
        raise
else :
    print ( "Enter a name or number for explanation of the parameter\n" )
    for key in pinfo :
        print ( pinfo [ str ( key ) ] [ desc ] . split ( "\n" ) [ 1 ] [ 2 : - 10 ] )


signature:
def update_assembly ( data ) :
    """Create a new Assembly ( ) and convert as many of our old params to the new
 version as we can . Also report out any parameters that are removed
 and what their values are ."""

implementation:
print ( "##############################################################" )
print ( "Updating assembly to current version" )
# # New assembly object to update pdate from .
new_assembly = ip . Assembly ( "update" , quiet = True )
# # Hackersonly dict gets automatically overwritten
# # Always use the current version for params in this dict .
data . _hackersonly = deepcopy ( new_assembly . _hackersonly )
new_params = set ( new_assembly . paramsdict . keys ( ) )
my_params = set ( data . paramsdict . keys ( ) )
# # Find all params in loaded assembly that aren ' t in the new assembly .
# # Make a new dict that doesn ' t include anything in removed _ params
removed_params = my_params . difference ( new_params )
for i in removed_params :
    print ( "Removing parameter: {} = {}" . format ( i , data . paramsdict [ i ] ) )
# # Find all params that are in the new paramsdict and not in the old one .
# # If the set isn ' t emtpy then we create a new dictionary based on the new
# # assembly parameters and populated with currently loaded assembly values .
# # Conditioning on not including any removed params . Magic .
added_params = new_params . difference ( my_params )
for i in added_params :
    print ( "Adding parameter: {} = {}" . format ( i , new_assembly . paramsdict [ i ] ) )
print ( "\nPlease take note of these changes. Every effort is made to\n" + "ensure compatibility across versions of ipyrad. See online\n" + "documentation for further details about new parameters." )
time . sleep ( 5 )
print ( "##############################################################" )
if added_params :
    for i in data . paramsdict :
        if i not in removed_params :
            new_assembly . paramsdict [ i ] = data . paramsdict [ i ]
    data . paramsdict = deepcopy ( new_assembly . paramsdict )
data . save ( )
return data


signature:
def save_json2 ( data ) :
    """save to json ."""

implementation:
# # convert everything to dicts
# # skip _ ipcluster cuz it ' s made new .
datadict = OrderedDict ( [ ( "outfiles" , data . __dict__ [ "outfiles" ] ) , ( "stats_files" , dict ( data . __dict__ [ "stats_files" ] ) ) , ( "stats_dfs" , data . __dict__ [ "stats_dfs" ] ) ] )


signature:
def save_json ( data ) :
    """Save assembly and samples as json"""

implementation:
# # data as dict
# # # # skip _ ipcluster because it ' s made new
# # # # skip _ headers because it ' s loaded new
# # # # statsfiles save only keys
# # # # samples save only keys
datadict = OrderedDict ( [ ( "_version" , data . __dict__ [ "_version" ] ) , ( "_checkpoint" , data . __dict__ [ "_checkpoint" ] ) , ( "name" , data . __dict__ [ "name" ] ) , ( "dirs" , data . __dict__ [ "dirs" ] ) , ( "paramsdict" , data . __dict__ [ "paramsdict" ] ) , ( "samples" , data . __dict__ [ "samples" ] . keys ( ) ) , ( "populations" , data . __dict__ [ "populations" ] ) , ( "database" , data . __dict__ [ "database" ] ) , ( "clust_database" , data . __dict__ [ "clust_database" ] ) , ( "outfiles" , data . __dict__ [ "outfiles" ] ) , ( "barcodes" , data . __dict__ [ "barcodes" ] ) , ( "stats_files" , data . __dict__ [ "stats_files" ] ) , ( "_hackersonly" , data . __dict__ [ "_hackersonly" ] ) , ] )
# # sample dict
sampledict = OrderedDict ( [ ] )
for key , sample in data . samples . iteritems ( ) :
    sampledict [ key ] = sample . _to_fulldict ( )
# # json format it using cumstom Encoder class
fulldumps = json . dumps ( { "assembly" : datadict , "samples" : sampledict } , cls = Encoder , sort_keys = False , indent = 4 , separators = ( "," , ":" ) , )
# # save to file
assemblypath = os . path . join ( data . dirs . project , data . name + ".json" )
if not os . path . exists ( data . dirs . project ) :
    os . mkdir ( data . dirs . project )
# # protect save from interruption
done = 0
while not done :
    try :
        with open ( assemblypath , 'w' ) as jout :
            jout . write ( fulldumps )
        done = 1
    except ( KeyboardInterrupt , SystemExit ) :
        print ( '.' )
        continue


signature:
def load_json ( path , quiet = False , cli = False ) :
    """Load a json serialized object and ensure it matches to the current
 Assembly object format"""

implementation:
# # load the JSON string and try with name + . json
checkfor = [ path + ".json" , path ]
for inpath in checkfor :
    inpath = inpath . replace ( "~" , os . path . expanduser ( "~" ) )
    try :
        with open ( inpath , 'r' ) as infile : # # uses _ tup _ and _ byte to ensure ascii and tuples are correct
            fullj = json . loads ( infile . read ( ) , object_hook = _tup_and_byte )
    except IOError :
        pass
# # create a new empty Assembly
try :
    oldname = fullj [ "assembly" ] . pop ( "name" )
    olddir = fullj [ "assembly" ] [ "dirs" ] [ "project" ]
    oldpath = os . path . join ( olddir , os . path . splitext ( oldname ) [ 0 ] + ".json" )
    null = ip . Assembly ( oldname , quiet = True , cli = cli )
except ( UnboundLocalError , AttributeError ) as inst :
    raise IPyradWarningExit ( """
    Could not find saved Assembly file (.json) in expected location.
    Checks in: [project_dir]/[assembly_name].json
    Checked: {}
    """ . format ( inpath ) )
# # print msg with shortpath
if not quiet :
    oldpath = oldpath . replace ( os . path . expanduser ( "~" ) , "~" )
    print ( "{}loading Assembly: {}" . format ( null . _spacer , oldname ) )
    print ( "{}from saved path: {}" . format ( null . _spacer , oldpath ) )
# # First get the samples . Create empty sample dict of correct length
samplekeys = fullj [ "assembly" ] . pop ( "samples" )
null . samples = { name : "" for name in samplekeys }
# # Next get paramsdict and use set _ params to convert values back to
# # the correct dtypes . Allow set _ params to fail because the object will
# # be subsequently updated by the params from the params file , which may
# # correct any errors / incompatibilities in the old params file
oldparams = fullj [ "assembly" ] . pop ( "paramsdict" )
for param , val in oldparams . iteritems ( ) : # # a fix for backward compatibility with deprecated options
    if param not in [ "assembly_name" , "excludes" , "outgroups" ] :
        try :
            null . set_params ( param , val )
        except IPyradWarningExit as inst : # null . set _ params ( param , " " )
            LOGGER . warning ( """
    Load assembly error setting params. Not critical b/c new params file may
    correct the problem. Recorded here for debugging:
    {}
    """ . format ( inst ) )
# # Import the hackersonly dict . In this case we don ' t have the nice
# # set _ params so we ' re shooting from the hip to reset the values
try :
    oldhackersonly = fullj [ "assembly" ] . pop ( "_hackersonly" )
    for param , val in oldhackersonly . iteritems ( ) :
        if val == None :
            null . _hackersonly [ param ] = None
        else :
            null . _hackersonly [ param ] = val
except Exception as inst :
    LOGGER . warning ( """
    Load assembly error resetting hackersonly dict element. We will just use
    the default value in the current assembly.""" )
# # Check remaining attributes of Assembly and Raise warning if attributes
# # do not match up between old and new objects
newkeys = null . __dict__ . keys ( )
oldkeys = fullj [ "assembly" ] . keys ( )
# # find shared keys and deprecated keys
sharedkeys = set ( oldkeys ) . intersection ( set ( newkeys ) )
lostkeys = set ( oldkeys ) . difference ( set ( newkeys ) )
# # raise warning if there are lost / deprecated keys
if lostkeys :
    LOGGER . warning ( """
    load_json found {a} keys that are unique to the older Assembly.
        - assembly [{b}] v.[{c}] has: {d}
        - current assembly is v.[{e}]
        """ . format ( a = len ( lostkeys ) , b = oldname , c = fullj [ "assembly" ] [ "_version" ] , d = lostkeys , e = null . _version ) )
# # load in remaining shared Assembly attributes to null
for key in sharedkeys :
    null . __setattr__ ( key , fullj [ "assembly" ] [ key ] )
# # load in svd results if they exist
try :
    if fullj [ "assembly" ] [ "svd" ] :
        null . __setattr__ ( "svd" , fullj [ "assembly" ] [ "svd" ] )
        null . svd = ObjDict ( null . svd )
except Exception :
    LOGGER . debug ( "skipping: no svd results present in old assembly" )
# # Now , load in the Sample objects json dicts
sample_names = fullj [ "samples" ] . keys ( )
if not sample_names :
    raise IPyradWarningExit ( """
    No samples found in saved assembly. If you are just starting a new
    assembly the file probably got saved erroneously, so it's safe to try
    removing the assembly file (e.g., rm {}.json) and restarting.

    If you fully completed step 1 and you see this message you should probably
    contact the developers.
    """ . format ( inpath ) )
sample_keys = fullj [ "samples" ] [ sample_names [ 0 ] ] . keys ( )
stats_keys = fullj [ "samples" ] [ sample_names [ 0 ] ] [ "stats" ] . keys ( )
stats_dfs_keys = fullj [ "samples" ] [ sample_names [ 0 ] ] [ "stats_dfs" ] . keys ( )
ind_statkeys = [ fullj [ "samples" ] [ sample_names [ 0 ] ] [ "stats_dfs" ] [ i ] . keys ( ) for i in stats_dfs_keys ]
ind_statkeys = list ( itertools . chain ( * ind_statkeys ) )
# # check against a null sample
nsamp = ip . Sample ( )
newkeys = nsamp . __dict__ . keys ( )
newstats = nsamp . __dict__ [ "stats" ] . keys ( )
newstatdfs = nsamp . __dict__ [ "stats_dfs" ] . keys ( )
newindstats = [ nsamp . __dict__ [ "stats_dfs" ] [ i ] . keys ( ) for i in newstatdfs ]
newindstats = list ( itertools . chain ( * [ i . values for i in newindstats ] ) )
# # different in attributes ?
diffattr = set ( sample_keys ) . difference ( newkeys )
diffstats = set ( stats_keys ) . difference ( newstats )
diffindstats = set ( ind_statkeys ) . difference ( newindstats )
# # Raise warning if any oldstats were lost or deprecated
alldiffs = diffattr . union ( diffstats ) . union ( diffindstats )
if any ( alldiffs ) :
    LOGGER . warning ( """
    load_json found {a} keys that are unique to the older Samples.
        - assembly [{b}] v.[{c}] has: {d}
        - current assembly is v.[{e}]
        """ . format ( a = len ( alldiffs ) , b = oldname , c = fullj [ "assembly" ] [ "_version" ] , d = alldiffs , e = null . _version ) )
# # save stats and statsfiles to Samples
for sample in null . samples : # # create a null Sample
    null . samples [ sample ] = ip . Sample ( )
    # # save stats
    sdat = fullj [ "samples" ] [ sample ] [ 'stats' ]
    # # Reorder the keys so they ascend by step , only include
    # # stats that are actually in the sample . newstats is a
    # # list of the new sample stat names , and stats _ keys
    # # are the names of the stats from the json file .
    newstats = [ x for x in newstats if x in stats_keys ]
    null . samples [ sample ] . stats = pd . Series ( sdat ) . reindex ( newstats )
    # # save stats _ dfs
    for statskey in stats_dfs_keys :
        null . samples [ sample ] . stats_dfs [ statskey ] = pd . Series ( fullj [ "samples" ] [ sample ] [ "stats_dfs" ] [ statskey ] ) . reindex ( nsamp . __dict__ [ "stats_dfs" ] [ statskey ] . keys ( ) )
    # # save Sample files
    for filehandle in fullj [ "samples" ] [ sample ] [ "files" ] . keys ( ) :
        null . samples [ sample ] . files [ filehandle ] = fullj [ "samples" ] [ sample ] [ "files" ] [ filehandle ]
# # build the Assembly object stats _ dfs
for statskey in stats_dfs_keys :
    indstat = null . _build_stat ( statskey )
    if not indstat . empty :
        null . stats_dfs [ statskey ] = indstat
# # add remaning attributes to null Samples
shared_keys = set ( sample_keys ) . intersection ( newkeys )
shared_keys . discard ( "stats" )
shared_keys . discard ( "files" )
shared_keys . discard ( "stats_files" )
shared_keys . discard ( "stats_dfs" )
for sample in null . samples : # # set the others
    for key in shared_keys :
        null . samples [ sample ] . __setattr__ ( key , fullj [ "samples" ] [ sample ] [ key ] )
# # ensure objects are object dicts
null . dirs = ObjDict ( null . dirs )
null . stats_files = ObjDict ( null . stats_files )
null . stats_dfs = ObjDict ( null . stats_dfs )
null . populations = ObjDict ( null . populations )
null . outfiles = ObjDict ( null . outfiles )
return null


signature:
def _tup_and_byte ( obj ) :
    """wat"""

implementation:
# if this is a unicode string , return its string representation
if isinstance ( obj , unicode ) :
    return obj . encode ( 'utf-8' )
# if this is a list of values , return list of byteified values
if isinstance ( obj , list ) :
    return [ _tup_and_byte ( item ) for item in obj ]
# if this is a dictionary , return dictionary of byteified keys and values
# but only if we haven ' t already byteified it
if isinstance ( obj , dict ) :
    if "__tuple__" in obj :
        return tuple ( _tup_and_byte ( item ) for item in obj [ "items" ] )
    else :
        return { _tup_and_byte ( key ) : _tup_and_byte ( val ) for key , val in obj . iteritems ( ) }
# if it ' s anything else , return it in its original form
return obj


signature:
def baba_panel_plot ( ttree , tests , boots , show_tip_labels = True , show_test_labels = True , use_edge_lengths = False , collapse_outgroup = False , pct_tree_x = 0.4 , pct_tree_y = 0.2 , alpha = 3.0 , * args , ** kwargs ) :
    """signature . . ."""

implementation:
# # create Panel plot object and set height & width
bootsarr = np . array ( boots )
panel = Panel ( ttree , tests , bootsarr , alpha )
if not kwargs . get ( "width" ) :
    panel . kwargs [ "width" ] = min ( 1000 , 50 * len ( panel . tree ) )
if not kwargs . get ( "height" ) :
    panel . kwargs [ "height" ] = min ( 1000 , 50 * len ( panel . tests ) )
# # update defaults with kwargs & update size based on ntips & ntests
kwargs . update ( dict ( pct_tree_x = pct_tree_x , pct_tree_y = pct_tree_y ) )
panel . kwargs . update ( kwargs )
# # create a canvas and a single cartesian coord system
canvas = toyplot . Canvas ( height = panel . kwargs [ 'height' ] , width = panel . kwargs [ 'width' ] )
axes = canvas . cartesian ( bounds = ( "10%" , "90%" , "5%" , "95%" ) )
axes . show = False
# # add panels to axes
panel . panel_tree ( axes )
panel . panel_test ( axes )
panel . panel_tip_labels ( axes )
if isinstance ( boots , np . ndarray ) :
    panel . panel_results ( axes )
return canvas , axes , panel


signature:
def depthplot ( data , samples = None , dims = ( None , None ) , canvas = ( None , None ) , xmax = 50 , log = False , outprefix = None , use_maxdepth = False ) :
    """plots histogram of coverages across clusters"""

implementation:
# # select samples to be plotted , requires depths info
if not samples :
    samples = data . samples . keys ( )
    samples . sort ( )
subsamples = OrderedDict ( [ ( i , data . samples [ i ] ) for i in samples ] )
# # get canvas dimensions based on n - samples
if any ( dims ) : # # user - supplied dimensions ( . . . )
    print ( "userdims" )
else :
    if len ( subsamples ) <= 4 : # # set dimension to N samples
        dims = ( 1 , len ( subsamples ) )
    else :
        dims = ( len ( subsamples ) / 4 , 4 )
# # create canvas
if any ( canvas ) :
    print ( "usercanvas" )
    canvas = toyplot . Canvas ( width = canvas [ 0 ] , height = canvas [ 1 ] )
else :
    canvas = toyplot . Canvas ( width = 200 * dims [ 1 ] , height = 150 * dims [ 0 ] )
# # get all of the data arrays
for panel , sample in enumerate ( subsamples ) : # # statistical called bins
    statdat = subsamples [ sample ] . depths
    statdat = statdat [ statdat >= data . paramsdict [ "mindepth_statistical" ] ]
    if use_maxdepth :
        statdat = { i : j for ( i , j ) in statdat if i < data . paramsdict [ "maxdepth" ] }
    sdat = np . histogram ( statdat , range ( 50 ) )
    # # majrule called bins
    statdat = subsamples [ sample ] . depths
    statdat = statdat [ statdat < data . paramsdict [ "mindepth_statistical" ] ]
    statdat = statdat [ statdat >= data . paramsdict [ "mindepth_majrule" ] ]
    if use_maxdepth :
        statdat = statdat [ statdat < data . paramsdict [ "maxdepth" ] ]
    mdat = np . histogram ( statdat , range ( 50 ) )
    # # excluded bins
    tots = data . samples [ sample ] . depths
    tots = tots [ tots < data . paramsdict [ "mindepth_majrule" ] ]
    if use_maxdepth :
        tots = tots [ tots < data . paramsdict [ "maxdepth" ] ]
    edat = np . histogram ( tots , range ( 50 ) )
    # # fill in each panel of canvas with a sample
    axes = canvas . cartesian ( grid = ( dims [ 0 ] , dims [ 1 ] , panel ) , gutter = 25 )
    axes . x . domain . xmax = xmax
    axes . label . text = sample
    if log :
        axes . y . scale = "log"
    # heights = np . column _ stack ( ( sdat , mdat , edat ) )
    axes . bars ( sdat )
    axes . bars ( edat )
    axes . bars ( mdat )
# # return objects to be saved . . .
if outprefix :
    toyplot . html . render ( canvas , fobj = outprefix + ".html" )
    toyplot . svg . render ( canvas , fobj = outprefix + ".svg" )


signature:
def _parse_00 ( ofile ) :
    """return 00 outfile as a pandas DataFrame"""

implementation:
with open ( ofile ) as infile : # # read in the results summary from the end of the outfile
    arr = np . array ( [ " " ] + infile . read ( ) . split ( "Summary of MCMC results\n\n\n" ) [ 1 : ] [ 0 ] . strip ( ) . split ( ) )
    # # reshape array
    rows = 12
    cols = ( arr . shape [ 0 ] + 1 ) / rows
    arr = arr . reshape ( rows , cols )
    # # make into labeled data frame
    df = pd . DataFrame ( data = arr [ 1 : , 1 : ] , columns = arr [ 0 , 1 : ] , index = arr [ 1 : , 0 ] , ) . T
    return df


signature:
def _parse_01 ( ofiles , individual = False ) :
    """a subfunction for summarizing results"""

implementation:
# # parse results from outfiles
cols = [ ]
dats = [ ]
for ofile in ofiles : # # parse file
    with open ( ofile ) as infile :
        dat = infile . read ( )
    lastbits = dat . split ( ".mcmc.txt\n\n" ) [ 1 : ]
    results = lastbits [ 0 ] . split ( "\n\n" ) [ 0 ] . split ( )
    # # get shape from . . .
    shape = ( ( ( len ( results ) - 3 ) / 4 ) , 4 )
    dat = np . array ( results [ 3 : ] ) . reshape ( shape )
    cols . append ( dat [ : , 3 ] . astype ( float ) )
if not individual : # # get mean results across reps
    cols = np . array ( cols )
    cols = cols . sum ( axis = 0 ) / len ( ofiles )
    # 10.
    dat [ : , 3 ] = cols . astype ( str )
    # # format as a DF
    df = pd . DataFrame ( dat [ : , 1 : ] )
    df . columns = [ "delim" , "prior" , "posterior" ]
    nspecies = 1 + np . array ( [ list ( i ) for i in dat [ : , 1 ] ] , dtype = int ) . sum ( axis = 1 )
    df [ "nspecies" ] = nspecies
    return df
else : # # get mean results across reps
# return cols
    res = [ ]
    for i in xrange ( len ( cols ) ) :
        x = dat
        x [ : , 3 ] = cols [ i ] . astype ( str )
        x = pd . DataFrame ( x [ : , 1 : ] )
        x . columns = [ 'delim' , 'prior' , 'posterior' ]
        nspecies = 1 + np . array ( [ list ( i ) for i in dat [ : , 1 ] ] , dtype = int ) . sum ( axis = 1 )
        x [ "nspecies" ] = nspecies
        res . append ( x )
    return res


signature:
def _load_existing_results ( self , name , workdir ) :
    """Load existing results files for an object with this workdir and name .
 This does NOT reload the parameter settings for the object . . ."""

implementation:
# # get mcmcs
path = os . path . realpath ( os . path . join ( self . workdir , self . name ) )
mcmcs = glob . glob ( path + "_r*.mcmc.txt" )
outs = glob . glob ( path + "_r*.out.txt" )
trees = glob . glob ( path + "_r*.tre" )
for mcmcfile in mcmcs :
    if mcmcfile not in self . files . mcmcfiles :
        self . files . mcmcfiles . append ( mcmcfile )
for outfile in outs :
    if outfile not in self . files . outfiles :
        self . files . outfiles . append ( outfile )
for tree in trees :
    if tree not in self . files . treefiles :
        self . files . treefiles . append ( tree )


signature:
def run ( self , ipyclient , nreps = 1 , quiet = False , randomize_order = False , force = False , ) :
    """Submits bpp jobs to run on a cluster ( ipyparallel Client ) .
 The seed for the random number generator if not set is randomly
 drawn , and if multiple reps are submitted ( nreps > 1 ) then each will
 draw a subsequent random seeds after that . An ipyclient connection
 is required . Asynchronous result objects are stored in the bpp
 object submitting the jobs .
 Parameters :
 nreps ( int ) :
 submits nreps replicate jobs to the cluster each with a different
 random seed drawn starting from the starting seed .
 ipyclient ( ipyparallel . Client )
 an ipyparallel . Client object connected to a running cluster .
 quiet ( bool ) :
 whether to print that the jobs have been submitted
 randomize _ order ( bool ) :
 if True then when maxloci is set this will randomly sample a
 different set of N loci in each replicate , rather than sampling
 just the first N loci < maxloci .
 force ( bool ) :
 Overwrite existing files with the same name . Default = False , skip
 over existing files ."""

implementation:
# # is this running algorithm 00?
is_alg00 = ( not self . params . infer_sptree ) and ( not self . params . infer_delimit )
# # clear out pre - existing files for this object
self . files . mcmcfiles = [ ]
self . files . outfiles = [ ]
self . files . treefiles = [ ]
self . asyncs = [ ]
# # initiate random seed
np . random . seed ( self . params . seed )
# # load - balancer
lbview = ipyclient . load_balanced_view ( )
# # send jobs
for job in xrange ( nreps ) : # # make repname and make ctl filename
    self . _name = "{}_r{}" . format ( self . name , job )
    ctlhandle = os . path . realpath ( os . path . join ( self . workdir , "{}.ctl.txt" . format ( self . _name ) ) )
    # # skip if ctlfile exists
    if ( not force ) and ( os . path . exists ( ctlhandle ) ) :
        print ( "Named ctl file already exists. Use force=True to" + " overwrite\nFilename:{}" . format ( ctlhandle ) )
    else : # # change seed and ctl for each rep , this writes into the ctl
    # # file the correct name for the other files which share the
    # # same rep number in their names .
    # self . params . _ seed = np . random . randint ( 0 , 1e9 , 1 ) [ 0]
        self . _write_mapfile ( )
        # if randomize _ order :
        self . _write_seqfile ( randomize_order = randomize_order )
        ctlfile = self . _write_ctlfile ( )
        # # submit to engines
        async = lbview . apply ( _call_bpp , * ( self . _kwargs [ "binary" ] , ctlfile , is_alg00 ) )
        self . asyncs . append ( async )
        # # save tree file if alg 00
        if is_alg00 :
            self . files . treefiles . append ( ctlfile . rsplit ( ".ctl.txt" , 1 ) [ 0 ] + ".tre" )
if self . asyncs and ( not quiet ) :
    sys . stderr . write ( "submitted {} bpp jobs [{}] ({} loci)\n" . format ( nreps , self . name , self . _nloci ) )


signature:
def write_bpp_files ( self , randomize_order = False , quiet = False ) :
    """Writes bpp files ( . ctl , . seq , . imap ) to the working directory .
 Parameters :
 randomize _ order ( bool ) :
 whether to randomize the locus order , this will allow you to
 sample different subsets of loci in different replicates when
 using the filters . maxloci option .
 quiet ( bool ) :
 whether to print info to stderr when finished ."""

implementation:
# # remove any old jobs with this same job name
self . _name = self . name
oldjobs = glob . glob ( os . path . join ( self . workdir , self . _name + "*.ctl.txt" ) )
for job in oldjobs :
    os . remove ( job )
# # check params types
# # write tmp files for the job
self . _write_seqfile ( randomize_order = randomize_order )
self . _write_mapfile ( )
# name = True )
self . _write_ctlfile ( )
if not quiet :
    sys . stderr . write ( "input files created for job {} ({} loci)\n" . format ( self . _name , self . _nloci ) )


signature:
def _write_ctlfile ( self ) : # , rep = None ) :
    """write outfile with any args in argdict"""

implementation:
# # A string to store ctl info
ctl = [ ]
# # write the top header info
ctl . append ( "seed = {}" . format ( self . params . seed ) )
ctl . append ( "seqfile = {}" . format ( self . seqfile ) )
ctl . append ( "Imapfile = {}" . format ( self . mapfile ) )
path = os . path . realpath ( os . path . join ( self . workdir , self . _name ) )
mcmcfile = "{}.mcmc.txt" . format ( path )
outfile = "{}.out.txt" . format ( path )
if mcmcfile not in self . files . mcmcfiles :
    self . files . mcmcfiles . append ( mcmcfile )
if outfile not in self . files . outfiles :
    self . files . outfiles . append ( outfile )
ctl . append ( "mcmcfile = {}" . format ( mcmcfile ) )
ctl . append ( "outfile = {}" . format ( outfile ) )
# # number of loci ( checks that seq file exists and parses from there )
ctl . append ( "nloci = {}" . format ( self . _nloci ) )
ctl . append ( "usedata = {}" . format ( self . params . usedata ) )
ctl . append ( "cleandata = {}" . format ( self . params . cleandata ) )
# # infer species tree
if self . params . infer_sptree :
    ctl . append ( "speciestree = 1 0.4 0.2 0.1" )
else :
    ctl . append ( "speciestree = 0" )
# # infer delimitation ( with algorithm 1 by default )
ctl . append ( "speciesdelimitation = {} {} {}" . format ( self . params . infer_delimit , self . params . delimit_alg [ 0 ] , " " . join ( [ str ( i ) for i in self . params . delimit_alg [ 1 : ] ] ) ) )
# # get tree values
nspecies = str ( len ( self . imap ) )
species = " " . join ( sorted ( self . imap ) )
ninds = " " . join ( [ str ( len ( self . imap [ i ] ) ) for i in sorted ( self . imap ) ] )
ctl . append ( SPECIESTREE . format ( nspecies , species , ninds , self . tree . write ( format = 9 ) ) )
# # priors
ctl . append ( "thetaprior = {} {}" . format ( * self . params . thetaprior ) )
ctl . append ( "tauprior = {} {} {}" . format ( * self . params . tauprior ) )
# # other values , fixed for now
ctl . append ( "finetune = 1: {}" . format ( " " . join ( [ str ( i ) for i in self . params . finetune ] ) ) )
# CTL . append ( " finetune = 1 : 1 0.002 0.01 0.01 0.02 0.005 1.0 " )
ctl . append ( "print = 1 0 0 0" )
ctl . append ( "burnin = {}" . format ( self . params . burnin ) )
ctl . append ( "sampfreq = {}" . format ( self . params . sampfreq ) )
ctl . append ( "nsample = {}" . format ( self . params . nsample ) )
# # write out the ctl file
ctlhandle = os . path . realpath ( "{}.ctl.txt" . format ( os . path . join ( self . workdir , self . _name ) ) )
# if isinstance ( rep , int ) :
# ctlhandle = os . path . realpath (
# " { } - r { } . ctl . txt " . format ( os . path . join ( self . workdir , self . _ name ) , rep ) )
# else :
# ctlhandle = os . path . realpath (
# " { } . ctl . txt " . format ( os . path . join ( self . workdir , self . _ name ) ) )
with open ( ctlhandle , 'w' ) as out :
    out . write ( "\n" . join ( ctl ) )
return ctlhandle


signature:
def copy ( self , name , load_existing_results = False ) :
    """Returns a copy of the bpp object with the same parameter settings
 but with the files . mcmcfiles and files . outfiles attributes cleared ,
 and with a new ' name ' attribute .
 Parameters
 name ( str ) :
 A name for the new copied bpp object that will be used for the
 output files created by the object ."""

implementation:
# # make deepcopy of self . _ _ dict _ _ but do not copy async objects
subdict = { i : j for i , j in self . __dict__ . iteritems ( ) if i != "asyncs" }
newdict = copy . deepcopy ( subdict )
# # make back into a bpp object
if name == self . name :
    raise Exception ( "new object must have a different 'name' than its parent" )
newobj = Bpp ( name = name , data = newdict [ "files" ] . data , workdir = newdict [ "workdir" ] , guidetree = newdict [ "tree" ] . write ( ) , imap = { i : j for i , j in newdict [ "imap" ] . items ( ) } , copied = True , load_existing_results = load_existing_results , )
# # update special dict attributes but not files
for key , val in newobj . params . __dict__ . iteritems ( ) :
    newobj . params . __setattr__ ( key , self . params . __getattribute__ ( key ) )
for key , val in newobj . filters . __dict__ . iteritems ( ) :
    newobj . filters . __setattr__ ( key , self . filters . __getattribute__ ( key ) )
# # new object must have a different name than it ' s parent
return newobj


signature:
def summarize_results ( self , individual_results = False ) :
    """Prints a summarized table of results from replicate runs , or ,
 if individual _ result = True , then returns a list of separate
 dataframes for each replicate run ."""

implementation:
# # return results depending on algorithm
# # algorithm 00
if ( not self . params . infer_delimit ) & ( not self . params . infer_sptree ) :
    if individual_results : # # return a list of parsed CSV results
        return [ _parse_00 ( i ) for i in self . files . outfiles ]
    else : # # concatenate each CSV and then get stats w / describe
        return pd . concat ( [ pd . read_csv ( i , sep = '\t' , index_col = 0 ) for i in self . files . mcmcfiles ] ) . describe ( ) . T
# # algorithm 01
if self . params . infer_delimit & ( not self . params . infer_sptree ) :
    return _parse_01 ( self . files . outfiles , individual = individual_results )
# # others
else :
    return "summary function not yet ready for this type of result"


signature:
def persistent_popen_align3 ( data , samples , chunk ) :
    """notes"""

implementation:
# # data are already chunked , read in the whole thing
with open ( chunk , 'rb' ) as infile :
    clusts = infile . read ( ) . split ( "//\n//\n" ) [ : - 1 ]
# # snames to ensure sorted order
samples . sort ( key = lambda x : x . name )
snames = [ sample . name for sample in samples ]
# # make a tmparr to store metadata ( this can get huge , consider using h5)
maxlen = data . _hackersonly [ "max_fragment_length" ] + 20
indels = np . zeros ( ( len ( samples ) , len ( clusts ) , maxlen ) , dtype = np . bool_ )
duples = np . zeros ( len ( clusts ) , dtype = np . bool_ )
# # create a persistent shell for running muscle in .
proc = sps . Popen ( [ "bash" ] , stdin = sps . PIPE , stdout = sps . PIPE , universal_newlines = True )
# # iterate over clusters until finished
allstack = [ ]
# istack = [ ]
for ldx in xrange ( len ( clusts ) ) : # # new alignment string for read1s and read2s
    aligned = [ ]
    istack = [ ]
    lines = clusts [ ldx ] . strip ( ) . split ( "\n" )
    names = lines [ : : 2 ]
    seqs = lines [ 1 : : 2 ]
    align1 = ""
    align2 = ""
    # # we don ' t allow seeds with no hits to make it here , currently
    # if len ( names ) = = 1:
    # aligned . append ( clusts [ ldx ] . replace ( " > " , " " ) . strip ( ) )
    # # find duplicates and skip aligning but keep it for downstream .
    if len ( names ) != len ( set ( [ x . rsplit ( "_" , 1 ) [ 0 ] for x in names ] ) ) :
        duples [ ldx ] = 1
        istack = [ "{}\n{}" . format ( i [ 1 : ] , j ) for i , j in zip ( names , seqs ) ]
        # aligned . append ( clusts [ ldx ] . replace ( " > " , " " ) . strip ( ) )
    else : # # append counter to names because muscle doesn ' t retain order
        names = [ ">{};*{}" . format ( j [ 1 : ] , i ) for i , j in enumerate ( names ) ]
        try : # # try to split names on nnnn splitter
            clust1 , clust2 = zip ( * [ i . split ( "nnnn" ) for i in seqs ] )
            # # make back into strings
            cl1 = "\n" . join ( itertools . chain ( * zip ( names , clust1 ) ) )
            cl2 = "\n" . join ( itertools . chain ( * zip ( names , clust2 ) ) )
            # # store allele ( lowercase ) info
            shape = ( len ( seqs ) , max ( [ len ( i ) for i in seqs ] ) )
            arrseqs = np . zeros ( shape , dtype = "S1" )
            for row in range ( arrseqs . shape [ 0 ] ) :
                seqsrow = seqs [ row ]
                arrseqs [ row , : len ( seqsrow ) ] = list ( seqsrow )
            amask = np . char . islower ( arrseqs )
            save_alleles = np . any ( amask )
            # # send align1 to the bash shell
            # # TODO : check for pipe - overflow here and use files for i / o
            cmd1 = "echo -e '{}' | {} -quiet -in - ; echo {}" . format ( cl1 , ipyrad . bins . muscle , "//" )
            print ( cmd1 , file = proc . stdin )
            # # read the stdout by line until splitter is reached
            for line in iter ( proc . stdout . readline , "//\n" ) :
                align1 += line
            # # send align2 to the bash shell
            # # TODO : check for pipe - overflow here and use files for i / o
            cmd2 = "echo -e '{}' | {} -quiet -in - ; echo {}" . format ( cl2 , ipyrad . bins . muscle , "//" )
            print ( cmd2 , file = proc . stdin )
            # # read the stdout by line until splitter is reached
            for line in iter ( proc . stdout . readline , "//\n" ) :
                align2 += line
            # # join the aligned read1 and read2 and ensure name order match
            la1 = align1 [ 1 : ] . split ( "\n>" )
            la2 = align2 [ 1 : ] . split ( "\n>" )
            dalign1 = dict ( [ i . split ( "\n" , 1 ) for i in la1 ] )
            dalign2 = dict ( [ i . split ( "\n" , 1 ) for i in la2 ] )
            keys = sorted ( dalign1 . keys ( ) , key = DEREP )
            keys2 = sorted ( dalign2 . keys ( ) , key = DEREP )
            # # Make sure R1 and R2 actually exist for each sample . If not
            # # bail out of this cluster .
            if not len ( keys ) == len ( keys2 ) :
                LOGGER . error ( "R1 and R2 results differ in length: " + "\nR1 - {}\nR2 - {}" . format ( keys , keys2 ) )
                continue
            # # impute allele ( lowercase ) info back into alignments
            for kidx , key in enumerate ( keys ) :
                concatseq = dalign1 [ key ] . replace ( "\n" , "" ) + "nnnn" + dalign2 [ key ] . replace ( "\n" , "" )
                # # impute alleles
                if save_alleles :
                    newmask = np . zeros ( len ( concatseq ) , dtype = np . bool_ )
                    # # check for indels and impute to amask
                    indidx = np . where ( np . array ( list ( concatseq ) ) == "-" ) [ 0 ]
                    if indidx . size :
                        allrows = np . arange ( amask . shape [ 1 ] )
                        mask = np . ones ( allrows . shape [ 0 ] , dtype = np . bool_ )
                        for idx in indidx :
                            if idx < mask . shape [ 0 ] :
                                mask [ idx ] = False
                        not_idx = allrows [ mask == 1 ]
                        # # fill in new data into all other spots
                        newmask [ not_idx ] = amask [ kidx , : not_idx . shape [ 0 ] ]
                    else :
                        newmask = amask [ kidx ]
                    # # lower the alleles
                    concatarr = np . array ( list ( concatseq ) )
                    concatarr [ newmask ] = np . char . lower ( concatarr [ newmask ] )
                    concatseq = concatarr . tostring ( )
                    # LOGGER . info ( concatseq )
                # # fill list with aligned data
                aligned . append ( "{}\n{}" . format ( key , concatseq ) )
            # # put into a dict for writing to file
            # aligned = [ ]
            # for key in keys :
            # aligned . append ( " \ n " . join (
            # [ key ,
            # dalign1 [ key ] . replace ( " \ n " , " " ) + " nnnn " + \
            # dalign2 [ key ] . replace ( " \ n " , " " ) ] ) )
        except IndexError as inst :
            LOGGER . debug ( "Error in PE - ldx: {}" . format ( ) )
            LOGGER . debug ( "Vars: {}" . format ( dict ( globals ( ) , ** locals ( ) ) ) )
            raise
        except ValueError : # # make back into strings
            cl1 = "\n" . join ( [ "\n" . join ( i ) for i in zip ( names , seqs ) ] )
            # # store allele ( lowercase ) info
            shape = ( len ( seqs ) , max ( [ len ( i ) for i in seqs ] ) )
            arrseqs = np . zeros ( shape , dtype = "S1" )
            for row in range ( arrseqs . shape [ 0 ] ) :
                seqsrow = seqs [ row ]
                arrseqs [ row , : len ( seqsrow ) ] = list ( seqsrow )
            amask = np . char . islower ( arrseqs )
            save_alleles = np . any ( amask )
            # # send align1 to the bash shell ( TODO : check for pipe - overflow )
            cmd1 = "echo -e '{}' | {} -quiet -in - ; echo {}" . format ( cl1 , ipyrad . bins . muscle , "//" )
            print ( cmd1 , file = proc . stdin )
            # # read the stdout by line until splitter is reached
            for line in iter ( proc . stdout . readline , "//\n" ) :
                align1 += line
            # # ensure name order match
            la1 = align1 [ 1 : ] . split ( "\n>" )
            dalign1 = dict ( [ i . split ( "\n" , 1 ) for i in la1 ] )
            keys = sorted ( dalign1 . keys ( ) , key = DEREP )
            # # put into dict for writing to file
            for kidx , key in enumerate ( keys ) :
                concatseq = dalign1 [ key ] . replace ( "\n" , "" )
                # # impute alleles
                if save_alleles :
                    newmask = np . zeros ( len ( concatseq ) , dtype = np . bool_ )
                    # # check for indels and impute to amask
                    indidx = np . where ( np . array ( list ( concatseq ) ) == "-" ) [ 0 ]
                    if indidx . size :
                        allrows = np . arange ( amask . shape [ 1 ] )
                        mask = np . ones ( allrows . shape [ 0 ] , dtype = np . bool_ )
                        for idx in indidx :
                            if idx < mask . shape [ 0 ] :
                                mask [ idx ] = False
                        not_idx = allrows [ mask == 1 ]
                        # # fill in new data into all other spots
                        newmask [ not_idx ] = amask [ kidx , : not_idx . shape [ 0 ] ]
                    else :
                        newmask = amask [ kidx ]
                    # # lower the alleles
                    concatarr = np . array ( list ( concatseq ) )
                    concatarr [ newmask ] = np . char . lower ( concatarr [ newmask ] )
                    concatseq = concatarr . tostring ( )
                # # fill list with aligned data
                aligned . append ( "{}\n{}" . format ( key , concatseq ) )
            # # put aligned locus in list
            # aligned . append ( " \ n " . join ( inner _ aligned ) )
        # # enforce maxlen on aligned seqs
        aseqs = np . vstack ( [ list ( i . split ( "\n" ) [ 1 ] ) for i in aligned ] )
        LOGGER . info ( "\naseqs here: %s" , aseqs )
        # # index names by snames order
        sidxs = [ snames . index ( key . rsplit ( "_" , 1 ) [ 0 ] ) for key in keys ]
        thislen = min ( maxlen , aseqs . shape [ 1 ] )
        for idx in xrange ( aseqs . shape [ 0 ] ) : # # enter into stack
            newn = aligned [ idx ] . split ( ";" , 1 ) [ 0 ]
            # newn = key [ idx ] . split ( " ; " , 1 ) [ 0]
            istack . append ( "{}\n{}" . format ( newn , aseqs [ idx , : thislen ] . tostring ( ) ) )
            # # name index in sorted list ( indels order )
            sidx = sidxs [ idx ]
            indels [ sidx , ldx , : thislen ] = aseqs [ idx , : thislen ] == "-"
    if istack :
        allstack . append ( "\n" . join ( istack ) )
        # LOGGER . debug ( " \ n \ nSTACK ( % s ) \ n % s \ n " , duples [ ldx ] , " \ n " . join ( istack ) )
# # cleanup
proc . stdout . close ( )
if proc . stderr :
    proc . stderr . close ( )
proc . stdin . close ( )
proc . wait ( )
# LOGGER . info ( " \ n \ nALLSTACK % s \ n " , " \ n " . join ( i ) for i in allstack [ : 5 ] ] )
# # write to file after
odx = chunk . rsplit ( "_" ) [ - 1 ]
alignfile = os . path . join ( data . tmpdir , "align_{}.fa" . format ( odx ) )
with open ( alignfile , 'wb' ) as outfile :
    outfile . write ( "\n//\n//\n" . join ( allstack ) + "\n" )
    os . remove ( chunk )
# # save indels array to tmp dir
ifile = os . path . join ( data . tmpdir , "indels_{}.tmp.npy" . format ( odx ) )
np . save ( ifile , indels )
dfile = os . path . join ( data . tmpdir , "duples_{}.tmp.npy" . format ( odx ) )
np . save ( dfile , duples )


signature:
def multi_muscle_align ( data , samples , ipyclient ) :
    """Sends the cluster bits to nprocessors for muscle alignment . They return
 with indel . h5 handles to be concatenated into a joint h5."""

implementation:
LOGGER . info ( "starting alignments" )
# # get client
lbview = ipyclient . load_balanced_view ( )
start = time . time ( )
printstr = " aligning clusters     | {} | s6 |"
elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
progressbar ( 20 , 0 , printstr . format ( elapsed ) , spacer = data . _spacer )
# # submit clustbits as jobs to engines . The chunkfiles are removed when they
# # are finished so this job can even be restarted if it was half finished ,
# # though that is probably rare .
path = os . path . join ( data . tmpdir , data . name + ".chunk_*" )
clustbits = glob . glob ( path )
jobs = { }
for idx in xrange ( len ( clustbits ) ) :
    args = [ data , samples , clustbits [ idx ] ]
    jobs [ idx ] = lbview . apply ( persistent_popen_align3 , * args )
allwait = len ( jobs )
elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
progressbar ( 20 , 0 , printstr . format ( elapsed ) , spacer = data . _spacer )
# # print progress while bits are aligning
while 1 :
    finished = [ i . ready ( ) for i in jobs . values ( ) ]
    fwait = sum ( finished )
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( allwait , fwait , printstr . format ( elapsed ) , spacer = data . _spacer )
    time . sleep ( 0.1 )
    if all ( finished ) :
        break
# # check for errors in muscle _ align _ across
keys = jobs . keys ( )
for idx in keys :
    if not jobs [ idx ] . successful ( ) :
        LOGGER . error ( "error in persistent_popen_align %s" , jobs [ idx ] . exception ( ) )
        raise IPyradWarningExit ( "error in step 6 {}" . format ( jobs [ idx ] . exception ( ) ) )
    del jobs [ idx ]
print ( "" )


signature:
def concatclusts ( outhandle , alignbits ) :
    """concatenates sorted aligned cluster tmpfiles and removes them ."""

implementation:
with gzip . open ( outhandle , 'wb' ) as out :
    for fname in alignbits :
        with open ( fname ) as infile :
            out . write ( infile . read ( ) + "//\n//\n" )


signature:
def build_indels ( data , samples , ipyclient ) :
    """Builds the indels array and catclust . gz file from the aligned clusters .
 Building catclust is very fast . Entering indels into h5 array is a bit
 slow but can probably be sped up . ( todo ) . NOT currently parallelized ."""

implementation:
# # progress bars
lbview = ipyclient . load_balanced_view ( )
start = time . time ( )
printstr = " database indels       | {} | s6 |"
njobs = len ( glob . glob ( os . path . join ( data . tmpdir , "align_*.fa" ) ) ) + 1
# # build tmparrs
async = lbview . apply ( build_tmp_h5 , * ( data , samples ) )
# # track progress
while 1 :
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    ready = bool ( async . ready ( ) )
    progressbar ( njobs , ready , printstr . format ( elapsed ) , spacer = data . _spacer )
    if ready :
        break
    else :
        time . sleep ( 0.1 )
# # check for errors
if not async . successful ( ) :
    raise IPyradWarningExit ( async . result ( ) )
# # start subfunc
async = lbview . apply ( sub_build_indels , * ( data , samples ) )
prog = 1
while 1 :
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    if async . stdout :
        prog = int ( async . stdout . split ( ) [ - 1 ] ) + 1
    progressbar ( njobs , prog , printstr . format ( elapsed ) , spacer = data . _spacer )
    if async . ready ( ) :
        break
    else :
        time . sleep ( 0.1 )
# # check for errors
if not async . successful ( ) :
    raise IPyradWarningExit ( async . result ( ) )
print ( "" )
# # prepare for next substep by removing the singlecat result files if
# # they exist .
snames = [ i . name for i in samples ]
snames . sort ( )
smpios = [ os . path . join ( data . dirs . across , i + '.tmp.h5' ) for i in snames ]
for smpio in smpios :
    if os . path . exists ( smpio ) :
        os . remove ( smpio )


signature:
def sub_build_indels ( data , samples ) :
    """sub func in ` build _ indels ( ) ` ."""

implementation:
# # get file handles
indelfiles = glob . glob ( os . path . join ( data . tmpdir , "indels_*.tmp.npy" ) )
alignbits = glob . glob ( os . path . join ( data . tmpdir , "align_*.fa" ) )
# # sort into input order by chunk names
indelfiles . sort ( key = lambda x : int ( x . rsplit ( "_" , 1 ) [ - 1 ] [ : - 8 ] ) )
alignbits . sort ( key = lambda x : int ( x . rsplit ( "_" , 1 ) [ - 1 ] [ : - 3 ] ) )
LOGGER . info ( "indelfiles %s" , indelfiles )
LOGGER . info ( "alignbits %s" , alignbits )
chunksize = int ( indelfiles [ 0 ] . rsplit ( "_" , 1 ) [ - 1 ] [ : - 8 ] )
# # concatenate finished seq clusters into a tmp file
outhandle = os . path . join ( data . dirs . across , data . name + "_catclust.gz" )
concatclusts ( outhandle , alignbits )
# # get dims for full indel array
maxlen = data . _hackersonly [ "max_fragment_length" ] + 20
nloci = get_nloci ( data )
LOGGER . info ( "maxlen inside build is %s" , maxlen )
LOGGER . info ( "nloci for indels %s" , nloci )
# # INIT TEMP INDEL ARRAY
# # build an indel array for ALL loci in cat . clust . gz ,
# # chunked so that individual samples can be pulled out
ipath = os . path . join ( data . dirs . across , data . name + ".tmp.indels.hdf5" )
with h5py . File ( ipath , 'w' ) as io5 :
    iset = io5 . create_dataset ( "indels" , shape = ( len ( samples ) , nloci , maxlen ) , dtype = np . bool_ , chunks = ( 1 , chunksize , maxlen ) )
    # # again make sure names are ordered right
    samples . sort ( key = lambda x : x . name )
    # iset . attrs [ " chunksize " ] = ( 1 , data . nloci , maxlen )
    iset . attrs [ "samples" ] = [ i . name for i in samples ]
    # # enter all tmpindel arrays into full indel array
    done = 0
    init = 0
    for indf in indelfiles :
        end = int ( indf . rsplit ( "_" , 1 ) [ - 1 ] [ : - 8 ] )
        inarr = np . load ( indf )
        LOGGER . info ( 'inarr shape %s' , inarr . shape )
        LOGGER . info ( 'iset shape %s' , iset . shape )
        iset [ : , init : end , : ] = inarr [ : , : end - init ]
        init += end - init
        done += 1
        print ( done )


signature:
def call_cluster ( data , noreverse , ipyclient ) :
    """distributes ' cluster ( ) ' function to an ipyclient to make sure it runs
 on a high memory node ."""

implementation:
# # Find host with the most engines , for now just using first .
lbview = ipyclient . load_balanced_view ( )
# # request engine data , skips busy engines .
asyncs = { }
for eid in ipyclient . ids :
    engine = ipyclient [ eid ]
    if not engine . outstanding :
        asyncs [ eid ] = engine . apply ( socket . gethostname )
# # get results
hosts = { }
for key in asyncs :
    hosts [ key ] = asyncs [ key ] . get ( )
# # count them
results = { }
for eid , hostname in hosts . items ( ) :
    if hostname in results :
        results [ hostname ] . append ( eid )
    else :
        results [ hostname ] = [ eid ]
# # which is largest
hosts = sorted ( results . items ( ) , key = lambda x : len ( x [ 1 ] ) , reverse = True )
_ , eids = hosts [ 0 ]
bighost = ipyclient [ eids [ 0 ] ]
# # nthreads is len eids , or ipcluster . threads , unless ipcluster . threads
# # is really small , then we assume threads should not apply here .
# # ipyrad - p params . txt - s 6 - c 20 would give :
# # min ( 20 , max ( 2 , 10 ) ) = 8
# # while
# # ipyrad - p params . txt - s 6 - c 20 - t 4 would give :
# # min ( 20 , max ( 4 , 10 ) ) = 10
# # and
# # ipyrad - p params . txt - s 6 - c 20 - t 15 would give :
# # min ( 20 , max ( 15 , 10 ) ) = 15
# # and
# # ipyrad - p params . txt - s 6 - c 16 - - MPI ( on 2 X 8 - core nodes ) would give :
# # min ( 8 , max ( 2 , 10 ) ) = 8
nthreads = min ( len ( eids ) , max ( data . _ipcluster [ "threads" ] , 10 ) )
# # submit job to the host with the most
async = bighost . apply ( cluster , * ( data , noreverse , nthreads ) )
# async = lbview . apply ( cluster , * ( data , noreverse , nthreads ) )
# # track progress
prog = 0
start = time . time ( )
printstr = " clustering across     | {} | s6 |"
while 1 :
    if async . stdout :
        prog = int ( async . stdout . split ( ) [ - 1 ] )
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( 100 , prog , printstr . format ( elapsed ) , spacer = data . _spacer )
    if async . ready ( ) :
        progressbar ( 100 , prog , printstr . format ( elapsed ) , spacer = data . _spacer )
        print ( "" )
        break
    else :
        time . sleep ( 0.5 )
# # store log result
ipyclient . wait ( )
data . stats_files . s6 = os . path . join ( data . dirs . across , "s6_cluster_stats.txt" )


signature:
def cluster ( data , noreverse , nthreads ) :
    """Calls vsearch for clustering across samples ."""

implementation:
# # input and output file handles
cathaplos = os . path . join ( data . dirs . across , data . name + "_catshuf.tmp" )
uhaplos = os . path . join ( data . dirs . across , data . name + ".utemp" )
hhaplos = os . path . join ( data . dirs . across , data . name + ".htemp" )
logfile = os . path . join ( data . dirs . across , "s6_cluster_stats.txt" )
# # parameters that vary by datatype
# # ( too low of cov values yield too many poor alignments )
strand = "plus"
cov = 0.75
# #0.90
if data . paramsdict [ "datatype" ] in [ "gbs" , "2brad" ] :
    strand = "both"
    cov = 0.60
elif data . paramsdict [ "datatype" ] == "pairgbs" :
    strand = "both"
    cov = 0.75
# #0.90
# # nthreads is calculated in ' call _ cluster ( ) '
cmd = [ ipyrad . bins . vsearch , "-cluster_smallmem" , cathaplos , "-strand" , strand , "-query_cov" , str ( cov ) , "-minsl" , str ( 0.5 ) , "-id" , str ( data . paramsdict [ "clust_threshold" ] ) , "-userout" , uhaplos , "-notmatched" , hhaplos , "-userfields" , "query+target+qstrand" , "-maxaccepts" , "1" , "-maxrejects" , "0" , "-fasta_width" , "0" , "-threads" , str ( nthreads ) , "-fulldp" , "-usersort" , "-log" , logfile ]
# # override reverse clustering option
if noreverse :
    strand = "plus"
# - leftjust "
try : # # this seems to start vsearch on a different pid than the engine
# # and so it ' s hard to kill . . .
    LOGGER . info ( cmd )
    ( dog , owner ) = pty . openpty ( )
    proc = sps . Popen ( cmd , stdout = owner , stderr = owner , close_fds = True )
    prog = 0
    newprog = 0
    while 1 :
        isdat = select . select ( [ dog ] , [ ] , [ ] , 0 )
        if isdat [ 0 ] :
            dat = os . read ( dog , 80192 )
        else :
            dat = ""
        if "Clustering" in dat :
            try :
                newprog = int ( dat . split ( ) [ - 1 ] [ : - 1 ] )
            # # may raise value error when it gets to the end
            except ValueError :
                pass
        # # break if done
        # # catches end chunk of printing if clustering went really fast
        elif "Clusters:" in dat :
            LOGGER . info ( "ended vsearch tracking loop" )
            break
        else :
            time . sleep ( 0.1 )
        # # print progress
        if newprog != prog :
            print ( newprog )
            prog = newprog
    # # another catcher to let vsearch cleanup after clustering is done
    proc . wait ( )
    print ( 100 )
except KeyboardInterrupt :
    LOGGER . info ( "interrupted vsearch here: %s" , proc . pid )
    os . kill ( proc . pid , 2 )
    raise KeyboardInterrupt ( )
except sps . CalledProcessError as inst :
    raise IPyradWarningExit ( """
        Error in vsearch: \n{}\n{}""" . format ( inst , sps . STDOUT ) )
except OSError as inst :
    raise IPyradWarningExit ( """
        Failed to allocate pty: \n{}""" . format ( inst ) )
finally :
    data . stats_files . s6 = logfile


signature:
def build_h5_array ( data , samples , nloci ) :
    """Sets up all of the h5 arrays that we will fill .
 The catg array of prefiltered loci is 4 - dimensional ( Big ) , so one big
 array would overload memory , we need to fill it in slices .
 This will be done in multicat ( singlecat ) and fill _ superseqs ."""

implementation:
# # sort to ensure samples will be in alphabetical order , tho they should be .
samples . sort ( key = lambda x : x . name )
# # get maxlen dim
maxlen = data . _hackersonly [ "max_fragment_length" ] + 20
LOGGER . info ( "maxlen inside build_h5_array is %s" , maxlen )
LOGGER . info ( "nloci inside build_h5_array is %s" , nloci )
# # open new h5 handle
data . clust_database = os . path . join ( data . dirs . across , data . name + ".clust.hdf5" )
io5 = h5py . File ( data . clust_database , 'w' )
# # chunk to approximately 2 chunks per core
chunks = ( ( nloci // ( data . cpus * 2 ) ) + ( nloci % ( data . cpus * 2 ) ) )
# # Number of elements in hdf5 chunk may not exceed 500MB
# # This is probably not actually optimal , to have such
# # enormous chunk sizes , could probably explore efficiency
# # of smaller chunk sizes on very very large datasets
chunklen = chunks * len ( samples ) * maxlen * 4
while chunklen > int ( 500e6 ) :
    chunks = ( chunks // 2 ) + ( chunks % 2 )
    chunklen = chunks * len ( samples ) * maxlen * 4
LOGGER . info ( "chunks in build_h5_array: %s" , chunks )
data . chunks = chunks
LOGGER . info ( "nloci is %s" , nloci )
LOGGER . info ( "chunks is %s" , data . chunks )
# # INIT FULL CATG ARRAY
# # store catgs with a . 10 loci chunk size
supercatg = io5 . create_dataset ( "catgs" , ( nloci , len ( samples ) , maxlen , 4 ) , dtype = np . uint32 , chunks = ( chunks , 1 , maxlen , 4 ) , compression = "gzip" )
superseqs = io5 . create_dataset ( "seqs" , ( nloci , len ( samples ) , maxlen ) , dtype = "|S1" , # dtype = np . uint8,
chunks = ( chunks , len ( samples ) , maxlen ) , compression = 'gzip' )
superalls = io5 . create_dataset ( "nalleles" , ( nloci , len ( samples ) ) , dtype = np . uint8 , chunks = ( chunks , len ( samples ) ) , compression = "gzip" )
superchroms = io5 . create_dataset ( "chroms" , ( nloci , 3 ) , dtype = np . int64 , chunks = ( chunks , 3 ) , compression = "gzip" )
# # allele count storage
supercatg . attrs [ "chunksize" ] = ( chunks , 1 , maxlen , 4 )
supercatg . attrs [ "samples" ] = [ i . name for i in samples ]
superseqs . attrs [ "chunksize" ] = ( chunks , len ( samples ) , maxlen )
superseqs . attrs [ "samples" ] = [ i . name for i in samples ]
superalls . attrs [ "chunksize" ] = ( chunks , len ( samples ) )
superalls . attrs [ "samples" ] = [ i . name for i in samples ]
superchroms . attrs [ "chunksize" ] = ( chunks , len ( samples ) )
superchroms . attrs [ "samples" ] = [ i . name for i in samples ]
# # array for pair splits locations , dup and ind filters
io5 . create_dataset ( "splits" , ( nloci , ) , dtype = np . uint16 )
io5 . create_dataset ( "duplicates" , ( nloci , ) , dtype = np . bool_ )
# # close the big boy
io5 . close ( )


signature:
def fill_dups_arr ( data ) :
    """fills the duplicates array from the multi _ muscle _ align tmp files"""

implementation:
# # build the duplicates array
duplefiles = glob . glob ( os . path . join ( data . tmpdir , "duples_*.tmp.npy" ) )
duplefiles . sort ( key = lambda x : int ( x . rsplit ( "_" , 1 ) [ - 1 ] [ : - 8 ] ) )
# # enter the duplicates filter into super h5 array
io5 = h5py . File ( data . clust_database , 'r+' )
dfilter = io5 [ "duplicates" ]
# # enter all duple arrays into full duplicates array
init = 0
for dupf in duplefiles :
    end = int ( dupf . rsplit ( "_" , 1 ) [ - 1 ] [ : - 8 ] )
    inarr = np . load ( dupf )
    dfilter [ init : end ] = inarr
    init += end - init
    # os . remove ( dupf )
# del inarr
# # continued progress bar
LOGGER . info ( "all duplicates: %s" , dfilter [ : ] . sum ( ) )
io5 . close ( )


signature:
def build_tmp_h5 ( data , samples ) :
    """build tmp h5 arrays that can return quick access for nloci"""

implementation:
# # get samples and names , sorted
snames = [ i . name for i in samples ]
snames . sort ( )
# # Build an array for quickly indexing consens reads from catg files .
# # save as a npy int binary file .
uhandle = os . path . join ( data . dirs . across , data . name + ".utemp.sort" )
bseeds = os . path . join ( data . dirs . across , data . name + ".tmparrs.h5" )
# # send as first async1 job
get_seeds_and_hits ( uhandle , bseeds , snames )


signature:
def get_nloci ( data ) :
    """return nloci from the tmp h5 arr"""

implementation:
bseeds = os . path . join ( data . dirs . across , data . name + ".tmparrs.h5" )
with h5py . File ( bseeds ) as io5 :
    return io5 [ "seedsarr" ] . shape [ 0 ]


signature:
def get_seeds_and_hits ( uhandle , bseeds , snames ) :
    """builds a seeds and hits ( uarr ) array of ints from the utemp . sort file .
 Saves outputs to files . . ."""

implementation:
# # Get max name length . Allow for trailing _ + up to 9 digits
# # of numbers of loci ( an astronomical number of unique loci )
maxlen_names = np . max ( map ( len , snames ) ) + 10
# # read in the utemp . sort file
updf = np . loadtxt ( uhandle , dtype = "S" . format ( maxlen_names ) )
# # Get seeds for all matches from usort
seeds = np . unique ( updf [ : , 1 ] )
seedsarr = np . column_stack ( [ np . arange ( len ( seeds ) ) , [ i . rsplit ( "_" , 1 ) [ 0 ] for i in seeds ] , [ i . rsplit ( "_" , 1 ) [ 1 ] for i in seeds ] ] )
seedsarr [ : , 1 ] = [ snames . index ( i ) for i in seedsarr [ : , 1 ] ]
seedsarr = seedsarr . astype ( np . int64 )
LOGGER . info ( "got a seedsarr %s" , seedsarr . shape )
# # Get matches from usort and create an array for fast entry
uarr = np . zeros ( ( updf . shape [ 0 ] , 3 ) , dtype = np . int64 )
idx = - 1
lastloc = None
for ldx in xrange ( updf . shape [ 0 ] ) :
    tloc = updf [ ldx , 1 ]
    if tloc != lastloc :
        idx += 1
    uarr [ ldx , 0 ] = idx
    lastloc = tloc
# # create a column with sample index
uarr [ : , 1 ] = [ int ( snames . index ( i . rsplit ( "_" , 1 ) [ 0 ] ) ) for i in updf [ : , 0 ] ]
# # create a column with only consens index for sample
uarr [ : , 2 ] = [ int ( i . rsplit ( "_" , 1 ) [ 1 ] ) for i in updf [ : , 0 ] ]
uarr = uarr . astype ( np . int64 )
LOGGER . info ( "got a uarr %s" , uarr . shape )
# # save as h5 to we can grab by sample slices
with h5py . File ( bseeds , 'w' ) as io5 :
    io5 . create_dataset ( "seedsarr" , data = seedsarr , dtype = np . int64 )
    io5 . create_dataset ( "uarr" , data = uarr , dtype = np . int64 )


signature:
def new_multicat ( data , samples , ipyclient ) :
    """Calls ' singlecat ( ) ' for all samples to build index files ."""

implementation:
# # track progress
LOGGER . info ( "in the multicat" )
start = time . time ( )
printstr = " indexing clusters     | {} | s6 |"
# # Build the large h5 array . This will write a new HDF5 file and overwrite
# # existing data .
nloci = get_nloci ( data )
build_h5_array ( data , samples , nloci )
# # parallel client ( reserve engine 0 for data entry ) , if / else here in case
# # user has only one engine .
if len ( ipyclient ) > 1 :
    filler = ipyclient . load_balanced_view ( targets = [ 0 ] )
    smallview = ipyclient . load_balanced_view ( targets = ipyclient . ids [ 1 : : 2 ] )
else :
    filler = ipyclient . load_balanced_view ( targets = [ 0 ] )
    smallview = ipyclient . load_balanced_view ( targets = [ 0 ] )
# # First submit a sleeper job as temp _ flag for cleanups
last_sample = 0
cleanups = { }
cleanups [ last_sample ] = filler . apply ( time . sleep , 0.0 )
# # fill the duplicates filter array
async = smallview . apply ( fill_dups_arr , data )
while 1 :
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( 20 , 0 , printstr . format ( elapsed ) , spacer = data . _spacer )
    time . sleep ( 0.1 )
    if async . ready ( ) :
        break
if not async . successful ( ) :
    raise IPyradWarningExit ( async . result ( ) )
# # Get all existing . tmp . h5 files . If files exist then assume that we are
# # restarting an interrupted job . We need to check for each one whether it
# # has it finished being built , and whether it has been written to the
# # large array yet .
snames = [ i . name for i in samples ]
snames . sort ( )
smpios = { i : os . path . join ( data . dirs . across , i + '.tmp.h5' ) for i in snames }
# # send ' singlecat ( ) ' jobs to engines
bseeds = os . path . join ( data . dirs . across , data . name + ".tmparrs.h5" )
jobs = { }
for sample in samples :
    sidx = snames . index ( sample . name )
    args = ( data , sample , bseeds , sidx , nloci )
    # # Only build it if it doesn ' t already exist . Singlecat removes
    # # unfinished files if interrupted , so . tmp . h5 should not exist
    # # unless the file is ready to be entered .
    if not os . path . exists ( smpios [ sample . name ] ) :
        jobs [ sample . name ] = smallview . apply ( singlecat , * args )
# # track progress of singlecat jobs and submit writing jobs for finished
# # singlecat files ( . tmp . h5 ) .
alljobs = len ( jobs )
while 1 : # # check for finished jobs
    curkeys = jobs . keys ( )
    for key in curkeys :
        async = jobs [ key ]
        if async . ready ( ) :
            if async . successful ( ) : # # submit cleanup for finished job
                args = ( data , data . samples [ key ] , snames . index ( key ) )
                with filler . temp_flags ( after = cleanups [ last_sample ] ) :
                    cleanups [ key ] = filler . apply ( write_to_fullarr , * args )
                    last_sample = key
                    del jobs [ key ]
            else :
                if not async . successful ( ) :
                    raise IPyradWarningExit ( async . result ( ) )
    # # print progress or break
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( alljobs , alljobs - len ( jobs ) , printstr . format ( elapsed ) , spacer = data . _spacer )
    time . sleep ( 0.1 )
    if not jobs :
        break
# # add the dask _ chroms func for reference data
if 'reference' in data . paramsdict [ "assembly_method" ] :
    with filler . temp_flags ( after = cleanups . values ( ) ) :
        cleanups [ 'ref' ] = filler . apply ( dask_chroms , * ( data , samples ) )
# # - - - - - print breakline between indexing and writing database - - - - -
print ( "" )
# # track progress of databaseing
start = time . time ( )
printstr = " building database     | {} | s6 |"
while 1 :
    finished = [ i for i in cleanups . values ( ) if i . ready ( ) ]
    elapsed = datetime . timedelta ( seconds = int ( time . time ( ) - start ) )
    progressbar ( len ( cleanups ) , len ( finished ) , printstr . format ( elapsed ) , spacer = data . _spacer )
    time . sleep ( 0.1 )
    # # break if one failed , or if finished
    if not all ( [ i . successful ( ) for i in finished ] ) :
        break
    if len ( cleanups ) == len ( finished ) :
        break
# # check for errors
for job in cleanups :
    if cleanups [ job ] . ready ( ) :
        if not cleanups [ job ] . successful ( ) :
            raise IPyradWarningExit ( ( job , cleanups [ job ] . result ( ) ) )