<commit_before>import asyncio
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError

from datapackage_pipelines.specs.specs import resolve_executor
from datapackage_pipelines.status import status
from ..utilities.extended_json import json

from .runners import runner_config


async def enqueue_errors(step, process, queue):
    out = process.stderr
    while True:
        line = await out.readline()
        if line == b'':
            break
        line = line.decode('utf8').rstrip()
        if len(line) != 0:
            line = "{}: {}".format(step['run'], line)
            logging.info(line)
            await queue.put(line)


async def dequeue_errors(queue, out):
    while True:
        line = await queue.get()
        if line is None:
            break
        out.append(line)
        if len(out) > 1000:
            out.pop(0)


async def collect_stats(infile):
    reader = asyncio.StreamReader()
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await asyncio.get_event_loop() \
        .connect_read_pipe(lambda: reader_protocol, infile)
    count = 0
    dp = None
    stats = None
    while True:
        try:
            line = await reader.readline()
        except ValueError:
            logging.error('Too large stats object!')
            break
        if line == b'':
            break
        stats = line
        if dp is None:
            try:
                dp = json.loads(line.decode('ascii'))
            except JSONDecodeError:
                break
        count += 1

    transport.close()

    if dp is None or count == 0:
        return {}

    try:
        stats = json.loads(stats.decode('ascii'))
    except JSONDecodeError:
        stats = {}

    return stats


def create_process(args, cwd, wfd, rfd):<commit_msg>improve the logging of exception "Too large stats object!" - to provide some more details<commit_after>import asyncio
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError

from datapackage_pipelines.specs.specs import resolve_executor
from datapackage_pipelines.status import status
from ..utilities.extended_json import json

from .runners import runner_config


async def enqueue_errors(step, process, queue):
    out = process.stderr
    while True:
        line = await out.readline()
        if line == b'':
            break
        line = line.decode('utf8').rstrip()
        if len(line) != 0:
            line = "{}: {}".format(step['run'], line)
            logging.info(line)
            await queue.put(line)


async def dequeue_errors(queue, out):
    while True:
        line = await queue.get()
        if line is None:
            break
        out.append(line)
        if len(out) > 1000:
            out.pop(0)


async def collect_stats(infile):
    reader = asyncio.StreamReader()
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await asyncio.get_event_loop() \
        .connect_read_pipe(lambda: reader_protocol, infile)
    count = 0
    dp = None
    stats = None
    while True:
        try:
            line = await reader.readline()
        except ValueError:
            logging.exception('Too large stats object!')
            break
        if line == b'':
            break
        stats = line
        if dp is None:
            try:
                dp = json.loads(line.decode('ascii'))
            except JSONDecodeError:
                break
        count += 1

    transport.close()

    if dp is None or count == 0:
        return {}

    try:
        stats = json.loads(stats.decode('ascii'))
    except JSONDecodeError:
        stats = {}

    return stats


def create_process(args, cwd, wfd, rfd):
{'fraction_non_alphanumeric': 0.09177718832891246, 'fraction_numerical': 0.005305039787798408, 'mean_word_length': 2.1062602965403623, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3614936', 'n_tokens_mistral': 1101, 'n_tokens_neox': 1052, 'n_words': 308}