o Nrf#@sdZddlmZddlmZmZddlmZmZm Z m Z m Z ddl m Z ddlmZddlmZdd lmZdd lmZgd Z e ZeZd(ddZGdddZGdddZeddfd)d&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]keyKeyTypereturnrcCsJ||}|dur||}|D]}|||}q|S|}|D]}||}q|SN)rrrrchunkdatafuncrrX/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/cooler/parallel.pyapply_pipeline4s  r c@sneZdZdZd)dd Zd*d d Zd+ddZd,ddZd-ddZd.ddZ d/ddZ e fd0d d!Z d1d&d'Z d(S)2MultiplexDataPipea* 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 cCs&||_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$rrr__init__}s   zMultiplexDataPipe.__init__rcCs,||j|j|j}t|j|_|j|_|Sr) __class__rr"r$r%rr&)r'otherrrr__copy__s zMultiplexDataPipe.__copy__dictcCs|j}|dd|S)Nr$)__dict__copypop)r'drrr __reduce__s  zMultiplexDataPipe.__reduce__ Iterator[Any]cCs t|Sriterrun)r'rrr__iter__s zMultiplexDataPipe.__iter__rCallable[[Any], Any]cCs ||_|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'rrrrrszMultiplexDataPipe.prepare0Callable[[Any], Any] | Callable[[Any, Any], Any]cOsb|}|s|rt|g|Ri|g}nzt|}Wn ty'|g}Ynw|j|7_|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. )r+rr% TypeErrorr)r'rargskwargsr*Zaddonrrrpipes   zMultiplexDataPipe.pipe Iterable[Any]cCs"tt|j|j|j}|||jS)zR Run the pipeline Output depends on map implementation. )rr rr&rr$r")r'Zpipelinerrrr5szMultiplexDataPipe.runcombineCallable[[Iterable], Sequence] Sequence[Any]cOs|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`` r3)r'r>r:r;rrrgatherszMultiplexDataPipe.gatherbinopCallable[[Any, Any], Any]initrcCst|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 )rr4r5)r'rBrDrrrrszMultiplexDataPipe.reduceN)rrr"r#r$r )rr!)rr,)rr2)rr7rr!)rr8rr!)rr=)r>r?rr@)rBrCrDrrr)__name__ __module__ __qualname____doc__r(r+r1r6rr<r5r%rArrrrrr!Fs 6       r!c@s(eZdZ   ddd d ZdddZdS) chunkgetterFTclrr include_chromsbool include_binsuse_lockcCs||_||_||_||_dSr)coolerrKrMrN)r'rJrKrMrNrrrr(s zchunkgetter.__init__spantuple[int, int]rdict[str, Any]cCs|\}}i}zM|jrt|jd.}|jr"t|ddd|d<|jr/t|ddd|d<t|d||dd|d<Wdn1sEwYW|jrRt|S|jr\tww)NrZchromsT)as_dictZbinsZpixels) rNracquirerOopenrKrrMrelease)r'rPlohirgrprrr__call__ s& zchunkgetter.__call__N)FTF)rJr rKrLrMrLrNrL)rPrQrrR)rErFrGr(r[rrrrrIs   rIiNrJr r$r chunksizeintspans Iterable[tuple[int, int]] | NonecKs8|durtdt|jd|}tt|fi|||dS)NrZnnz)rr"r$)rr]infor!rI)rJr$r\r^r;rrrr2sr) rrrrrrrrrr) rJr r$r r\r]r^r_rr!)rH __future__r functoolsrrtypingrrrrr Z multiprocessr Z_typingr apir corerutilr__all__rrr r!rIr$rrrrrs*       N!