<s> from operator import add , mul <EOL> import os <EOL> from time import sleep <EOL> from dask . diagnostics import Profiler , ResourceProfiler , CacheProfiler <EOL> from dask . threaded import get <EOL> from dask . utils import ignoring , tmpfile <EOL> from dask . compatibility import apply <EOL> import pytest <EOL> try : <EOL> import bokeh <EOL> except : <EOL> bokeh = None <EOL> try : <EOL> import psutil <EOL> except : <EOL> psutil = None <EOL> prof = Profiler ( ) <EOL> dsk = { '<STR_LIT:a>' : <NUM_LIT:1> , <EOL> '<STR_LIT:b>' : <NUM_LIT:2> , <EOL> '<STR_LIT:c>' : ( add , '<STR_LIT:a>' , '<STR_LIT:b>' ) , <EOL> '<STR_LIT:d>' : ( mul , '<STR_LIT:a>' , '<STR_LIT:b>' ) , <EOL> '<STR_LIT:e>' : ( mul , '<STR_LIT:c>' , '<STR_LIT:d>' ) } <EOL> dsk2 = { '<STR_LIT:a>' : <NUM_LIT:1> , '<STR_LIT:b>' : <NUM_LIT:2> , '<STR_LIT:c>' : ( lambda a , b : sleep ( <NUM_LIT:0.1> ) or ( a + b ) , '<STR_LIT:a>' , '<STR_LIT:b>' ) } <EOL> def test_profiler ( ) : <EOL> with prof : <EOL> out = get ( dsk , '<STR_LIT:e>' ) <EOL> assert out == <NUM_LIT:6> <EOL> prof_data = sorted ( prof . results , key = lambda d : d . key ) <EOL> keys = [ i . key for i in prof_data ] <EOL> assert keys == [ '<STR_LIT:c>' , '<STR_LIT:d>' , '<STR_LIT:e>' ] <EOL> tasks = [ i . task for i in prof_data ] <EOL> assert tasks == [ ( add , '<STR_LIT:a>' , '<STR_LIT:b>' ) , ( mul , '<STR_LIT:a>' , '<STR_LIT:b>' ) , ( mul , '<STR_LIT:c>' , '<STR_LIT:d>' ) ] <EOL> prof . clear ( ) <EOL> assert prof . results == [ ] <EOL> def test_profiler_works_under_error ( ) : <EOL> div = lambda x , y : x / y <EOL> dsk = { '<STR_LIT:x>' : ( div , <NUM_LIT:1> , <NUM_LIT:1> ) , '<STR_LIT:y>' : ( div , '<STR_LIT:x>' , <NUM_LIT:2> ) , '<STR_LIT:z>' : ( div , '<STR_LIT:y>' , <NUM_LIT:0> ) } <EOL> with ignoring ( ZeroDivisionError ) : <EOL> with prof : <EOL> out = get ( dsk , '<STR_LIT:z>' ) <EOL> assert all ( len ( v ) == <NUM_LIT:5> for v in prof . results ) <EOL> assert len ( prof . results ) == <NUM_LIT:2> <EOL> def test_two_gets ( ) : <EOL> with prof : <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> n = len ( prof . results ) <EOL> dsk2 = { '<STR_LIT:x>' : ( add , <NUM_LIT:1> , <NUM_LIT:2> ) , '<STR_LIT:y>' : ( add , '<STR_LIT:x>' , '<STR_LIT:x>' ) } <EOL> with prof : <EOL> get ( dsk2 , '<STR_LIT:y>' ) <EOL> m = len ( prof . results ) <EOL> with prof : <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> get ( dsk2 , '<STR_LIT:y>' ) <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> assert len ( prof . results ) == n + m + n <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_resource_profiler ( ) : <EOL> with ResourceProfiler ( dt = <NUM_LIT> ) as rprof : <EOL> out = get ( dsk2 , '<STR_LIT:c>' ) <EOL> results = rprof . results <EOL> assert all ( isinstance ( i , tuple ) and len ( i ) == <NUM_LIT:3> for i in results ) <EOL> rprof . clear ( ) <EOL> assert rprof . results == [ ] <EOL> rprof . close ( ) <EOL> assert not rprof . _tracker . is_alive ( ) <EOL> with pytest . raises ( AssertionError ) : <EOL> with rprof : <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> def test_cache_profiler ( ) : <EOL> with CacheProfiler ( ) as cprof : <EOL> out = get ( dsk2 , '<STR_LIT:c>' ) <EOL> results = cprof . results <EOL> assert all ( isinstance ( i , tuple ) and len ( i ) == <NUM_LIT:5> for i in results ) <EOL> cprof . clear ( ) <EOL> assert cprof . results == [ ] <EOL> tics = [ <NUM_LIT:0> ] <EOL> def nbytes ( res ) : <EOL> tics [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> return tics [ <NUM_LIT:0> ] <EOL> with CacheProfiler ( nbytes ) as cprof : <EOL> out = get ( dsk2 , '<STR_LIT:c>' ) <EOL> results = cprof . results <EOL> assert tics [ - <NUM_LIT:1> ] == len ( results ) <EOL> assert tics [ - <NUM_LIT:1> ] == results [ - <NUM_LIT:1> ] . metric <EOL> assert cprof . _metric_name == '<STR_LIT>' <EOL> assert CacheProfiler ( metric = nbytes , metric_name = '<STR_LIT:foo>' ) . _metric_name == '<STR_LIT:foo>' <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_unquote ( ) : <EOL> from dask . diagnostics . profile_visualize import unquote <EOL> from dask . delayed import to_task_dasks <EOL> f = lambda x : to_task_dasks ( x ) [ <NUM_LIT:0> ] <EOL> t = { '<STR_LIT:a>' : <NUM_LIT:1> , '<STR_LIT:b>' : <NUM_LIT:2> , '<STR_LIT:c>' : <NUM_LIT:3> } <EOL> assert unquote ( f ( t ) ) == t <EOL> t = { '<STR_LIT:a>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , '<STR_LIT:b>' : <NUM_LIT:2> , '<STR_LIT:c>' : <NUM_LIT:3> } <EOL> assert unquote ( f ( t ) ) == t <EOL> t = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> assert unquote ( f ( t ) ) == t <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_pprint_task ( ) : <EOL> from dask . diagnostics . profile_visualize import pprint_task <EOL> keys = set ( [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' , '<STR_LIT:d>' , '<STR_LIT:e>' ] ) <EOL> assert pprint_task ( ( add , '<STR_LIT:a>' , <NUM_LIT:1> ) , keys ) == '<STR_LIT>' <EOL> assert pprint_task ( ( add , ( add , '<STR_LIT:a>' , <NUM_LIT:1> ) ) , keys ) == '<STR_LIT>' <EOL> res = '<STR_LIT>' <EOL> assert pprint_task ( ( sum , [ <NUM_LIT:1> , '<STR_LIT:b>' , ( add , '<STR_LIT:a>' , <NUM_LIT:1> ) ] ) , keys ) == res <EOL> assert pprint_task ( ( sum , ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) ) , keys ) == '<STR_LIT>' <EOL> assert len ( pprint_task ( ( sum , list ( keys ) * <NUM_LIT:100> ) , keys ) ) < <NUM_LIT:100> <EOL> assert pprint_task ( ( sum , list ( keys ) * <NUM_LIT:100> ) , keys ) == '<STR_LIT>' <EOL> assert pprint_task ( ( sum , [ <NUM_LIT:1> , <NUM_LIT:2> , ( sum , [ '<STR_LIT:a>' , <NUM_LIT:4> ] ) , <NUM_LIT:5> , <NUM_LIT:6> ] * <NUM_LIT:100> ) , keys ) == '<STR_LIT>' <EOL> assert pprint_task ( ( sum , [ <NUM_LIT:1> , <NUM_LIT:2> , ( sum , [ '<STR_LIT:a>' , ( sum , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ] ) , <NUM_LIT:5> , <NUM_LIT:6> ] ) , <EOL> keys ) == '<STR_LIT>' <EOL> def foo ( w , x , y = ( ) , z = <NUM_LIT:3> ) : <EOL> return w + x + sum ( y ) + z <EOL> task = ( apply , foo , ( tuple , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) , <EOL> ( dict , [ [ '<STR_LIT:y>' , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ] , [ '<STR_LIT:z>' , '<STR_LIT:c>' ] ] ) ) <EOL> assert pprint_task ( task , keys ) == '<STR_LIT>' <EOL> task = ( apply , foo , ( tuple , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) , <EOL> ( dict , [ [ '<STR_LIT:y>' , [ '<STR_LIT:a>' , <NUM_LIT:1> ] ] , [ '<STR_LIT:z>' , <NUM_LIT:1> ] ] ) ) <EOL> assert pprint_task ( task , keys ) == '<STR_LIT>' <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_profiler_plot ( ) : <EOL> with prof : <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> p = prof . visualize ( plot_width = <NUM_LIT> , <EOL> plot_height = <NUM_LIT> , <EOL> tools = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> show = False , save = False ) <EOL> assert p . plot_width == <NUM_LIT> <EOL> assert p . plot_height == <NUM_LIT> <EOL> assert len ( p . tools ) == <NUM_LIT:1> <EOL> assert isinstance ( p . tools [ <NUM_LIT:0> ] , bokeh . models . HoverTool ) <EOL> assert p . title == "<STR_LIT>" <EOL> prof . clear ( ) <EOL> prof . visualize ( show = False , save = False ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_resource_profiler_plot ( ) : <EOL> with ResourceProfiler ( dt = <NUM_LIT> ) as rprof : <EOL> get ( dsk2 , '<STR_LIT:c>' ) <EOL> p = rprof . visualize ( plot_width = <NUM_LIT> , <EOL> plot_height = <NUM_LIT> , <EOL> tools = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> show = False , save = False ) <EOL> assert p . plot_width == <NUM_LIT> <EOL> assert p . plot_height == <NUM_LIT> <EOL> assert len ( p . tools ) == <NUM_LIT:1> <EOL> assert isinstance ( p . tools [ <NUM_LIT:0> ] , bokeh . models . HoverTool ) <EOL> assert p . title == "<STR_LIT>" <EOL> rprof . clear ( ) <EOL> rprof . visualize ( show = False , save = False ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_cache_profiler_plot ( ) : <EOL> with CacheProfiler ( metric_name = '<STR_LIT>' ) as cprof : <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> p = cprof . visualize ( plot_width = <NUM_LIT> , <EOL> plot_height = <NUM_LIT> , <EOL> tools = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> show = False , save = False ) <EOL> assert p . plot_width == <NUM_LIT> <EOL> assert p . plot_height == <NUM_LIT> <EOL> assert len ( p . tools ) == <NUM_LIT:1> <EOL> assert isinstance ( p . tools [ <NUM_LIT:0> ] , bokeh . models . HoverTool ) <EOL> assert p . title == "<STR_LIT>" <EOL> assert p . axis [ <NUM_LIT:1> ] . axis_label == '<STR_LIT>' <EOL> cprof . clear ( ) <EOL> cprof . visualize ( show = False , save = False ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_plot_multiple ( ) : <EOL> from dask . diagnostics . profile_visualize import visualize <EOL> from bokeh . models import GridPlot <EOL> with ResourceProfiler ( dt = <NUM_LIT> ) as rprof : <EOL> with prof : <EOL> get ( dsk2 , '<STR_LIT:c>' ) <EOL> p = visualize ( [ prof , rprof ] , label_size = <NUM_LIT:50> , <EOL> title = "<STR_LIT>" , show = False , save = False ) <EOL> assert isinstance ( p , GridPlot ) <EOL> assert len ( p . children ) == <NUM_LIT:2> <EOL> assert p . children [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . title == "<STR_LIT>" <EOL> assert p . children [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . xaxis [ <NUM_LIT:0> ] . axis_label is None <EOL> assert p . children [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] . title is None <EOL> assert p . children [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] . xaxis [ <NUM_LIT:0> ] . axis_label == '<STR_LIT>' <EOL> prof . clear ( ) <EOL> rprof . clear ( ) <EOL> visualize ( [ prof , rprof ] , show = False , save = False ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_saves_file ( ) : <EOL> with tmpfile ( '<STR_LIT:html>' ) as fn : <EOL> with prof : <EOL> get ( dsk , '<STR_LIT:e>' ) <EOL> prof . visualize ( show = False , file_path = fn ) <EOL> assert os . path . exists ( fn ) <EOL> with open ( fn ) as f : <EOL> assert '<STR_LIT:html>' in f . read ( ) . lower ( ) <EOL> @ pytest . mark . skipif ( "<STR_LIT>" ) <EOL> def test_get_colors ( ) : <EOL> from dask . diagnostics . profile_visualize import get_colors <EOL> from bokeh . palettes import Blues9 , Blues5 , BrBG3 <EOL> from itertools import cycle <EOL> funcs = list ( range ( <NUM_LIT:11> ) ) <EOL> cmap = get_colors ( '<STR_LIT>' , funcs ) <EOL> lk = dict ( zip ( funcs , cycle ( Blues9 ) ) ) <EOL> assert cmap == [ lk [ i ] for i in funcs ] <EOL> funcs = list ( range ( <NUM_LIT:5> ) ) <EOL> cmap = get_colors ( '<STR_LIT>' , funcs ) <EOL> lk = dict ( zip ( funcs , Blues5 ) ) <EOL> assert cmap == [ lk [ i ] for i in funcs ] <EOL> funcs = [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> ] <EOL> cmap = get_colors ( '<STR_LIT>' , funcs ) <EOL> lk = dict ( zip ( [ <NUM_LIT:0> , <NUM_LIT:1> ] , BrBG3 ) ) <EOL> assert cmap == [ lk [ i ] for i in funcs ] </s>
