<s> from __future__ import with_statement <EOL> import threading <EOL> from watchdog . utils import BaseThread <EOL> from watchdog . utils . compat import queue <EOL> from watchdog . utils . bricks import SkipRepeatsQueue <EOL> DEFAULT_EMITTER_TIMEOUT = <NUM_LIT:1> <EOL> DEFAULT_OBSERVER_TIMEOUT = <NUM_LIT:1> <EOL> class EventQueue ( SkipRepeatsQueue ) : <EOL> """<STR_LIT>""" <EOL> class ObservedWatch ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path , recursive ) : <EOL> self . _path = path <EOL> self . _is_recursive = recursive <EOL> @ property <EOL> def path ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _path <EOL> @ property <EOL> def is_recursive ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _is_recursive <EOL> @ property <EOL> def key ( self ) : <EOL> return self . path , self . is_recursive <EOL> def __eq__ ( self , watch ) : <EOL> return self . key == watch . key <EOL> def __ne__ ( self , watch ) : <EOL> return self . key != watch . key <EOL> def __hash__ ( self ) : <EOL> return hash ( self . key ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( <EOL> self . path , self . is_recursive ) <EOL> class EventEmitter ( BaseThread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , event_queue , watch , timeout = DEFAULT_EMITTER_TIMEOUT ) : <EOL> BaseThread . __init__ ( self ) <EOL> self . _event_queue = event_queue <EOL> self . _watch = watch <EOL> self . _timeout = timeout <EOL> @ property <EOL> def timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _timeout <EOL> @ property <EOL> def watch ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _watch <EOL> def queue_event ( self , event ) : <EOL> """<STR_LIT>""" <EOL> self . _event_queue . put ( ( event , self . watch ) ) <EOL> def queue_events ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> def run ( self ) : <EOL> try : <EOL> while self . should_keep_running ( ) : <EOL> self . queue_events ( self . timeout ) <EOL> finally : <EOL> pass <EOL> class EventDispatcher ( BaseThread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , timeout = DEFAULT_OBSERVER_TIMEOUT ) : <EOL> BaseThread . __init__ ( self ) <EOL> self . _event_queue = EventQueue ( ) <EOL> self . _timeout = timeout <EOL> @ property <EOL> def timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _timeout <EOL> @ property <EOL> def event_queue ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _event_queue <EOL> def dispatch_events ( self , event_queue , timeout ) : <EOL> """<STR_LIT>""" <EOL> def run ( self ) : <EOL> while self . should_keep_running ( ) : <EOL> try : <EOL> self . dispatch_events ( self . event_queue , self . timeout ) <EOL> except queue . Empty : <EOL> continue <EOL> class BaseObserver ( EventDispatcher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , emitter_class , timeout = DEFAULT_OBSERVER_TIMEOUT ) : <EOL> EventDispatcher . __init__ ( self , timeout ) <EOL> self . _emitter_class = emitter_class <EOL> self . _lock = threading . RLock ( ) <EOL> self . _watches = set ( ) <EOL> self . _handlers = dict ( ) <EOL> self . _emitters = set ( ) <EOL> self . _emitter_for_watch = dict ( ) <EOL> def _add_emitter ( self , emitter ) : <EOL> self . _emitter_for_watch [ emitter . watch ] = emitter <EOL> self . _emitters . add ( emitter ) <EOL> def _remove_emitter ( self , emitter ) : <EOL> del self . _emitter_for_watch [ emitter . watch ] <EOL> self . _emitters . remove ( emitter ) <EOL> emitter . stop ( ) <EOL> try : <EOL> emitter . join ( ) <EOL> except RuntimeError : <EOL> pass <EOL> def _clear_emitters ( self ) : <EOL> for emitter in self . _emitters : <EOL> emitter . stop ( ) <EOL> for emitter in self . _emitters : <EOL> try : <EOL> emitter . join ( ) <EOL> except RuntimeError : <EOL> pass <EOL> self . _emitters . clear ( ) <EOL> self . _emitter_for_watch . clear ( ) <EOL> def _add_handler_for_watch ( self , event_handler , watch ) : <EOL> if watch not in self . _handlers : <EOL> self . _handlers [ watch ] = set ( ) <EOL> self . _handlers [ watch ] . add ( event_handler ) <EOL> def _remove_handlers_for_watch ( self , watch ) : <EOL> del self . _handlers [ watch ] <EOL> @ property <EOL> def emitters ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _emitters <EOL> def start ( self ) : <EOL> for emitter in self . _emitters : <EOL> emitter . start ( ) <EOL> super ( BaseObserver , self ) . start ( ) <EOL> def schedule ( self , event_handler , path , recursive = False ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> watch = ObservedWatch ( path , recursive ) <EOL> self . _add_handler_for_watch ( event_handler , watch ) <EOL> if self . _emitter_for_watch . get ( watch ) is None : <EOL> emitter = self . _emitter_class ( event_queue = self . event_queue , <EOL> watch = watch , <EOL> timeout = self . timeout ) <EOL> self . _add_emitter ( emitter ) <EOL> if self . is_alive ( ) : <EOL> emitter . start ( ) <EOL> self . _watches . add ( watch ) <EOL> return watch <EOL> def add_handler_for_watch ( self , event_handler , watch ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _add_handler_for_watch ( event_handler , watch ) <EOL> def remove_handler_for_watch ( self , event_handler , watch ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _handlers [ watch ] . remove ( event_handler ) <EOL> def unschedule ( self , watch ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> emitter = self . _emitter_for_watch [ watch ] <EOL> del self . _handlers [ watch ] <EOL> self . _remove_emitter ( emitter ) <EOL> self . _watches . remove ( watch ) <EOL> def unschedule_all ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _handlers . clear ( ) <EOL> self . _clear_emitters ( ) <EOL> self . _watches . clear ( ) <EOL> def on_thread_stop ( self ) : <EOL> self . unschedule_all ( ) <EOL> def dispatch_events ( self , event_queue , timeout ) : <EOL> event , watch = event_queue . get ( block = True , timeout = timeout ) <EOL> with self . _lock : <EOL> for handler in list ( self . _handlers . get ( watch , [ ] ) ) : <EOL> if handler in self . _handlers . get ( watch , [ ] ) : <EOL> handler . dispatch ( event ) <EOL> event_queue . task_done ( ) </s>
