10 from htsworkflow.util import mount
14 from pyinotify import EventsCodes
16 from benderjab import rpc
19 def get_top_dir(root, path):
21 Return the directory in path that is a subdirectory of root.
24 >>> print get_top_dir('/a/b/c', '/a/b/c/d/e/f')
26 >>> print get_top_dir('/a/b/c/', '/a/b/c/d/e/f')
28 >>> print get_top_dir('/a/b/c', '/g/e/f')
30 >>> print get_top_dir('/a/b/c', '/a/b/c')
33 if path.startswith(root):
34 subpath = path[len(root):]
35 if subpath.startswith('/'):
37 return subpath.split(os.path.sep)[0]
41 class WatcherEvents(object):
42 # two events need to be tracked
43 # one to send startCopy
44 # one to send OMG its broken
45 # OMG its broken needs to stop when we've seen enough
47 # this should be per runfolder.
53 class Handler(pyinotify.ProcessEvent):
54 def __init__(self, watchmanager, bot, ipar=False):
56 ipar flag indicates we should wait for ipar to finish, instead of
57 just the run finishing
60 self.watchmanager = watchmanager
64 self.last_file = 'IPAR_Netcopy_Complete.txt'.lower()
66 self.last_file = "run.completed".lower()
68 def process_IN_CREATE(self, event):
69 for wdd in self.bot.wdds:
70 for watch_path in self.bot.watchdirs:
71 if event.path.startswith(watch_path):
72 target = get_top_dir(watch_path, event.path)
73 self.last_event.setdefault(watch_path, {})[target] = time.time()
75 msg = "Create: %s %s" % (event.path, event.name)
77 if event.name.lower() == self.last_file:
79 self.bot.sequencingFinished(event.path)
81 logging.error("Couldn't send sequencingFinished")
84 def process_IN_DELETE(self, event):
85 logging.debug("Remove: %s" % os.path.join(event.path, event.name))
87 def process_IN_UNMOUNT(self, event):
88 pathname = os.path.join(event.path, event.name)
89 logging.debug("IN_UNMOUNT: %s" % (pathname,))
90 self.bot.unmount_watch(event.path)
92 class SpoolWatcher(rpc.XmlRpcBot):
94 Watch a directory and send a message when another process is done writing.
96 This monitors a directory tree using inotify (linux specific) and
97 after some files having been written will send a message after <timeout>
98 seconds of no file writing.
100 (Basically when the solexa machine finishes dumping a round of data
101 this'll hopefully send out a message saying hey look theres data available
104 # these params need to be in the config file
105 # I wonder where I should put the documentation
107 # `watchdirs` - list of directories to monitor for modifications
108 # `profile` - specify which .htsworkflow profile to use
109 # `write_timeout` - how many seconds to wait for writes to finish to
111 # `notify_timeout` - how often to timeout from notify
113 def __init__(self, section=None, configfile=None):
114 #if configfile is None:
115 # self.configfile = "~/.htsworkflow"
116 super(SpoolWatcher, self).__init__(section, configfile)
118 self.cfg['watchdirs'] = None
119 self.cfg['write_timeout'] = 10
120 self.cfg['notify_users'] = None
121 self.cfg['notify_runner'] = None
122 self.cfg['wait_for_ipar'] = 0
124 self.notify_timeout = 0.001
125 self.wm = pyinotify.WatchManager()
127 # keep track if the specified mount point is currently mounted
128 self.mounted_points = {}
129 # keep track of which mount points tie to which watch directories
130 # so maybe we can remount them.
131 self.mounts_to_watches = {}
133 self.notify_users = None
134 self.notify_runner = None
136 self.eventTasks.append(self.process_notify)
138 def read_config(self, section=None, configfile=None):
139 super(SpoolWatcher, self).read_config(section, configfile)
141 self.watchdirs = shlex.split(self._check_required_option('watchdirs'))
142 self.write_timeout = int(self.cfg['write_timeout'])
143 self.wait_for_ipar = int(self.cfg['wait_for_ipar'])
144 logging.debug('wait for ipar: ' + str(self.cfg['wait_for_ipar']))
146 self.notify_users = self._parse_user_list(self.cfg['notify_users'])
148 self.notify_runner = \
149 self._parse_user_list(self.cfg['notify_runner'],
150 require_resource=True)
151 except bot.JIDMissingResource:
152 msg = 'need a full jabber ID + resource for xml-rpc destinations'
154 raise bot.JIDMissingResource(msg)
156 self.handler = Handler(self.wm, self, self.wait_for_ipar)
157 self.notifier = pyinotify.Notifier(self.wm, self.handler)
159 def add_watch(self, watchdirs=None):
161 start watching watchdir or self.watchdir
162 we're currently limited to watching one directory tree.
164 # the one tree limit is mostly because self.wdd is a single item
165 # but managing it as a list might be a bit more annoying
166 if watchdirs is None:
167 watchdirs = self.watchdirs
169 mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
170 # rec traverses the tree and adds all the directories that are there
172 # auto_add will add in new directories as they are created
174 mount_location = mount.find_mount_point_for(w)
175 self.mounted_points[mount_location] = True
176 mounts = self.mounts_to_watches.get(mount_location, [])
179 self.mounts_to_watches[mount_location] = mounts
181 logging.info(u"Watching:"+unicode(w))
182 self.wdds.append(self.wm.add_watch(w, mask, rec=True, auto_add=True))
184 def unmount_watch(self, event_path):
185 # remove backwards so we don't get weirdness from
186 # the list getting shorter
187 for i in range(len(self.wdds),0, -1):
189 logging.info(u'unmounting: '+unicode(wdd.items()))
190 self.wm.rm_watch(wdd.values())
194 def process_notify(self, *args):
195 # process the queue of events as explained above
196 self.notifier.process_events()
197 #check events waits timeout
198 if self.notifier.check_events(self.notify_timeout):
199 # read notified events and enqeue them
200 self.notifier.read_events()
201 # should we do something?
202 # has something happened?
203 for watchdir, last_events in self.handler.last_event.items():
204 logging.debug('last_events: %s %s' % (watchdir, last_events))
205 for last_event_dir, last_event_time in last_events.items():
206 time_delta = time.time() - last_event_time
207 if time_delta > self.write_timeout:
208 self.startCopy(watchdir, last_event_dir)
209 self.handler.last_event[watchdir] = {}
210 # handle unmounted filesystems
211 for mount_point, was_mounted in self.mounted_points.items():
212 if not was_mounted and mount.is_mounted(mount_point):
213 # we've been remounted. Huzzah!
215 for watch in self.mounts_to_watches[mount_point]:
216 self.add_watch(watch)
218 "%s was remounted, restarting watch" % \
221 self.mounted_points[mount_point] = True
223 def _parser(self, msg, who):
225 Parse xmpp chat messages
227 help = u"I can send [copy] message, or squencer [finished]"
228 if re.match(u"help", msg):
230 elif re.match("copy", msg):
232 reply = u"sent copy message"
233 elif re.match(u"finished", msg):
236 self.sequencingFinished(words[1])
237 reply = u"sending sequencing finished for %s" % (words[1])
239 reply = u"need runfolder name"
241 reply = u"I didn't understand '%s'" %(msg)
244 def start(self, daemonize):
249 super(SpoolWatcher, self).start(daemonize)
255 # destroy the inotify's instance on this interrupt (stop monitoring)
257 super(SpoolWatcher, self).stop()
259 def startCopy(self, watchdir=None, event_path=None):
260 logging.debug("writes seem to have stopped")
261 logging.debug("watchdir = %s, event_path = %s" % (watchdir, event_path))
262 if self.notify_runner is not None:
263 for r in self.notify_runner:
264 self.rpc_send(r, tuple(), 'startCopy')
265 if self.notify_users is not None:
266 for u in self.notify_users:
267 self.send(u, 'startCopy %s %s' % (watchdir, event_path))
269 def sequencingFinished(self, run_dir):
270 # need to strip off self.watchdirs from rundir I suspect.
271 logging.info("run.completed in " + str(run_dir))
272 pattern = self.watch_dir
273 if pattern[-1] != os.path.sep:
274 pattern += os.path.sep
275 stripped_run_dir = re.sub(pattern, "", run_dir)
276 logging.debug("stripped to " + stripped_run_dir)
277 if self.notify_users is not None:
278 for u in self.notify_users:
279 self.send(u, 'Sequencing run %s finished' % (stripped_run_dir))
280 if self.notify_runner is not None:
281 for r in self.notify_runner:
282 self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
286 return bot.main(args)
288 if __name__ == "__main__":
289 sys.exit(main(sys.argv[1:]))
292 # send messages to copier specifying which mount to copy