<commit_before>from treeCl.constants import PARALLEL_PROFILE
from treeCl.utils import setup_progressbar

__author__ = 'kgori'


def async_avail():
    from IPython import parallel
    try:
        client = parallel.Client(PARALLEL_PROFILE)
        return len(client) > 0
    except IOError:
        return False
    except Exception:
        return False


def get_client():
    from IPython import parallel
    try:
        client = parallel.Client(profile=PARALLEL_PROFILE)
        return client if len(client) > 0 else None
    except IOError:
        return None
    except Exception:
        return None


def parallel_map(client, task, args, message, batchsize=1):
    """
    Helper to map a function over a sequence of inputs, in parallel, with progress meter.
    :param client: IPython.parallel.Client instance
    :param task: Function
    :param args: Must be a list of tuples of arguments that the task function will be mapped onto.
                 If the function takes a single argument, it still must be a 1-tuple.
    :param message: String for progress bar
    :param batchsize: Jobs are shipped in batches of this size. Higher numbers mean less network traffic,
                      but longer execution time per job.
    :return: IPython.parallel.AsyncMapResult
    """
    njobs = len(args)
    nproc = len(client)
    view = client.load_balanced_view()
    message += ' ({} proc)'.format(nproc)
    pbar = setup_progressbar(message, njobs, simple_progress=True)
    pbar.start()
    map_result = view.map(task, *list(zip(*args)), chunksize=batchsize)
    while not map_result.ready():
        map_result.wait(1)
        pbar.update(map_result.progress)
    pbar.finish()
    return map_result


def sequential_map(task, args, message):
    """
    Helper to map a function over a sequence of inputs, sequentially, with progress meter.<commit_msg>Add writing results to disk as they are generated as an option to pll_task<commit_after>from treeCl.constants import PARALLEL_PROFILE
from treeCl.utils import setup_progressbar

__author__ = 'kgori'


def async_avail():
    from IPython import parallel
    try:
        client = parallel.Client(PARALLEL_PROFILE)
        return len(client) > 0
    except IOError:
        return False
    except Exception:
        return False


def get_client():
    from IPython import parallel
    try:
        client = parallel.Client(profile=PARALLEL_PROFILE)
        return client if len(client) > 0 else None
    except IOError:
        return None
    except Exception:
        return None


def parallel_map(client, task, args, message, batchsize=1, background=False):
    """
    Helper to map a function over a sequence of inputs, in parallel, with progress meter.
    :param client: IPython.parallel.Client instance
    :param task: Function
    :param args: Must be a list of tuples of arguments that the task function will be mapped onto.
                 If the function takes a single argument, it still must be a 1-tuple.
    :param message: String for progress bar
    :param batchsize: Jobs are shipped in batches of this size. Higher numbers mean less network traffic,
                      but longer execution time per job.
    :return: IPython.parallel.AsyncMapResult
    """
    njobs = len(args)
    nproc = len(client)
    view = client.load_balanced_view()
    message += ' ({} proc)'.format(nproc)
    pbar = setup_progressbar(message, njobs, simple_progress=True)
    if not background:
        pbar.start()
    map_result = view.map(task, *list(zip(*args)), chunksize=batchsize)
    if background:
        return map_result
    while not map_result.ready():
        map_result.wait(1)
        pbar.update(map_result.progress)
    pbar.finish()
    return map_result


def sequential_map(task, args, message):
    """
    Helper to map a function over a sequence of inputs, sequentially, with progress meter.
{'fraction_non_alphanumeric': 0.07453096890259574, 'fraction_numerical': 0.0025700334104343356, 'mean_word_length': 2.9714285714285715, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5253284', 'n_tokens_mistral': 1111, 'n_tokens_neox': 1080, 'n_words': 392}