<commit_before>import types
import traceback
from PythonQt import QtCore, QtGui
from ddapp.timercallback import TimerCallback
from ddapp.simpletimer import SimpleTimer
from ddapp import callbacks


class AsyncTaskQueue(object):

    TASK_STARTED_SIGNAL = 'TASK_STARTED_SIGNAL'
    TASK_ENDED_SIGNAL = 'TASK_ENDED_SIGNAL'
    TASK_PAUSED_SIGNAL = 'TASK_PAUSED_SIGNAL'
    TASK_FAILED_SIGNAL = 'TASK_FAILED_SIGNAL'
    TASK_EXCEPTION_SIGNAL = 'TASK_EXCEPTION_SIGNAL'

    class PauseException(Exception):
        pass

    class FailException(Exception):
        pass

    def __init__(self):
        self.tasks = []
        self.generators = []
        self.timer = TimerCallback(targetFps=10)
        self.timer.callback = self.callbackLoop
        self.callbacks = callbacks.CallbackRegistry([self.TASK_STARTED_SIGNAL,
                                                     self.TASK_ENDED_SIGNAL,
                                                     self.TASK_PAUSED_SIGNAL,
                                                     self.TASK_FAILED_SIGNAL,
                                                     self.TASK_EXCEPTION_SIGNAL])
        self.currentTask = None
        self.isRunning = False

    def reset(self):
        assert not self.isRunning
        assert not self.generators
        self.tasks = []

    def start(self):
        self.isRunning = True
        self.timer.start()

    def stop(self):
        self.isRunning = False
        self.currentTask = None
        self.generators = []
        self.timer.stop()

    def wrapGenerator(self, generator):
        def generatorWrapper():
            return generator
        return generatorWrapper

    def addTask(self, task):

        if isinstance(task, types.GeneratorType):
            task = self.wrapGenerator(task)

        assert hasattr(task, '__call__')
        self.tasks.append(task)

    def callbackLoop(self):

        try:
            for i in xrange(10):
                self.doWork()
            if not self.tasks:
                self.stop()

        except AsyncTaskQueue.PauseException:
            assert self.currentTask
            self.callbacks.process(self.TASK_PAUSED_SIGNAL, self, self.currentTask)
            self.stop()
        except AsyncTaskQueue.FailException:
            assert self.currentTask
            self.callbacks.process(self.TASK_FAILED_SIGNAL, self, self.currentTask)
            self.stop()
        except:
            assert self.currentTask
            self.callbacks.process(self.TASK_EXCEPTION_SIGNAL, self, self.currentTask)
            self.stop()
            raise

        return self.isRunning

    def popTask(self):
        assert not self.isRunning
        assert not self.currentTask
        if self.tasks:
            self.tasks.pop(0)

    def completePreviousTask(self):
        assert self.currentTask
        self.tasks.remove(self.currentTask)
        self.callbacks.process(self.TASK_ENDED_SIGNAL, self, self.currentTask)
        self.currentTask = None

    def startNextTask(self):
        self.currentTask = self.tasks[0]
        self.callbacks.process(self.TASK_STARTED_SIGNAL, self, self.currentTask)
        result = self.currentTask()
        if isinstance(result, types.GeneratorType):
            self.generators.insert(0, result)

    def doWork(self):
        if self.generators:
            self.handleGenerator(self.generators[0])
        else:
            if self.currentTask:
                self.completePreviousTask()
            if self.tasks:
                self.startNextTask()

    def handleGenerator(self, generator):
        try:
            result = generator.next()
        except StopIteration:
            self.generators.remove(generator)
        else:
            if isinstance(result, types.GeneratorType):
                self.generators.insert(0, result)

    def connectTaskStarted(self, func):
        return self.callbacks.connect(self.TASK_STARTED_SIGNAL, func)

    def disconnectTaskStarted(self, callbackId):
        self.callbacks.disconnect(callbackId)<commit_msg>add queue start/stop signals to AsyncTaskQueue<commit_after>import types
import traceback
from PythonQt import QtCore, QtGui
from ddapp.timercallback import TimerCallback
from ddapp.simpletimer import SimpleTimer
from ddapp import callbacks


class AsyncTaskQueue(object):

    QUEUE_STARTED_SIGNAL = 'QUEUE_STARTED_SIGNAL'
    QUEUE_STOPPED_SIGNAL = 'QUEUE_STOPPED_SIGNAL'
    TASK_STARTED_SIGNAL = 'TASK_STARTED_SIGNAL'
    TASK_ENDED_SIGNAL = 'TASK_ENDED_SIGNAL'
    TASK_PAUSED_SIGNAL = 'TASK_PAUSED_SIGNAL'
    TASK_FAILED_SIGNAL = 'TASK_FAILED_SIGNAL'
    TASK_EXCEPTION_SIGNAL = 'TASK_EXCEPTION_SIGNAL'

    class PauseException(Exception):
        pass

    class FailException(Exception):
        pass

    def __init__(self):
        self.tasks = []
        self.generators = []
        self.timer = TimerCallback(targetFps=10)
        self.timer.callback = self.callbackLoop
        self.callbacks = callbacks.CallbackRegistry([self.QUEUE_STARTED_SIGNAL,
                                                     self.QUEUE_STOPPED_SIGNAL,
                                                     self.TASK_STARTED_SIGNAL,
                                                     self.TASK_ENDED_SIGNAL,
                                                     self.TASK_PAUSED_SIGNAL,
                                                     self.TASK_FAILED_SIGNAL,
                                                     self.TASK_EXCEPTION_SIGNAL])
        self.currentTask = None
        self.isRunning = False

    def reset(self):
        assert not self.isRunning
        assert not self.generators
        self.tasks = []

    def start(self):
        self.isRunning = True
        self.callbacks.process(self.QUEUE_STARTED_SIGNAL, self)
        self.timer.start()

    def stop(self):
        self.isRunning = False
        self.currentTask = None
        self.generators = []
        self.timer.stop()
        self.callbacks.process(self.QUEUE_STOPPED_SIGNAL, self)

    def wrapGenerator(self, generator):
        def generatorWrapper():
            return generator
        return generatorWrapper

    def addTask(self, task):

        if isinstance(task, types.GeneratorType):
            task = self.wrapGenerator(task)

        assert hasattr(task, '__call__')
        self.tasks.append(task)

    def callbackLoop(self):

        try:
            for i in xrange(10):
                self.doWork()
            if not self.tasks:
                self.stop()

        except AsyncTaskQueue.PauseException:
            assert self.currentTask
            self.callbacks.process(self.TASK_PAUSED_SIGNAL, self, self.currentTask)
            self.stop()
        except AsyncTaskQueue.FailException:
            assert self.currentTask
            self.callbacks.process(self.TASK_FAILED_SIGNAL, self, self.currentTask)
            self.stop()
        except:
            assert self.currentTask
            self.callbacks.process(self.TASK_EXCEPTION_SIGNAL, self, self.currentTask)
            self.stop()
            raise

        return self.isRunning

    def popTask(self):
        assert not self.isRunning
        assert not self.currentTask
        if self.tasks:
            self.tasks.pop(0)

    def completePreviousTask(self):
        assert self.currentTask
        self.tasks.remove(self.currentTask)
        self.callbacks.process(self.TASK_ENDED_SIGNAL, self, self.currentTask)
        self.currentTask = None

    def startNextTask(self):
        self.currentTask = self.tasks[0]
        self.callbacks.process(self.TASK_STARTED_SIGNAL, self, self.currentTask)
        result = self.currentTask()
        if isinstance(result, types.GeneratorType):
            self.generators.insert(0, result)

    def doWork(self):
        if self.generators:
            self.handleGenerator(self.generators[0])
        else:
            if self.currentTask:
                self.completePreviousTask()
            if self.tasks:
                self.startNextTask()

    def handleGenerator(self, generator):
        try:
            result = generator.next()
        except StopIteration:
            self.generators.remove(generator)
        else:
            if isinstance(result, types.GeneratorType):
                self.generators.insert(0, result)

    def connectQueueStarted(self, func):
        return self.callbacks.connect(self.QUEUE_STARTED_SIGNAL, func)

    def disconnectQueueStarted(self, callbackId):
        self.callbacks.disconnect(callbackId)

    def connectQueueStopped(self, func):
        return self.callbacks.connect(self.QUEUE_STOPPED_SIGNAL, func)

    def disconnectQueueStopped(self, callbackId):
        self.callbacks.disconnect(callbackId)

    def connectTaskStarted(self, func):
        return self.callbacks.connect(self.TASK_STARTED_SIGNAL, func)

    def disconnectTaskStarted(self, callbackId):
        self.callbacks.disconnect(callbackId)
{'fraction_non_alphanumeric': 0.07692307692307693, 'fraction_numerical': 0.002015451797111186, 'mean_word_length': 2.008420343550017, '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': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7143858', 'n_tokens_mistral': 2489, 'n_tokens_neox': 2408, 'n_words': 454}