13 from pyinotify import EventsCodes
15 from benderjab import rpc
18 s_cycles = re.compile('No. Cycles: (?P<cycles>[0-9]+)')
20 def get_cycles(run_dir="."):
22 Find the number of cycles from the Recipe*.xml file found in run_dir.
25 file_path_list = glob.glob(os.path.join(run_dir, "Recipe*.xml"))
28 if len(file_path_list) == 0:
29 msg = "Recipe xml file not found."
32 elif len(file_path_list) > 1:
33 msg = "%s Recipe files found, expected 1." % (len(file_path_list))
36 f = open(file_path_list[0], 'r')
38 #Find the line of the file with the cycle number in it.
40 mo = s_cycles.search(line)
46 # Process the line with the cycle number in it.
47 cycle_num = int(mo.group('cycles'))
53 class WatcherEvents(object):
54 # two events need to be tracked
55 # one to send startCopy
56 # one to send OMG its broken
57 # OMG its broken needs to stop when we've seen enough
59 # this should be per runfolder.
65 class Handler(pyinotify.ProcessEvent):
66 def __init__(self, watchmanager, bot):
67 self.last_event_time = None
68 self.watchmanager = watchmanager
71 def process_IN_CREATE(self, event):
72 self.last_event_time = time.time()
73 msg = "Create: %s" % os.path.join(event.path, event.name)
74 if event.name.lower() == "run.completed":
76 self.bot.sequencingFinished(event.path)
78 logging.error("Couldn't send sequencingFinished")
81 def process_IN_DELETE(self, event):
82 logging.debug("Remove: %s" % os.path.join(event.path, event.name))
84 def process_IN_UNMOUNT(self, event):
85 self.bot.unmount_watch()
87 class SpoolWatcher(rpc.XmlRpcBot):
89 Watch a directory and send a message when another process is done writing.
91 This monitors a directory tree using inotify (linux specific) and
92 after some files having been written will send a message after <timeout>
93 seconds of no file writing.
95 (Basically when the solexa machine finishes dumping a round of data
96 this'll hopefully send out a message saying hey look theres data available
99 # these params need to be in the config file
100 # I wonder where I should put the documentation
102 # `watchdir` - which directory tree to monitor for modifications
103 # `profile` - specify which .gaworkflow profile to use
104 # `write_timeout` - how many seconds to wait for writes to finish to
106 # `notify_timeout` - how often to timeout from notify
108 def __init__(self, section=None, configfile=None):
109 #if configfile is None:
110 # self.configfile = "~/.gaworkflow"
111 super(SpoolWatcher, self).__init__(section, configfile)
113 self.cfg['watchdir'] = None
114 self.cfg['write_timeout'] = 10
115 self.cfg['notify_users'] = None
116 self.cfg['notify_runner'] = None
118 self.notify_timeout = 0.001
119 self.wm = pyinotify.WatchManager()
120 self.handler = Handler(self.wm, self)
121 self.notifier = pyinotify.Notifier(self.wm, self.handler)
124 self.notify_users = None
125 self.notify_runner = None
127 self.eventTasks.append(self.process_notify)
129 def read_config(self, section=None, configfile=None):
130 super(SpoolWatcher, self).read_config(section, configfile)
132 self.watch_dir = self._check_required_option('watchdir')
133 self.write_timeout = int(self.cfg['write_timeout'])
135 self.notify_users = self._parse_user_list(self.cfg['notify_users'])
137 self.notify_runner = \
138 self._parse_user_list(self.cfg['notify_runner'],
139 require_resource=True)
140 except bot.JIDMissingResource:
141 msg = 'need a full jabber ID + resource for xml-rpc destinations'
143 raise bot.JIDMissingResource(msg)
145 def add_watch(self, watchdir=None):
147 start watching watchdir or self.watch_dir
148 we're currently limited to watching one directory tree.
150 # the one tree limit is mostly because self.wdd is a single item
151 # but managing it as a list might be a bit more annoying
153 watchdir = self.watch_dir
154 logging.info("Watching:"+str(watchdir))
155 mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
156 # rec traverses the tree and adds all the directories that are there
158 # auto_add will add in new directories as they are created
159 self.wdd = self.wm.add_watch(watchdir, mask, rec=True, auto_add=True)
161 def unmount_watch(self):
162 if self.wdd is not None:
163 logging.debug("disabling watch")
164 logging.debug(str(self.wdd))
165 self.wm.rm_watch(self.wdd.values())
168 def process_notify(self, *args):
169 # process the queue of events as explained above
170 self.notifier.process_events()
171 #check events waits timeout
172 if self.notifier.check_events(self.notify_timeout):
173 # read notified events and enqeue them
174 self.notifier.read_events()
175 # should we do something?
176 last_event_time = self.handler.last_event_time
177 if last_event_time is not None:
178 time_delta = time.time() - last_event_time
179 if time_delta > self.write_timeout:
181 self.handler.last_event_time = None
183 def _parser(self, msg, who):
185 Parse xmpp chat messages
187 help = u"I can send [copy] message, or squencer [finished]"
188 if re.match(u"help", msg):
190 elif re.match("copy", msg):
192 reply = u"sent copy message"
193 elif re.match(u"finished", msg):
196 self.sequencingFinished(words[1])
197 reply = u"sending sequencing finished for %s" % (words[1])
199 reply = u"need runfolder name"
201 reply = u"I didn't understand '%s'" %(msg)
204 def start(self, daemonize):
209 super(SpoolWatcher, self).start(daemonize)
215 # destroy the inotify's instance on this interrupt (stop monitoring)
217 super(SpoolWatcher, self).stop()
220 logging.debug("writes seem to have stopped")
221 if self.notify_runner is not None:
222 for r in self.notify_runner:
223 self.rpc_send(r, tuple(), 'startCopy')
225 def sequencingFinished(self, run_dir):
226 # need to strip off self.watch_dir from rundir I suspect.
227 logging.info("run.completed in " + str(run_dir))
228 pattern = self.watch_dir
229 if pattern[-1] != os.path.sep:
230 pattern += os.path.sep
231 stripped_run_dir = re.sub(pattern, "", run_dir)
232 logging.debug("stripped to " + stripped_run_dir)
233 if self.notify_users is not None:
234 for u in self.notify_users:
235 self.send(u, 'Sequencing run %s finished' % (stripped_run_dir))
236 if self.notify_runner is not None:
237 for r in self.notify_runner:
238 self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
242 return bot.main(args)
244 if __name__ == "__main__":
245 sys.exit(main(sys.argv[1:]))