CdE_lddlmZddlZddlZddlZddlZddlmZddlm Z m Z m Z ddl m Z ddlmZmZmZmZddlmZddlmZmZmZmZmZdd lmZdd lmZdd lm Z m!Z!dd l"m#Z#dd l$m%Z%m&Z&m'Z'm(Z(m)Z)ddgZ*ej+dedZ,dZ-dZ.dZ/dZ0dddZed$dZ1dZ2dZ3dZ4Gddee%Z5d%dZ6Gdd e5Z7Gd!d"e5Z8ej9ej:ej;ej<ej=ej>ej?ej@ejAejBejCejDejEejFejGejHejIejJejKejLejMejNejOfD]ZPe5QeP e5QejRn #eS$rYnwxYwd#ZTdS)&) annotationsN)Iterator)fields is_dataclassreplace)partial)concatcurrymergeunique)config)DaskMethodsMixin dont_optimizeis_dask_collectionnamed_schedulersreplace_name_in_key)tokenize) globalmethod)flattenquote)HighLevelGraph)OperatorMethodMixinapplyfuncnameis_namedtuple_instance methodcallerDelayeddelayedthreadssynccBtt|}|sdg|z}|S)z,Unzip a list of lists into ``nout`` outputs.)listzip)lsnoutouts ,lib/python3.11/site-packages/dask/delayed.pyunzipr)'s) sBx..C dTk Jct|sJdt|z}|}|\}}|||f|zi}t j|||g}t ||S)N finalize- dependencies)rr __dask_keys____dask_postcompute__rfrom_collectionsr) collectionnamekeysfinalizeargslayergraphs r(r5r5/s j ) )))) *-- -D  # # % %D4466NHd Hd#d* +E  +D%zl S S SE 4  r*cttr jffStrt }|j|ffStt rt t}|tt tfvrmtdDd\}}t|}t tt|t}|tur||f}||fS|tur9tdD\}}t|f|fS|t"ur1tjjjg\}}t"g|R|fSt+rtfdt-D\}}|sdfS fdt-D}t/fi|nG#t0$r}t1d|d |d }~wt2$r}t3d|d |d }~wwxYwt4|dt|ff|fSt7r$td D\}}|g|R|fSdfS) aNormalize a python object and merge all sub-graphs. - Replace ``Delayed`` with their keys - Convert literals to things the schedulers can handle - Extract dask graphs from all enclosed values Parameters ---------- expr : object The object to be normalized. This function knows how to handle dask collections, as well as most builtin python types. Returns ------- task : normalized task to be run collections : a tuple of collections Examples -------- >>> import dask >>> a = delayed(1, 'a') >>> b = delayed(2, 'b') >>> task, collections = unpack_collections([a, b, 3]) >>> task ['a', 'b', 3] >>> collections (Delayed('a'), Delayed('b')) >>> task, collections = unpack_collections({a: 1, b: 2}) >>> task (, [['a', 1], ['b', 2]]) >>> collections (Delayed('a'), Delayed('b')) c34K|]}t|VdSN)unpack_collections.0es r( z%unpack_collections..js+"G"GQ#5a#8#8"G"G"G"G"G"Gr*keycg|] \}}||g Sr"r"r>kvs r( z&unpack_collections..ss /P/P/P41aA/P/P/Pr*crg|]3}t|j|jt|jg4Sr"hasattrr3getattrr>fexprs r(rHz&unpack_collections..|N   4(( qv../   r*r"cpi|]2}t|j|jt|j3Sr"rJrMs r( z&unpack_collections..sK4((af--r*zFailed to unpack z> instance. Note that using a custom __init__ is not supported.NzF instance. Note that using fields with `init=False` are not supported.cg|]}|Sr"r"r>rGs r(rHz&unpack_collections..s/@/@/@a/@/@/@r*) isinstancer_keyrr5rtupletyper#setr)r r iddictr<itemsslicestartstopsteprrr TypeError ValueErrorrr)rO finalizedtypr6 collections_fieldsr?s` r(r<r<:s%F$  "y4'!!$,TNN ~ |++$!!T{{ t**C tUC   !"G"G$"G"G"GKKkDzzF6+#6#6B???@@ d??;D[   d{{./P/P4::<

t-jjjg\}}t8ft#|z|fSifS) aNormalize a python object and merge all sub-graphs. - Replace ``Delayed`` with their keys - Convert literals to things the schedulers can handle - Extract dask graphs from all enclosed values Parameters ---------- expr : object The object to be normalized. This function knows how to handle ``Delayed``s, as well as most builtin python types. Returns ------- task : normalized task to be run dask : a merged dask graph that forms the dag for this task Examples -------- >>> import dask >>> a = delayed(1, 'a') >>> b = delayed(2, 'b') >>> task, dask = to_task_dask([a, b, 3]) # doctest: +SKIP >>> task # doctest: +SKIP ['a', 'b', 3] >>> dict(dask) # doctest: +SKIP {'a': 1, 'b': 2} >>> task, dasks = to_task_dask({a: 1, b: 2}) # doctest: +SKIP >>> task # doctest: +SKIP (dict, [['a', 1], ['b', 2]]) >>> dict(dask) # doctest: +SKIP {'a': 1, 'b': 2} zYThe dask.delayed.to_dask_dask function has been Deprecated in favor of unpack_collectionsrA) stacklevelr,Tpure__dask_optimize__c34K|]}t|VdSr;) to_task_daskr=s r(r@zto_task_dask..s(;;\!__;;;;;;r*cg|] \}}||g Sr"r"rEs r(rHz to_task_dask..s !B!B!BTQ1a&!B!B!Br*crg|]3}t|j|jt|jg4Sr"rJrMs r(rHz to_task_dask..rPr*r"cg|]}|Sr"r"rTs r(rHz to_task_dask..s!2!2!2!!2!2!2r*) warningswarnrUrrCdaskrrr/rLrr0update__dask_graph__rr#rXrWrYr)r r[rmr\rrrrr]r^r_r`) rOr3r4optr5r6dskrddaskss ` r(rmrmsF M 4 $  #x""$Xd6666!!##d/??2244$h%,- 33t**,,d33444Sy$!!Dzz t**C tUC   ;;d;;;Q?? eDzzEll!Tkkc{{d S/AA d{{ !B!BTZZ\\!B!B!BCC cd|S  D 3          csBt -s22d##! !2!2T!2!2!233 c|d||S   e|| $*di!CDD cx%++%s** 8Or*ric|tjdd}|r t|i|Stt jS)arMapping function from task -> consistent name. Parameters ---------- args : object Python objects that summarize the task. pure : boolean, optional If True, a consistent hash function is tried on the input. If this fails, then a unique identifier is used. If False (default), then a unique identifier is always used. N delayed_pureF)r get _tokenizestruuiduuid4)rjr6kwargss r(rrsI |z.%00 !$)&)))4:<<   r*Tc|t|tr|St|s|rt|\}}nt |}t }|.t |tur|dkstd|z||ur[|sF |j }n$#t$rt |j }YnwxYwt|||}|d|}t||||S|s(t |j dt||}||i} tj|| |} t|| |S)aHWraps a function or object to produce a ``Delayed``. ``Delayed`` objects act as proxies for the object they wrap, but all operations on them are done lazily by building up a dask graph internally. Parameters ---------- obj : object The function or object to wrap name : string or hashable, optional The key to use in the underlying graph for the wrapped object. Defaults to hashing content. Note that this only affects the name of the object wrapped by this call to delayed, and *not* the output of delayed function calls - for that use ``dask_key_name=`` as described below. .. note:: Because this ``name`` is used as the key in task graphs, you should ensure that it uniquely identifies ``obj``. If you'd like to provide a descriptive name that is still unique, combine the descriptive name with :func:`dask.base.tokenize` of the ``array_like``. See :ref:`graphs` for more. pure : bool, optional Indicates whether calling the resulting ``Delayed`` object is a pure operation. If True, arguments to the call are hashed to produce deterministic keys. If not provided, the default is to check the global ``delayed_pure`` setting, and fallback to ``False`` if unset. nout : int, optional The number of outputs returned from calling the resulting ``Delayed`` object. If provided, the ``Delayed`` output of the call can be iterated into ``nout`` objects, allowing for unpacking of results. By default iteration over ``Delayed`` objects will error. Note, that ``nout=1`` expects ``obj`` to return a tuple of length 1, and consequently for ``nout=0``, ``obj`` should return an empty tuple. traverse : bool, optional By default dask traverses builtin python collections looking for dask objects passed to ``delayed``. For large collections this can be expensive. If ``obj`` doesn't contain any dask objects, set ``traverse=False`` to avoid doing this traversal. Examples -------- Apply to functions to delay execution: >>> from dask import delayed >>> def inc(x): ... return x + 1 >>> inc(10) 11 >>> x = delayed(inc, pure=True)(10) >>> type(x) == Delayed True >>> x.compute() 11 Can be used as a decorator: >>> @delayed(pure=True) ... def add(a, b): ... return a + b >>> add(1, 2).compute() 3 ``delayed`` also accepts an optional keyword ``pure``. If False, then subsequent calls will always produce a different ``Delayed``. This is useful for non-pure functions (such as ``time`` or ``random``). >>> from random import random >>> out1 = delayed(random, pure=False)() >>> out2 = delayed(random, pure=False)() >>> out1.key == out2.key False If you know a function is pure (output only depends on the input, with no global state), then you can set ``pure=True``. This will attempt to apply a consistent name to the output, but will fallback on the same behavior of ``pure=False`` if this fails. >>> @delayed(pure=True) ... def add(a, b): ... return a + b >>> out1 = add(1, 2) >>> out2 = add(1, 2) >>> out1.key == out2.key True Instead of setting ``pure`` as a property of the callable, you can also set it contextually using the ``delayed_pure`` setting. Note that this influences the *call* and not the *creation* of the callable: >>> @delayed ... def mul(a, b): ... return a * b >>> import dask >>> with dask.config.set(delayed_pure=True): ... print(mul(1, 2).key == mul(1, 2).key) True >>> with dask.config.set(delayed_pure=False): ... print(mul(1, 2).key == mul(1, 2).key) False The key name of the result of calling a delayed object is determined by hashing the arguments by default. To explicitly set the name, you can use the ``dask_key_name`` keyword when calling the function: >>> add(1, 2) # doctest: +SKIP Delayed('add-3dce7c56edd1ac2614add714086e950f') >>> add(1, 2, dask_key_name='three') Delayed('three') Note that objects with the same key name are assumed to have the same result. If you set the names explicitly you should make sure your key names are different for different results. >>> add(1, 2, dask_key_name='three') Delayed('three') >>> add(2, 1, dask_key_name='three') Delayed('three') >>> add(2, 2, dask_key_name='four') Delayed('four') ``delayed`` can also be applied to objects to make operations on them lazy: >>> a = delayed([1, 2, 3]) >>> isinstance(a, Delayed) True >>> a.compute() [1, 2, 3] The key name of a delayed object is hashed by default if ``pure=True`` or is generated randomly if ``pure=False`` (default). To explicitly set the name, you can use the ``name`` keyword. To ensure that the key is unique you should include the tokenized value as well, or otherwise ensure that it's unique: >>> from dask.base import tokenize >>> data = [1, 2, 3] >>> a = delayed(data, name='mylist-' + tokenize(data)) >>> a # doctest: +SKIP Delayed('mylist-55af65871cb378a4fa6de1660c3e8fb7') Delayed results act as a proxy to the underlying object. Many operators are supported: >>> (a + [1, 2]).compute() [1, 2, 3, 1, 2] >>> a[1].compute() 2 Method and attribute access also works: >>> a.count(2).compute() 1 Note that if a method doesn't exist, no error will be thrown until runtime: >>> res = a.not_a_real_method() # doctest: +SKIP >>> res.compute() # doctest: +SKIP AttributeError("'list' object has no attribute 'not_a_real_method'") "Magic" methods (e.g. operators and attribute access) are assumed to be pure, meaning that subsequent calls must return the same results. This behavior is not overrideable through the ``delayed`` call, but can be modified using other ways as described below. To invoke an impure attribute or operator, you'd need to use it in a delayed function with ``pure=False``: >>> class Incrementer: ... def __init__(self): ... self._n = 0 ... @property ... def n(self): ... self._n += 1 ... return self._n ... >>> x = delayed(Incrementer()) >>> x.n.key == x.n.key True >>> get_n = delayed(lambda x: x.n, pure=False) >>> get_n(x).key == get_n(x).key False In contrast, methods are assumed to be impure by default, meaning that subsequent calls may return different results. To assume purity, set ``pure=True``. This allows sharing of any intermediate values. >>> a.count(2, pure=True).key == a.count(2, pure=True).key True As with function calls, method calls also respect the global ``delayed_pure`` setting and support the ``dask_key_name`` keyword: >>> a.count(2, dask_key_name="count_2") Delayed('count_2') >>> import dask >>> with dask.config.set(delayed_pure=True): ... print(a.count(2).key == a.count(2).key) True Nrz3nout must be None or a non-negative integer, got %sri-rjr&r-)rUrrr<rrYrXintrb__name__AttributeErrorr DelayedLeafrr1) objr3rjr&traversetaskreprefixtokenr7r8s r(rr s{Z#w #(.s33kkSzzee LT$ZZ3..4199NQUUVVV s{{ ' ,! , , ,c+ ,S$T222E&&u&&D34d;;;; G3ii(FF8Dt+D+D+DFFDt /e+VVVtUD)))sBB98B9c|||Sr;r")methodselfothers r(_swaprs 6%  r*c,tt|S)z@Wrapper to create 'right' version of operator given left version)rr)rs r(rightrs 5& ! !!r*c t|ttfs|g}t|ts$tjt ||d}|tt|}|SNr"r-)rUr#rYrr1rZcullr)rwr4rs r(optimizerss dT3K ( (v c> * *M-bggsLLL ((3wt}}%% & &C Jr*ceZdZdZdZddZedZedZdZ dZ d Z d Z e eZeed Zd ZdZdddZdZdZdZdZdZdZdZdZddddZdZeZ dZ!e"d dZ#e#Z$dS)!rzpRepresents a value to be computed by dask. Equivalent to the output from a single key in a dask graph. )rV_dask_length_layerNc||_||_||_|p||_t |t r:|j|jvr.td|jdt|jdSdS)NzLayer z% not in the HighLevelGraph's layers: ) rVrrrrUrlayersrbr#)rrCrwlengthr7s r(__init__zDelayed.__init__s   ls c> * * t{#*/L/L]]]4PSPZK[K[]]   /L/Lr*c|jSr;)rVrs r(rCz Delayed.keys yr*c|jSr;)rrs r(rsz Delayed.dask s zr*c|jSr;)rsrs r(ruzDelayed.__dask_graph__$s yr*c|jgSr;rBrs r(r/zDelayed.__dask_keys__'s zr*c|jfSr;)rrs r(__dask_layers__zDelayed.__dask_layers__*s  ~r*c|jSr;rBrs r(__dask_tokenize__zDelayed.__dask_tokenize__-s xr*delayed_optimizerBctdfSNr") single_keyrs r(r0zDelayed.__dask_postcompute__3s 2~r*c|jdfSr)_rebuildrs r(__dask_postpersist__zDelayed.__dask_postpersist__6s}b  r*)renamec|rt|j|n|j}t|tr:t |jdkr"t t|j}nd}t|||j |S)N)r7) rrCrUrlenrnextiterrr)rrwrrCr7s r(rzDelayed._rebuild9s}7=K!$(F33348 c> * * s3:!/C/C cj))**EEEsCU;;;;r*c2dt|jdS)NzDelayed())reprrCrs r(__repr__zDelayed.__repr__Es+$tx..++++r*c*t|jSr;)hashrCrs r(__hash__zDelayed.__hash__HsDH~~r*c:tt|Sr;)dirrXrs r(__dir__zDelayed.__dir__Ks4::r*c|drtd|d|dkrtjdt ||S)N_z Attribute z not found visualisezOdask.delayed objects have no `visualise` method. Perhaps you meant `visualize`?) startswithrrqrr DelayedAttr)rattrs r( __getattr__zDelayed.__getattr__Nsi ??3   @ !>d!>!>!>?? ? ;   M1    4&&&r*cz t|||dS#t$rtdwxYwNzDelayed objects are immutable)object __setattr__rra)rrvals r(rzDelayed.__setattr__\sQ =   tT3 / / / / / = = = ;<< <  =s :c tdrra)rindexrs r( __setitem__zDelayed.__setitem__gs7888r*c#xK|jtdt|jD] }||V dS)Nz6Delayed objects of unspecified length are not iterable)rrarange)ris r(__iter__zDelayed.__iter__jsO < TUU Ut|$$  Aq'MMMM  r*c<|jtd|jS)Nz3Delayed objects of unspecified length have no len())rrars r(__len__zDelayed.__len__ps! < QRR R|r*)rj dask_key_namecjtt|}||||||S||||S)Nri)r)rr)rrjrr6rfuncs r(__call__zDelayed.__call__usGu4(((  $4dF-HHH HtD$'''r*c td)Nz)Truth of Delayed objects is not supportedrrs r(__bool__zDelayed.__bool__{sCDDDr*c4||Stj||Sr;)types MethodType)rinstanceclss r(__get__zDelayed.__get__s  Kh///r*FcRt|rt|n|dfdS)NTric|i|Sr;r")r6rrs r(z.Delayed._get_binary_operator..svvt'>v'>'>r*)rr)ropinvrs @r(_get_binary_operatorzDelayed._get_binary_operators3c1rr===>>>>>r*NN)F)%r __module__ __qualname____doc__ __slots__rpropertyrCrsrur/rr staticmethod DEFAULT_GET__dask_scheduler__rrrkr0rrrrrrrrrrrr __nonzero__r classmethodr_get_unary_operatorr"r*r(rrs 7I    XX&k22$ X3EFFF!!!'+ < < < < <,,, ' ' ' = = =999  $(t((((( EEEK000 ???[?/r*c@|dd}|d|}|4dt|t|g|Rd|i|}n|}t t t |d\}} tt| } |r@t |\} } | | t|t|| f} n|f|z} tj ||| i| } ||nd}t|| |S)Nrrjz{}-{}rAr-)r)popformatrrr)mapr<r#r extendrrr1r)r func_tokenr6rrjr&rr3args2re dask_kwargs collections2rr8s r( call_functionrs=JJ55M ::fd # #D~~ TNN Z <$ < < $>! \<(((tT%[[+6w  + tTl   E#44D 4t , , ,,r*cleZdZdZdfd ZedZdZedZedZxZ S) r)_obj_pure_noutNcvt|d||_||_||_dSr;)superrrrr)rrrCrjr& __class__s r(rzDelayedLeaf.__init__s6 d###   r*cRtj|j|j|jidSr)rr1rVrrs r(rszDelayedLeaf.dasks.. I 49-B    r*cTt|j|j|||j|jS)Nr)rrrVrrrr6rs r(rzDelayedLeaf.__call__s. Ity$TZdj    r*c|jjSr;)rrrs r(rzDelayedLeaf.__name__s y!!r*c|jjSr;)rrrs r(rzDelayedLeaf.__doc__s y  r*r) rrrrrrrsrr __classcell__rs@r(rrs*I   X    ""X"!!X!!!!!r*rcHeZdZdZfdZfdZedZdZxZ S)r)r_attrcdt||dz}t|d||_||_dS)Nz getattr-%sTri)rrrrr )rrrrCrs r(rzDelayedAttr.__init__sHXc4d;;;; d###  r*c|dkr|jdkrtdt|S)NdtypezAttribute dtype not found)r rrr)rrrs r(rzDelayedAttr.__getattr__sA 7??tzW44 !<== =ww""4(((r*c|jt|jj|jfi}t j|j||jgS)Nr-)rVrLrr rr1)rr7s r(rszDelayedAttr.dasksBWdindjAB. IuDI;    r*cftt|j|j|jf|z|Sr;)rrr rrs r(rzDelayedAttr.__call__s2  $ $dj49,2Ev   r*) rrrrrrrrsrr r s@r(rrs!I )))))  X        r*rc|dS)z6Pick out the only element of this list, a list of keysrr")seqs r(rr s q6Mr*)NNNTr)U __future__roperatorrr~rqcollections.abcr dataclassesrrr functoolsrtlzr r r r rsr dask.baserrrrrrr| dask.contextr dask.corerrdask.highlevelgraphr dask.utilsrrrrr__all__r{rr)r5r<rmrrrrrrrrabsnegposinvertaddsubmulfloordivtruedivmodpowand_or_xorlshiftrshifteqgegtneleltgetitemr_bind_operatormatmulrrr"r*r(r:s"""""" $$$$$$5555555555,,,,,,,,,,,,,+++++%%%%%%$$$$$$$$...... i #"9.>v.FGG    aaaHWWWt!!!!!*e*e*e*e*P""" B/B/B/B/B/ 3B/B/B/J----:!!!!!'!!!:     '   B L L L O L L L   L L M L L O O K K K K K K / B2 2  8?++++   D s F&&F.-F.