<commit_before># -*- coding: utf-8 -*-
from __future__ import print_function

import csv
import multiprocessing
import os
import platform
import signal
import sys
import threading
from multiprocessing.managers import SyncManager
from time import time

import requests
import six
from six.moves import queue

from datarobot_batch_scoring import __version__
from datarobot_batch_scoring.consts import (WriterQueueMsg,
                                            ProgressQueueMsg,
                                            SENTINEL,
                                            REPORT_INTERVAL)
from datarobot_batch_scoring.network import Network
from datarobot_batch_scoring.reader import (fast_to_csv_chunk,
                                            slow_to_csv_chunk, peek_row,
                                            Shovel, auto_sampler,
                                            investigate_encoding_and_dialect,
                                            decode_reader_state)
from datarobot_batch_scoring.utils import (acquire_api_token,
                                           make_validation_call)
from datarobot_batch_scoring.writer import (WriterProcess, RunContext,
                                            decode_writer_state)

if six.PY2:  # pragma: no cover
    from contextlib2 import ExitStack
elif six.PY3:  # pragma: no cover
    from contextlib import ExitStack


MAX_BATCH_SIZE = 5 * 1024 ** 2


def manager_init():
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)


def format_usage(rusage):
    if not rusage:
        return ""
    else:
        return " User time: {utime:.3f} System time: {stime:.3f} " \
               "RSS: {rss}".format(**rusage)


def run_batch_predictions(base_url, base_headers, user, pwd,
                          api_token, create_api_token,
                          pid, lid, import_id, n_retry, concurrent,
                          resume, n_samples,
                          out_file, keep_cols, delimiter,
                          dataset, pred_name,
                          timeout, ui, fast_mode, auto_sample,
                          dry_run, encoding, skip_dialect,
                          skip_row_id=False,
                          output_delimiter=None,
                          max_batch_size=None, compression=None,
                          field_size_limit=None):

    if field_size_limit is not None:
        csv.field_size_limit(field_size_limit)

    if max_batch_size is None:
        max_batch_size = MAX_BATCH_SIZE

    multiprocessing.freeze_support()
    t1 = time()
    queue_size = concurrent * 2
    #  provide version info and system info in user-agent
    base_headers['User-Agent'] = 'datarobot_batch_scoring/{}|' \
                                 'Python/{}|{}|system/{}|concurrency/{}' \
                                 ''.format(__version__,
                                           sys.version.split(' ')[0],
                                           requests.utils.default_user_agent(),
                                           platform.system(),
                                           concurrent)

    with ExitStack() as stack:
        if os.name is 'nt':
            #  Windows requires an additional manager process. The locks
            #  and queues it creates are proxies for objects that exist within
            #  the manager itself. It does not perform as well so we only
            #  use it when necessary.

            manager = SyncManager()
            manager.start(initializer=manager_init)

            conc_manager = stack.enter_context(manager)
        else:
            #  You're on a nix of some sort and don't need a manager process.
            conc_manager = multiprocessing
        network_queue = conc_manager.Queue(queue_size)
        network_deque = conc_manager.Queue(queue_size)
        writer_queue = conc_manager.Queue(queue_size)
        progress_queue = conc_manager.Queue()

        shovel_status = conc_manager.Value('c', b'-', lock=False)
        network_status = conc_manager.Value('c', b'-', lock=False)
        writer_status = conc_manager.Value('c', b'-', lock=False)
        abort_flag = conc_manager.Value('b', 0, lock=False)

        base_headers['content-type'] = 'text/csv; charset=utf8'
        if compression:
            base_headers['Content-Encoding'] = 'gzip'
        if import_id:
            endpoint = base_url + '/'.join((import_id, 'predict'))
        else:
            endpoint = base_url + '/'.join((pid, lid, 'predict'))
        encoding = investigate_encoding_and_dialect(
            dataset=dataset,
            sep=delimiter, ui=ui,
            fast=fast_mode,
            encoding=encoding,
            skip_dialect=skip_dialect,
            output_delimiter=output_delimiter)
        if auto_sample:
            #  override n_sample
            n_samples = auto_sampler(dataset, encoding, ui)
            ui.info('auto_sample: will use batches of {} rows'
                    ''.format(n_samples))
        # Make a sync request to check authentication and fail early
        first_row = peek_row(dataset, delimiter, ui, fast_mode, encoding)
        ui.debug('First row for auth request: {}'.format(first_row))
        if fast_mode:
            chunk_formatter = fast_to_csv_chunk
        else:
            chunk_formatter = slow_to_csv_chunk
        first_row_data = chunk_formatter(first_row.data, first_row.fieldnames)
        first_row = first_row._replace(data=first_row_data)

        if keep_cols:
            if not all(c in first_row.fieldnames for c in keep_cols):
                ui.fatal('keep_cols "{}" not in columns {}.'.format(
                    [c for c in keep_cols if c not in first_row.fieldnames],
                    first_row.fieldnames))

        if not dry_run:
<commit_msg>Use set operations to perform difference (faster on large column count)<commit_after># -*- coding: utf-8 -*-
from __future__ import print_function

import csv
import multiprocessing
import os
import platform
import signal
import sys
import threading
from multiprocessing.managers import SyncManager
from time import time

import requests
import six
from six.moves import queue

from datarobot_batch_scoring import __version__
from datarobot_batch_scoring.consts import (WriterQueueMsg,
                                            ProgressQueueMsg,
                                            SENTINEL,
                                            REPORT_INTERVAL)
from datarobot_batch_scoring.network import Network
from datarobot_batch_scoring.reader import (fast_to_csv_chunk,
                                            slow_to_csv_chunk, peek_row,
                                            Shovel, auto_sampler,
                                            investigate_encoding_and_dialect,
                                            decode_reader_state)
from datarobot_batch_scoring.utils import (acquire_api_token,
                                           make_validation_call)
from datarobot_batch_scoring.writer import (WriterProcess, RunContext,
                                            decode_writer_state)

if six.PY2:  # pragma: no cover
    from contextlib2 import ExitStack
elif six.PY3:  # pragma: no cover
    from contextlib import ExitStack


MAX_BATCH_SIZE = 5 * 1024 ** 2


def manager_init():
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)


def format_usage(rusage):
    if not rusage:
        return ""
    else:
        return " User time: {utime:.3f} System time: {stime:.3f} " \
               "RSS: {rss}".format(**rusage)


def run_batch_predictions(base_url, base_headers, user, pwd,
                          api_token, create_api_token,
                          pid, lid, import_id, n_retry, concurrent,
                          resume, n_samples,
                          out_file, keep_cols, delimiter,
                          dataset, pred_name,
                          timeout, ui, fast_mode, auto_sample,
                          dry_run, encoding, skip_dialect,
                          skip_row_id=False,
                          output_delimiter=None,
                          max_batch_size=None, compression=None,
                          field_size_limit=None):

    if field_size_limit is not None:
        csv.field_size_limit(field_size_limit)

    if max_batch_size is None:
        max_batch_size = MAX_BATCH_SIZE

    multiprocessing.freeze_support()
    t1 = time()
    queue_size = concurrent * 2
    #  provide version info and system info in user-agent
    base_headers['User-Agent'] = 'datarobot_batch_scoring/{}|' \
                                 'Python/{}|{}|system/{}|concurrency/{}' \
                                 ''.format(__version__,
                                           sys.version.split(' ')[0],
                                           requests.utils.default_user_agent(),
                                           platform.system(),
                                           concurrent)

    with ExitStack() as stack:
        if os.name is 'nt':
            #  Windows requires an additional manager process. The locks
            #  and queues it creates are proxies for objects that exist within
            #  the manager itself. It does not perform as well so we only
            #  use it when necessary.

            manager = SyncManager()
            manager.start(initializer=manager_init)

            conc_manager = stack.enter_context(manager)
        else:
            #  You're on a nix of some sort and don't need a manager process.
            conc_manager = multiprocessing
        network_queue = conc_manager.Queue(queue_size)
        network_deque = conc_manager.Queue(queue_size)
        writer_queue = conc_manager.Queue(queue_size)
        progress_queue = conc_manager.Queue()

        shovel_status = conc_manager.Value('c', b'-', lock=False)
        network_status = conc_manager.Value('c', b'-', lock=False)
        writer_status = conc_manager.Value('c', b'-', lock=False)
        abort_flag = conc_manager.Value('b', 0, lock=False)

        base_headers['content-type'] = 'text/csv; charset=utf8'
        if compression:
            base_headers['Content-Encoding'] = 'gzip'
        if import_id:
            endpoint = base_url + '/'.join((import_id, 'predict'))
        else:
            endpoint = base_url + '/'.join((pid, lid, 'predict'))
        encoding = investigate_encoding_and_dialect(
            dataset=dataset,
            sep=delimiter, ui=ui,
            fast=fast_mode,
            encoding=encoding,
            skip_dialect=skip_dialect,
            output_delimiter=output_delimiter)
        if auto_sample:
            #  override n_sample
            n_samples = auto_sampler(dataset, encoding, ui)
            ui.info('auto_sample: will use batches of {} rows'
                    ''.format(n_samples))
        # Make a sync request to check authentication and fail early
        first_row = peek_row(dataset, delimiter, ui, fast_mode, encoding)
        ui.debug('First row for auth request: {}'.format(first_row))
        if fast_mode:
            chunk_formatter = fast_to_csv_chunk
        else:
            chunk_formatter = slow_to_csv_chunk
        first_row_data = chunk_formatter(first_row.data, first_row.fieldnames)
        first_row = first_row._replace(data=first_row_data)

        if keep_cols:
            # If any columns appear in `keep_cols` that are not in
            # `first_row.fieldnames`, it is a fatal error.
            extra_cols = set(keep_cols) - set(first_row.fieldnames)
            if extra_cols:
                u.fatal('keep_cols "{}" not in columns {}.'.format(
                    list(sorted(extra_cols)),
                    first_row.fieldnames))

        if not dry_run:

{'fraction_non_alphanumeric': 0.07556717550813866, 'fraction_numerical': 0.00286750442776419, 'mean_word_length': 1.7538318625174176, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 56, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2125861', 'n_tokens_mistral': 3190, 'n_tokens_neox': 3069, 'n_words': 769}