<s> from heapq import heappush , heappop <EOL> from itertools import count <EOL> from threading import Lock <EOL> from atom . api import ( <EOL> Atom , Bool , Typed , ForwardTyped , Tuple , Dict , Callable , Value , List , <EOL> observe <EOL> ) <EOL> class ScheduledTask ( Atom ) : <EOL> """<STR_LIT>""" <EOL> _callback = Callable ( ) <EOL> _args = Tuple ( ) <EOL> _kwargs = Dict ( ) <EOL> _result = Value ( ) <EOL> _valid = Bool ( True ) <EOL> _pending = Bool ( True ) <EOL> _notify = Callable ( ) <EOL> def __init__ ( self , callback , args , kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _callback = callback <EOL> self . _args = args <EOL> self . _kwargs = kwargs <EOL> def _execute ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if self . _valid : <EOL> self . _result = self . _callback ( * self . _args , ** self . _kwargs ) <EOL> if self . _notify is not None : <EOL> self . _notify ( self . _result ) <EOL> finally : <EOL> del self . _notify <EOL> self . _pending = False <EOL> def notify ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _notify = callback <EOL> def pending ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _pending <EOL> def unschedule ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _valid = False <EOL> def result ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _result <EOL> class ProxyResolver ( Atom ) : <EOL> """<STR_LIT>""" <EOL> factories = Dict ( ) <EOL> def resolve ( self , name ) : <EOL> """<STR_LIT>""" <EOL> factory = self . factories . get ( name ) <EOL> if factory is not None : <EOL> return factory ( ) <EOL> def StyleSheet ( ) : <EOL> """<STR_LIT>""" <EOL> from enaml . styling import StyleSheet <EOL> return StyleSheet <EOL> class Application ( Atom ) : <EOL> """<STR_LIT>""" <EOL> resolver = Typed ( ProxyResolver ) <EOL> style_sheet = ForwardTyped ( StyleSheet ) <EOL> _task_heap = List ( ) <EOL> _counter = Value ( factory = count ) <EOL> _heap_lock = Value ( factory = Lock ) <EOL> _instance = None <EOL> @ staticmethod <EOL> def instance ( ) : <EOL> """<STR_LIT>""" <EOL> return Application . _instance <EOL> def __new__ ( cls , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if Application . _instance is not None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> self = super ( Application , cls ) . __new__ ( cls , * args , ** kwargs ) <EOL> Application . _instance = self <EOL> return self <EOL> def _process_task ( self , task ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> task . _execute ( ) <EOL> finally : <EOL> self . _next_task ( ) <EOL> def _next_task ( self ) : <EOL> """<STR_LIT>""" <EOL> heap = self . _task_heap <EOL> with self . _heap_lock : <EOL> if heap : <EOL> priority , ignored , task = heappop ( heap ) <EOL> self . deferred_call ( self . _process_task , task ) <EOL> @ observe ( '<STR_LIT>' ) <EOL> def _clear_destroyed_style_sheet ( self , change ) : <EOL> """<STR_LIT>""" <EOL> self . style_sheet = None <EOL> @ observe ( '<STR_LIT>' ) <EOL> def _invalidate_style_cache ( self , change ) : <EOL> """<STR_LIT>""" <EOL> if change [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> from enaml . styling import StyleCache <EOL> StyleCache . _app_sheet_changed ( ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def deferred_call ( self , callback , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def timed_call ( self , ms , callback , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def is_main_thread ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def create_mime_data ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def resolve_proxy_class ( self , declaration_class ) : <EOL> """<STR_LIT>""" <EOL> resolver = self . resolver <EOL> for base in declaration_class . mro ( ) : <EOL> name = base . __name__ <EOL> cls = resolver . resolve ( name ) <EOL> if cls is not None : <EOL> return cls <EOL> def create_proxy ( self , declaration ) : <EOL> """<STR_LIT>""" <EOL> cls = self . resolve_proxy_class ( type ( declaration ) ) <EOL> if cls is not None : <EOL> return cls ( declaration = declaration ) <EOL> msg = "<STR_LIT>" <EOL> msg += "<STR_LIT>" <EOL> d_name = type ( declaration ) . __name__ <EOL> a_name = type ( self ) . __name__ <EOL> raise TypeError ( msg % ( d_name , a_name ) ) <EOL> def schedule ( self , callback , args = None , kwargs = None , priority = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if args is None : <EOL> args = ( ) <EOL> if kwargs is None : <EOL> kwargs = { } <EOL> task = ScheduledTask ( callback , args , kwargs ) <EOL> heap = self . _task_heap <EOL> with self . _heap_lock : <EOL> needs_start = len ( heap ) == <NUM_LIT:0> <EOL> item = ( - priority , self . _counter . next ( ) , task ) <EOL> heappush ( heap , item ) <EOL> if needs_start : <EOL> if self . is_main_thread ( ) : <EOL> self . _next_task ( ) <EOL> else : <EOL> self . deferred_call ( self . _next_task ) <EOL> return task <EOL> def has_pending_tasks ( self ) : <EOL> """<STR_LIT>""" <EOL> heap = self . _task_heap <EOL> with self . _heap_lock : <EOL> has_pending = len ( heap ) > <NUM_LIT:0> <EOL> return has_pending <EOL> def destroy ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stop ( ) <EOL> Application . _instance = None <EOL> def deferred_call ( callback , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> app = Application . instance ( ) <EOL> if app is None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> app . deferred_call ( callback , * args , ** kwargs ) <EOL> def timed_call ( ms , callback , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> app = Application . instance ( ) <EOL> if app is None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> app . timed_call ( ms , callback , * args , ** kwargs ) <EOL> def is_main_thread ( ) : <EOL> """<STR_LIT>""" <EOL> app = Application . instance ( ) <EOL> if app is None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return app . is_main_thread ( ) <EOL> def schedule ( callback , args = None , kwargs = None , priority = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> app = Application . instance ( ) <EOL> if app is None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return app . schedule ( callback , args , kwargs , priority ) </s>
