jpayne@68: package clump;
jpayne@68: 
jpayne@68: import java.io.File;
jpayne@68: import java.io.IOException;
jpayne@68: import java.util.ArrayList;
jpayne@68: 
jpayne@68: import fileIO.FileFormat;
jpayne@68: import fileIO.ReadWrite;
jpayne@68: import sort.SortByName;
jpayne@68: import stream.ConcurrentReadInputStream;
jpayne@68: import stream.ConcurrentReadOutputStream;
jpayne@68: import stream.Read;
jpayne@68: import structures.ListNum;
jpayne@68: 
jpayne@68: public class StreamToOutput {
jpayne@68: 	
jpayne@68: 	public StreamToOutput(FileFormat ffin1, FileFormat ffin2, ConcurrentReadOutputStream[] rosa_, KmerComparator old, boolean sortByName_, boolean incrementComparator){
jpayne@68: 		final ConcurrentReadInputStream cris=ConcurrentReadInputStream.getReadInputStream(-1, false, ffin1, ffin2, null, null);
jpayne@68: 		cris.start();
jpayne@68: 		rosa=rosa_;
jpayne@68: 		kc=(incrementComparator ? new KmerComparator(old.k, old.seed+1, old.border-1, old.hashes, false, false) : old);
jpayne@68: 		sortByName=sortByName_;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	public StreamToOutput(ConcurrentReadInputStream cris_, ConcurrentReadOutputStream[] rosa_, KmerComparator old, boolean sortByName_, boolean incrementComparator){
jpayne@68: 		cris=cris_;
jpayne@68: 		rosa=rosa_;
jpayne@68: 		kc=(incrementComparator ? new KmerComparator(old.k, old.seed+1, old.border-1, old.hashes, false, false) : old);
jpayne@68: 		sortByName=sortByName_;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	public boolean process(){
jpayne@68: 		if(rosa==null || rosa.length==0){return errorState;}
jpayne@68: 		
jpayne@68: 		File temp=null;
jpayne@68: 		if(sortByName){
jpayne@68: 			try {
jpayne@68: 				temp=File.createTempFile("temp_namesort_", ".fq.gz");
jpayne@68: 			} catch (IOException e) {
jpayne@68: 				// TODO Auto-generated catch block
jpayne@68: 				e.printStackTrace();
jpayne@68: 			}
jpayne@68: 			SortByName sbn=new SortByName(new String[] {"out="+temp.getAbsolutePath()});
jpayne@68: 			sbn.processInner(cris);
jpayne@68: 			FileFormat ff=FileFormat.testInput(temp.getAbsolutePath(), null, false);
jpayne@68: 			cris=ConcurrentReadInputStream.getReadInputStream(-1, false, ff, null, null, null);
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		if(rosa.length==1){
jpayne@68: 			processSingle(cris);
jpayne@68: 		}else{
jpayne@68: 			processMulti(cris);
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		errorState|=ReadWrite.closeStream(cris);
jpayne@68: 		if(temp!=null){
jpayne@68: 			temp.delete();
jpayne@68: 		}
jpayne@68: 		return errorState;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	public void processSingle(ConcurrentReadInputStream cris){
jpayne@68: 		
jpayne@68: 		ListNum<Read> ln=cris.nextList();
jpayne@68: 		ArrayList<Read> reads=(ln!=null ? ln.list : null);
jpayne@68: 		
jpayne@68: 		while(ln!=null && reads!=null && reads.size()>0){//ln!=null prevents a compiler potential null access warning
jpayne@68: 			if(rosa!=null){rosa[0].add(reads, ln.id);}
jpayne@68: 			
jpayne@68: 			for(Read r : reads){
jpayne@68: 				readsIn+=r.pairCount();
jpayne@68: 				basesIn+=r.pairLength();
jpayne@68: 			}
jpayne@68: 			
jpayne@68: 			cris.returnList(ln);
jpayne@68: 			
jpayne@68: 			ln=cris.nextList();
jpayne@68: 			reads=(ln!=null ? ln.list : null);
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		if(ln!=null){
jpayne@68: 			cris.returnList(ln.id, ln.list==null || ln.list.isEmpty());
jpayne@68: 		}
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	public void processMulti(ConcurrentReadInputStream cris){
jpayne@68: 		final int groups=rosa.length;
jpayne@68: 		
jpayne@68: 		@SuppressWarnings("unchecked")
jpayne@68: 		final ArrayList<Read>[] out=new ArrayList[groups];
jpayne@68: 		for(int i=0; i<out.length; i++){
jpayne@68: 			out[i]=new ArrayList<Read>();
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		ListNum<Read> ln=cris.nextList();
jpayne@68: 		ArrayList<Read> reads=(ln!=null ? ln.list : null);
jpayne@68: 		
jpayne@68: 		while(ln!=null && reads!=null && reads.size()>0){//ln!=null prevents a compiler potential null access warning
jpayne@68: 			for(Read r : reads){
jpayne@68: 				long kmer=kc.hash(r, null, 0, false);
jpayne@68: 				int group=(int)(kmer%groups);
jpayne@68: 				out[group].add(r);
jpayne@68: 				
jpayne@68: 				readsIn+=r.pairCount();
jpayne@68: 				basesIn+=r.pairLength();
jpayne@68: 			}
jpayne@68: 			for(int group=0; group<groups; group++){
jpayne@68: 				rosa[group].add(out[group], ln.id);
jpayne@68: 				out[group]=new ArrayList<Read>();
jpayne@68: 			}
jpayne@68: 			
jpayne@68: 			cris.returnList(ln);
jpayne@68: 			
jpayne@68: 			ln=cris.nextList();
jpayne@68: 			reads=(ln!=null ? ln.list : null);
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		if(ln!=null){
jpayne@68: 			cris.returnList(ln.id, ln.list==null || ln.list.isEmpty());
jpayne@68: 		}
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	long readsIn=0;
jpayne@68: 	long basesIn=0;
jpayne@68: 	
jpayne@68: //	final FileFormat ffin1;
jpayne@68: //	final FileFormat ffin2;
jpayne@68: 	ConcurrentReadInputStream cris;
jpayne@68: 	ConcurrentReadOutputStream[] rosa;
jpayne@68: 	final KmerComparator kc;
jpayne@68: 	final boolean sortByName;
jpayne@68: 	
jpayne@68: 	boolean errorState=false;
jpayne@68: 	
jpayne@68: }