<commit_before>#!/usr/bin/python

"""
Task
"""

import sys

from time import time as _time
from heapq import heapify, heappush, heappop

from .singleton import SingletonLogging
from .debugging import DebugContents, Logging, ModuleLogger, bacpypes_debugging

# some debugging
_debug = 0
_log = ModuleLogger(globals())

# globals
_task_manager = None
_unscheduled_tasks = []

# only defined for linux platforms
if 'linux' in sys.platform:
    from .event import WaitableEvent
    #
    #   _Trigger
    #
    #   An instance of this class is used in the task manager to break
    #   the asyncore.loop() call.  In this case, handle_read will
    #   immediately "clear" the event.
    #

    class _Trigger(WaitableEvent, Logging):

        def handle_read(self):
            if _debug: _Trigger._debug("handle_read")

            # read in the character, highlander
            data = self.recv(1)
            if _debug: _Trigger._debug("    - data: %r", data)

#
#   _Task
#

class _Task(DebugContents, Logging):

    _debug_contents = ('taskTime', 'isScheduled')

    def __init__(self):
        self.taskTime = None
        self.isScheduled = False

    def install_task(self, when=None, delta=None):
        global _task_manager, _unscheduled_tasks

        # check for delta from now
        if (when is None) and (delta is not None):
            if not _task_manager:
                raise RuntimeError("no task manager")

            when = _task_manager.get_time() + delta

        # fallback to the inited value
        if when is None:
            when = self.taskTime
        if when is None:
            raise RuntimeError("schedule missing, use zero for 'now'")
        self.taskTime = when

        # pass along to the task manager
        if not _task_manager:
            _unscheduled_tasks.append(self)
        else:
            _task_manager.install_task(self)

    def process_task(self):
        raise RuntimeError("process_task must be overridden")

    def suspend_task(self):
        global _task_manager

        # pass along to the task manager
        if not _task_manager:
            _unscheduled_tasks.remove(self)
        else:
            _task_manager.suspend_task(self)

    def resume_task(self):
        global _task_manager

        _task_manager.resume_task(self)

    def __lt__(self, other):
        return id(self) < id(other)

#
#   OneShotTask
#

class OneShotTask(_Task):

    def __init__(self, when=None):
        _Task.__init__(self)
        self.taskTime = when

#
#   OneShotDeleteTask
#

class OneShotDeleteTask(_Task):

    def __init__(self, when=None):
        _Task.__init__(self)
        self.taskTime = when

#
#   OneShotFunction
#

@bacpypes_debugging
def OneShotFunction(fn, *args, **kwargs):

    class OneShotFunctionTask(OneShotDeleteTask):

        def process_task(self):
            OneShotFunction._debug("process_task %r %s %s", fn, repr(args), repr(kwargs))
            fn(*args, **kwargs)

    task = OneShotFunctionTask()

    # if there is no task manager, postpone the install
    if not _task_manager:
        _unscheduled_tasks.append(task)
    else:
        task.install_task(_task_manager.get_time())

    return task

#
#   FunctionTask
#

def FunctionTask(fn, *args, **kwargs):
    _log.debug("FunctionTask %r %r %r", fn, args, kwargs)

    class _FunctionTask(OneShotDeleteTask):

        def process_task(self):
            _log.debug("process_task (%r %r %r)", fn, args, kwargs)
            fn(*args, **kwargs)

    task = _FunctionTask()
    _log.debug("    - task: %r", task)

    return task

#
#   RecurringTask
#

class RecurringTask(_Task):

    _debug_contents = ('taskInterval',)

    def __init__(self, interval=None):
        _Task.__init__(self)
        self.taskInterval = interval

    def install_task(self, interval=None):
        global _task_manager, _unscheduled_tasks

        # set the interval if it hasn't already been set
        if interval is not None:
            self.taskInterval = interval
        if self.taskInterval is None:
            raise RuntimeError("interval unset, use ctor or install_task parameter")
        if self.taskInterval <= 0.0:
            raise RuntimeError("interval must be greater than zero")

        # if there is no task manager, postpone the install
        if not _task_manager:
            _unscheduled_tasks.append(self)

        else:
            # get ready for the next interval (aligned)
            now = _task_manager.get_time()
            interval = self.taskInterval / 1000.0
            self.taskTime = now + interval - (now % interval)

            # install it
            _task_manager.install_task(self)

#
#   RecurringFunctionTask
#

@bacpypes_debugging
def RecurringFunctionTask(interval, fn, *args, **kwargs):
    if _debug: RecurringFunctionTask._debug("RecurringFunctionTask %r %r %r", fn, args, kwargs)

    class _RecurringFunctionTask(RecurringTask):
        def __init__(self, interval):
            RecurringTask.__init__(self, interval)

        def process_task(self):
            if _debug: RecurringFunctionTask._debug("process_task %r %r %r", fn, args, kwargs)
            fn(*args, **kwargs)

    task = _RecurringFunctionTask(interval)
    if _debug: RecurringFunctionTask._debug("    - task: %r", task)

    return task

#<commit_msg>add some debugging<commit_after>#!/usr/bin/python

"""
Task
"""

import sys

from time import time as _time
from heapq import heapify, heappush, heappop

from .singleton import SingletonLogging
from .debugging import DebugContents, Logging, ModuleLogger, bacpypes_debugging

# some debugging
_debug = 0
_log = ModuleLogger(globals())

# globals
_task_manager = None
_unscheduled_tasks = []

# only defined for linux platforms
if 'linux' in sys.platform:
    from .event import WaitableEvent
    #
    #   _Trigger
    #
    #   An instance of this class is used in the task manager to break
    #   the asyncore.loop() call.  In this case, handle_read will
    #   immediately "clear" the event.
    #

    class _Trigger(WaitableEvent, Logging):

        def handle_read(self):
            if _debug: _Trigger._debug("handle_read")

            # read in the character, highlander
            data = self.recv(1)
            if _debug: _Trigger._debug("    - data: %r", data)

#
#   _Task
#

class _Task(DebugContents, Logging):

    _debug_contents = ('taskTime', 'isScheduled')

    def __init__(self):
        self.taskTime = None
        self.isScheduled = False

    def install_task(self, when=None, delta=None):
        global _task_manager, _unscheduled_tasks

        # check for delta from now
        if (when is None) and (delta is not None):
            if not _task_manager:
                raise RuntimeError("no task manager")

            when = _task_manager.get_time() + delta

        # fallback to the inited value
        if when is None:
            when = self.taskTime
        if when is None:
            raise RuntimeError("schedule missing, use zero for 'now'")
        self.taskTime = when

        # pass along to the task manager
        if not _task_manager:
            _unscheduled_tasks.append(self)
        else:
            _task_manager.install_task(self)

    def process_task(self):
        raise RuntimeError("process_task must be overridden")

    def suspend_task(self):
        global _task_manager

        # pass along to the task manager
        if not _task_manager:
            _unscheduled_tasks.remove(self)
        else:
            _task_manager.suspend_task(self)

    def resume_task(self):
        global _task_manager

        _task_manager.resume_task(self)

    def __lt__(self, other):
        return id(self) < id(other)

#
#   OneShotTask
#

class OneShotTask(_Task):

    def __init__(self, when=None):
        _Task.__init__(self)
        self.taskTime = when

#
#   OneShotDeleteTask
#

class OneShotDeleteTask(_Task):

    def __init__(self, when=None):
        _Task.__init__(self)
        self.taskTime = when

#
#   OneShotFunction
#

@bacpypes_debugging
def OneShotFunction(fn, *args, **kwargs):

    class OneShotFunctionTask(OneShotDeleteTask):

        def process_task(self):
            OneShotFunction._debug("process_task %r %s %s", fn, repr(args), repr(kwargs))
            fn(*args, **kwargs)

    task = OneShotFunctionTask()

    # if there is no task manager, postpone the install
    if not _task_manager:
        _unscheduled_tasks.append(task)
    else:
        task.install_task(_task_manager.get_time())

    return task

#
#   FunctionTask
#

def FunctionTask(fn, *args, **kwargs):
    _log.debug("FunctionTask %r %r %r", fn, args, kwargs)

    class _FunctionTask(OneShotDeleteTask):

        def process_task(self):
            _log.debug("process_task (%r %r %r)", fn, args, kwargs)
            fn(*args, **kwargs)

    task = _FunctionTask()
    _log.debug("    - task: %r", task)

    return task

#
#   RecurringTask
#

@bacpypes_debugging
class RecurringTask(_Task):

    _debug_contents = ('taskInterval',)

    def __init__(self, interval=None):
        if _debug: RecurringTask._debug("__init__ interval=%r", interval)
        _Task.__init__(self)

        # save the interval, but do not automatically install
        self.taskInterval = interval

    def install_task(self, interval=None):
        if _debug: RecurringTask._debug("install_task interval=%r", interval)
        global _task_manager, _unscheduled_tasks

        # set the interval if it hasn't already been set
        if interval is not None:
            self.taskInterval = interval
        if self.taskInterval is None:
            raise RuntimeError("interval unset, use ctor or install_task parameter")
        if self.taskInterval <= 0.0:
            raise RuntimeError("interval must be greater than zero")

        # if there is no task manager, postpone the install
        if not _task_manager:
            if _debug: RecurringTask._debug("    - no task manager")
            _unscheduled_tasks.append(self)

        else:
            # get ready for the next interval (aligned)
            now = _task_manager.get_time()
            interval = self.taskInterval / 1000.0
            self.taskTime = now + interval - (now % interval)
            if _debug: RecurringTask._debug("    - task time: %r", self.taskTime)

            # install it
            _task_manager.install_task(self)

#
#   RecurringFunctionTask
#

@bacpypes_debugging
def RecurringFunctionTask(interval, fn, *args, **kwargs):
    if _debug: RecurringFunctionTask._debug("RecurringFunctionTask %r %r %r", fn, args, kwargs)

    class _RecurringFunctionTask(RecurringTask):
        def __init__(self, interval):
            RecurringTask.__init__(self, interval)

        def process_task(self):
            if _debug: RecurringFunctionTask._debug("process_task %r %r %r", fn, args, kwargs)
            fn(*args, **kwargs)

    task = _RecurringFunctionTask(interval)
    if _debug: RecurringFunctionTask._debug("    - task: %r", task)

    return task

#
{'fraction_non_alphanumeric': 0.09094212773689471, 'fraction_numerical': 0.001635322976287817, 'mean_word_length': 2.5555555555555554, 'pattern_counts': {'":': 0, '<': 7, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5071747', 'n_tokens_mistral': 3353, 'n_tokens_neox': 3277, 'n_words': 904}