#NEWLINE# def cull( dsk , keys ) : #NEWLINE# #INDENT# " Return new dask with only the tasks required to calculate keys.\n\n    In other words, remove unnecessary tasks from dask.\n    ``keys`` may be a single key or list of keys.\n\n    Examples\n    --------\n    >>> d = {'x': 1, 'y': (inc, 'x'), 'out': (add, 'x', 10)}\n    >>> cull(d, 'out')  # doctest: +SKIP\n    {'x': 1, 'out': (add, 'x', 10)}\n    " #NEWLINE# if ( not isinstance ( keys , ( list , set ) ) ) : #NEWLINE# #INDENT# keys = [ keys ] #NEWLINE# #UNINDENT# nxt = set ( flatten ( keys ) ) #NEWLINE# seen = nxt #NEWLINE# while nxt : #NEWLINE# #INDENT# cur = nxt #NEWLINE# nxt = set ( ) #NEWLINE# for item in cur : #NEWLINE# #INDENT# for dep in get_dependencies ( dsk , item ) : #NEWLINE# #INDENT# if ( dep not in seen ) : #NEWLINE# #INDENT# nxt . add ( dep ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# seen . update ( nxt ) #NEWLINE# #UNINDENT# return dict ( ( ( k , v ) for ( k , v ) in dsk . items ( ) if ( k in seen ) ) )