o Nrf@s4ddlmZddlZddlZddlmZddlmZddlm Z ddl m Z m Z m Z ddlmZmZddlmZmZmZmZmZmZmZdd lmZmZd d Zd d Zd'ddZddZd(ddZ d)ddZ!ddZ"ddZ#d*ddZ$Gdd d eZ%e%j&Z'dde'e'e'e'e'e'fd!d"Z(d#d$Z)Gd%d&d&Z*dS)+) annotationsN)Iterable)Enum)Any)configcoreutils)normalize_tokentokenize)flattenget_dependencies ishashableistask reverse_dictsubstoposort)GraphKeyc Cst|ttfs |g}t}t}i}ttt|}|rKg}|D]&}t||dd}||||<|||<|D]} | |vrE|| || q5q |}|s||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 r addappend) dskkeysseen dependenciesoutworknew_workkZdependencies_kdr%Z/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/dask/optimization.pyculls*    r'cCst|d}|tur"dd|dddD}||dd|S|turZt|ddkrZt|ddtrZdd|dddD}||ddd|f|dddSdS) zCreate new keys for fused tasksrcSg|]}t|qSr%r key_split.0xr%r%r& Nz5default_fused_linear_keys_renamer..N-cSr(r%r)r+r%r%r&r.Rr/)typestrrjointuplelenr)rtypnamesr%r%r&!default_fused_linear_keys_renamerJs  *r:Tcs:|durt|tst|ts|g}tt|}|dur$fddD}i}t}D]<}||}t|dk}|D]-} |durI| |vrI|| q9| |vrV|| =|| q9|r^|| q9| |vrf||| <q9q+g} ttt| } |r| \} }| |g} ||vr| |}| |=| |||vs| | | vr| | } || =| | | | vs| | |sudd| D}|durt} n |durd} n|} i}t}t}d}| D]l} | dur| | }|duo|vo||v}| } | }| r | }||| | ||| t|| |}|| |} | s|| |rA|||<||| <|| ||<|h|| <|| q||| <qو D]\}}||vrW|||<qJ|r| D]%\}}||@D]}||}||||t||||||<qhq`|dur||D] }||=||=q||fS)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. Nci|] }|t|ddqSTrr r,r#rr%r& zfuse_linear..r2cSi|] \}}|t|qSr%rr,r#vr%r%r&r@TF)rrrr r7rrmapreverseditemspopitempoprreverser:updateremover)rrr rename_keys child2parent unfusibleparentdepshas_many_childrenchildchains parent2childchain key_renamerrvZfusedaliasesZ is_renamedZnew_keyvalkeyold_keyr%r?r& fuse_linearYs0                          r_cCs8|durtSt|tr|St|ttfs|g}t|SN)rrr)r-r%r%r& _flat_sets rac s0rttttrddDt|}dur'fddD|r8|fddDtfdd|Dd}i}|D]&}|}||@D]}||vr`||} n|} t ||| }qU|||<qI| } D]\}}|| vr||@D] } t || || }q|| |<qx| S) 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'))} cSrBr%rCrDr%r%r&r@rFzinline..Nci|]}|t|qSr%r=r>r?r%r&r@c3s8|]\}}t|r|vs|st|s|VqdSr`)r rrDrrr%r& szinline..csi|] }|vr||qSr%r%r>r?r%r&r@s)r) rnextitervaluesrrIrarMrrcopy) rrinline_constantsrZ replaceorderZkeysubsr]r\depreplaceZdsk2itemr%rdr&inlines:  rnFcssStt|durfddD}t|fddfddD}|rBt|||d|D]}|=q<S) 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} Ncrbr%r=r>r?r%r&r@Yrcz$inline_functions..cs&zt|WStyYdSw)NF) functions_ofissubset TypeError)rE)fast_functionsr%r& inlinable\s  z#inline_functions..inlinablecs4g|]\}}t|r|r|vr|r|qSr%)rrD) dependentsrsoutputr%r&r.bsz$inline_functions..)rjr)rrrIrn)rrurrrjrrr#r%)rtrrrrsrur&inline_functions6s$ rvcCst|dr |j}t|ds|S)Nfunc)hasattrrw)rwr%r%r&unwrap_partialqs  rycCsvt}|g}tth}|r9g}|D]$}t||vr4t|r/|t|d||ddq||q|}|s |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]) rr2N)rrr6r3rrryextend)taskfuncsr!Zsequence_typesr"r%r%r&rows    roxc st|}t|}t|}rd8fdd}|tur?t|}dd|D}||t|}||d |}||S|t ur|t |dkr~t |dtrt|}dd|D}||t|}||dd |}||f|d d Sd Sd Sd S) 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. cs>rt|krt|ddd}|dd|}|S)Nr-r1)r7hash)Zkey_nameZ name_hashmax_fused_key_lengthr%r&_enforce_max_key_limitsz:default_fused_keys_renamer.._enforce_max_key_limitcSh|]}t|qSr%r)r>r%r%r& r/z-default_fused_keys_renamer..r1rcSrr%r)r>r%r%r&rr/r2N) rHrfr3r4rr*discardsortedrr5r6r7r) rritZ first_keyr8rZ first_namer9Zconcatenated_namer%rr&default_fused_keys_renamers.     "    rc@seZdZdZdddZdS)Defaultrreturnr4cCsdS)Nz r%selfr%r%r&__repr__szDefault.__repr__Nrr4)__name__ __module__ __qualname__tokenrr%r%r%r&rsrc > stddur |fS|dur"t|ts"t|ts|g}tt|}|tur1td}|tus1J|tur@td}|tus@J|turOtd}|tusOJ|durW|d}|turftd}|tusfJ|durud|t|d }|turtd }|tusJ|durd}|r|s|fS|turtd }|tusJ|d urt } n|durd} n t |st d |} | du}|durˇfddD} n fdd| D} i} | D] \} } | D]}|| vr| g| |<q| | | qt| | | <qdd| D}|r ||8} D]\} }t|tur*t|tjtfs*|| q|sB|r>tdd| DrB| fS}i}g}g}| j}|j}|j}|j}|j}|j }|j}|j }|j}|j}|r|}||||vr| |d}||vsy||||| |@ |d}||kr|| |@} | r|| |}|d}|| |@} | s||||||r|gndd d d d| ||fn@|| |}!|!|}"|!|"} t| }#|#d kr|\}$}%}&}'}(})}*}+t|+},|*|,d krdkrnn|,d }*|"|+O}"t|"|,k}-|-s|*d 7}*|)|*|'|kr|-s0|'|krt||$|%}.|!|$|!||$O}!||$=||$|r[|& ||&||<||$d|r|-ro|||.|&|'|(|)|*|"fn|||.|&|'d |(|)d |*|"fn|.||<n|%||$<||$|r|*t|d krt|d }*|||||r|gndd |(d |*|"fnVn\g}&d }'d}(d}/d})d}*t}+d}0||# d}1||# d=|1D]7\}2}2}2}3}4}5}6}7|3d kr|/d 7}/n|3|'kr|3}'|(|47}(|)|57})|*|67}*t|7|0kr t|7}0|+|7O}+qt|+},|*t|#d td|,|07}*|*|,d kr1dkr7nn|,d }*|"|+O}"t|"|,k}-|-sH|*d 7}*|)|*|'|kr|/|kr|(|kr|'|kr|-sh|'|krʈ|}.t}8|1D]+}9|9d}:t|.|:|9d }.||:=|8||:O}8||:|r||:d|&|9dqq|!| 8}!|!|8O}!|r|& ||&||<|r|||.|&|'d |(|)d |*|"fnH|.||<nK|1D]}9|9d ||9d<||9dq|r |(|kr|}(|*t|d krt|d }*|||||r|gndd |(d |*|"fnn| |d}q|sm|r#t ||| ||| rS| D](\};}<| |<}=|=durQ|=|vrQ||;||=<|=||;<| |;| |=<|=h| |;<q*|| fS)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-widthr2zoptimization.fuse.subgraphszoptimization.fuse.rename-keysTz)rename_keys must be a boolean or callablecr;r<r=r>r?r%r&r@?rAzfuse..cs.i|]\}}|t|tr|nt|ddqSr<)rrr rDr?r%r&r@AscSs h|] \}}t|dkr|qS)r2)r7)r,r#valsr%r%r&rOs zfuse..css |] }tt|dkVqdS)r2N)r7r)r,rEr%r%r&reXszfuse..rr0)!rgetrrrr _defaultmathlogrcallablerqrIrr3r6numbersNumberr4rallrhrirKrrNrzr7rintminmax_inplace_fuse_subgraphs)>rrrZ ave_width max_width max_heightZmax_depth_new_edgesrOZfuse_subgraphsrYrSZrdepsr#rrEZ reduciblerZ fused_treesZ info_stackZchildren_stackZdeps_popZ reducible_addZ reducible_popZreducible_removeZfused_trees_popZinfo_stack_appendZinfo_stack_popZchildren_stack_appendZchildren_stack_extendZchildren_stack_poprRrUchildrenZ deps_parentedgesZ num_childrenZ child_keyZ child_taskZ child_keysheightwidthZ num_nodesZfudgeZchildren_edgesZnum_children_edgesZ no_new_edgesr\Znum_single_nodesZ max_num_edgesZ children_info_Z cur_heightZ cur_widthZ cur_num_nodesZ cur_fudgeZ cur_edgesZ children_depsZ child_infoZ cur_childZroot_keyZ fused_keysaliasr%r?r&fuses^H                                                          y   rcsi}t}D]<}||}t|dk} |D]-} |dur%| |vr%|| q| |vr2|| =|| q| r:|| q| |vrB||| <qqg} dd|D} |r|\} }| |g} ||vrp||}| |=| |||vs_| | | vr| | } || =| | | | vsxd}| D]}|t|7}|dkr| | nq|sQ| D]W} fdd| D}| d}|| d}||<| ddD]}||=|=qt |}t |||f||<|rg}| D]}||d}|r| |q||q|||<qdS) zKSubroutine of fuse. Mutates dsk, dependencies, and fused_trees inplacer2NcSsi|]\}}||qSr%r%rDr%r%r&r@r/z+_inplace_fuse_subgraphs..rcsi|]}||qSr%r%r>r?r%r&r@r/r0F) rr7rrIrJrKrrLrr6SubgraphCallablerz)rrrrrOrPrQrRrSrTrUrVrWrXZntasksr]ZsubgraphoutkeyZ inkeys_setr#inkeysZchain2Zsubchainr%r?r&rusx              rc@seZdZUdZded<ded<ded<ded <eeZ d#d$d dZd%ddZd&ddZ d'ddZ d(ddZ d)ddZ d*d!d"Z d S)+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, ...]rr4nameN Iterable[Key] str | NonecCs:||_||_t||_|durdt|||j}||_dS)Nzsubgraph_callable-)rrr6rr r)rrrrrr%r%r&__init__s   zSubgraphCallable.__init__rcCs|jSr`)rrr%r%r&rszSubgraphCallable.__repr__otherrboolcCs<t|t|uo|j|jko|j|jkot|jt|jkSr`)r3rrrr)rrr%r%r&__eq__s  zSubgraphCallable.__eq__argscGsJt|t|jkstdt|jt|ft|j|jtt|j|S)NzExpected %d args, got %d) r7r ValueErrorrrrrrzip)rrr%r%r&__call__szSubgraphCallable.__call__r6cCst|j|j|j|jffSr`)rrrrrrr%r%r& __reduce__zSubgraphCallable.__reduce__rcCst|jt|j|jfSr`)rr frozensetrrrr%r%r&__hash__rzSubgraphCallable.__hash__objectcCsdt|j|j|j|jfS)Nr)r rrrrrr%r%r&__dask_tokenize__s z"SubgraphCallable.__dask_tokenize__r`)rrrrrrrrr)rrrr)rrrr)rr6)rr)rr)rrr__doc____annotations__r6 __slots__rrrrrrrr%r%r%r&rs     r)NNT)NTN)NFN)r})+ __future__rrrcollections.abcrenumrtypingrZdaskrrrZ dask.baser r Z dask.corer r r rrrrZ dask.typingrrr'r:r_rarnrvryrorrrrrrrr%r%r%r&sJ    $ 3  C ; ' 1G