tf#dZddlmZddlmZmZddlmZmZm Z m Z m Z ddl m Z ddlmZddlmZdd lmZdd lmZgd Z e ZeZ dd ZGd dZGddZeddf ddZy)zS Experimental API for developing split-apply-combine style algorithms on coolers. ) annotations)partialreduce)AnyCallableIterableIteratorSequence)Lock) MapFunctor)Cooler)get) partition)rsplitlockcp||}|||}|D] }|||} |S|}|D] }||} |SN)funcspreparerkeychunkdatafuncs Y/var/www/html/software/conda/envs/higlass/lib/python3.12/site-packages/cooler/parallel.pyapply_pipeliner4sb HEu~ %Dt$D % K D:D  KceZdZdZ d dZd dZddZddZddZ ddZ ddZ e f dd Z dd Z y )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 c\||_t||_||_g|_d|_y)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)rlistkeysmapr_prepare)selfrr#r$s r__init__zMultiplexDataPipe.__init__}s+*J   rc|j|j|j|j}t |j |_|j |_|Sr) __class__rr#r$r"rr%)r&others r__copy__zMultiplexDataPipe.__copy__s@txxDHH=4::&  rc^|jj}|jdd|S)Nr$)__dict__copypop)r&ds r __reduce__zMultiplexDataPipe.__reduce__s' MM    eTrc4t|jSriterrun)r&s r__iter__zMultiplexDataPipe.__iter__sDHHJrc||_|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&rs rrzMultiplexDataPipe.prepares*  rc|j}|s|rt|g|i|g}n t|}|xj|z c_|S#t$r|g}Y%wxYw)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. )r+rr" TypeErrorr)r&rargskwargsr*addons rpipezMultiplexDataPipe.pipesf(  6T3D3F34E T   u    s A AActt|j|j|j}|j ||j S)zR Run the pipeline Output depends on map implementation. )rrrr%rr$r#)r&pipelines rr5zMultiplexDataPipe.runs5>4::t}}dhhOxx$)),,rcH|t|jg|i|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`` r3)r&combiner:r;s rgatherzMultiplexDataPipe.gathers$&tDHHJ'9$9&99rcJt|t|j|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 )rr4r5)r&binopinits rrzMultiplexDataPipe.reduces(eT$((*-t44rN)rCallable[[KeyType], Any]r#zIterable[KeyType]r$r )returnr )rGdict)rGz Iterator[Any])rzCallable[[Any], Any]rGr )rz0Callable[[Any], Any] | Callable[[Any, Any], Any]rGr )rGz Iterable[Any])rAzCallable[[Iterable], Sequence]rGz Sequence[Any])rDzCallable[[Any, Any], Any]rErrGr)__name__ __module__ __qualname____doc__r'r+r1r6rr=r5r"rBrrrrr r Fs4l %  6   0>  >-37:/:  :*5(55 5rr c2eZdZ d ddZddZy) chunkgetterc<||_||_||_||_yr)coolerinclude_chroms include_binsuse_lock)r&clrrQrRrSs rr'zchunkgetter.__init__s" ,(  rc|\}}i} |jrtj|jj d5}|j rt |dd|d<|jrt |dd|d<t |d||d|d<ddd|jrtj|S#1swY+xYw#|jrtjwwxYw)NrchromsT)as_dictbinspixels) rSracquirerPopenrQrrRrelease)r&spanlohirgrps r__call__zchunkgetter.__call__ sB }} !!#& K#&&&)#h-&FE(O$$$'F T$BE&M"%c(mRT"Jh  K}}   K K}} s$;CACCC C#C1N)FTF)rTrrQboolrRrcrSrc)r^ztuple[int, int]rGzdict[str, Any])rIrJrKr'rbrrrrNrNs= %! !  ! ! !  !rrNiNc ||#tdt|jd|}tt |fi|||S)Nrnnz)rr#r$)rintinfor rN)rTr$ chunksizespansr;s rrr2sD }!S%19=   &v &   r) rzlist[Callable]rzCallable[[Any], Any] | NonerrFrKeyTyperGr) rTrr$r rhrfriz Iterable[tuple[int, int]] | NonerGr )rL __future__r functoolsrrtypingrrrr r multiprocessr _typingr apircorerutilr__all__rrjrr rNr$rrrrrts #%>> (6 v   ( "    $J5J5ZB.2       ,   r