10 from pyinotify import EventsCodes
12 from benderjab import rpc
14 class WatcherEvents(object):
15 # two events need to be tracked
16 # one to send startCopy
17 # one to send OMG its broken
18 # OMG its broken needs to stop when we've seen enough
20 # this should be per runfolder.
26 class Handler(pyinotify.ProcessEvent):
27 def __init__(self, watchmanager, bot):
28 self.last_event_time = None
29 self.watchmanager = watchmanager
32 def process_IN_CREATE(self, event):
33 self.last_event_time = time.time()
34 msg = "Create: %s" % os.path.join(event.path, event.name)
35 if event.name.lower() == "run.completed":
37 self.bot.sequencingFinished(event.path)
39 logging.error("Couldn't send sequencingFinished")
42 def process_IN_DELETE(self, event):
43 logging.debug("Remove: %s" % os.path.join(event.path, event.name))
45 def process_IN_UNMOUNT(self, event):
46 self.bot.unmount_watch()
48 class SpoolWatcher(rpc.XmlRpcBot):
50 Watch a directory and send a message when another process is done writing.
52 This monitors a directory tree using inotify (linux specific) and
53 after some files having been written will send a message after <timeout>
54 seconds of no file writing.
56 (Basically when the solexa machine finishes dumping a round of data
57 this'll hopefully send out a message saying hey look theres data available
60 # these params need to be in the config file
61 # I wonder where I should put the documentation
63 # `watchdir` - which directory tree to monitor for modifications
64 # `profile` - specify which .gaworkflow profile to use
65 # `write_timeout` - how many seconds to wait for writes to finish to
67 # `notify_timeout` - how often to timeout from notify
69 def __init__(self, section=None, configfile=None):
70 #if configfile is None:
71 # self.configfile = "~/.gaworkflow"
72 super(SpoolWatcher, self).__init__(section, configfile)
74 self.cfg['watchdir'] = None
75 self.cfg['write_timeout'] = 10
76 self.cfg['notify_users'] = None
77 self.cfg['notify_runner'] = None
79 self.notify_timeout = 0.001
80 self.wm = pyinotify.WatchManager()
81 self.handler = Handler(self.wm, self)
82 self.notifier = pyinotify.Notifier(self.wm, self.handler)
85 self.notify_users = None
86 self.notify_runner = None
88 self.eventTasks.append(self.process_notify)
90 def read_config(self, section=None, configfile=None):
91 super(SpoolWatcher, self).read_config(section, configfile)
93 self.watch_dir = self._check_required_option('watchdir')
94 self.write_timeout = int(self.cfg['write_timeout'])
96 self.notify_users = self._parse_user_list(self.cfg['notify_users'])
98 self.notify_runner = \
99 self._parse_user_list(self.cfg['notify_runner'],
100 require_resource=True)
101 except bot.JIDMissingResource:
102 msg = 'need a full jabber ID + resource for xml-rpc destinations'
104 raise bot.JIDMissingResource(msg)
106 def add_watch(self, watchdir=None):
108 start watching watchdir or self.watch_dir
109 we're currently limited to watching one directory tree.
111 # the one tree limit is mostly because self.wdd is a single item
112 # but managing it as a list might be a bit more annoying
114 watchdir = self.watch_dir
115 logging.info("Watching:"+str(watchdir))
116 mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
117 # rec traverses the tree and adds all the directories that are there
119 # auto_add will add in new directories as they are created
120 self.wdd = self.wm.add_watch(watchdir, mask, rec=True, auto_add=True)
122 def unmount_watch(self):
123 if self.wdd is not None:
124 logging.debug("disabling watch")
125 logging.debug(str(self.wdd))
126 self.wm.rm_watch(self.wdd.values())
129 def process_notify(self, *args):
130 # process the queue of events as explained above
131 self.notifier.process_events()
132 #check events waits timeout
133 if self.notifier.check_events(self.notify_timeout):
134 # read notified events and enqeue them
135 self.notifier.read_events()
136 # should we do something?
137 last_event_time = self.handler.last_event_time
138 if last_event_time is not None:
139 time_delta = time.time() - last_event_time
140 if time_delta > self.write_timeout:
142 self.handler.last_event_time = None
144 def _parser(self, msg, who):
146 Parse xmpp chat messages
148 help = u"I can send [copy] message, or squencer [finished]"
149 if re.match(u"help", msg):
151 elif re.match("copy", msg):
153 reply = u"sent copy message"
154 elif re.match(u"finished", msg):
157 self.sequencingFinished(words[1])
158 reply = u"sending sequencing finished for %s" % (words[1])
160 reply = u"need runfolder name"
162 reply = u"I didn't understand '%s'" %(msg)
165 def start(self, daemonize):
170 super(SpoolWatcher, self).start(daemonize)
176 # destroy the inotify's instance on this interrupt (stop monitoring)
178 super(SpoolWatcher, self).stop()
181 logging.debug("writes seem to have stopped")
182 if self.notify_runner is not None:
183 for r in self.notify_runner:
184 self.rpc_send(r, tuple(), 'startCopy')
186 def sequencingFinished(self, run_dir):
187 # need to strip off self.watch_dir from rundir I suspect.
188 logging.info("run.completed in " + str(run_dir))
189 pattern = self.watch_dir
190 if pattern[-1] != os.path.sep:
191 pattern += os.path.sep
192 stripped_run_dir = re.sub(pattern, "", run_dir)
193 logging.debug("stripped to " + stripped_run_dir)
194 if self.notify_users is not None:
195 for u in self.notify_users:
196 self.send(u, 'Sequencing run %s finished' % (stripped_run_dir))
197 if self.notify_runner is not None:
198 for r in self.notify_runner:
199 self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
203 return bot.main(args)
205 if __name__ == "__main__":
206 sys.exit(main(sys.argv[1:]))