def _execute_load_graph_streaming(filename):
 scripts = utils.scriptpath()
 infile = utils.copy_test_data(filename)
 in_dir = os.path.dirname(infile)
 args = u'-x 1e7 -N 2 -k 20 out -'
 cmd = u'cat {infile} | {scripts}/load-graph.py {args}'.format(infile=infile, scripts=scripts, args=args)
 (status, out, err) = utils.run_shell_cmd(cmd, in_directory=in_dir)
 if (status != 0):
  print(out)
  print(err)
  assert (status == 0), status
 assert (u'Total number of unique k-mers: 3960' in err), err
 ht_file = os.path.join(in_dir, u'out')
 assert os.path.exists(ht_file), ht_file
 tagset_file = os.path.join(in_dir, u'out.tagset')
 assert os.path.exists(tagset_file), tagset_file
 ht = khmer.load_nodegraph(ht_file)
 ht.load_tagset(tagset_file)
 subset = ht.do_subset_partition(0, 0)
 x = ht.subset_count_partitions(subset)
 assert (x == (1, 0)), x

def _execute_load_graph_streaming ( filename ) : scripts = utils . scriptpath ( ) infile = utils . copy_test_data ( filename ) in_dir = os . path . dirname ( infile ) args = u'-x 1e7 -N 2 -k 20 out -' cmd = u'cat {infile} | {scripts}/load-graph.py {args}' . format ( infile = infile , scripts = scripts , args = args ) ( status , out , err ) = utils . run_shell_cmd ( cmd , in_directory = in_dir ) if ( status != 0 ) : print ( out ) print ( err ) assert ( status == 0 ) , status assert ( u'Total number of unique k-mers: 3960' in err ) , err ht_file = os . path . join ( in_dir , u'out' ) assert os . path . exists ( ht_file ) , ht_file tagset_file = os . path . join ( in_dir , u'out.tagset' ) assert os . path . exists ( tagset_file ) , tagset_file ht = khmer . load_nodegraph ( ht_file ) ht . load_tagset ( tagset_file ) subset = ht . do_subset_partition ( 0 , 0 ) x = ht . subset_count_partitions ( subset ) assert ( x == ( 1 , 0 ) ) , x
helper function for the matrix of streaming tests using screed via filter-abund-single .