o Df"~@sddlZddlZddlZddlZddlZddlZddlmZmZm Z m Z m Z m Z m Z ddlmZmZddlmZmZmZddlmZddlZddlmZddlZddlZddlmZmZdd lmZm Z m!Z!m"Z"dd l#m$Z$dd l%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6e7Z8Gd d d Z9GdddZ:GdddZ;GdddeZGdddeZ?GdddeZ@GdddZAGdddeZBGd d!d!eBZCGd"d#d#eBZDdS)$N)ListOptionalBinaryIOTextIOAnyTupleDict)ABCabstractmethod)ProcessPipeQueue)Path) Connection)Progress FileOpener)SingleEndModifierPairedEndModifierPairedEndModifierWrapperModificationInfo) Statistics) RedirectorPairedRedirectorNoFilterPairedNoFilterInfoFileWriterRestFileWriterWildcardFileWriterTooShortReadFilterTooLongReadFilterNContentFilterMaximumExpectedErrorsFilter CasavaFilterDiscardTrimmedFilterDiscardUntrimmedFilter DemultiplexerPairedDemultiplexerCombinatorialDemultiplexerc@s:eZdZd dedeedefddZdd Zdd d ZdS) InputFilesNFfile1file2 interleavedcC||_||_||_dSN)r*r+r,)selfr*r+r,r0Z/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/cutadapt/pipeline.py__init__ zInputFiles.__init__cCstj|j|j|jddS)Nr)r+r,mode)dnaioopenr*r+r,r/r0r0r1r7#szInputFiles.openreturncC&|j|jdur|jdSdSr.)r*closer+r8r0r0r1r;&  zInputFiles.closeNFr9N) __name__ __module__ __qualname__rrboolr2r7r;r0r0r0r1r)sr)c@s:eZdZd dedeedefddZded efd d Z dS) InputPathsNFpath1path2r,cCr-r.)rDrEr,)r/rDrEr,r0r0r1r2-r3zInputPaths.__init__ file_openerr9cCs$||j|jd\}}t|||jS)Nrb)Z xopen_pairrDrEr)r,)r/rFr*r+r0r0r1r72szInputPaths.openr=) r?r@rAstrrrBr2rr)r7r0r0r0r1rC,srCc#@seZdZdZ                ddeedeedeedeedeedeed eed eed eed eed eedeeeefdeeeefdeeeeefefdeeeeefefdee f ddZ ddZ dddZ dddZ dS) OutputFilesz The attributes are either None or open file-like objects except for demultiplex_out and demultiplex_out2, which are dictionaries that map an adapter name to file-like objects. Noutout2 untrimmed untrimmed2 too_short too_short2too_long too_long2inforestwildcarddemultiplex_outdemultiplex_out2combinatorial_outcombinatorial_out2 force_fastacCsd||_||_||_||_||_||_||_||_| |_| |_ | |_ | |_ | |_ ||_ ||_||_dSr.)rJrKrLrMrNrOrPrQrRrSrTrUrVrWrXrY)r/rJrKrLrMrNrOrPrQrRrSrTrUrVrWrXrYr0r0r1r2=s  zOutputFiles.__init__c cs|j|j|j|j|j|j|j|j|j|j |j f D] }|dur"|Vq|j |j |j |jfD]}|durC|D] }|dus?J|Vq7q-dSr.)rJrKrLrMrNrOrPrQrRrSrTrUrVrWrXvalues)r/fZoutsr0r0r1__iter__as2   zOutputFiles.__iter__r9cCst|jd}dD]}t||durt||tqdD]%}t||durAt||tt||D] \}}tt|||<q3q|S)zl Create a new OutputFiles instance that has BytesIO instances for each non-None output file rY) rJrKrLrMrNrOrPrQrRrSrTN)rUrVrWrX)rIrYgetattrsetattrioBytesIOdictitems)r/resultattrkvr0r0r1 as_bytesiozs zOutputFiles.as_bytesiocCs.|D]}|tjus|tjjurq|qdS)z*Close all output files that are not stdoutN)sysstdoutbufferr;r/r[r0r0r1r;s  zOutputFiles.close)NNNNNNNNNNNNNNNN)r9rIr>)r?r@rA__doc__rrrrHrrBr2r\rhr;r0r0r0r1rI7sl      $ rIc @seZdZdZdZdZdefddZdede d d fd d Z e d)d e de e de efddZde d d fddZd*ddZd*ddZd*ddZd*ddZed efddZe d+ded eeee effdd Ze d!d"Ze d#d$Ze de fd%d&Ze de fd'd(Zd S),PipelinezU Processing pipeline that loops over reads and applies modifiers and filters rFrFcCsXd|_g|_d|_d|_d|_g|_d|_d|_d|_d|_ d|_ d|_ d|_ ||_ dSr=)_reader_filters_infiles _outfiles_demultiplexer_textiowrappers_minimum_length_maximum_lengthmax_nmax_expected_errorsdiscard_casavadiscard_trimmeddiscard_untrimmedrFr/rFr0r0r1r2s zPipeline.__init__infilesoutfilesr9NcCs||_||_||dSr.)rqr7ro _set_output)r/r}r~r0r0r1 connect_ios zPipeline.connect_iofiler+rYcCdSr.r0r/rr+rYr0r0r1 _open_writerszPipeline._open_writercCsg|_||_|}t|jft|jft|jffD]\}}|r5t |}|j ||j |d||dq|j |j|jtf|j|j|jtffD]C\}}}}|durSqH|r[|||nd} |dduri||dnd} t|dkr~|ddur~||d} nd} |j || | | dqH|jdurt|j} } |j |d| | |jdur|jjstdnt|j} } |j |d| | |jrt } } |j |d| | t!|j"t!|j#t!|j$dudkrt%d|j&dus|j'dur|(||_)|j |j)nG|*} |j"r|j |dt+t+n)|j#r*|j | dt,t,n|j$rB||j$|j-} |j | | t,t,|j |.|t/d|jdS)Nrr)filterZfilter2zAIgnoring option --max-ee as input does not contain quality valueszXdiscard_trimmed, discard_untrimmed and outfiles.untrimmed must not be set simultaneouslyz Filters: %s)0rprr_filter_wrapperrrSrrRrrTr` TextIOWrapperrtappendrurNrOrrvrPrQr rlenrwr!rxrodelivers_qualitiesloggerwarningr"ryr#intrzr{rL ValueErrorrUrW_create_demultiplexerrs_untrimmed_filter_wrapperr$r%rM _final_filterdebug)r/r~Zfilter_wrapperZ filter_classoutfileZ textiowrapperlengthsr*r+writerf1f2Zuntrimmed_filter_wrapperZuntrimmed_writerr0r0r1rsn       & zPipeline._set_outputcCs:|jD]}|q|jdusJ|jD]}|qdSr.)rtflushrrrlr0r0r1rs    zPipeline.flushcCs||dSr.) _close_input _close_outputr8r0r0r1r; s zPipeline.closecCr:r.)ror;rqr8r0r0r1rr<zPipeline._close_inputcCs0|jD]}|q|jdusJ|jdSr.)rtr;rrrlr0r0r1rs  zPipeline._close_outputcCs|jdusJ|jjSr.)rorr8r0r0r1uses_qualitiesszPipeline.uses_qualitiesprogresscCrr.r0)r/rr0r0r1 process_reads zPipeline.process_readscCrr.r0r8r0r0r1r$rzPipeline._filter_wrappercCrr.r0r8r0r0r1r(rz"Pipeline._untrimmed_filter_wrappercCrr.r0r/r~r0r0r1r,rzPipeline._final_filtercCrr.r0rr0r0r1r0rzPipeline._create_demultiplexerNNr>r.)r?r@rArmZ n_adapterspairedrr2r)rIrr rrrBrrrr;rrpropertyrrrrrrrrrr0r0r0r1rnsB  H   $  rnc seZdZdZdeffdd ZdefddZd!d ed e e e e e ffd d Z  d"d e de e de efddZddZddZdefddZdefddZeddZejddZeddZejd dZZS)#SingleEndPipelinez2 Processing pipeline for single-end reads rFcst|g|_dSr.)superr2 _modifiersr| __class__r0r1r29s  zSingleEndPipeline.__init__modifiercCs |durtd|j|dS)NzModifier must not be None)rrrr/rr0r0r1add=szSingleEndPipeline.addNrr9cCsd}d}|jD]5}|d7}|ddkr|r|||t|7}t|}|jD]}|||}q'|jD] }|||r;nq2q||dfS)z#Run the pipeline. Return statisticsrr'Nroupdaterrrrp)r/rnZtotal_bpreadrRrfilter_r0r0r1rBs         zSingleEndPipeline.process_readsrr+rYcCsB|dusJt|tttfrJ|jj|d|j|rddSddS)Nwfasta)r5 qualities fileformat isinstancerHbytesrrFZdnaio_open_raise_limitrrr0r0r1rSs zSingleEndPipeline._open_writercCtSr.rr8r0r0r1r^z!SingleEndPipeline._filter_wrappercCrr.rr8r0r0r1rarz+SingleEndPipeline._untrimmed_filter_wrapperr~cCs2|jdur |jdus J|j|j|jd}t|SNr])rKrJrrYrr/r~rr0r0r1rdszSingleEndPipeline._final_filtercCsdt}|jdur|j|j|jd|d<|jdusJ|jD]\}}|j||jd||<qt|Sr)rbrLrrYrUrcr&)r/r~writersnamerr0r0r1ris z'SingleEndPipeline._create_demultiplexercC|jSr.rur8r0r0r1minimum_lengthrz SingleEndPipeline.minimum_lengthcC"|dus t|dks J||_dSNrrrur/valuer0r0r1rv cCrr.rvr8r0r0r1maximum_length{rz SingleEndPipeline.maximum_lengthcCrrrrvrr0r0r1rrr.r)r?r@rArmrr2rrrrrrrrrBrrrrIrrrrsetterr __classcell__r0r0rr1r5s4"     rc seZdZdZdZdeffdd Zdeedeedd fd d Z d edd fd dZ d e dd fddZ d'de deeeeeffddZ d(dedeedeefddZd'ddZddZddZdd Zed!d"Zejd#d"Zed$d%Zejd&d%ZZS))PairedEndPipelinez3 Processing pipeline for paired-end reads. TrFcs(t|g|_||_d|_d|_dSr=)rr2r_pair_filter_moderooverride_untrimmed_pair_filter)r/pair_filter_moderFrr0r1r2s  zPairedEndPipeline.__init__ modifier1 modifier2r9NcCs.|dur |dur td|jt||dS)z Add a modifier for R1 and R2. One of them can be None, in which case the modifier will only be added for the respective read. NzNot both modifiers can be None)rrrr)r/rrr0r0r1rszPairedEndPipeline.addrcCs(|dusJ|jt|t|dS)z5 Add one modifier for both R1 and R2 N)rrrcopyrr0r0r1add_boths zPairedEndPipeline.add_bothcCs|j|dS)zBAdd a Modifier (without wrapping it in a PairedEndModifierWrapper)N)rrrr0r0r1add_paired_modifiersz%PairedEndPipeline.add_paired_modifierrc Csd}d}d}|jdus J|jD]G\}}|d7}|ddkr%|r%|||t|7}|t|7}t|}t|}|jD] } | ||||\}}q<|jD] } | ||||rVnqKq|||fS)Nrrrr) r/rr total1_bp total2_bpread1Zread2Zinfo1Zinfo2rrr0r0r1rs(      zPairedEndPipeline.process_readsrr+rYcCsH||fD] }t|tttfrJq|jj||d|j|rdnd|dudS)Nrr)r+r5rrr,r)r/rr+rYr[r0r0r1rs  zPairedEndPipeline._open_writercCs|dur|j}tjt|dS)Nr)r functoolspartialr)r/rr0r0r1rsz!PairedEndPipeline._filter_wrappercCs|jr |jddS|S)z Return a different filter wrapper when adapters were given only for R1 or only for R2 (then override_untrimmed_pair_filter will be set) Zbothr)rrr8r0r0r1rs z+PairedEndPipeline._untrimmed_filter_wrappercCs|j|j|j|jd}t|Sr)rrJrKrYrrr0r0r1rszPairedEndPipeline._final_filtercsfdd}jdur3jdurjdusJt}jD]\}}||j|||<q t|St}jdurD|jj|d<jD]\}}||j|||<qIt |S)Ncsj||jdSr)rrY)rr+r~r/r0r1 open_writersz.open_writer) rWrLrMrbrcrXr(rUrVr')r/r~rrkeyrJrrr0rr1rs  z'PairedEndPipeline._create_demultiplexercCrr.rr8r0r0r1rrz PairedEndPipeline.minimum_lengthcCrNrrrr0r0r1rrcCrr.rr8r0r0r1rrz PairedEndPipeline.maximum_lengthcCrrrrr0r0r1rrr.r)r?r@rArmrrr2rrrrrrrrrrrrBrrrrrrrrrrr0r0rr1rs: "     rcsDeZdZdZdedeedeffdd ZddZd d d Z Z S) ReaderProcessa> Read chunks of FASTA or FASTQ data (single-end or paired) and send to a worker. The reader repeatedly - reads a chunk from the file(s) - reads a worker index from the Queue - sends the chunk to connections[index] and finally sends the stop token -1 ("poison pills") to all connections. pathrEopenercs8t||_||_||_||_||_||_||_dS)z queue -- a Queue of worker indices. A worker writes its own index into this queue to notify the reader that it is ready to receive more data. connections -- a list of Connection objects, one for each worker. N) rr2rrE connectionsqueue buffer_sizestdin_fd_opener)r/rrErrrrrrr0r1r2s  zReaderProcess.__init__c Csf|jdkrtjt|jt_zz|j|jdP}|j rO|j|j d }t t |||j D] \}\}}||||q1Wdn1sIwYnt t ||j D] \}}|||qXWdn1smwYtt|jD]}|j}|j|dqyWdSty} z|jD]} | d| | tfqWYd} ~ dSd} ~ ww)NrG)rristdinr;osfdopenrZxopenrrE enumerater6Zread_paired_chunksrsend_to_worker read_chunksrangerrrgetsend Exception traceback format_exc) r/r[r chunk_indexchunk1chunk2chunk_ worker_indexe connectionr0r0r1run#s:     zReaderProcess.runNcCsB|j}|j|}|||||dur||dSdSr.)rrrr send_bytes)r/rrrrrr0r0r1r<s    zReaderProcess.send_to_workerr.) r?r@rArmrHrrr2rrrr0r0rr1rs  rcsneZdZdZdededededededed e ffd d Z d d Z de fddZ dededefddZZS) WorkerProcessa7 The worker repeatedly reads chunks of data from the read_pipe, runs the pipeline on it and sends the processed chunks to the write_pipe. To notify the reader process that it wants data, it puts its own identifier into the need_work_queue before attempting to read data from the read_pipe. id_pipelinetwo_input_filesinterleaved_input orig_outfiles read_pipe write_pipeneed_work_queuec sBt||_||_||_||_||_||_||_| |_ dSr.) rr2_id _pipeline_two_input_files_interleaved_input _read_pipe _write_pipe_need_work_queuerh_original_outfiles) r/rrrrrrr r rr0r1r2Ms zWorkerProcess.__init__c CsPzt} |j|j|j}|dkrnH|dkr)|j\}}td|||}|j }|j |||j \}}} |j t||| g|j j} || 7}||||q|j j} tdd|j jrmdnd| g} || 7}|jd|j|WdSty}z|jd|j|tfWYd}~dSd}~ww)NTrr%sr)rrputr rrecvrerror_make_input_filesrrhr rrrcollectrp_send_outfilesrrrrrrr) r/statsrrtb_strr}r~rZbp1Zbp2 cur_statsmZmodifier_statsr0r0r1rds<       "zWorkerProcess.runr9cCsD|j}t|}|jr|j}t|}nd}t|||jdS)N)r,)r recv_bytesr`rar r)r)r/datainputZinput2r0r0r1rs    zWorkerProcess._make_input_filesr~rn_readscCsR|j||j||D]}|t|tjsJ|}|j|qdSr.)rrrrr`ragetvaluer)r/r~rr!r[Zprocessed_chunkr0r0r1rs  zWorkerProcess._send_outfiles)r?r@rArmrrnrBrIrr r2rr)rrrr0r0rr1rEs, " rc@s(eZdZdZddZddZddZdS) OrderedChunkWriterz We may receive chunks of processed data from worker processes in any order. This class writes them to an output file in the correct order. cCst|_d|_||_dS)Nr)rb_chunks_current_index_outfile)r/rr0r0r1r2s zOrderedChunkWriter.__init__cCsV||j|<|j|jvr)|j|j|j|j|j=|jd7_|j|jvs dSdS)z rN)r$r%r&write)r/rindexr0r0r1r's   zOrderedChunkWriter.writecCs|j Sr.)r$r8r0r0r1wrote_everythingsz#OrderedChunkWriter.wrote_everythingN)r?r@rArmr2r'r)r0r0r0r1r#s  r#c@sJeZdZdZdedefddZeddZedd Z d d Z d d Z dS)PipelineRunnerz$ A read processing pipeline rrcCs||_||_dSr.)r  _progress)r/rrr0r0r1r2s zPipelineRunner.__init__cCrr.r0r8r0r0r1rrzPipelineRunner.runcCrr.r0r8r0r0r1r;rzPipelineRunner.closecCs|Sr.r0r8r0r0r1 __enter__rzPipelineRunner.__enter__cGs |dSr.)r;)r/argsr0r0r1__exit__s zPipelineRunner.__exit__N) r?r@rArmrnrr2r rr;r,r.r0r0r0r1r*s   r*cseZdZdZ ddededededede d e ffd d Z dde de e de dd fddZdeeeeeffddZdefddZdddZZS)ParallelPipelineRunnera Run a Pipeline in parallel - When connect_io() is called, a reader process is spawned. - When run() is called, as many worker processes as requested are spawned. - In the main process, results are written to the output files in the correct order, and statistics are aggregated. If a worker needs work, it puts its own index into a Queue() (_need_work_queue). The reader process listens on this queue and sends the raw data to the worker that has requested work. For sending the data from reader to worker, a Connection() is used. There is one such connection for each worker (self._pipes). For sending the processed data from the worker to the main process, there is a second set of connections, again one for each worker. When the reader is finished, it sends 'poison pills' to all workers. When a worker receives this, it sends a poison pill to the main process, followed by a Statistics object that contains statistics about all the reads processed by that worker. @rr}r~rr n_workersrcsFt||||_t|_||_||_||_||j |j |j dSr.) rr2 _n_workersr r _buffer_sizerrr _assign_inputrDrEr,)r/rr}r~rrr1rrr0r1r2s zParallelPipelineRunner.__init__NFrDrEr,r9cCs|du|_||_ddt|jD}t|\|_}ztj}Wn t j y,d}Ynwt |||j ||j |j||_d|j_|jdS)NcSsg|]}tddqS)FZduplex)r ).0rr0r0r1 sz8ParallelPipelineRunner._assign_input..rT)r rrr2zip _connectionsrirfilenor`UnsupportedOperationrrrr3_reader_processdaemonstart)r/rDrEr,rZconnwr:r0r0r1r4s   z$ParallelPipelineRunner._assign_inputc Csvg}g}t|jD]-}tdd\}}||t||j|j|j|j|j |||j }d|_ | ||q ||fS)NFr5T) rr2r rrr r rrrr9rr=r>)r/workersrr(Zconn_rZconn_wZworkerr0r0r1_start_workers s  z%ParallelPipelineRunner._start_workerscCsf|\}}g}|jD] }|t|q t}d}|rtj|}|D]k}|} | dkrN|} |dkrD|\} } t d| | || 7}| |q$| dkr`|\} } t d| | |} | dkrv|\} } t d| | || 7}|j ||D] }|}||| qq$|s|D]}|sJq|D]}|q|j|j ||S)Nrrrr)r@rrrr#rmultiprocessingrwaitrrrremover+rrr'r)joinr<stop)r/r?rrr[rrZready_connectionsrrrrrZchunk_nrrrr0r0r1rsT           %   zParallelPipelineRunner.runcC|jdSr.)rrr;r8r0r0r1r;NzParallelPipelineRunner.close)r0r=r>)r?r@rArmrnrCrIrrrr2rHrrBr4rrrrr@rrr;rr0r0rr1r/s@ 4r/csJeZdZdZdedededeffdd Zde fd d Z dd d Z Z S)SerialPipelineRunnerz) Run a Pipeline on a single core rr}r~rcs t|||j||dSr.)rr2r r)r/rr}r~rrr0r1r2WszSerialPipelineRunner.__init__r9cCsZ|jj|jd\}}}|jr|j|t|jdd}|dus!Jt|||||jjS)N)rr)r rr+rEr^rrrp)r/rrr modifiersr0r0r1ras   zSerialPipelineRunner.runNcCrFr.)r r;r8r0r0r1r;jrGzSerialPipelineRunner.closer>) r?r@rArmrnr)rIrr2rrr;rr0r0rr1rHRs  rH)Er`rrirloggingrtypingrrrrrrrabcr r rAr r r pathlibrZmultiprocessing.connectionrrr6utilsrrrIrrrrreportrfiltersrrrrrrrrr r!r"r#r$r%r&r'r( getLoggerrr)rCrIrnrrrrr#r*r/rHr0r0r0r1sD$   L ]"P>W