ó ßüÚ\c@`s ddlmZmZmZddlmZddlZddlmZddl m Z m Z ddl m Z ddlmZd d lmZmZmZmZejeƒZed „Zd efd „ƒYZdefd„ƒYZd„Zdefd„ƒYZdefd„ƒYZdS(i(tprint_functiontdivisiontabsolute_import(t defaultdictN(t default_timer(tgroupbytvalmap(tgeni(tSchedulerPlugini(t key_splittkey_split_groupt log_errorsttokeycC`sºtƒ}tƒ}t|ƒ}x|r¯|jƒ}|j}||krNq!n| rd|jrdq!n|jdk r|j|ƒ|sq!qn|j|ƒ|j|j ƒq!W||fS(s¢ All keys that need to compute for these keys to finish. If *complete* is false, omit tasks that are busy processing or have finished executing. N( tsettlisttpoptkeytwho_hast exceptiontNonetaddtextendt dependencies(ttaskstcompletetoutterrorststackttsR((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pytdependent_keyss"          tProgresscB`sSeZdZdded„Zejd„ƒZd„Zd„Z ddd„Z RS( s­ Tracks progress of a set of keys or futures On creation we provide a set of keys or futures that interest us as well as a scheduler. We traverse through the scheduler's dependencies to find all relevant keys on which our keys depend. We then plug into the scheduler to learn when our keys become available in memory at which point we record their completion. State ----- keys: set Set of keys that are not yet computed all_keys: set Set of all keys that we track This class performs no visualization. However it is used by other classes, notably TextProgressBar and ProgressWidget, which do perform visualization. igš™™™™™¹?cC`sd„|Dƒ|_d„|jDƒ|_||_||_||_||_d|_tƒ|_t|_ d|_ i|_ dS(NcS`s.h|]$}t|dƒr$|jn|’qS(R(thasattrR(t.0tk((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pys As cS`sh|]}t|ƒ’qS((R (R R!((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pys Bs i( tkeyst schedulerRt_minimumt_dtt last_durationRt _start_timetFalset_runningRtstatustextra(tselfR"R#tminimumtdtR((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyt__init__@s        cc`sI|j}x'|j|jjƒs2tjdƒVq Wg|D]}|jj|^q:}d|_|jj|ƒt|d|j ƒ\|_ }|j s«|j j ƒ|_nt|dt ƒ\|_}|j j |ƒ|j||j @O_|js|jddddƒntjdƒx'|D]}|j|dddtƒq"WdS(Ngš™™™™™©?RRRsSet up Progress keysterred(R"tissubsetR#RRtsleepRt add_pluginRRtall_keystcopyR(tupdatetstoptloggertdebugt transitiontTrue(R,R"R!RRt_((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pytsetupMs" #     cO`sí||jkr`|dkr`|dkr`tjd|ƒ|jj|ƒ|js`|jƒq`n||jkr¥|dkr¥tjdƒ|jd|dd|ƒn||jkré|dkrétjd |ƒ|jdtd|ƒndS( Nt processingtmemorysProgress sees key %sR0sProgress sees task erredRRt forgottens,A task was cancelled (%s), stopping progress(R"R8R9tremoveR7R4R;(R,Rtstarttfinishtargstkwargs((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR:is'  cC`s|jƒdS(N(R7(R,R#((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pytrestartyscC`s|||jjkr(|jjj|ƒn|rbd|_|jji|jj|d6|d6ƒn d|_tjdƒdS(NterrorRRtfinishedsRemove Progress plugin( R#tpluginsRAR*R+R6t exceptionsR8R9(R,RR((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR7|s + N( t__name__t __module__t__doc__R(R/Rt coroutineR=R:RFRR7(((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR,s    t MultiProgresscB`sAeZdZdedded„Zejd„ƒZ d„Z RS(sº Progress variant that keeps track of different groups of keys See Progress for most details. This only adds a function ``func=`` that splits keys. This defaults to ``key_split`` which aligns with naming conventions chosen in the dask project (tuples, hyphens, etc..) State ----- keys: dict Maps group name to set of not-yet-complete keys for that group all_keys: dict Maps group name to set of all keys for that group Examples -------- >>> split = lambda s: s.split('-')[0] >>> p = MultiProgress(['y-2'], func=split) # doctest: +SKIP >>> p.keys # doctest: +SKIP {'x': {'x-1', 'x-2', 'x-3'}, 'y': {'y-1', 'y-2'}} igš™™™™™¹?c C`s2||_tj|||d|d|d|ƒdS(NR-R.R(tfuncRR/(R,R"R#RPR-R.R((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR/žs cc`sÁ|j}x'|j|jjƒs2tjdƒVq Wg|D]}|jj|^q:}d|_|jj|ƒt|d|j ƒ\|_ }|j s«|j j ƒ|_nt|dt ƒ\|_}|j j |ƒ|j||j @O_|js|jddddƒnttt|j|jƒƒ|_ttt|j|j ƒƒ|_ x3|j D](}||jkrZtƒ|j|R?R0sProgress sees task erredRRR@s,A task was cancelled (%s), stopping progress( R"tgetRPRRAtanytvaluesR7R8R9R4R;(R,RRBRCRDREtsR!((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR:Ès    "% "N( RKRLRMRR R(R/RRNR=R:(((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyRO‡s"cC`sjt|dƒ\}}t|dƒ\}}|rCdj|||ƒS|rYdj||ƒSdj|ƒSdS(s¹Format seconds into a human readable form. >>> format_time(10.4) '10.4s' >>> format_time(1000.4) '16min 40.4s' >>> format_time(100000.4) '27hr 46min 40.4s' i<s {0:2.0f}hr {1:2.0f}min {2:4.1f}ss{0:2.0f}min {1:4.1f}ss {0:4.1f}sN(tdivmodtformat(tttmRTth((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyt format_timeÞs t AllProgresscB`s)eZdZd„Zd„Zd„ZRS(s. Keep track of all keys, grouped by key_split cC`sÓttƒ|_td„ƒ|_td„ƒ|_||_xƒ|jjjƒD]o}|j}|j }|j|j |ƒ|j|j|j |ƒ|jdk rO|j|c|j7÷tcS`s ttƒS(N(RR (((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR\øR]( RR talltnbyteststateR#RRSRtprefixRRR3(R,R#RRRa((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyR/õs   c O`sH|jj|}|j}|j|j|ƒy|j||j|ƒWntk r\nX|dkrˆ|j|c|jp‚d8cS`sh|]}t|jƒ’qS((R R(R tdts((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pys Cs ( R R#RR R"RgR_RhRiRR RR(R,RR!RtgRntd((s?lib/python2.7/site-packages/distributed/diagnostics/progress.pyRk;s )   c O`stƒ€|jj|}t|ƒ}||jkrH|j||ƒn|j|}||j|kr|j|j|ƒn||cd8<|dkr®||cd7s " [W 2