DUf#dZddlmZddlmZmZddlmZmZm Z m Z m Z ddl m Z ddlmZddlmZdd lmZdd lmZgd Z e ZeZd&dZGddZGddZeddfd'd%ZdS)(zS Experimental API for developing split-apply-combine style algorithms on coolers. ) annotations)partialreduce)AnyCallableIterableIteratorSequence)Lock) MapFunctor)Cooler)get) partition)rsplitlockfuncslist[Callable]prepareCallable[[Any], Any] | NonerCallable[[KeyType], Any]keyKeyTypereturnrc~||}|||}|D]}|||}n|}|D] }||}|SN)rrrrchunkdatafuncs L/var/www/html/software/conda/lib/python3.11/site-packages/cooler/parallel.pyapply_pipeliner"4su CHHEwu~~ % %D4t$$DD %  D4::DD Kc^eZdZdZd dZd!d Zd"d Zd#dZd$dZd%dZ d&dZ e fd'dZ d(dZ dS))MultiplexDataPipea* Create an extendable pipeline of callables to be applied independently to each of a collection of inputs and produce a collection of outputs. New tasks are appended with the ``pipe`` method. Pipeline execution can be multiplexed using any ``map`` implementation, e.g. multiprocessing Pool.map or ipyparallel view.map for distributed execution. Depending on the ``map`` implementation results may be * yielded sequentially or online: Python 3 ``map`` or ``itertools.imap``, ``Pool.imap``, ``Pool.imap_unordered`` * gathered and returned once all outputs are finished: Python 2 ``map``, ``Pool.map``, ``Pool.map_async`` The pipeline can be run using one of: * ``gather``: Results are gathered and combined after all pipelines complete. * ``reduce``: Results are sequentially folded using a binary operator. This can save on memory when using a sequential or online ``map`` implementation. Both methods are blocking, regardless of the blocking behavior of the ``map`` implementation (e.g., ``Pool.map_async``). Notes ----- Python's multiprocessing module uses Pickle for serialization, which has several limitations. Consider using a parallel map implementation that uses a more versatile serializer, such as dill or cloudpickle. See also -------- http://stackoverflow.com/a/26521507 for a discussion of the differences between multiprocessing Pool implementations. Examples -------- >>> X = np.arange(30) >>> spans = [(0, 10), (10, 20), (20, 30)] >>> dp = MultiplexDataPipe(lambda span: X[span[0]:span[1]], spans, map) >>> dp = dp.pipe(lambda x: x - 1).pipe(sum) >>> dp.gather() [35, 135, 235] >>> dp.reduce(lambda x, y: x + y, 0) 405 rrkeysIterable[KeyType]mapr cf||_t||_||_g|_d|_dS)aN Parameters ---------- get : callable Callable used to be used by workers that fetches the data corresponding to any of the provided keys keys : iterable Keys corresponding to input data map : callable Implementation of a map functor N)rlistr&r(r_prepare)selfrr&r(s r!__init__zMultiplexDataPipe.__init__}s1*JJ   r#rc||j|j|j}t |j|_|j|_|Sr) __class__rr&r(r*rr+)r,others r!__copy__zMultiplexDataPipe.__copy__s<txDH==4:&&  r#dictcd|j}|dd|S)Nr()__dict__copypop)r,ds r! __reduce__zMultiplexDataPipe.__reduce__s. M    eTr# Iterator[Any]cDt|Sriterrun)r,s r!__iter__zMultiplexDataPipe.__iter__sDHHJJr#r Callable[[Any], Any]c||_|S)ae Prepend a task that initializes the data for transformation. This optional step allows one to keep the original data chunk pristine. The callable ``func`` should return an initial object to pass along the pipeline for transformation. Subsequent callables in the pipeline will take two arguments instead of one: * chunk: original data chunk * data: transformed data passed along the pipeline Parameters ---------- func : function/callable Returns ------- A new datapipe with the initializer set. )r+)r,r s r!rzMultiplexDataPipe.prepares*  r#0Callable[[Any], Any] | Callable[[Any, Any], Any]c|}|s|rt|g|Ri|g}n$ t|}n#t$r|g}YnwxYw|xj|z c_|S)a Append new task(s) to the pipeline Parameters ---------- func : function/callable or sequence of callables If a single function is provided, additional positional and keyword arguments can be provided and will be curried into the function. Returns ------- A new datapipe with the additional task(s) appended. )r1rr* TypeErrorr)r,r argskwargsr0addons r!pipezMultiplexDataPipe.pipes(   6 T3D333F334EE T      u  s< A  A  Iterable[Any]ctt|j|j|j}|||jS)zR Run the pipeline Output depends on map implementation. )rr"rr+rr(r&)r,pipelines r!r=zMultiplexDataPipe.runs3>4:t}dhOOxx$),,,r#combineCallable[[Iterable], Sequence] Sequence[Any]cX|t|g|Ri|S)a Run the pipeline and gather outputs Parameters ---------- combine : callable, optional Callable to consume the output. Default is builtin list. Returns ------- Output of ``combine`` r;)r,rKrDrEs r!gatherzMultiplexDataPipe.gathers4&wtDHHJJ''9$999&999r#binopCallable[[Any, Any], Any]initrcbt|t||S)ad Run the pipeline and fold outputs cumulatively as they are returned Parameters ---------- binop : binary operation A function of two arguments that returns a single value. init : object The initial value of the accumulation. Returns ------- Reduced output )rr<r=)r,rPrRs r!rzMultiplexDataPipe.reduces&(eT$((**--t444r#N)rrr&r'r(r )rr%)rr2)rr9)r r?rr%)r rArr%)rrH)rKrLrrM)rPrQrRrrr)__name__ __module__ __qualname____doc__r-r1r8r>rrGr=r*rOrrr#r!r%r%Fs44l6      0>----37:::::*555555r#r%c&eZdZ ddd ZddZdS) chunkgetterFTclrrinclude_chromsbool include_binsuse_lockc>||_||_||_||_dSr)coolerr[r]r^)r,rZr[r]r^s r!r-zchunkgetter.__init__s& ,(  r#spantuple[int, int]rdict[str, Any]c|\}}i} |jrt|jd5}|jrt |dd|d<|jrt |dd|d<t |d||d|d<dddn #1swxYwY|jrtn&#|jrtwwxYw|S)NrchromsT)as_dictbinspixels) r^racquirer`openr[rr]release)r,ralohirgrps r!__call__zchunkgetter.__call__ sQB }  !!#&& K#&G&)#h-&F&F&FE(O$C$'F T$B$B$BE&M"%c(mRT"J"J"Jh  K K K K K K K K K K K K K K K}  }    s0:CAB." C.B22C5B26C#C=N)FTF)rZrr[r\r]r\r^r\)rarbrrc)rTrUrVr-rprr#r!rYrYsL %! ! ! ! ! !r#rYiNrZrr(r chunksizeintspans Iterable[tuple[int, int]] | Nonec |)tdt|jd|}tt |fi|||S)Nrnnz)rr&r()rrrinfor%rY)rZr(rqrsrEs r!rr2sX }!S%119==   & &v & &     r#) rrrrrrrrrr) rZrr(r rqrrrsrtrr%)rW __future__r functoolsrrtypingrrrr r multiprocessr _typingr apircorerutilr__all__rrr"r%rYr(rrr#r!rs} #"""""%%%%%%%%>>>>>>>>>>>>>> ( ( (6 tvv $J5J5J5J5J5J5J5J5ZB.2       r#