import time
from htsworkflow.util import mount
+from htsworkflow.automation.solexa import is_runfolder, get_top_dir
# this uses pyinotify
import pyinotify
from pyinotify import EventsCodes
+IN_CREATE = EventsCodes.ALL_FLAGS['IN_CREATE']
+IN_UNMOUNT = EventsCodes.ALL_FLAGS['IN_UNMOUNT']
from benderjab import rpc
-def is_runfolder(name):
- """
- Is it a runfolder?
-
- >>> print is_runfolder('090630_HWUSI-EAS999_0006_30LNFAAXX')
- True
- >>> print is_runfolder('hello')
- False
- """
- if re.match("[0-9]{6}_.*", name):
- return True
- else:
- return False
-
-def get_top_dir(root, path):
- """
- Return the directory in path that is a subdirectory of root.
- e.g.
-
- >>> print get_top_dir('/a/b/c', '/a/b/c/d/e/f')
- d
- >>> print get_top_dir('/a/b/c/', '/a/b/c/d/e/f')
- d
- >>> print get_top_dir('/a/b/c', '/g/e/f')
- None
- >>> print get_top_dir('/a/b/c', '/a/b/c')
- <BLANKLINE>
- """
- if path.startswith(root):
- subpath = path[len(root):]
- if subpath.startswith('/'):
- subpath = subpath[1:]
- return subpath.split(os.path.sep)[0]
- else:
- return None
-
class WatcherEvent(object):
"""
Track information about a file event
return u"<WatchEvent: %s %s %s>" % (time.ctime(self.time), self.event_root, complete)
class Handler(pyinotify.ProcessEvent):
- def __init__(self, watchmanager, bot, completion_file=None):
+ def __init__(self, watchmanager, bot, completion_files=None):
"""
Completion file contains current "completion" filename
"""
self.last_event = {}
self.watchmanager = watchmanager
self.bot = bot
- if completion_file is not None:
- completion_file = completion_file.lower()
- self.completion_file = completion_file
+ self.log = bot.log
+ if completion_files is not None:
+ completion_files = [ x.lower() for x in completion_files ]
+ self.completion_files = completion_files
def process_IN_CREATE(self, event):
for wdd in self.bot.wdds:
runfolder = os.path.join(watch_path, target)
if not is_runfolder(target):
- logging.debug("Skipping %s, not a runfolder" % (target,))
+ self.log.debug("Skipping %s, not a runfolder" % (target,))
continue
# grab the previous events for this watch path
# main analysis, which means this completion code might get tripped because of it
# so we need to make sure we're getting the completion file in the root of the
# runfolder
- if (self.completion_file == event.name.lower() and event.path == runfolder) \
+ event_name = event.name.lower()
+ if (event_name in self.completion_files and event.path == runfolder) \
or run_already_complete:
self.last_event[watch_path][target].complete = True
msg += "(completed)"
- logging.debug(msg)
+ self.log.debug(msg)
def process_IN_DELETE(self, event):
- logging.debug("Remove: %s" % os.path.join(event.path, event.name))
+ self.log.debug("Remove: %s" % os.path.join(event.path, event.name))
pass
def process_IN_UNMOUNT(self, event):
pathname = os.path.join(event.path, event.name)
- logging.debug("IN_UNMOUNT: %s" % (pathname,))
+ self.log.debug("IN_UNMOUNT: %s" % (pathname,))
self.bot.unmount_watch(event.path)
class SpoolWatcher(rpc.XmlRpcBot):
# `write_timeout` - how many seconds to wait for writes to finish to
# the spool
# `notify_timeout` - how often to timeout from notify
- # `completion_file` - what file indicates we've finished sequencing
+ # `completion_files` - what files indicates we've finished sequencing
# defaults to: netcopy_complete.txt
def __init__(self, section=None, configfile=None):
self.cfg['write_timeout'] = 10
self.cfg['notify_users'] = None
self.cfg['notify_runner'] = None
- self.cfg['completion_file'] = 'netcopy_complete.txt'
+ self.cfg['completion_files'] = 'ImageAnalysis_Netcopy_complete_READ2.txt ImageAnalysis_Netcopy_complete_SINGLEREAD.txt'
self.watchdirs = []
self.watchdir_url_map = {}
self.watchdir_url_map[watchdir] = self.cfg.get(watchdir, watchdir)
self.write_timeout = int(self.cfg['write_timeout'])
- self.completion_file = self.cfg['completion_file']
+ self.completion_files = shlex.split(self.cfg['completion_files'])
self.notify_users = self._parse_user_list(self.cfg['notify_users'])
try:
# create the watch managers if we need them
if self.wm is None:
self.wm = pyinotify.WatchManager()
- self.handler = Handler(self.wm, self, self.completion_file)
+ self.handler = Handler(self.wm, self, self.completion_files)
self.notifier = pyinotify.Notifier(self.wm, self.handler)
# the one tree limit is mostly because self.wdd is a single item
if watchdirs is None:
watchdirs = self.watchdirs
- mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
+ mask = IN_CREATE | IN_UNMOUNT
# rec traverses the tree and adds all the directories that are there
# at the start.
# auto_add will add in new directories as they are created
mounts.append(w)
self.mounts_to_watches[mount_location] = mounts
- logging.info(u"Watching:"+unicode(w))
+ self.log.info(u"Watching:"+unicode(w))
self.wdds.append(self.wm.add_watch(w, mask, rec=True, auto_add=True))
def unmount_watch(self, event_path):
# the list getting shorter
for i in range(len(self.wdds),0, -1):
wdd = self.wdds[i]
- logging.info(u'unmounting: '+unicode(wdd.items()))
+ self.log.info(u'unmounting: '+unicode(wdd.items()))
self.wm.rm_watch(wdd.values())
del self.wdds[i]
self.mounted = False
if root_copy_url[-1] != '/':
root_copy_url += '/'
copy_url = root_copy_url + list_event_dir
- logging.debug('Copy url: %s' % (copy_url,))
+ self.log.debug('Copy url: %s' % (copy_url,))
return copy_url
def process_notify(self, *args):
# restart the watch
for watch in self.mounts_to_watches[mount_point]:
self.add_watch(watch)
- logging.info(
+ self.logg.info(
"%s was remounted, restarting watch" % \
(mount_point)
)
super(SpoolWatcher, self).stop()
def startCopy(self, copy_url=None):
- logging.debug("writes seem to have stopped")
+ self.log.debug("writes seem to have stopped")
if self.notify_runner is not None:
for r in self.notify_runner:
self.rpc_send(r, tuple([copy_url]), 'startCopy')
def sequencingFinished(self, run_dir):
# need to strip off self.watchdirs from rundir I suspect.
- logging.info("run.completed in " + str(run_dir))
+ 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)
else:
stripped_run_dir = run_dir
- logging.debug("stripped to " + stripped_run_dir)
+ self.log.debug("stripped to " + stripped_run_dir)
if self.notify_users is not None:
for u in self.notify_users:
self.send(u, 'Sequencing run %s finished' % \
projects / htsworkflow.git / blobdiff