LgXddlmZddlZddlZddlmZddlmZddlm Z ddl m Z m Z m Z ddlmZddlmZmZdd lmZmZmZmZmZmZmZddlmZmZdd lmZmZd Zd Z dd Z!dZ"ddZ# ddZ$dZ%dZ&ddZ'GddeZ(e(jRZ*dde*e*e*e*e*e*fdZ+dZ,GddZ-y)) annotationsN)Iterable)Enum)Any)configcoreutils) GraphNode)normalize_tokentokenize)flattenget_dependencies ishashableistask reverse_dictsubstoposort)GraphKeycTt|ttfs|g}t}t}i}ttt |}|rWg}|D]K}t ||d}||||<|||<|D])} | |vs|j | |j| +M|}|rW||fS)a<Return new dask with only the tasks required to calculate keys. In other words, remove unnecessary tasks from dask. ``keys`` may be a single key or list of keys. Examples -------- >>> def inc(x): ... return x + 1 >>> def add(x, y): ... return x + y >>> d = {'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)} >>> dsk, dependencies = cull(d, 'out') >>> dsk # doctest: +ELLIPSIS {'out': (, 'x', 10), 'x': 1} >>> dependencies # doctest: +ELLIPSIS {'out': ['x'], 'x': []} Returns ------- dsk: culled dask graph dependencies: Dict mapping {key: [deps]}. Useful side effect to accelerate other optimizations, notably fuse. Tas_list) isinstancelistsetdictr raddappend) dskkeysseen dependenciesoutworknew_workkdependencies_kds 1lib/python3.12/site-packages/dask/optimization.pycullr*s6 dT3K (v 5D6L C GDM" #D A-c1dCNVCF,LO#D=HHQKOOA&$    ct|d}|turM|dddDcgc]}tj|}}|j |ddj |S|t urt|ddkDrpt|ddtrZ|dddDcgc]}tj|}}|j |dddj |f|dddzSycc}wcc}w)zCreate new keys for fused tasksrN-) typestrr key_splitrjointuplelenr)r typxnamess r)!default_fused_linear_keys_renamerr9Ls tAw-C cz-1%1R%[9[#[9 T!Wxx #d1g,*z$q'!*c/J-1%1R%[9[#[9 T!WQZ !DGABK//::s C.C3c |7t|ts't|ts|g}tt|}||Dcic]}|t ||d}}i}t}|D]k}||}t |dkD} |D]Q} || |vr|j | | |vr|| =|j | 4| r|j | H| |vsM||| <Smg} ttt|j} |r|j\} }| |g} ||vr*|j|}| |=| j|||vr*| j| | vr*| j| } || =| j| | | vr*| j| |r|jDcic]\}}|t|}}}|durt}n |durd}n|}i}t}t}d}| D]} ||| }|duxr ||vxr||v}| j} || }| rm| j}||j!|j| ||j#| t%||| |}|j | |} | rm|j | |r*||<||| <|| ||<|h|| <|j | ||| <|jD]\}}||vs |||<|rn|jD]I\}}||zD]<}||}|j#||j |t%||||||<>K|||z D]}||=||= ||fScc}wcc}}w)aReturn new dask graph with linear sequence of tasks fused together. If specified, the keys in ``keys`` keyword argument are *not* fused. Supply ``dependencies`` from output of ``cull`` if available to avoid recomputing dependencies. **This function is mostly superseded by ``fuse``** Parameters ---------- dsk: dict keys: list dependencies: dict, optional {key: [list-of-keys]}. Must be a list to provide count of each key This optional input often comes from ``cull`` rename_keys: bool or func, optional Whether to rename fused keys with ``default_fused_linear_keys_renamer`` or not. Renaming fused keys can keep the graph more understandable and comprehensive, but it comes at the cost of additional processing. If False, then the top-most key will be used. For advanced usage, a func is also accepted, ``new_key = rename_keys(fused_key_list)``. Examples -------- >>> def inc(x): ... return x + 1 >>> def add(x, y): ... return x + y >>> d = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')} >>> dsk, dependencies = fuse(d) >>> dsk # doctest: +SKIP {'a-b-c': (inc, (inc, 1)), 'c': 'a-b-c'} >>> dsk, dependencies = fuse(d, rename_keys=False) >>> dsk # doctest: +ELLIPSIS {'c': (, (, 1))} >>> dsk, dependencies = fuse(d, keys=['b'], rename_keys=False) >>> dsk # doctest: +ELLIPSIS {'b': (, 1), 'c': (, 'b')} Returns ------- dsk: output graph with keys fused dependencies: dict mapping dependencies after fusion. Useful side effect to accelerate other downstream optimizations. NTrr/F)rrrr rr5rrmapreverseditemspopitempoprreverser9updateremover)rr r" rename_keysr& child2parent unfusibleparentdepshas_many_childrenchildchains parent2childchainv key_renamerrvfusedaliases is_renamednew_keyvalkeyold_keys r) fuse_linearrW[s`  4 5$%6D74=!KNO3a+CDAA3 OLIF#IMEETM e$,& ' e$" e$i'&, U#FHl&8&8&:;?*>$!QAs1vI*>L?d7   !  B EEeGJ  "!%(Gt#Ps(:Pwb?P  %jYY[F  ' ' (8(8(? @  ' ' .s6{E3/C IIe E   % BwKBuI$0$7L !#*)L  KK BuI/0IIKS e BsG %++-IC'>W+ G$!r#w93 *.  ~sG %& | qPH@s M+)M0c| tSt|tr|St|ttfs|g}t|SN)rrr)r7s r) _flat_setrZs;yu As  D#; ' C q6Mr+cr]tttjtr-j Dcic]\}}|t |c}}t|}Dcic]}|t|c}|r)|jfdj Dt|Dcic] }|vs||c}}i}|D]5}|} ||zD]} | |vr|| } n| } t| | | } | ||<7|j} j D],\}} || vs ||zD]} t| | || } | | |<.| Scc}}wcc}wcc}w)aReturn new dask with the given keys inlined with their values. Inlines all constants if ``inline_constants`` keyword is True. Note that the constant keys will remain in the graph, to remove them follow ``inline`` with ``cull``. Examples -------- >>> def inc(x): ... return x + 1 >>> def add(x, y): ... return x + y >>> d = {'x': 1, 'y': (inc, 'x'), 'z': (add, 'x', 'y')} >>> inline(d) # doctest: +ELLIPSIS {'x': 1, 'y': (, 1), 'z': (, 1, 'y')} >>> inline(d, keys='y') # doctest: +ELLIPSIS {'x': 1, 'y': (, 1), 'z': (, 1, (, 1))} >>> inline(d, keys='y', inline_constants=False) # doctest: +ELLIPSIS {'x': 1, 'y': (, 'x'), 'z': (, 'x', (, 'x'))} c3fK|](\}}t|r|vs|st|s|*ywrY)rr).0r&rMr"rs r) zinline..s7 #11 !s(LOFSTI #s.1)r") rnextitervaluesrr=rrZrrArrcopy)rr inline_constantsr"r&rM replaceorderkeysubsrUrTdepreplacedsk2items` ` r)inlinerjs2 4\-@-@-B(C#DdK.:.@.@.BC.Bda3q6 .BC T?D=@AS+C33SA       -DqAHCFD-LLG#h,s++Cg~!#,c(sC)C ,   <<>DIIKS d?|C003gdm41DI  KKD B .sE&3E,> E1E1cD sStt|Dcic]}|t|}}t| fd}jDcgc]\}}|||s|}}}|rt ||||D]}|=Scc}wcc}}w)aInline cheap functions into larger operations Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> double = lambda x: x * 2 >>> dsk = {'out': (add, 'i', 'd'), # doctest: +SKIP ... 'i': (inc, 'x'), ... 'd': (double, 'y'), ... 'x': 1, 'y': 1} >>> inline_functions(dsk, [], [inc]) # doctest: +SKIP {'out': (add, (inc, 'x'), 'd'), 'd': (double, 'y'), 'x': 1, 'y': 1} Protect output keys. In the example below ``i`` is not inlined because it is marked as an output key. >>> inline_functions(dsk, ['i', 'out'], [inc, double]) # doctest: +SKIP {'out': (add, 'i', (double, 'y')), 'i': (inc, 'x'), 'x': 1, 'y': 1} ct|tsHt|r=|vr9|r4 t|j rt fd|Dsyyy#t $rYywxYw)Nc3DK|]}t|tywrYrr )r]r(rs r)r^z6inline_functions..inlinable..fs!K;JaJs1vy1? TF)rr r functions_ofissubsetany TypeError)rUtask dependentsrfast_functionsoutputs r) inlinablez#inline_functions..inlinable^sw4+t 6!3 %..~>sK;Ec?KH   s1A A'&A')rcr")rrrr=rj) rrwrvrcr"r&rxrMr rus ``` @r)inline_functionsry8s6  [F(N=@AS+C33S Al+J ))+ 9+$!Q1aA+D 9  (8| AA J7B& :sB!B3BcPt|dr|j}t|dr|S)Nfunc)hasattrr{)r{s r)unwrap_partialr}ys& $ yy $  Kr+ct}|g}tth}|rjg}|D]^}t||vst |r2|j t |d|j|ddN|j|`|}|rj|S)a1Set of functions contained within nested task Examples -------- >>> inc = lambda x: x + 1 >>> add = lambda x, y: x + y >>> mul = lambda x, y: x * y >>> task = (add, (mul, 1, 2), (inc, 3)) # doctest: +SKIP >>> functions_of(task) # doctest: +SKIP set([add, mul, inc]) rr/N)rrr4r0rrr}extend)rtfuncsr$sequence_typesr%s r)rprps EE 6DE]N DDz^+$<IInT!W56OODH-OOD)   Lr+ct|}t|}t|}rdzfd}|tur}t j |}|Dchc]}t j |}}|j |t|}|j|dj|} || S|turt|dkDrt|dtrt j |}|Dchc]}t j |}}|j |t|}|j|ddj|} || f|ddzSyyycc}wcc}w)zCreate new keys for ``fuse`` tasks. The optional parameter `max_fused_key_length` is used to limit the maximum string length for each renamed key. If this parameter is set to `None`, there is no limit. c\r(t|kDrt|ddd}|dd|}|S)Nr7r.)r5hash)key_name name_hashmax_fused_key_lengths r)_enforce_max_key_limitz:default_fused_keys_renamer.._enforce_max_key_limitsF CM4H$H>!,bq1I"#8$89:!I;GHr+r.rr/N) r<r_r0r1r r2discardsortedrr3r4r5r) r rit first_keyr6r first_namer&r8concatenated_names ` r)default_fused_keys_renamerrsN $BRI y/C!  cz__Y/ -/0R#R0 j!u  YHHUO%&788 #i.1,IaL#1N__Y/ -/0R#R0 j!u  Yq\"HHUO&'89;imKK2O, 11s E'6E,ceZdZdZddZy)Defaultrcy)Nz selfs r)__repr__zDefault.__repr__sr+Nreturnr1)__name__ __module__ __qualname__tokenrrr+r)rrs Er+rc xtjddur|fS|7t|ts't|ts|g}tt |}|t urtjd}|t usJ|t urtjd}|t usJ|t urtjd}|t usJ||dz}|t urtjd}|t usJ|d|tj|d zzz}|t urtjd }|t usJ|d}|r|s|fS|t urtjd }|t usJ|d urt} n|durd} nt|s td |} | du}|D cic]} | t| d } } nA|jD cic]&\} } | t| tr| n t| d (} } } i} | jD]9\} }|D]!} | | vr| g| | <| | j| #t|| | <;t}| jD]\} }t|d k(s| |xsdvs| vs#t| t r7t#| t$us$t| t&j(t*fsot-fd|Dr|j/| |s&|r t1d| j3Dr| fSj5}i}g}g}| j6}|j.}|j6}|j8}|j6}|j}|j6}|j}|j:}|j6}|r|}||||vr | |d}||vr ||||| |z |d}||k7rK|| |z} | r|| |}|d}|| |z} | r||||||r|gndd d d d| ||z fn|| |}!|!|z }"|!|"z } t| }#|#d k(rG|\}$}%}&}'}(})}*}+t|+},|*|,d z cxkDrdk\rnn|,d z }*|"|+z}"t|"|,k(}-|-s|*d z }*|)|*z|'z |kr|-s|'|krt|t st=||$|%}.|!j9|$|!||$z}!||$=||$|r|&j||&||<||$d|r,|-r|||.|&|'|(|)|*|"fn|||.|&|'d z|(|)d z|*|"fng|.||<nj|%||$<||$|r9|*t?|d z kDrt?|d z }*|||||r|gndd |(d |*|"fnn g}&d }'d}(d}/d})d}*t}+d}0||# d}1||# d=|1D]J\}2}2}2}3}4}5}6}7|3d k(r|/d z }/n|3|'kDr|3}'|(|4z }(|)|5z })|*|6z }*t|7|0kDr t|7}0|+|7z}+Lt|+},|*tA|#d z tCd|,|0z z }*|*|,d z cxkDrdk\rnn|,d z }*|"|+z}"t|"|,k(}-|-s|*d z }*|)|*z|'z |kr|/|kr|(|kr|'|kr|-s|'|krt|t st-fd| Ds|}.t}8|1D]M}9|9d}:t=|.|:|9d }.||:=|8||:z}8||:|s1||:d|&j;|9dO|!| z}!|!|8z}!|r|&j||&||<|r|||.|&|'d z|(|)d z|*|"fne|.||<ni|1D]}9|9d ||9d<||9d|r?|(|kDr|}(|*t?|d z kDrt?|d z }*|||||r|gndd |(d |*|"fnn | |d}|r|rtE||| ||| rC|jD]0\};}<| |<}=|=|=|vs||;||=<|=||;<| |;| |=<|=h| |;<2|| fScc} wcc} } w)am Fuse tasks that form reductions; more advanced than ``fuse_linear`` This trades parallelism opportunities for faster scheduling by making tasks less granular. It can replace ``fuse_linear`` in optimization passes. This optimization applies to all reductions--tasks that have at most one dependent--so it may be viewed as fusing "multiple input, single output" groups of tasks into a single task. There are many parameters to fine tune the behavior, which are described below. ``ave_width`` is the natural parameter with which to compare parallelism to granularity, so it should always be specified. Reasonable values for other parameters will be determined using ``ave_width`` if necessary. Parameters ---------- dsk: dict dask graph keys: list or set, optional Keys that must remain in the returned dask graph dependencies: dict, optional {key: [list-of-keys]}. Must be a list to provide count of each key This optional input often comes from ``cull`` ave_width: float (default 1) Upper limit for ``width = num_nodes / height``, a good measure of parallelizability. dask.config key: ``optimization.fuse.ave-width`` max_width: int (default infinite) Don't fuse if total width is greater than this. dask.config key: ``optimization.fuse.max-width`` max_height: int or None (default None) Don't fuse more than this many levels. Set to None to dynamically adjust to ``1.5 + ave_width * log(ave_width + 1)``. dask.config key: ``optimization.fuse.max-height`` max_depth_new_edges: int or None (default None) Don't fuse if new dependencies are added after this many levels. Set to None to dynamically adjust to ave_width * 1.5. dask.config key: ``optimization.fuse.max-depth-new-edges`` rename_keys: bool or func, optional (default True) Whether to rename the fused keys with ``default_fused_keys_renamer`` or not. Renaming fused keys can keep the graph more understandable and comprehensive, but it comes at the cost of additional processing. If False, then the top-most key will be used. For advanced usage, a function to create the new name is also accepted. dask.config key: ``optimization.fuse.rename-keys`` fuse_subgraphs : bool or None, optional (default None) Whether to fuse multiple tasks into ``SubgraphCallable`` objects. Set to None to let the default optimizer of individual dask collections decide. If no collection-specific default exists, None defaults to False. dask.config key: ``optimization.fuse.subgraphs`` Returns ------- dsk output graph with keys fused dependencies dict mapping dependencies after fusion. Useful side effect to accelerate other downstream optimizations. zoptimization.fuse.activeFNzoptimization.fuse.ave-widthzoptimization.fuse.max-heightz%optimization.fuse.max-depth-new-edgesg?zoptimization.fuse.max-widthr/zoptimization.fuse.subgraphszoptimization.fuse.rename-keysTz)rename_keys must be a boolean or callablerrc3DK|]}t|tywrYrn)r]rMrs r)r^zfuse.._sDt! 3q695troc3JK|]}tt|dk7yw)r/N)r5r)r]rMs r)r^zfuse..ds!KNq#c!f+"2Ns!#rr-c3DK|]}t|tywrYrn)r] child_keyrs r)r^zfuse..7s$(19I!+3y>9 E19ro)#rgetrrrr _defaultmathlogrcallablersrr=rr5r r0r4numbersNumberr1rrrallrarbr?rBrrintminmax_inplace_fuse_subgraphs)>rr r" ave_width max_width max_heightmax_depth_new_edgesrCfuse_subgraphsrNr&rGrMrdepsvals reduciblerO fused_trees info_stackchildren_stackdeps_pop reducible_add reducible_popreducible_removefused_trees_popinfo_stack_appendinfo_stack_popchildren_stack_appendchildren_stack_extendchildren_stack_poprFrIchildren deps_parentedges num_childrenr child_task child_keysheightwidth num_nodesfudgechildren_edgesnum_children_edges no_new_edgesrTnum_single_nodes max_num_edges children_info_ cur_height cur_width cur_num_nodes cur_fudge cur_edges children_deps child_info cur_childroot_key fused_keysaliass>` r)fusersT Pzz,-6L    4 5$%6D74=!HJJ<= (((XZZ >? )))h&$jj)PQ"(222"'#oHJJ<= ((()dhhy1}&=== !$ABX--- JL  hjj!@A (***d0   k "CDD! T)KCFG3a#CD993G%**, ,1 Jq$'q-=c1d-S S,  E::<4A~3aa"  d)Q  I;;=4 IN$*"%Ss1vy1c!f&*SVgnnc=R*SDtDD MM! !  c!KELLN!KKDy BKJNxxHMMMMMM ''!ooO"))^^N*11*11'++ f !6]1%F ! f%i$v,67"2&E$tE{2)(3"F*2.E(4;6H  #$"5 #.DU i/  #$"6l #i/&."8} 1$'( !""!&),^)<&1A5:: 2Q 6^+E#&u:1C#CL' "U*f4 A&2E)E *3v; B #3v; :F#**95#x ':: yM(3&&--f52M"L=>2' "!%!!%?,1,'&0%/F*!]2 *y>M9,/ NM&)3'*-^)<&S$q(#a1Cm1S*TE1A5:: 2Q 6^+E#&u:1C#CL' "U*f4 A, 9!Y."j0&2E)E *3v; B$'(19(%"&k(+ *7J(21 I"&sIz!}"EC "9 )Xi-@@M,Y7* / 4 @ * 1 1*Q- @+8$x/ #}4 &&--f52r?rr@rr4SubgraphCallabler)rr r"rrCrDrErFrGrHrIrJr&rMrKrLntasksrUsubgraphoutkey inkeys_setinkeyschain2subchains r)rrsw LIF#IMEETM e$,& ' e$" e$i'&, U#F%1%7%7%9:%9TQAqD%9L: $,,. v $!%%f-FV$ LL  $  |# $$U+EU# LL |# C fSX& &Fz e$  *',-u!As1vIu-q-9r,CC \&)qrAQA z"'&&ACfLF  F&??1e4MM(+MM!$  #)K /-;..s  H% HceZdZUdZded<ded<ded<ded <eeZ d dd Zdd Zdd Z ddZ ddZ ddZ ddZ y )raDCreate a callable object from a dask graph. Parameters ---------- dsk : dict A dask graph outkey : Dask key The output key from the graph inkeys : list A list of keys to be used as arguments to the callable. name : str, optional The name to use for the function. rrrrztuple[Key, ...]rr1nameNc||_||_t||_|dt |||jz}||_y)Nzsubgraph_callable-)rrr4rr r)rrrrrs r)__init__zSubgraphCallable.__init__s@ Fm <'(3 *LLD r+c|jSrY)rrs r)rzSubgraphCallable.__repr__s yyr+ct|t|uxra|j|jk(xrF|j|jk(xr+t|jt|jk(SrY)r0rrrr)rothers r)__eq__zSubgraphCallable.__eq__sb J$u+ % 6 UZZ' 6 u||+ 6DKK C $55  r+c ,t|t|jk(s,tdt|jt|fztj|j |j tt|j|S)NzExpected %d args, got %d) r5r ValueErrorrrrrrzip)rargss r)__call__zSubgraphCallable.__call__sf4yC ,,73t{{;KSQUY:WWX Xxx$++tC T4J/KLLr+cjt|j|j|j|jffSrY)rrrrrrs r) __reduce__zSubgraphCallable.__reduce__s&$((DKKdii!PPPr+clt|jt|j|jfSrY)rr frozensetrrrs r)__hash__zSubgraphCallable.__hash__ s%T[[)DKK"8$))DEEr+crdt|j|j|j|jfS)Nr)r rrrrrs r)__dask_tokenize__z"SubgraphCallable.__dask_tokenize__ s0  DHH % KK KK II   r+rY)rrrrrz Iterable[Key]rz str | Noner)rrrbool)rrrr)rr4)rr)robject)rrr__doc____annotations__r4 __slots__rrrrrrrrr+r)rrsu  J K  Io&IRV"%/<DN M QF r+r)NNT)NTN)NFN)x). __future__rrrcollections.abcrenumrtypingrdaskrrr dask._task_specr dask.baser r dask.corer rrrrrr dask.tokenize dask.typingrrr*r9rWrZrjryr}rprrrrrrrrr+r)rs" $$$%/4"0f Nb?FLP>B >"LNd ==  ~BD)N< < r+