σ ίόΪ\c@`s}ddlmZmZmZddlZddlmZmZddlm Z ddl m Z ddl m Z mZdd lmZdd lmZmZdd lmZejeƒZd „Ze jd „ƒZd„Zddd„Zd„Zidd6dd6dd6dd6ed6Zidd6dd6dd6dd6dd6Z idd6dd6dd6dd6d d6Z!d!„Z"dS("i(tprint_functiontdivisiontabsolute_importN(tvalmaptmerge(tgeni(t AllProgressi(tconnecttcoerce_to_address(t Scheduler(t key_splittcolor_of(tdumps_functionc`sFtittˆjƒd6ˆjd6‡fd†ddddgDƒƒS(Ntalltnbytesc`s)i|]}ttˆj|ƒ|“qS((Rtlentstate(t.0R(t allprogress(sFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pys s tmemoryterredtreleasedt processing(RRRR R(t schedulerR((RsFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pytcountss  cc`srt|ƒ}t|ƒV}|jidd6ttƒd6ttƒd6|d6ttjƒd6ƒVtj |ƒ‚dS(s# Open a TCP connection to scheduler, receive progress messages The messages coming back are dicts containing counts of key groups:: {'inc': {'all': 5, 'memory': 2, 'erred': 0, 'released': 1}, 'dec': {'all': 1, 'memory': 0, 'erred': 0, 'released': 0}} Parameters ---------- address: address of scheduler interval: time between batches, in seconds Examples -------- >>> stream = yield eventstream('127.0.0.1:8786', 0.100) # doctest: +SKIP >>> print(yield read(stream)) # doctest: +SKIP tfeedtoptsetuptfunctiontintervaltteardownN( RRtwriteR RRR t remove_pluginRtReturn(taddressRtcomm((sFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pytprogress_streams    cC`svt|jƒƒ}t|ƒ}igd6gd6gd6gd6gd6gd6gd6gd6}|sf|Sd }x|D]ϋ}|}||||}||d }|dj||d ƒ|djt|||d d ƒƒ|dj|ƒ|dj|ƒ|dj|ƒ|djt|ƒƒ|dj|ƒ||d kr]|dj|ƒqs|djdƒqsW|S(sχ Convert nbytes message into rectangle placements >>> nbytes_bar({'inc': 1000, 'dec': 3000}) # doctest: +NORMALIZE_WHITESPACE {'names': ['dec', 'inc'], 'left': [0, 0.75], 'center': [0.375, 0.875], 'right': [0.75, 1.0]} tnamettexttlefttrighttcentertcolortpercenttMBiii@Bidgš™™™™™Ή?t(tsumtvaluestsortedtappendtroundR (RttotaltnamestdR(R%R'R)((sFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pyt nbytes_bar>s:    &iic`sΒd}t|dd|djdtƒ‰ˆ|| ‰tˆƒ}‡fd†|jƒDƒ}ˆ|d>> msg = {'all': {'inc': 5, 'dec': 1, 'add': 4}, ... 'memory': {'inc': 2, 'dec': 0, 'add': 1}, ... 'erred': {'inc': 0, 'dec': 1, 'add': 0}, ... 'released': {'inc': 1, 'dec': 0, 'add': 1}, ... 'processing': {'inc': 1, 'dec': 0, 'add': 2}} >>> progress_quads(msg, nrows=2) # doctest: +SKIP {'name': ['inc', 'add', 'dec'], 'left': [0, 0, 1], 'right': [0.9, 0.9, 1.9], 'top': [0, -1, 0], 'bottom': [-.8, -1.8, -.8], 'released': [1, 1, 0], 'memory': [2, 1, 0], 'erred': [0, 0, 1], 'processing': [1, 0, 2], 'done': ['3 / 5', '2 / 4', '1 / 1'], 'released-loc': [.2/.9, .25 / 0.9, 1], 'memory-loc': [3 / 5 / .9, .5 / 0.9, 1], 'erred-loc': [3 / 5 / .9, .5 / 0.9, 1.9], 'processing-loc': [4 / 5, 1 / 1, 1]}} gΝΜΜΜΜΜμ?R tkeytreversec`s>i|]4\}}gˆD]}|j|dƒ^q|“qS(i(tget(RtktvR%(R4(sFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pys ‰s R%ii s...s show-nameR'R(ttopgš™™™™™ι?tbottomR*s released-locs memory-locs erred-locsprocessing-loctdoneRRRRs%d / %d( R0R9tTrueRtitemstrangeR tzipR1(tmsgtnrowstncolstwidthtnR5R%titrtmtetptatltrltmlteltplR>((R4sFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pytprogress_quadsls<#  ='+(,#     FcC`s2|ddkr*t|dƒ}t|ƒSdSdS(NtstatustOKR7tblack(R R (RCtsplit((sFlib/python2.7/site-packages/distributed/diagnostics/progress_stream.pytcolor_of_message©s tredttransfertoranges disk-writes disk-readtgrayt deserializetcomputegš™™™™™Ω?s transfer-s disk-write-s disk-read-s deserialize-R-c C`ss|d}t|ƒ}|jdgƒ}x>|D]6\}}}t|} t| ƒtk ri| |ƒ} n|dj||ddƒ|djd||ƒ|dj|ƒ|djt||ƒ|dj| ƒ|d jt|ƒ|d j|d ƒd |d |d f} |d j| ƒ| |krPt|ƒd|| sB  ! .=