CdL<dZddlmZddlZddlmZmZmZddlm Z m Z m Z ddl m Z mZmZmZmZmZddlmZddlmZdd lmZmZdd lmZmZmZd Ze d Zdd d5dZ d6dZ!d7dZ" d8d9dZ#ddd dd!d:d&Z$d;d/Z%ddd d0dzTools to modify already existing dask graphs. Unlike in :mod:`dask.optimization`, the output collections produced by this module are typically not functionally equivalent to their inputs. ) annotationsN)CallableHashableSet)AnyLiteralTypeVar) clone_keyget_collection_namesget_name_from_keyreplace_name_in_keytokenizeunpack_collections) blockwise)flatten)Delayeddelayed)HighLevelGraphLayerMaterializedLayer)bind checkpointclonewait_onT split_everyrfloat | Literal[False] | Nonereturnrcdn(dur$tdkrtdt|\}}t|dkrt |dSt t jfd|DS) a5Build a :doc:`delayed` which waits until all chunks of the input collection(s) have been computed before returning None. Parameters ---------- collections Zero or more Dask collections or nested data structures containing zero or more collections split_every: int >= 2 or False, optional Determines the depth of the recursive aggregation. If greater than the number of input keys, the aggregation will be performed in multiple steps; the depth of the aggregation graph will be :math:`log_{split_every}(input keys)`. Setting to a low value can reduce cache size and network transfers, at the cost of more CPU and a larger dask graph. Set to False to disable. Defaults to 8. Returns ------- :doc:`delayed` yielding None NFz(split_every must be False, None, or >= 2rc38K|]}t|VdSN)_checkpoint_one).0crs 7lib/python3.11/site-packages/dask/graph_manipulation.py zcheckpoint..Bs- C C!oa-- C C C C C C)int ValueErrorrlenr&rchunksr)r collections_s` r)rrs2 E ! !+&& ??GHH H'5NK ;1{1~{;;;)wv()) C C C C{ C C C  r+cDt|}d|z}t|} t|t|nZ#t$rM|t j|fi}tj|||f}t||cYSwxYwg}t}g} t|D]} dt| |z} | | tt j| | |} | t| z } |tj| | |fi} |r`t#| |krM|t#| f}t j| d|f| |<| |d|gz} |rt#| |kMt j| f| |<|t|| i||itj|}t||S)Nz checkpoint- dependencieszcheckpoint_map-)rr __dask_keys__next StopIterationr/rrfrom_collectionsrsetr add_build_map_layerlistget_output_keysappendr.merge) collectionrtokname keys_iterlayerdskdsks map_namesmap_keys prev_namemap_name map_layer reduce_layerks r)r&r&FsJ :  C 3 D 002233I" Y Y """):+C+C+E+EFG-dE VVVtS!!!!! " DIH)*55   $x 3'?'?? h$V%6 8ZXX D2244555  +):-       L 0#h--+55 3|$$ %!,h| |.DE QKLL)QC/ 0#h--+55!+X6LKKl34BSTTTUUU   %C 4  sAAB-,B-boolc ddlm}t||rdSn#t$rYnwxYw ddlm}t||rdSn#t$rYnwxYw ddlm}m}t|||fS#t$rYdSwxYw)zReturn True if _map_blocks can be sped up via blockwise operations; False otherwise. FIXME this returns False for collections that wrap around around da.Array, such as pint.Quantity, xarray DataArray, Dataset, and Variable. r)BagT)Array) DataFrameSeriesF) dask.bagrP isinstance ImportError dask.arrayrQdask.dataframerRrS)r@rPrQrRrSs r)_can_apply_blockwiserYts       j# & & 4        $$$$$$ j% ( ( 4       44444444*y&&9::: uus/ ((A AAA00 A>=A>funcrrIstrnew_namer4tuple[Delayed, ...]rct|ro |j}n#t$r |jf}YnwxYwt dt |D}|rdd|Dini}t ||f|i|d|St d|Dtfdt| DS)aApply func to all keys of collection. Create a Blockwise layer whenever possible; fall back to MaterializedLayer otherwise. Parameters ---------- func Callable to be invoked on the graph node prev_name : str name of the layer to map from; in case of dask base collections, this is the collection name. Note how third-party collections, e.g. xarray.Dataset, can have multiple names. new_name : str name of the layer to map to collection Arbitrary dask collection dependencies Zero or more Delayed objects, which will be passed as arbitrary variadic args to func after the collection's chunk c3 K|] \}}|V dSr%rZ)r'ir1s r)r*z#_build_map_layer..s&;;da;;;;;;r+_depscg|] }|j SrZkeyr'ds r) z$_build_map_layer..s888aAE888r+) numblocksr4c3$K|] }|jV dSr%rdrfs r)r*z#_build_map_layer..s$551555555r+cfi|]-}t|kt|i|fz.SrZ)r r )r'rMdep_keysr[r]rIs r) z$_build_map_layer..sP   $Q''944$A 8'<==ay8?S444r+) rYriAttributeError npartitionstuple enumeraterrrr5) r[rIr]r@r4riindiceskwargsrls ``` @r)r;r;sS4J''  2",II 2 2 2#/1III 2;;i &:&:;;;;;=IQ'88<88899r       !),%      55 55555         !9!9!;!;<<      s 22T)omitseed assume_layersrchildrenruHashable | Nonervc^ tjj|t||nd t |\}}|rd|D t nt d|D t |\}}| fd|DdS)a Make ``children`` collection(s), optionally omitting sub-collections, dependent on ``parents`` collection(s). Two examples follow. The first example creates an array ``b2`` whose computation first computes an array ``a`` completely and then computes ``b`` completely, recomputing ``a`` in the process: >>> import dask >>> import dask.array as da >>> a = da.ones(4, chunks=2) >>> b = a + 1 >>> b2 = bind(b, a) >>> len(b2.dask) 9 >>> b2.compute() array([2., 2., 2., 2.]) The second example creates arrays ``b3`` and ``c3``, whose computation first computes an array ``a`` and then computes the additions, this time not recomputing ``a`` in the process: >>> c = a + 2 >>> b3, c3 = bind((b, c), a, omit=a) >>> len(b3.dask), len(c3.dask) (7, 7) >>> dask.compute(b3, c3) (array([2., 2., 2., 2.]), array([3., 3., 3., 3.])) Parameters ---------- children Dask collection or nested structure of Dask collections parents Dask collection or nested structure of Dask collections omit Dask collection or nested structure of Dask collections seed Hashable used to seed the key regeneration. Omit to default to a random number that will produce different keys at every call. assume_layers True Use a fast algorithm that works at layer level, which assumes that all collections in ``children`` and ``omit`` #. use :class:`~dask.highlevelgraph.HighLevelGraph`, #. define the ``__dask_layers__()`` method, and #. never had their graphs squashed and rebuilt between the creation of the ``omit`` collections and the ``children`` collections; in other words if the keys of the ``omit`` collections can be found among the keys of the ``children`` collections, then the same must also hold true for the layers. False Use a slower algorithm that works at keys level, which makes none of the above assumptions. split_every See :func:`checkpoint` Returns ------- Same as ``children`` Dask collection or structure of dask collection equivalent to ``children``, which compute to the same values. All nodes of ``children`` will be regenerated, up to and excluding the nodes of ``omit``. Nodes immediately above ``omit``, or the leaf nodes if the collections in ``omit`` are not found, are prevented from computing until all collections in ``parents`` have been fully computed. The keys of the regenerated nodes will be different from the original ones, so that they can be used within the same graph. Nrc@h|]}|D]}|SrZ)__dask_layers__)r'collrDs r) zbind..$s1RRR4;O;O;Q;QRR%uRRRRr+c@h|]}|D]}|SrZ)__dask_graph__)r'r|res r)r}zbind..)s1KKKTT5H5H5J5JKKcSKKKKr+c 6g|]}t|SrZ) _bind_one)r'childblocker omit_keys omit_layersrus r)rhzbind..-s9    eWk9d C C   r+r)uuiduuid4bytesrrr9) rwparentsrtrurvrr1unpacked_childrenrepackrrrs ` @@@r)rrs\ |z||!9@8K 7 4444QU !&&GD!LRRRRR EE ee KKTKKK 28 < <v 6       *         r+rrDelayed | Nonerset[str]r set[Hashable]rct|}|s|S|}i}i}t|trG t |} nu#t $r|} YnUwxYwt|dkrtt|} n t|} tj | |}| h} | |z } |D]8} |j| } n#t$rYwxYw| | z} 9|g|j}|}t|tsJ||j||jnd}t }| r| }t+|}||vr,|j|} |j|}||z }||z}| |z} ||z}| | |\||<}fd|D|z}|r|||||<| |rB|} | |vr|j| }||z}||| <|j| || <|B|\}}|t||g|Rdfd|DiS)Nr#ru)keysrubind_toc2h|]}t|S)rr )r'deprus r)r}z_bind_one..xs3   *-Ic % % %   r+renamec2i|]}|t|SrZr)r'rIrus r)rmz_bind_one..s%WWW) 9Y55WWWr+)r rrUrr9r{rncopyr.r6iterrr8get_all_external_keyslayersKeyErrorr=reupdater4popr rr:__dask_postpersist__)rrrrruprev_coll_namesrE new_layersnew_depslayers_to_clonehlg_name clone_keys layer_namerD blocker_key blocker_dsklayers_to_copy_verbatimprev_layer_namenew_layer_name layer_depslayer_deps_to_clonelayer_deps_to_omitis_boundnew_deprebuildargss ` r)rr4s+511O     C#%J$&H#~&& % 5!%"7"7"9"9::OO 5 5 5-2244OOO 5   1 $ $D1122HH1H-h<<#***,,y8J!..  Jz*EE    H e++--- k ,,.. +~66666+,--- 01111 !ee +)--//"?>>> Z ' '  ?+%o6 (;6'+5..#55/4{{$ 0;0 0 , >"H    1D     % KK $ $ $#* + +6 "8,0022  # # %j1 :-)!$J!7 : "8..00MGT 7z8,,     XWWWWWW  s$!A%%BB6 D DD)rtrurvcdt|d|||}t|dkr|dn|S)a Clone dask collections, returning equivalent collections that are generated from independent calculations. Examples -------- (tokens have been simplified for the sake of brevity) >>> import dask.array as da >>> x_i = da.asarray([1, 1, 1, 1], chunks=2) >>> y_i = x_i + 1 >>> z_i = y_i + 2 >>> dict(z_i.dask) # doctest: +SKIP {('array-1', 0): array([1, 1]), ('array-1', 1): array([1, 1]), ('add-2', 0): (, ('array-1', 0), 1), ('add-2', 1): (, ('array-1', 1), 1), ('add-3', 0): (, ('add-2', 0), 1), ('add-3', 1): (, ('add-2', 1), 1)} >>> w_i = clone(z_i, omit=x_i) >>> w_i.compute() array([4, 4, 4, 4]) >>> dict(w_i.dask) # doctest: +SKIP {('array-1', 0): array([1, 1]), ('array-1', 1): array([1, 1]), ('add-4', 0): (, ('array-1', 0), 1), ('add-4', 1): (, ('array-1', 1), 1), ('add-5', 0): (, ('add-4', 0), 1), ('add-5', 1): (, ('add-4', 1), 1)} The typical usage pattern for clone() is the following: >>> x = cheap_computation_with_large_output() # doctest: +SKIP >>> y = expensive_and_long_computation(x) # doctest: +SKIP >>> z = wrap_up(clone(x), y) # doctest: +SKIP In the above code, the chunks of x will be forgotten as soon as they are consumed by the chunks of y, and then they'll be regenerated from scratch at the very end of the computation. Without clone(), x would only be computed once and then kept in memory throughout the whole computation of y, needlessly consuming memory. Parameters ---------- collections Zero or more Dask collections or nested structures of Dask collections omit Dask collection or nested structure of Dask collections which will not be cloned seed See :func:`bind` assume_layers See :func:`bind` Returns ------- Same as ``collections`` Dask collections of the same type as the inputs, which compute to the same value, or nested structures equivalent to the inputs, where the original collections have been replaced. The keys of the regenerated nodes in the new collections will be different from the original ones, so that they can be used within the same graph. N)rrtrurvr#r)rr.)rtrurvr0outs r)rrsFz T4}   C%%**3q663r+ct|d|ifdt|\}}|fd|D}t|dkr|dn|S)a7Ensure that all chunks of all input collections have been computed before computing the dependents of any of the chunks. The following example creates a dask array ``u`` that, when used in a computation, will only proceed when all chunks of the array ``x`` have been computed, but otherwise matches ``x``: >>> import dask.array as da >>> x = da.ones(10, chunks=5) >>> u = wait_on(x) The following example will create two arrays ``u`` and ``v`` that, when used in a computation, will only proceed when all chunks of the arrays ``x`` and ``y`` have been computed but otherwise match ``x`` and ``y``: >>> x = da.ones(10, chunks=5) >>> y = da.zeros(10, chunks=5) >>> u, v = wait_on(x, y) Parameters ---------- collections Zero or more Dask collections or nested structures of Dask collections split_every See :func:`checkpoint` Returns ------- Same as ``collections`` Dask collection of the same type as the input, which computes to the same value, or a nested structure equivalent to the input where the original collections have been replaced. The keys of the regenerated nodes of the new collections will be different from the original ones, so that they can be used within the same graph. rc xt| }g}i}t|D]e}dt||z}|||<ttj||| f}|t j||| fft j|}| \}} ||g| Rd|iS)Nzwait_on-r3r) rr r;r/rr>rr8r?r) r|rArFrrIr]rDrErrrs r) block_onezwait_on..block_onestW%%-d33  I!HY$<$<.s!777d))D//777r+r#r)rrr.)rr0unpackedrrrrs @@r)rrsN+?;??G22222&*;7Hf &7777h777 8 8C%%**3q663r+cBeZdZdZed dZed dZdS) r/z*Callables to be inserted in the Dask graphnoderrc|S)zDummy graph node of :func:`bind` and :func:`wait_on`. Wait for both node and all variadic args to complete; then return node. rZ)rrrss r)rz chunks.binds  r+NonecdS)zrDummy graph node of :func:`checkpoint`. Wait for all variadic args to complete; then return None. NrZ)rrss r)rzchunks.checkpoint"s r+N)rrrr)rr)__name__ __module__ __qualname____doc__ staticmethodrrrZr+r)r/r/sX44\    \   r+r/)rrrr)rr)rrN)rZ) r[rrIr\r]r\r4r^rr) rwrrurxrvrNrrrr) rrrrrrrrrurrr)rurrvrN)rr))r __future__rrcollections.abcrrrtypingrrr dask.baser r r r rrdask.blockwiser dask.corer dask.delayedrrdask.highlevelgraphrrr__all__rrr&rYr;rrrrr/rZr+r)rsd#""""" 3333333333((((((((((%$$$$$))))))))HHHHHHHHHH 4 GCLL 26& & & & & & R++++\D)+ 7 7 7 7 7 |  15f f f f f f R]]]]@"DPT@4@4@4@4@4@4J26>4>4>4>4>4>4B          r+