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
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