<s> """<STR_LIT>""" <EOL> import os . path <EOL> from . compat import get_platform_system <EOL> class FileModifyWatcher ( object ) : <EOL> """<STR_LIT>""" <EOL> supported_platforms = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def __init__ ( self , path_list ) : <EOL> """<STR_LIT>""" <EOL> self . file_list = set ( ) <EOL> self . watch_dirs = set ( ) <EOL> self . notify_dirs = set ( ) <EOL> for filename in path_list : <EOL> path = os . path . abspath ( filename ) <EOL> if os . path . isfile ( path ) : <EOL> self . file_list . add ( path ) <EOL> self . watch_dirs . add ( os . path . dirname ( path ) ) <EOL> else : <EOL> self . notify_dirs . add ( path ) <EOL> self . watch_dirs . add ( path ) <EOL> self . platform = get_platform_system ( ) <EOL> if self . platform not in self . supported_platforms : <EOL> msg = "<STR_LIT>" % self . platform <EOL> msg += ( "<STR_LIT>" % <EOL> ( self . supported_platforms , ) ) <EOL> raise Exception ( msg ) <EOL> def _handle ( self , event ) : <EOL> """<STR_LIT>""" <EOL> if self . platform == '<STR_LIT>' : <EOL> filename = event . name <EOL> elif self . platform == '<STR_LIT>' : <EOL> filename = event . pathname <EOL> if ( filename in self . file_list or <EOL> os . path . dirname ( filename ) in self . notify_dirs ) : <EOL> self . handle_event ( event ) <EOL> def handle_event ( self , event ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def _loop_darwin ( self ) : <EOL> """<STR_LIT>""" <EOL> from fsevents import Observer <EOL> from fsevents import Stream <EOL> from fsevents import IN_MODIFY <EOL> observer = Observer ( ) <EOL> handler = self . _handle <EOL> def fsevent_callback ( event ) : <EOL> if event . mask == IN_MODIFY : <EOL> handler ( event ) <EOL> for watch_this in self . watch_dirs : <EOL> stream = Stream ( fsevent_callback , watch_this , file_events = True ) <EOL> observer . schedule ( stream ) <EOL> observer . daemon = True <EOL> observer . start ( ) <EOL> try : <EOL> import time <EOL> while True : <EOL> time . sleep ( <NUM_LIT> ) <EOL> except ( SystemExit , KeyboardInterrupt ) : <EOL> pass <EOL> def _loop_linux ( self , loop_callback ) : <EOL> """<STR_LIT>""" <EOL> import pyinotify <EOL> handler = self . _handle <EOL> class EventHandler ( pyinotify . ProcessEvent ) : <EOL> def process_default ( self , event ) : <EOL> handler ( event ) <EOL> watch_manager = pyinotify . WatchManager ( ) <EOL> event_handler = EventHandler ( ) <EOL> notifier = pyinotify . Notifier ( watch_manager , event_handler ) <EOL> mask = pyinotify . IN_CLOSE_WRITE | pyinotify . IN_MOVED_TO <EOL> for watch_this in self . watch_dirs : <EOL> watch_manager . add_watch ( watch_this , mask ) <EOL> notifier . loop ( loop_callback ) <EOL> def loop ( self , loop_callback = None ) : <EOL> """<STR_LIT>""" <EOL> if self . platform == '<STR_LIT>' : <EOL> self . _loop_darwin ( ) <EOL> elif self . platform == '<STR_LIT>' : <EOL> self . _loop_linux ( loop_callback ) </s>
