<s> """<STR_LIT>""" <EOL> import sys <EOL> from time import time as _time <EOL> from heapq import heapify , heappush , heappop <EOL> from . singleton import SingletonLogging <EOL> from . debugging import DebugContents , Logging , ModuleLogger , bacpypes_debugging <EOL> _debug = <NUM_LIT:0> <EOL> _log = ModuleLogger ( globals ( ) ) <EOL> _task_manager = None <EOL> _unscheduled_tasks = [ ] <EOL> if sys . platform in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> from . event import WaitableEvent <EOL> class _Trigger ( WaitableEvent , Logging ) : <EOL> def handle_read ( self ) : <EOL> if _debug : _Trigger . _debug ( "<STR_LIT>" ) <EOL> data = self . recv ( <NUM_LIT:1> ) <EOL> if _debug : _Trigger . _debug ( "<STR_LIT>" , data ) <EOL> else : <EOL> _Trigger = None <EOL> class _Task ( DebugContents , Logging ) : <EOL> _debug_contents = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self ) : <EOL> self . taskTime = None <EOL> self . isScheduled = False <EOL> def install_task ( self , when = None , delta = None ) : <EOL> global _task_manager , _unscheduled_tasks <EOL> if ( when is None ) and ( delta is not None ) : <EOL> if not _task_manager : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> when = _task_manager . get_time ( ) + delta <EOL> if when is None : <EOL> when = self . taskTime <EOL> if when is None : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> self . taskTime = when <EOL> if not _task_manager : <EOL> _unscheduled_tasks . append ( self ) <EOL> else : <EOL> _task_manager . install_task ( self ) <EOL> def process_task ( self ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def suspend_task ( self ) : <EOL> global _task_manager <EOL> if not _task_manager : <EOL> _unscheduled_tasks . remove ( self ) <EOL> else : <EOL> _task_manager . suspend_task ( self ) <EOL> def resume_task ( self ) : <EOL> global _task_manager <EOL> _task_manager . resume_task ( self ) <EOL> def __lt__ ( self , other ) : <EOL> return id ( self ) < id ( other ) <EOL> class OneShotTask ( _Task ) : <EOL> def __init__ ( self , when = None ) : <EOL> _Task . __init__ ( self ) <EOL> self . taskTime = when <EOL> class OneShotDeleteTask ( _Task ) : <EOL> def __init__ ( self , when = None ) : <EOL> _Task . __init__ ( self ) <EOL> self . taskTime = when <EOL> @ bacpypes_debugging <EOL> def OneShotFunction ( fn , * args , ** kwargs ) : <EOL> class OneShotFunctionTask ( OneShotDeleteTask ) : <EOL> def process_task ( self ) : <EOL> OneShotFunction . _debug ( "<STR_LIT>" , fn , repr ( args ) , repr ( kwargs ) ) <EOL> fn ( * args , ** kwargs ) <EOL> task = OneShotFunctionTask ( ) <EOL> if not _task_manager : <EOL> _unscheduled_tasks . append ( task ) <EOL> else : <EOL> task . install_task ( _task_manager . get_time ( ) ) <EOL> return task <EOL> def FunctionTask ( fn , * args , ** kwargs ) : <EOL> _log . debug ( "<STR_LIT>" , fn , args , kwargs ) <EOL> class _FunctionTask ( OneShotDeleteTask ) : <EOL> def process_task ( self ) : <EOL> _log . debug ( "<STR_LIT>" , fn , args , kwargs ) <EOL> fn ( * args , ** kwargs ) <EOL> task = _FunctionTask ( ) <EOL> _log . debug ( "<STR_LIT>" , task ) <EOL> return task <EOL> @ bacpypes_debugging <EOL> class RecurringTask ( _Task ) : <EOL> _debug_contents = ( '<STR_LIT>' , ) <EOL> def __init__ ( self , interval = None ) : <EOL> if _debug : RecurringTask . _debug ( "<STR_LIT>" , interval ) <EOL> _Task . __init__ ( self ) <EOL> self . taskInterval = interval <EOL> def install_task ( self , interval = None ) : <EOL> if _debug : RecurringTask . _debug ( "<STR_LIT>" , interval ) <EOL> global _task_manager , _unscheduled_tasks <EOL> if interval is not None : <EOL> self . taskInterval = interval <EOL> if self . taskInterval is None : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if self . taskInterval <= <NUM_LIT:0.0> : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if not _task_manager : <EOL> if _debug : RecurringTask . _debug ( "<STR_LIT>" ) <EOL> _unscheduled_tasks . append ( self ) <EOL> else : <EOL> now = _task_manager . get_time ( ) <EOL> interval = self . taskInterval / <NUM_LIT> <EOL> self . taskTime = now + interval - ( now % interval ) <EOL> if _debug : RecurringTask . _debug ( "<STR_LIT>" , self . taskTime ) <EOL> _task_manager . install_task ( self ) <EOL> @ bacpypes_debugging <EOL> def RecurringFunctionTask ( interval , fn , * args , ** kwargs ) : <EOL> if _debug : RecurringFunctionTask . _debug ( "<STR_LIT>" , fn , args , kwargs ) <EOL> class _RecurringFunctionTask ( RecurringTask ) : <EOL> def __init__ ( self , interval ) : <EOL> RecurringTask . __init__ ( self , interval ) <EOL> def process_task ( self ) : <EOL> if _debug : RecurringFunctionTask . _debug ( "<STR_LIT>" , fn , args , kwargs ) <EOL> fn ( * args , ** kwargs ) <EOL> task = _RecurringFunctionTask ( interval ) <EOL> if _debug : RecurringFunctionTask . _debug ( "<STR_LIT>" , task ) <EOL> return task <EOL> @ bacpypes_debugging <EOL> def recurring_function ( interval ) : <EOL> def recurring_function_decorator ( fn ) : <EOL> class _RecurringFunctionTask ( RecurringTask ) : <EOL> def process_task ( self ) : <EOL> if _debug : recurring_function . _debug ( "<STR_LIT>" , fn ) <EOL> fn ( ) <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> fn ( * args , ** kwargs ) <EOL> task = _RecurringFunctionTask ( interval ) <EOL> task . install_task ( ) <EOL> return task <EOL> return recurring_function_decorator <EOL> class TaskManager ( SingletonLogging ) : <EOL> def __init__ ( self ) : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" ) <EOL> global _task_manager , _unscheduled_tasks <EOL> self . tasks = [ ] <EOL> if _Trigger : <EOL> self . trigger = _Trigger ( ) <EOL> else : <EOL> self . trigger = None <EOL> _task_manager = self <EOL> if _unscheduled_tasks : <EOL> for task in _unscheduled_tasks : <EOL> task . install_task ( ) <EOL> def get_time ( self ) : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" ) <EOL> return _time ( ) <EOL> def install_task ( self , task ) : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" , task , task . taskTime ) <EOL> if task . taskTime is None : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> if task . isScheduled : <EOL> self . suspend_task ( task ) <EOL> heappush ( self . tasks , ( task . taskTime , task ) ) <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" , self . tasks ) <EOL> task . isScheduled = True <EOL> if self . trigger : <EOL> self . trigger . set ( ) <EOL> def suspend_task ( self , task ) : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" , task ) <EOL> for i , ( when , curtask ) in enumerate ( self . tasks ) : <EOL> if task is curtask : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" ) <EOL> del self . tasks [ i ] <EOL> task . isScheduled = False <EOL> heapify ( self . tasks ) <EOL> break <EOL> else : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" ) <EOL> if self . trigger : <EOL> self . trigger . set ( ) <EOL> def resume_task ( self , task ) : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" , task ) <EOL> self . install_task ( task ) <EOL> def get_next_task ( self ) : <EOL> """<STR_LIT>""" <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" ) <EOL> now = _time ( ) <EOL> task = None <EOL> delta = None <EOL> if self . tasks : <EOL> when , nxttask = self . tasks [ <NUM_LIT:0> ] <EOL> if when <= now : <EOL> heappop ( self . tasks ) <EOL> task = nxttask <EOL> task . isScheduled = False <EOL> if self . tasks : <EOL> when , nxttask = self . tasks [ <NUM_LIT:0> ] <EOL> delta = max ( when - now , <NUM_LIT:0.0> ) <EOL> else : <EOL> delta = when - now <EOL> return ( task , delta ) <EOL> def process_task ( self , task ) : <EOL> if _debug : TaskManager . _debug ( "<STR_LIT>" , task ) <EOL> task . process_task ( ) <EOL> if isinstance ( task , RecurringTask ) : <EOL> task . install_task ( ) <EOL> elif isinstance ( task , OneShotDeleteTask ) : <EOL> del task </s>
