task/__init__.py

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

"""
task (Task Module)
==================

**Author:**

* Dirk Alders <sudo-dirk@mount-mockery.de>

**Description:**

    This Module supports helpfull classes for queues, tasks, ...

**Submodules:**

* :class:`task.crontab`
* :class:`task.delayed`
* :class:`task.periodic`
* :class:`task.queue`
* :class:`task.threaded_queue`

**Unittest:**

        See also the :download:`unittest <../../task/_testresults_/unittest.pdf>` documentation.
"""
__DEPENDENCIES__ = []

import logging
import sys
import threading
import time
if sys.version_info >= (3, 0):
    from queue import PriorityQueue
    from queue import Empty
else:
    from Queue import PriorityQueue
    from Queue import Empty

logger_name = 'TASK'
logger = logging.getLogger(logger_name)

__DESCRIPTION__ = """The Module {\\tt %s} is designed to help with task issues like periodic tasks, delayed tasks, queues, threaded queues and crontabs.
For more Information read the documentation.""" % __name__.replace('_', '\_')
"""The Module Description"""
__INTERPRETER__ = (2, 3)
"""The Tested Interpreter-Versions"""


class queue(object):
    """Class to execute queued methods.

    :param bool expire: The default value for expire. See also :py:func:`expire`.

    **Example:**

    .. literalinclude:: ../../task/_examples_/queue.py

    Will result to the following output:

    .. literalinclude:: ../../task/_examples_/queue.log
    """
    class job(object):
        def __init__(self, priority, callback, *args, **kwargs):
            self.priority = priority
            self.callback = callback
            self.args = args
            self.kwargs = kwargs

        def run(self, queue):
            self.callback(queue, *self.args, **self.kwargs)

        def __lt__(self, other):
            return self.priority < other.priority

    def __init__(self, expire=True):
        self.__expire = expire
        self.__stop = False
        self.queue = PriorityQueue()

    def clean_queue(self):
        """
        This Methods removes all jobs from the queue.

        .. note:: Be aware that already runnung jobs will not be terminated.
        """
        while not self.queue.empty():
            try:
                self.queue.get(False)
            except Empty:         # This block is hard to reach for a testcase, but is
                continue          # needed, if the thread runs dry while cleaning the queue.
            self.queue.task_done()

    def enqueue(self, priority, method, *args, **kwargs):
        """
        This enqueues a given callback.

        :param number priority: The priority indication number of this task. The lowest value will be queued first.
        :param method method: Method to be executed
        :param args args: Arguments to be given to method
        :param kwargs kwargs: Kewordsarguments to be given to method

        .. note:: Called method will get this instance as first argument, followed by :py:data:`args` und :py:data:`kwargs`.
        """
        self.queue.put(self.job(priority, method, *args, **kwargs))

    def qsize(self):
        return self.queue.qsize()

    def run(self):
        """
        This starts the execution of the queued methods.
        """
        self.__stop = False
        while not self.__stop:
            try:
                self.queue.get(timeout=0.1).run(self)
            except Empty:
                if self.__expire:
                    break
        if type(self) is threaded_queue:
            self.thread = None

    def expire(self):
        """
        This sets the expire flag. That means that the process will stop after queue gets empty.
        """
        self.__expire = True

    def stop(self):
        """
        This sets the stop flag. That means that the process will stop after finishing the active task.
        """
        self.__stop = True


class threaded_queue(queue):
    """Class to execute queued methods in a background thread (See also parent :py:class:`queue`).

    :param bool expire: The default value for expire. See also :py:func:`queue.expire`.

    **Example:**

    .. literalinclude:: ../../task/_examples_/threaded_queue.py

    Will result to the following output:

    .. literalinclude:: ../../task/_examples_/threaded_queue.log
    """
    def __init__(self, expire=False):
        queue.__init__(self, expire=expire)
        self.thread = None

    def run(self):
        if self.thread is None:
            self.thread = threading.Thread(target=self._start, args=())
            self.thread.daemon = True    # Daemonize thread
            self.thread.start()          # Start the execution

    def join(self):
        """
        This blocks till the queue is empty.

        .. note:: If the queue does not run dry, join will block till the end of the days.
        """
        self.expire()
        if self.thread is not None:
            self.thread.join()

    def stop(self):
        queue.stop(self)
        self.join()

    def _start(self):
        queue.run(self)


class periodic(object):
    """
    :param float cycle_time: Cycle time in seconds -- method will be executed every *cycle_time* seconds
    :param method method: Method to be executed
    :param args args: Arguments to be given to method
    :param kwargs kwargs: Kewordsarguments to be given to method

    Class to execute a method cyclicly.

    .. note:: Called method will get this instance as first argument, followed by :py:data:`args` und :py:data:`kwargs`.

    **Example:**

    .. literalinclude:: ../../task/_examples_/periodic.py

    Will result to the following output:

    .. literalinclude:: ../../task/_examples_/periodic.log
    """
    def __init__(self, cycle_time, method, *args, **kwargs):
        self._lock = threading.Lock()
        self._timer = None
        self.method = method
        self.cycle_time = cycle_time
        self.args = args
        self.kwargs = kwargs
        self._stopped = True
        self._last_tm = None
        self.dt = None

    def join(self, timeout=0.1):
        """
        This blocks till the cyclic task is terminated.

        :param float timeout: Cycle time for checking if task is stopped

        .. note:: Using join means that somewhere has to be a condition calling :py:func:`stop` to terminate.
        """
        while not self._stopped:
            time.sleep(timeout)

    def run(self):
        """
        This starts the cyclic execution of the given method.
        """
        if self._stopped:
            self._set_timer(force_now=True)

    def stop(self):
        """
        This stops the execution of any following task.
        """
        self._lock.acquire()
        self._stopped = True
        if self._timer is not None:
            self._timer.cancel()
        self._lock.release()

    def _set_timer(self, force_now=False):
        """
        This sets the timer for the execution of the next task.
        """
        self._lock.acquire()
        self._stopped = False
        if force_now:
            self._timer = threading.Timer(0, self._start)
        else:
            self._timer = threading.Timer(self.cycle_time, self._start)
        self._timer.start()
        self._lock.release()

    def _start(self):
        tm = time.time()
        if self._last_tm is not None:
            self.dt = tm - self._last_tm
        self._set_timer(force_now=False)
        self.method(self, *self.args, **self.kwargs)
        self._last_tm = tm


class delayed(periodic):
    """Class to execute a method a given time in the future. See also parent :py:class:`periodic`.

    :param float time: Delay time for execution of the given method
    :param method method: Method to be executed
    :param args args: Arguments to be given to method
    :param kwargs kwargs: Kewordsarguments to be given to method

    **Example:**

    .. literalinclude:: ../../task/_examples_/delayed.py

    Will result to the following output:

    .. literalinclude:: ../../task/_examples_/delayed.log
    """
    def run(self):
        """
        This starts the timer for the delayed execution.
        """
        self._set_timer(force_now=False)

    def _start(self):
        self.method(*self.args, **self.kwargs)
        self.stop()


class crontab(periodic):
    """Class to execute a callback at the specified time conditions. See also parent :py:class:`periodic`.

    :param accuracy: Repeat time in seconds for background task checking event triggering. This time is the maximum delay between specified time condition and the execution.
    :type accuracy: float

    **Example:**

    .. literalinclude:: ../../task/_examples_/crontab.py

    Will result to the following output:

    .. literalinclude:: ../../task/_examples_/crontab.log
    """
    ANY = '*'
    """Constant for matching every condition."""

    class cronjob(object):
        """Class to handle cronjob parameters and cronjob changes.

        :param minute: Minute for execution. Either 0...59, [0...59, 0...59, ...] or :py:const:`crontab.ANY` for every Minute.
        :type minute: int, list, str
        :param hour: Hour for execution. Either 0...23, [0...23, 0...23, ...] or :py:const:`crontab.ANY` for every Hour.
        :type hour: int, list, str
        :param day_of_month: Day of Month for execution. Either 0...31, [0...31, 0...31, ...] or :py:const:`crontab.ANY` for every Day of Month.
        :type day_of_month: int, list, str
        :param month: Month for execution. Either 0...12, [0...12, 0...12, ...] or :py:const:`crontab.ANY` for every Month.
        :type month: int, list, str
        :param day_of_week: Day of Week for execution. Either 0...6, [0...6, 0...6, ...] or :py:const:`crontab.ANY` for every Day of Week.
        :type day_of_week: int, list, str
        :param callback: The callback to be executed. The instance of :py:class:`cronjob` will be given as the first, args and kwargs as the following parameters.
        :type callback: func

        .. note:: This class should not be used stand alone. An instance will be created by adding a cronjob by using :py:func:`crontab.add_cronjob()`.
        """
        class all_match(set):
            """Universal set - match everything"""
            def __contains__(self, item):
                (item)
                return True

        def __init__(self, minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs):
            self.set_trigger_conditions(minute or crontab.ANY, hour or crontab.ANY, day_of_month or crontab.ANY, month or crontab.ANY, day_of_week or crontab.ANY)
            self.callback = callback
            self.args = args
            self.kwargs = kwargs
            self.__last_cron_check_time__ = None
            self.__last_execution__ = None

        def set_trigger_conditions(self, minute=None, hour=None, day_of_month=None, month=None, day_of_week=None):
            """This Method changes the execution parameters.

            :param minute: Minute for execution. Either 0...59, [0...59, 0...59, ...] or :py:const:`crontab.ANY` for every Minute.
            :type minute: int, list, str
            :param hour: Hour for execution. Either 0...23, [0...23, 0...23, ...] or :py:const:`crontab.ANY` for every Hour.
            :type hour: int, list, str
            :param day_of_month: Day of Month for execution. Either 0...31, [0...31, 0...31, ...] or :py:const:`crontab.ANY` for every Day of Month.
            :type day_of_month: int, list, str
            :param month: Month for execution. Either 0...12, [0...12, 0...12, ...] or :py:const:`crontab.ANY` for every Month.
            :type month: int, list, str
            :param day_of_week: Day of Week for execution. Either 0...6, [0...6, 0...6, ...] or :py:const:`crontab.ANY` for every Day of Week.
            :type day_of_week: int, list, str
            """
            if minute is not None:
                self.minute = self.__conv_to_set__(minute)
            if hour is not None:
                self.hour = self.__conv_to_set__(hour)
            if day_of_month is not None:
                self.day_of_month = self.__conv_to_set__(day_of_month)
            if month is not None:
                self.month = self.__conv_to_set__(month)
            if day_of_week is not None:
                self.day_of_week = self.__conv_to_set__(day_of_week)

        def __conv_to_set__(self, obj):
            if obj is crontab.ANY:
                return self.all_match()
            elif isinstance(obj, (int, long) if sys.version_info < (3,0) else (int)):
                return set([obj])
            else:
                return set(obj)

        def __execution_needed_for__(self, minute, hour, day_of_month, month, day_of_week):
            if self.__last_execution__ != [minute, hour, day_of_month, month, day_of_week]:
                if minute in self.minute and hour in self.hour and day_of_month in self.day_of_month and month in self.month and day_of_week in self.day_of_week:
                    return True
            return False

        def __store_execution_reminder__(self, minute, hour, day_of_month, month, day_of_week):
            self.__last_execution__ = [minute, hour, day_of_month, month, day_of_week]

        def cron_execution(self, tm):
            """This Methods executes the Cron-Callback, if a execution is needed for the given time (depending on the parameters on initialisation)

            :param tm: (Current) Time Value to be checked. The time needs to be given in seconds since 1970 (e.g. generated by int(time.time())).
            :type tm: int
            """
            if self.__last_cron_check_time__ is None:
                self.__last_cron_check_time__ = tm - 1
            #
            for t in range(self.__last_cron_check_time__ + 1, tm + 1):
                lt = time.localtime(t)
                if self.__execution_needed_for__(lt[4], lt[3], lt[2], lt[1], lt[6]):
                    self.callback(self, *self.args, **self.kwargs)
                    self.__store_execution_reminder__(lt[4], lt[3], lt[2], lt[1], lt[6])
                    break
            self.__last_cron_check_time__ = tm

    def __init__(self, accuracy=30):
        periodic.__init__(self, accuracy, self.__periodic__)
        self.__crontab__ = []

    def __periodic__(self, rt):
        (rt)
        tm = int(time.time())
        for cronjob in self.__crontab__:
            cronjob.cron_execution(tm)

    def add_cronjob(self, minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs):
        """This Method adds a cronjob to be executed.

        :param minute: Minute for execution. Either 0...59, [0...59, 0...59, ...] or :py:const:`crontab.ANY` for every Minute.
        :type minute: int, list, str
        :param hour: Hour for execution. Either 0...23, [0...23, 0...23, ...] or :py:const:`crontab.ANY` for every Hour.
        :type hour: int, list, str
        :param day_of_month: Day of Month for execution. Either 0...31, [0...31, 0...31, ...] or :py:const:`crontab.ANY` for every Day of Month.
        :type day_of_month: int, list, str
        :param month: Month for execution. Either 0...12, [0...12, 0...12, ...] or :py:const:`crontab.ANY` for every Month.
        :type month: int, list, str
        :param day_of_week: Day of Week for execution. Either 0...6, [0...6, 0...6, ...] or :py:const:`crontab.ANY` for every Day of Week.
        :type day_of_week: int, list, str
        :param callback: The callback to be executed. The instance of :py:class:`cronjob` will be given as the first, args and kwargs as the following parameters.
        :type callback: func

        .. note:: The ``callback`` will be executed with it's instance of :py:class:`cronjob` as the first parameter.
        """
        self.__crontab__.append(self.cronjob(minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs))