#NEWLINE# @ dispatch ( Expr , h5py . Dataset ) #NEWLINE# def compute_down( expr , data , map = None , **kwargs ) : #NEWLINE# #INDENT# ' Compute expressions on H5Py datasets by operating on chunks\n\n    This uses blaze.expr.split to break a full-array-computation into a\n    per-chunk computation and a on-aggregate computation.\n\n    This uses blaze.partition to pick out chunks from the h5py dataset, uses\n    compute(numpy) to compute on each chunk and then uses blaze.partition to\n    aggregate these (hopefully smaller) intermediate results into a local\n    numpy array.  It then performs a second operation (again given by\n    blaze.expr.split) on this intermediate aggregate\n\n    The expression must contain some sort of Reduction.  Both the intermediate\n    result and the final result are assumed to fit into memory\n    ' #NEWLINE# map = _get_map ( map ) #NEWLINE# leaf = expr . _leaves ( ) [ 0 ] #NEWLINE# if ( not any ( ( isinstance ( node , Reduction ) for node in path ( expr , leaf ) ) ) ) : #NEWLINE# #INDENT# raise MDNotImplementedError ( ) #NEWLINE# #UNINDENT# chunksize = kwargs . get ( 'chunksize' , data . chunks ) #NEWLINE# chunk = symbol ( 'chunk' , DataShape ( * ( chunksize + ( leaf . dshape . measure , ) ) ) ) #NEWLINE# ( ( chunk , chunk_expr ) , ( agg , agg_expr ) ) = split ( leaf , expr , chunk = chunk ) #NEWLINE# ( shape , dtype ) = to_numpy ( agg . dshape ) #NEWLINE# intermediate = np . empty ( shape = shape , dtype = dtype ) #NEWLINE# source_parts = list ( partitions ( data , chunksize = chunksize , keepdims = True ) ) #NEWLINE# target_parts = list ( partitions ( intermediate , chunksize = chunk_expr . shape , keepdims = True ) ) #NEWLINE# list ( map ( curry ( compute_chunk , data , intermediate , chunk , data ) , zip ( source_parts , target_parts ) ) ) #NEWLINE# return compute ( agg_expr , { #NEWLINE# #INDENT# agg : intermediate , #NEWLINE# #UNINDENT# } , return_type = 'native' )