a tDf@s2ddlmZddlZddlZddlmZddlmZddlm 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|}|rg}|D]L}t||dd}||||<|||<|D] } | |vrj|| || qjq@|}q4||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. TZas_list) isinstancelistsetdictr r addappend) dskkeysseen dependenciesoutworknew_workkZdependencies_kdr$^/nfs/NAS7/SABIOD/METHODE/ermites/ermites_venv/lib/python3.9/site-packages/dask/optimization.pyculls$  r&cCst|d}|turDdd|dddD}||dd|S|turt|ddkrt|ddtrdd|dddD}||ddd|f|dddSdSdS) zCreate new keys for fused tasksrcSsg|]}t|qSr$r key_split.0xr$r$r% Nz5default_fused_linear_keys_renamer..N-cSsg|]}t|qSr$r'r)r$r$r%r,Rr-)typestrrjointuplelenr)rtypnamesr$r$r%!default_fused_linear_keys_renamerJs  *r8TcsB|dur.t|ts.t|ts"|g}tt|}|durHfddD}i}t}D]x}||}t|dk}|D]Z} |dur| |vr|| qr| |vr|| =|| qr|r|| qr| |vrr||| <qrqVg} ttt| } |rd| \} }| |g} ||vr(| |}| |=| |q| | | vrX| | } || =| | q0| | qdd| D}|durt} n|durd} n|} i}t}t}d}| D]} | dur| | }|duo|vo||v}| } | }| rF| }||| | ||| t|| |}|| |} q|| |r|||<||| <|| ||<|h|| <|| n||| <q D]\}}||vr|||<q|r:| D]J\}}||@D]6}||}||||t||||||<qؐq|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. Ncsi|]}|t|ddqSTrr r*r"rr$r% r-zfuse_linear..r0cSsi|]\}}|t|qSr$rr*r"vr$r$r%r=r-TF)rrrr r5rrmapreverseditemspopitempoprreverser8updateremover)rrr rename_keys child2parent unfusibleparentdepshas_many_childrenchildchains parent2childchain key_renamerrvZfusedaliasesZ is_renamedZnew_keyvalkeyZold_keyr$r<r% fuse_linearYs0                              rXcCs8|durtSt|tr|St|ttfs0|g}t|SN)rrr)r+r$r$r% _flat_sets rZc s2r,ttttr,ddDt|}durNfddD|rp|fddDtfdd|Dd}i}|D]L}|}||@D]*}||vr||} n|} t ||| }q|||<q| } D]<\}}|| vr||@D]} t || || }q || |<q| 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'))} cSsi|]\}}|t|qSr$r>r?r$r$r%r=r-zinline..Ncsi|]}|t|qSr$r:r;r<r$r%r=r-c3s6|].\}}t|r|vs*|st|s|VqdSrY)r rr?rrr$r% s zinline..csi|]}|vr||qSr$r$r;r<r$r%r=r-)r) rnextitervaluesrrCrZrGrrcopy) rrinline_constantsrZ replaceorderZkeysubsrWrVdepreplaceZdsk2itemr$r[r%inlines8   reFcssStt|dur2fddD}t|fddfddD}|rt|||d|D] }|=qxS) 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} Ncsi|]}|t|qSr$r:r;r<r$r%r=Yr-z$inline_functions..cs*zt|WSty$YdS0dS)NF) functions_ofissubset TypeError)r@)fast_functionsr$r% inlinable\s z#inline_functions..inlinablecs4g|],\}}t|r|r|vr|r|qSr$)rr?) dependentsrjoutputr$r%r,bs z$inline_functions..)rar)rrrCre)rrlrirarrr"r$)rkrrirjrlr%inline_functions6s$ rmcCst|dr|j}q|S)Nfunc)hasattrrn)rnr$r$r%unwrap_partialqs rpcCstt}|g}tth}|rpg}|D]H}t||vr t|r^|t|d||ddq ||q |}q|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]) rr0N)rrr4r1rrrpextend)Ztaskfuncsr Zsequence_typesr!r$r$r%rfws   rfxc st|}t|}t|}r$d8fdd}|tur~t|}dd|D}||t|}||d |}||S|t urt |dkrt |dtrt|}dd|D}||t|}||dd |}||f|d d 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>r:t|kr:t|ddd}|dd|}|S)Nr+r/)r5hash)Zkey_nameZ name_hashmax_fused_key_lengthr$r%_enforce_max_key_limitsz:default_fused_keys_renamer.._enforce_max_key_limitcSsh|]}t|qSr$r'r;r$r$r% r-z-default_fused_keys_renamer..r/rcSsh|]}t|qSr$r'r;r$r$r%rzr-r0N) rBr]r1r2rr(discardsortedrr3r4r5r) rrxitZ first_keyr6ryZ first_namer7Zconcatenated_namer$rwr%default_fused_keys_renamers,     "   r~c@seZdZdZddddZdS)Defaultrr2returncCsdS)Nz r$selfr$r$r%__repr__szDefault.__repr__N)__name__ __module__ __qualname__tokenrr$r$r$r%rsrc > stddur|fS|durDt|tsDt|ts8|g}tt|}|turbtd}|tusbJ|turtd}|tusJ|turtd}|tusJ|dur|d}|turtd}|tusJ|durd|t|d }|tur td }|tus J|durd}|r&|s.|fS|turPtd }|tusPJ|d ur`t } n(|durpd} nt |st d n|} | du}|durfddD} nfdd| D} i} | D]F\} } | D]*}|| vr| g| |<n| | | qt| | | <qdd| D}|r8||8} D]4\} }t|tur@t|tjtfs@|| q@|s|rtdd| Dr| fS}i}g}g}| j}|j}|j}|j}|j}|j }|j}|j }|j}|j}|rH|}||||vr | |d}q||||| |@|d}||kr|| |@} | r|| |}|d}|| |@} qV||||||r|gndd d d d| ||fq8|| |}!|!|}"|!|"} t| }#|#d kr|\}$}%}&}'}(})}*}+t|+},|*|,d kr$dkr0nn|,d }*|"|+O}"t|"|,k}-|-sR|*d 7}*|)|*|'|kr |-st|'|kr t||$|%}.|!|$|!||$O}!||$=||$|r|& ||&||<||$d|r|-r|||.|&|'|(|)|*|"fn |||.|&|'d |(|)d |*|"fn |.||<qnb|%||$<||$|rD|*t|d krTt|d }*|||||rj|gndd |(d |*|"fnqng}&d }'d}(d}/d})d}*t}+d}0||# d}1||# d=|1D]n\}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 krvdkrnn|,d }*|"|+O}"t|"|,k}-|-s|*d 7}*|)|*|'|kr|/|kr|(|kr|'|kr|-s|'|kr|}.t}8|1D]V}9|9d}:t|.|:|9d }.||:=|8||:O}8||:|r||:d|&|9dq|!| 8}!|!|8O}!|rv|& ||&||<|r|||.|&|'d |(|)d |*|"fn |.||<qn|1D]"}9|9d ||9d<||9dq|rD|(|kr|}(|*t|d krt|d }*|||||r|gndd |(d |*|"fnq| |d}q8q|r^t ||| ||| r| D]P\};}<| |<}=|=durl|=|vrl||;||=<|=||;<| |;| |=<|=h| |;<ql|| 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-widthr0zoptimization.fuse.subgraphszoptimization.fuse.rename-keysTz)rename_keys must be a boolean or callablecsi|]}|t|ddqSr9r:r;r<r$r%r=?r-zfuse..cs.i|]&\}}|t|tr|n t|ddqSr9)rrr r?r<r$r%r=AscSs h|]\}}t|dkr|qS)r0)r5)r*r"valsr$r$r%rzOr-zfuse..css|]}tt|dkVqdS)r0N)r5r)r*r@r$r$r%r\Xr-zfuse..rr.)!rgetrrrr _defaultmathlogr~callablerhrCrr1r4numbersNumberr2r{allr_r`rErrHrqr5rintminmax_inplace_fuse_subgraphs)>rrrZ ave_width max_widthZ max_heightZmax_depth_new_edgesrIZfuse_subgraphsrSrMZrdepsr"rr@Z reduciblerT fused_treesZ info_stackZchildren_stackZdeps_popZ reducible_addZ reducible_popZreducible_removeZfused_trees_popZinfo_stack_appendZinfo_stack_popZchildren_stack_appendZchildren_stack_extendZchildren_stack_poprLrOchildrenZ deps_parentedgesZ num_childrenZ child_keyZ child_taskZ child_keysheightwidthZ num_nodesZfudgeZchildren_edgesZnum_children_edgesZ no_new_edgesrVZnum_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%fusesXH                                                          rcsi}t}D]x}||}t|dk} |D]Z} |durJ| |vrJ|| q*| |vrd|| =|| q*| rt|| q*| |vr*||| <q*qg} dd|D} |rD|\} }| |g} ||vr||}| |=| |q| | | vr | | } || =| | qd}| D].}|t|7}|dkr| | qqq| D]} fdd| D}| d}|| d}||<| ddD]}||=|=qt |}t |||f||<|rHg}| D].}||d}|r| |n ||q|||<qHdS) zKSubroutine of fuse. Mutates dsk, dependencies, and fused_trees inplacer0NcSsi|]\}}||qSr$r$r?r$r$r%r=r-z+_inplace_fuse_subgraphs..rcsi|]}||qSr$r$r;r<r$r%r=r-r.F) rr5rrCrDrErrFrr4SubgraphCallablerq)rrrrrIrJrKrLrMrNrOrPrQrRZntasksrWZsubgraphoutkeyZ inkeys_setr"inkeysZchain2Zsubchainr$r<r%rush                 rc@seZdZUdZded<ded<ded<ded <eeZd$ddd d d ddZddddZdddddZ dddddZ ddddZ dddd Z d!dd"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, ...]rr2nameNz Iterable[Key]z str | None)rrrrcCs:||_||_t||_|dur0dt|||j}||_dS)Nzsubgraph_callable-)rrr4rr r)rrrrrr$r$r%__init__s  zSubgraphCallable.__init__rcCs|jSrY)rrr$r$r%rszSubgraphCallable.__repr__rbool)otherrcCs<t|t|uo:|j|jko:|j|jko:t|jt|jkSrY)r1rrrr)rrr$r$r%__eq__s  zSubgraphCallable.__eq__)argsrcGsJt|t|jks,tdt|jt|ft|j|jtt|j|S)NzExpected %d args, got %d) r5r ValueErrorrrrrrzip)rrr$r$r%__call__szSubgraphCallable.__call__r4cCst|j|j|j|jffSrY)rrrrrrr$r$r% __reduce__szSubgraphCallable.__reduce__rcCst|jt|j|jfSrY)rvr frozensetrrrr$r$r%__hash__szSubgraphCallable.__hash__objectcCsdt|j|j|j|jfS)Nr)r rrrrrr$r$r%__dask_tokenize__s z"SubgraphCallable.__dask_tokenize__)N)rrr__doc____annotations__r4 __slots__rrrrrrrr$r$r$r%rs  r)NNT)NTN)NFN)rs)+ __future__rrrcollections.abcrenumrtypingrZdaskrrrZ dask.baser r Z dask.corer r r rrrrZ dask.typingrrr&r8rXrZrermrprfr~rrrrrrr$r$r$r%sH    $ 3  C ; ' 1G