turntable/spin.py
def batch(collection, method, processes=None, batch_size=None, quiet=False,
          kwargs_to_dump=None, args=None, **kwargs):
    '''Processes a collection in parallel batches, 
    each batch processes in series on a single process.
    Running batches in parallel can be more effficient that splitting a list across cores as in spin.parallel 
    because of parallel processing has high IO requirements.

    Parameters
    ----------
    collection : list
        i.e. list of Record objects
    method : method to call on each Record
    processes : int
        number of processes to run on [defaults to number of cores on machine]
    batch_size : int
        lenght of each batch [defaults to number of elements / number of processes]

    Returns
    -------
    collection : list
        list of Record objects after going through method called

    Example
    -------
    adding 2 to every number in a range

    >>> import turntable
    >>> collection = range(100)
    >>> def jam(record):
    >>>     return record + 2
    >>> collection = turntable.spin.batch(collection, jam)

    Note
    ----

    lambda functions do not work in parallel

    '''

    if processes is None:
        # default to the number of processes, not exceeding 20 or the number of
        # subjects
        processes = min(mp.cpu_count(), 20, len(collection))

    if batch_size is None:
        # floor divide rounds down to nearest int
        batch_size = max(len(collection) // processes, 1)
    print 'size of each batch =', batch_size

    mod = len(collection) % processes
    # batch_list is a list of cars broken in to batch size chunks
    batch_list = [collection[x:x + batch_size]
                  for x in xrange(0, len(collection) - mod, batch_size)]
    # remainder handling
    if mod != 0:
        batch_list[len(batch_list) - 1] += collection[-mod:]
    print 'number of batches =', len(batch_list)

    # New args
    if args is None:
        args = method
    else:
        if isinstance(args, tuple) == False:
            args = (args,)
        args = (method,) + args

    # Applying the mp method w/ or w/o dumping using the custom operator
    # method
    if kwargs_to_dump is None:
        res = parallel(
            batch_list,
            new_function_batch,
            processes=processes,
            args=args,
            **kwargs)
    else:
        res = process_dump(
            batch_list,
            new_function_batch,
            kwargs_to_dump,
            processes=processes,
            args=args,
            **kwargs)

    returnList = []
    for l in res:
        returnList += l

    # toc = time.time()
    # elapsed = toc-tic
    # if quiet is False:
    # 	if processes is None:
    # 		print "Total Elapsed time: %s  :-)" % str(elapsed)
    # 	else:
    # print "Total Elapsed time: %s  on %s processes :-)" %
    # (str(elapsed),str(processes))

    return returnList
def batch(collection, method, processes=None, batch_size=None, quiet=False,
          kwargs_to_dump=None, args=None, **kwargs):
    '''Processes a collection in parallel batches, 
    each batch processes in series on a single process.
    Running batches in parallel can be more effficient that splitting a list across cores as in spin.parallel 
    because of parallel processing has high IO requirements.

    Parameters
    ----------
    collection : list
        i.e. list of Record objects
    method : method to call on each Record
    processes : int
        number of processes to run on [defaults to number of cores on machine]
    batch_size : int
        lenght of each batch [defaults to number of elements / number of processes]

    Returns
    -------
    collection : list
        list of Record objects after going through method called

    Example
    -------
    adding 2 to every number in a range

    >>> import turntable
    >>> collection = range(100)
    >>> def jam(record):
    >>>     return record + 2
    >>> collection = turntable.spin.batch(collection, jam)

    Note
    ----

    lambda functions do not work in parallel

    '''

    if processes is None:
        # default to the number of processes, not exceeding 20 or the number of
        # subjects
        processes = min(mp.cpu_count(), 20, len(collection))

    if batch_size is None:
        # floor divide rounds down to nearest int
        batch_size = max(len(collection) // processes, 1)
    print 'size of each batch =', batch_size

    mod = len(collection) % processes
    # batch_list is a list of cars broken in to batch size chunks
    batch_list = [collection[x:x + batch_size]
                  for x in xrange(0, len(collection) - mod, batch_size)]
    # remainder handling
    if mod != 0:
        batch_list[len(batch_list) - 1] += collection[-mod:]
    print 'number of batches =', len(batch_list)

    # New args
    if args is None:
        args = method
    else:
        if isinstance(args, tuple) == False:
            args = (args,)
        args = (method,) + args

    # Applying the mp method w/ or w/o dumping using the custom operator
    # method
    if kwargs_to_dump is None:
        res = parallel(
            batch_list,
            new_function_batch,
            processes=processes,
            args=args,
            **kwargs)
    else:
        res = process_dump(
            batch_list,
            new_function_batch,
            kwargs_to_dump,
            processes=processes,
            args=args,
            **kwargs)

    returnList = []
    for l in res:
        returnList += l

    # toc = time.time()
    # elapsed = toc-tic
    # if quiet is False:
    # 	if processes is None:
    # 		print "Total Elapsed time: %s  :-)" % str(elapsed)
    # 	else:
    # print "Total Elapsed time: %s  on %s processes :-)" %
    # (str(elapsed),str(processes))

    return returnList
def batch ( collection , method , processes = None , batch_size = None , quiet = False , kwargs_to_dump = None , args = None , * * kwargs ) : if processes is None : # default to the number of processes, not exceeding 20 or the number of # subjects processes = min ( mp . cpu_count ( ) , 20 , len ( collection ) ) if batch_size is None : # floor divide rounds down to nearest int batch_size = max ( len ( collection ) // processes , 1 ) print 'size of each batch =' , batch_size mod = len ( collection ) % processes # batch_list is a list of cars broken in to batch size chunks batch_list = [ collection [ x : x + batch_size ] for x in xrange ( 0 , len ( collection ) - mod , batch_size ) ] # remainder handling if mod != 0 : batch_list [ len ( batch_list ) - 1 ] += collection [ - mod : ] print 'number of batches =' , len ( batch_list ) # New args if args is None : args = method else : if isinstance ( args , tuple ) == False : args = ( args , ) args = ( method , ) + args # Applying the mp method w/ or w/o dumping using the custom operator # method if kwargs_to_dump is None : res = parallel ( batch_list , new_function_batch , processes = processes , args = args , * * kwargs ) else : res = process_dump ( batch_list , new_function_batch , kwargs_to_dump , processes = processes , args = args , * * kwargs ) returnList = [ ] for l in res : returnList += l # toc = time.time() # elapsed = toc-tic # if quiet is False: # 	if processes is None: # 		print "Total Elapsed time: %s  :-)" % str(elapsed) # 	else: # print "Total Elapsed time: %s  on %s processes :-)" % # (str(elapsed),str(processes)) return returnList
Processes a collection in parallel batches, 
    each batch processes in series on a single process.
    Running batches in parallel can be more effficient that splitting a list across cores as in spin.parallel 
    because of parallel processing has high IO requirements.

    Parameters
    ----------
    collection : list
        i.e. list of Record objects
    method : method to call on each Record
    processes : int
        number of processes to run on [defaults to number of cores on machine]
    batch_size : int
        lenght of each batch [defaults to number of elements / number of processes]

    Returns
    -------
    collection : list
        list of Record objects after going through method called

    Example
    -------
    adding 2 to every number in a range

    >>> import turntable
    >>> collection = range(100)
    >>> def jam(record):
    >>>     return record + 2
    >>> collection = turntable.spin.batch(collection, jam)

    Note
    ----

    lambda functions do not work in parallel
Processes a collection in parallel batches each batch processes in series on a single process . Running batches in parallel can be more effficient that splitting a list across cores as in spin . parallel because of parallel processing has high IO requirements .