from benderjab import rpc
+LOGGER = logging.getLogger(__name__)
+
class WatcherEvent(object):
"""
Track information about a file event
self.time = time.time()
self.event_root = event_root
self.complete = False
-
+
def __unicode__(self):
if self.complete:
complete = "(completed)"
else:
complete = ""
- return u"<WatchEvent: %s %s %s>" % (time.ctime(self.time), self.event_root, complete)
+ return "<WatchEvent: %s %s %s>" % (time.ctime(self.time), self.event_root, complete)
class Handler(pyinotify.ProcessEvent):
def __init__(self, watchmanager, bot, completion_files=None):
if not is_runfolder(target):
self.log.debug("Skipping %s, not a runfolder" % (target,))
continue
-
+
# grab the previous events for this watch path
watch_path_events = self.last_event.setdefault(watch_path, {})
# if we've already seen an event in this directory (AKA runfolder)
# keep track if its already hit the "completed" flag
- if watch_path_events.has_key(target):
+ if target in watch_path_events:
run_already_complete = watch_path_events[target].complete
watch_path_events[target] = WatcherEvent(target)
class SpoolWatcher(rpc.XmlRpcBot):
"""
Watch a directory and send a message when another process is done writing.
-
+
This monitors a directory tree using inotify (linux specific) and
after some files having been written will send a message after <timeout>
seconds of no file writing.
-
+
(Basically when the solexa machine finishes dumping a round of data
this'll hopefully send out a message saying hey look theres data available
-
+
"""
# these params need to be in the config file
# I wonder where I should put the documentation
# `notify_timeout` - how often to timeout from notify
# `completion_files` - what files indicates we've finished sequencing
# defaults to: netcopy_complete.txt
-
+
def __init__(self, section=None, configfile=None):
#if configfile is None:
# self.configfile = "~/.htsworkflow"
super(SpoolWatcher, self).__init__(section, configfile)
-
+
self.cfg['watchdirs'] = None
self.cfg['write_timeout'] = 10
self.cfg['notify_users'] = None
self.cfg['notify_runner'] = None
self.cfg['completion_files'] = 'ImageAnalysis_Netcopy_complete_READ2.txt ImageAnalysis_Netcopy_complete_SINGLEREAD.txt'
-
+
self.watchdirs = []
self.watchdir_url_map = {}
self.notify_timeout = 0.001
- self.wm = None
+ self.wm = None
self.notify_users = None
self.notify_runner = None
self.wdds = []
# keep track of which mount points tie to which watch directories
# so maybe we can remount them.
self.mounts_to_watches = {}
-
+
self.eventTasks.append(self.process_notify)
def read_config(self, section=None, configfile=None):
- # Don't give in to the temptation to use logging functions here,
+ # Don't give in to the temptation to use logging functions here,
# need to wait until after we detach in start
super(SpoolWatcher, self).read_config(section, configfile)
-
+
self.watchdirs = shlex.split(self._check_required_option('watchdirs'))
# see if there's an alternate url that should be used for the watchdir
for watchdir in self.watchdirs:
self.write_timeout = int(self.cfg['write_timeout'])
self.completion_files = shlex.split(self.cfg['completion_files'])
-
+
self.notify_users = self._parse_user_list(self.cfg['notify_users'])
try:
self.notify_runner = \
mounts.append(w)
self.mounts_to_watches[mount_location] = mounts
- self.log.info(u"Watching:"+unicode(w))
+ self.log.info("Watching:"+str(w))
self.wdds.append(self.wm.add_watch(w, mask, rec=True, auto_add=True))
def unmount_watch(self, event_path):
- # remove backwards so we don't get weirdness from
+ # remove backwards so we don't get weirdness from
# the list getting shorter
for i in range(len(self.wdds),0, -1):
wdd = self.wdds[i]
- self.log.info(u'unmounting: '+unicode(wdd.items()))
- self.wm.rm_watch(wdd.values())
+ self.log.info('unmounting: '+str(list(wdd.items())))
+ self.wm.rm_watch(list(wdd.values()))
del self.wdds[i]
self.mounted = False
copy_url = root_copy_url + list_event_dir
self.log.debug('Copy url: %s' % (copy_url,))
return copy_url
-
+
def process_notify(self, *args):
if self.notifier is None:
# nothing to do yet
self.notifier.read_events()
# should we do something?
# has something happened?
- for watchdir, last_events in self.handler.last_event.items():
- for last_event_dir, last_event_detail in last_events.items():
+ for watchdir, last_events in list(self.handler.last_event.items()):
+ for last_event_dir, last_event_detail in list(last_events.items()):
time_delta = time.time() - last_event_detail.time
if time_delta > self.write_timeout:
- print "timeout", unicode(last_event_detail)
+ LOGGER.info("timeout: %s" % (str(last_event_detail),))
copy_url = self.make_copy_url(watchdir, last_event_dir)
self.startCopy(copy_url)
if last_event_detail.complete:
self.handler.last_event[watchdir] = {}
# handle unmounted filesystems
- for mount_point, was_mounted in self.mounted_points.items():
+ for mount_point, was_mounted in list(self.mounted_points.items()):
if not was_mounted and mount.is_mounted(mount_point):
# we've been remounted. Huzzah!
# restart the watch
"""
Parse xmpp chat messages
"""
- help = u"I can send [copy] message, or squencer [finished]"
- if re.match(u"help", msg):
+ help = "I can send [copy] message, or squencer [finished]"
+ if re.match("help", msg):
reply = help
- elif re.match("copy", msg):
+ elif re.match("copy", msg):
self.startCopy(msg)
- reply = u"sent copy message"
- elif re.match(u"finished", msg):
+ reply = "sent copy message"
+ elif re.match("finished", msg):
words = msg.split()
if len(words) == 2:
self.sequencingFinished(words[1])
- reply = u"sending sequencing finished for %s" % (words[1])
+ reply = "sending sequencing finished for %s" % (words[1])
else:
- reply = u"need runfolder name"
+ reply = "need runfolder name"
else:
- reply = u"I didn't understand '%s'" %(msg)
+ reply = "I didn't understand '%s'" %(msg)
return reply
-
+
def run(self):
"""
Start application
# after it's initialized.
self.add_watch()
super(SpoolWatcher, self).run()
-
+
def stop(self):
"""
shutdown application
if self.notifier is not None:
self.notifier.stop()
super(SpoolWatcher, self).stop()
-
+
def startCopy(self, copy_url=None):
self.log.debug("writes seem to have stopped")
if self.notify_runner is not None:
if self.notify_users is not None:
for u in self.notify_users:
self.send(u, 'startCopy %s.' % (copy_url,))
-
+
def sequencingFinished(self, run_dir):
# need to strip off self.watchdirs from rundir I suspect.
self.log.info("run.completed in " + str(run_dir))
for watch in self.watchdirs:
if not run_dir.startswith(watch):
- print "%s didn't start with %s" % (run_dir, watch)
+ LOGGER.info("%s didn't start with %s" % (run_dir, watch))
continue
if watch[-1] != os.path.sep:
watch += os.path.sep
def main(args=None):
bot = SpoolWatcher()
return bot.main(args)
-
+
if __name__ == "__main__":
ret = main(sys.argv[1:])
#sys.exit(ret)
projects / htsworkflow.git / blobdiff