ó ¦–Õ\c @`sŽddlmZmZmZddlZddlZddlmZddlm Z m Z ddl m Z ddl m Z mZmZmZmZmZmZddlmZmZd „Zd „Zeeed „Zd „Zeeed „Zeeed„Zd„Z d„Z!d„Z"d„Z#d„Z$eeeeeeeed„Z%d„Z&ej'dƒZ(d„Z)de*fd„ƒYZ+dS(i(tabsolute_importtdivisiontprint_functionN(tgetitemi(tconfigtcore(tunicode(tistasktget_dependenciestsubsttoposorttflattent reverse_dictt ishashable(taddtincc `s t|ttfƒs!|g}ng}tƒ}tƒ}ttt|ƒƒƒ}x¬|rÿg}||7}g|D]!}|tˆ|dtƒf^qq}|j|ƒxN|D]F\} } x7| D]/} | |kr¿|j| ƒ|j | ƒq¿q¿Wq¬W|}qTW‡fd†|Dƒ} | |fS(s 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 -------- >>> d = {'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)} >>> dsk, dependencies = cull(d, 'out') # doctest: +SKIP >>> dsk # doctest: +SKIP {'x': 1, 'out': (add, 'x', 10)} >>> dependencies # doctest: +SKIP {'x': set(), 'out': set(['x'])} Returns ------- dsk: culled dask graph dependencies: Dict mapping {key: [deps]}. Useful side effect to accelerate other optimizations, notably fuse. tas_listc`si|]}ˆ||“qS(((t.0tk(tdsk(s0lib/python2.7/site-packages/dask/optimization.pys 7s ( t isinstancetlisttsettdictR RtTruetupdateRtappend( Rtkeystout_keystseent dependenciestworktnew_workRtdepst_tdeplisttdtout((Rs0lib/python2.7/site-packages/dask/optimization.pytculls(     +     cC`st|dƒ}|tks(|tkrrg|ddd…D]}t|ƒ^q<}|j|dƒdj|ƒS|tkrt|dƒdkrt|ddttfƒrg|ddd…D]}t|ƒ^qÅ}|j|ddƒdj|ƒf|ddSdSdS(sCreate new keys for fused tasksiNiÿÿÿÿt-i( ttypetstrRt key_splitRtjointtupletlenRtNone(Rttyptxtnames((s0lib/python2.7/site-packages/dask/optimization.pyt!default_fused_linear_keys_renamer<s, ",c`s…|dk rLt|tƒ rLt|tƒs7|g}ntt|ƒƒ}n|dkrq‡fd†ˆDƒ}ni}tƒ}xµˆD]­}||}t|ƒdk}xˆ|D]€} |dk rÞ| |krÞ|j| ƒq°| |kr|| =|j| ƒq°|r|j| ƒq°| |kr°||| >> 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: +SKIP {'c': (inc, (inc, 1))} >>> dsk, dependencies = fuse(d, keys=['b'], rename_keys=False) >>> dsk # doctest: +SKIP {'b': (inc, 1), 'c': (inc, 'b')} Returns ------- dsk: output graph with keys fused dependencies: dict mapping dependencies after fusion. Useful side effect to accelerate other downstream optimizations. c`s(i|]}tˆ|dtƒ|“qS(R(RR(RR(R(s0lib/python2.7/site-packages/dask/optimization.pys |s icS`s%i|]\}}t|ƒ|“qS((R(RRtv((s0lib/python2.7/site-packages/dask/optimization.pys ¡s N(R.RRRR R-RRtmaptreversedtitemstpopitemtpopRtreverseRR2tFalseRtremoveR (RRRt rename_keyst child2parentt unfusibletparentR!thas_many_childrentchildtchainst parent2childtchaint key_renamertrvtfusedtaliasest is_renamedtnew_keytvaltkeytold_key((Rs0lib/python2.7/site-packages/dask/optimization.pyt fuse_linearLs¨*                                    " cC`sQ|dkrtƒSt|tƒr&|St|ttfƒsG|g}nt|ƒS(N(R.RRR(R0((s0lib/python2.7/site-packages/dask/optimization.pyt _flat_set×s  c `s¶ˆr@tttˆjƒƒƒtƒr@d„ˆjƒDƒ‰nt|ƒ}ˆdkrq‡fd†ˆDƒ‰n|r |j‡‡fd†ˆjƒDƒƒnt t ‡fd†|Dƒƒdˆƒ}i}xp|D]h}ˆ|}xK|ˆ|@D];}||kr||} n ˆ|} t ||| ƒ}qôW|||>> d = {'x': 1, 'y': (inc, 'x'), 'z': (add, 'x', 'y')} >>> inline(d) # doctest: +SKIP {'x': 1, 'y': (inc, 1), 'z': (add, 1, 'y')} >>> inline(d, keys='y') # doctest: +SKIP {'x': 1, 'y': (inc, 1), 'z': (add, 1, (inc, 1))} >>> inline(d, keys='y', inline_constants=False) # doctest: +SKIP {'x': 1, 'y': (inc, 1), 'z': (add, 'x', (inc, 'x'))} cS`s%i|]\}}t|ƒ|“qS((R(RRR3((s0lib/python2.7/site-packages/dask/optimization.pys õs c`s"i|]}tˆ|ƒ|“qS((R(RR(R(s0lib/python2.7/site-packages/dask/optimization.pys ús c3`sK|]A\}}t|ƒr'|ˆks?ˆ| rt|ƒ r|VqdS(N(R R(RRR3(RR(s0lib/python2.7/site-packages/dask/optimization.pys þs c3`s+|]!}|ˆkr|ˆ|fVqdS(N((RR(R(s0lib/python2.7/site-packages/dask/optimization.pys sRN( RtnexttitertvaluesRR6ROR.RR RR tcopy( RRtinline_constantsRt replaceordertkeysubsRLRKtdeptreplacetdsk2titem((RRs0lib/python2.7/site-packages/dask/optimization.pytinlineás4'    )        c `sõˆs ˆSt|ƒ}tˆƒ‰|dkrG‡fd†ˆDƒ}nt|ƒ}‡fd†}gˆjƒD]@\}}t|ƒro||ro||kro||ƒro|^qo} | rñtˆ| d|d|ƒ‰x| D] }ˆ|=qÝWnˆS(s† Inline cheap functions into larger operations Examples -------- >>> 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} c`s"i|]}tˆ|ƒ|“qS((R(RR(R(s0lib/python2.7/site-packages/dask/optimization.pys :s c`s0yt|ƒjˆƒSWntk r+tSXdS(N(t functions_oftissubsett TypeErrorR:(R3(tfast_functions(s0lib/python2.7/site-packages/dask/optimization.pyt inlinable>s RTRN(RR.R R6RR[( RtoutputR_RTRt dependentsR`RR3R((RR_s0lib/python2.7/site-packages/dask/optimization.pytinline_functionss"      :  cC`s#xt|dƒr|j}qW|S(Ntfunc(thasattrRd(Rd((s0lib/python2.7/site-packages/dask/optimization.pytunwrap_partialOs cC`sŸtƒ}|g}tth}xz|ršg}xa|D]Y}t|ƒ|kr4t|ƒr€|jt|dƒƒ||d7}q||7}q4q4W|}q!W|S(sÒ Set of functions contained within nested task Examples -------- >>> task = (add, (mul, 1, 2), (inc, 3)) # doctest: +SKIP >>> functions_of(task) # doctest: +SKIP set([add, mul, inc]) ii(RRR,R(RRRf(ttasktfuncsRtsequence_typesR ((s0lib/python2.7/site-packages/dask/optimization.pyR\Us       cC`sÌtƒ}x¼|jƒD]®\}}y„t|ƒr›|d|kr›|d|kr›t||dƒr›||dd|kr›||||dƒ||>> def load(store, partition, columns): ... pass >>> dsk = {'x': (load, 'store', 'part', ['a', 'b']), ... 'y': (getitem, 'x', 'a')} >>> merge = lambda t1, t2: (load, t2[1], t2[2], t1[2]) >>> dsk2 = fuse_selections(dsk, getitem, load, merge) >>> cull(dsk2, 'y')[0] {'y': (, 'store', 'part', 'a')} ii(RR6RR^(Rthead1thead2tmergeRYRR3((s0lib/python2.7/site-packages/dask/optimization.pytfuse_selectionsqs ,, c`st|t|‡fd†ƒS(s6 Fuse getitem with lower operation Parameters ---------- dsk: dict dask graph func: function A function in a task to merge place: int Location in task to insert the getitem key Examples -------- >>> def load(store, partition, columns): ... pass >>> dsk = {'x': (load, 'store', 'part', ['a', 'b']), ... 'y': (getitem, 'x', 'a')} >>> dsk2 = fuse_getitem(dsk, load, 3) # columns in arg place 3 >>> cull(dsk2, 'y')[0] {'y': (, 'store', 'part', 'a')} c`s+t|ˆ ƒ|dft|ˆdƒS(Nii(R,(tatb(tplace(s0lib/python2.7/site-packages/dask/optimization.pyt²t(RmR(RRdRp((Rps0lib/python2.7/site-packages/dask/optimization.pyt fuse_getitem›s cC`s$t|ƒ}t|ƒ}t|ƒ}|tks<|tkr‹t|ƒ}d„|Dƒ}|j|ƒt|ƒ}|j|ƒdj |ƒS|t kr t |ƒdkr t |dttfƒr t|ƒ}d„|Dƒ}|j|ƒt|ƒ}|j|dƒdj |ƒf|dSdS(s"Create new keys for ``fuse`` taskscS`sh|]}t|ƒ’qS((R*(RR((s0lib/python2.7/site-packages/dask/optimization.pys ¼s R'icS`sh|]}t|ƒ’qS((R*(RR((s0lib/python2.7/site-packages/dask/optimization.pys Äs iN( R5RPR(R)RR*tdiscardtsortedRR+R,R-R(Rtitt first_keyR/t first_nameR1((s0lib/python2.7/site-packages/dask/optimization.pytdefault_fused_keys_renamerµs$           c @ `s¬ |dk rLt|tƒ rLt|tƒs7|g}ntt|ƒƒ}n|dkrŽtjddƒdkryd}qŽtjddƒ}n|dkrÖtjddƒdkrÁtˆƒ}qÖtjddƒ}n|põtjddƒpõ|d}|p(tjddƒp(d|tj |dƒ}|dkrLtjdt ƒ}n| sZ| rdˆ|fS|dkr…tjdt ƒ}n|t kršt } n|t kr¯d} n|} | dk }|dkræ‡fd †ˆDƒ} n t |ƒ} i} xh| jƒD]Z\} } x;| D]3}|| kr:| g| ||'|kr? ˆ|}.tƒ}:xv|1D]n};|;d }<t|.|<|;dƒ}.||<=|:||<ƒO}:||<ƒ|rX||<dƒ|&j|;dƒqXqXW|!| 8}!|!|:O}!|rþ|&j|ƒ|&||| |>ƒ}?|?dk r5 |?|kr5 ||=||?<|?||=<| |=| |?<|?h| |=*s cS`s.h|]$\}}t|ƒdkr|’qS(i(R-(RRtvals((s0lib/python2.7/site-packages/dask/optimization.pys 7s cs`s'|]}tt|ƒƒdkVqdS(iN(R-R(RR3((s0lib/python2.7/site-packages/dask/optimization.pys ;siiÿÿÿÿiN(R.RRRR RtgetR-tmathtlogR:RRyRR6RtallRRRSR8RR;textendR tinttmintmaxt_inplace_fuse_subgraphs(@RRRt ave_widtht max_widtht max_heighttmax_depth_new_edgesR<R~RER!trdepsRR€R3t reducibleRFt fused_treest info_stacktchildren_stacktdeps_popt reducible_addt reducible_poptreducible_removetfused_trees_poptinfo_stack_appendtinfo_stack_poptchildren_stack_appendtchildren_stack_extendtchildren_stack_popR?RAtchildrent deps_parenttedgest num_childrent child_keyt child_taskt child_keystheighttwidtht num_nodestfudgetchildren_edgestnum_children_edgest no_new_edgesRKtnum_single_nodest max_num_edgest children_infotcur_keytcur_tasktcur_keyst cur_heightt cur_widtht cur_num_nodest cur_fudget cur_edgest children_depst child_infot cur_childtroot_keyt fused_keystalias((Rs0lib/python2.7/site-packages/dask/optimization.pytfuseËsŠ1                  #                      !     !         %       "        $                 -      c`sÂi}tƒ}xµˆD]­}||}t|ƒdk} xˆ|D]€} |dk rm| |krm|j| ƒq?| |kr|| =|j| ƒq?| r¦|j| ƒq?| |kr?||| s ic`si|]}ˆ||“qS(((RR(R(s0lib/python2.7/site-packages/dask/optimization.pys 3s iÿÿÿÿN(RR-R.RR6R7R8RR9RR,tSubgraphCallableR:R…(RRRRR<R=R>R?R!R@RARBRCRDtntasksRLtsubgraphtoutkeyt inkeys_setRtinkeystchain2tsubchain((Rs0lib/python2.7/site-packages/dask/optimization.pyR‰sh                  s[a-f]+cC`s`t|ƒtkr!|jƒ}nt|ƒtkr@|d}ny|jdƒ}|ddjƒs||djdƒ}n |d}xX|dD]L}|jƒrÜt|ƒdkoÇtj |ƒdk rÜ|d|7}q‘Pq‘Wt|ƒdkr t j d|ƒr dS|dd krB|jd ƒjƒdjd ƒd }n|SWnt k r[d SXdS(sQ >>> key_split('x') 'x' >>> key_split('x-1') 'x' >>> key_split('x-1-2-3') 'x' >>> key_split(('x-2', 1)) 'x' >>> key_split("('x-2', 1)") 'x' >>> key_split('hello-world-1') 'hello-world' >>> key_split(b'hello-world-1') 'hello-world' >>> key_split('ae05086432ca935f6eba409a8ecd4896') 'data' >>> key_split('>> key_split(None) 'Other' >>> key_split('x-abcdefab') # ignores hex 'x' >>> key_split('_(x)') # strips unpleasant characters 'x' iR's_'()"iii s [a-f0-9]{32}tdatatt.iÿÿÿÿtOtherN( R(tbytestdecodeR,tsplittisalphatstripR-t hex_patterntmatchR.tret Exception(tstwordstresulttword((s0lib/python2.7/site-packages/dask/optimization.pyR*Qs*  $) R½cB`sMeZdZd Zdd„Zd„Zd„Zd „Zd „Zd „Z RS( sDCreate a callable object from a dask graph. Parameters ---------- dsk : dict A dask graph outkey : hashable 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. RRÀRÂtnametsubgraph_callablecC`s(||_||_||_||_dS(N(RRÀRÂRÖ(tselfRRÀRÂRÖ((s0lib/python2.7/site-packages/dask/optimization.pyt__init__–s   cC`s|jS(N(RÖ(RØ((s0lib/python2.7/site-packages/dask/optimization.pyt__repr__œscC`sjt|ƒt|ƒkoi|j|jkoi|j|jkoit|jƒt|jƒkoi|j|jkS(N(R(RRÀRRÂRÖ(RØtother((s0lib/python2.7/site-packages/dask/optimization.pyt__eq__Ÿs cC`s ||k S(N((RØRÛ((s0lib/python2.7/site-packages/dask/optimization.pyt__ne__¦scG`snt|ƒt|jƒksCtdt|jƒt|ƒfƒ‚ntj|j|jtt|j|ƒƒƒS(NsExpected %d args, got %d( R-RÂt ValueErrorRRRRÀRtzip(RØtargs((s0lib/python2.7/site-packages/dask/optimization.pyt__call__©s "cC`s"t|j|j|j|jffS(N(R½RRÀRÂRÖ(RØ((s0lib/python2.7/site-packages/dask/optimization.pyt __reduce__°s(RRÀRÂRÖ( t__name__t __module__t__doc__t __slots__RÙRÚRÜRÝRáRâ(((s0lib/python2.7/site-packages/dask/optimization.pyR½†s      (,t __future__RRRR‚RÐtoperatorRRrRRt compatibilityRRRR R R R R t utils_testRRR&R2R.RRNROR[R:RcRfR\RmRsRyR¼R‰tcompileRÎR*tobjectR½(((s0lib/python2.7/site-packages/dask/optimization.pyts6  4 . ‹ : 3   *     ÿ9 I 5