#!/usr/bin/python2.4 ######################################################################## # # Time-stamp: <2006-06-07 13:53:47 Jeremy Hankins> # # Interface with wmii # # TODO: # # - Add caching to wmii.proglist # ######################################################################## from optparse import OptionParser import os, sys, re, string, signal, subprocess, dircache, stat def abort(reason, exitcode): program = os.path.basename(sys.argv[0]) print >>sys.stderr, '%s: error: %s' % (program, reason) sys.exit(exitcode) def warn(message): print >>sys.stderr, message def timeout_handler(signum, frame): raise TimedOutError() class TimedOutError(Exception): def __init__(self, value = "Timed Out"): self.value = value def __str__(self): return repr(self.value) class wmiiError(Exception): def __init__(self, value): self.value = value def __str__(self): return `self.value` class wmii_event(subprocess.Popen): def read(self): return string.rstrip(self.stdout.readline()) def close(self, timeout = 1): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) returncode = None try: self.stdout.close() returncode = self.wait() except TimedOutError, KeyboardInterrupt: warn("wmiir process %d killed." % self.pid) os.kill(self.pid, signal.SIGTERM) return(returncode) class wmii: """ Class for objects which interact with a running wmii instance. """ def event(self): """Return a wmii_event object from which events can be read. """ return(wmii_event(("wmiir", "read", "/event"), stdout = subprocess.PIPE)) def menu(self, menu, timeout = None): """Sends menu (a list) to the wmiimenu command and returns the output. By default there is no timeout, since wmiimenu is interactive. """ if timeout: signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: p = subprocess.Popen(("wmiimenu"), stdin = subprocess.PIPE, stdout = subprocess.PIPE) out = p.communicate("\n".join(menu + [""])) except TimedOutError: os.kill(p.pid, signal.SIGTERM) raise TimedOutError("Timed out calling wmiimenu.") if timeout: signal.alarm(0) if p.returncode == 1: return None elif p.returncode != 0: raise wmiiError("Error calling wmiimenu: %d" % p.returncode) return out[0] def read(self, file, timeout = 5): """Reads the contents of file using wmiir. The read is done with a timeout, so this is not appropriate for reading from /event. """ signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: p = subprocess.Popen(("wmiir", "read", "%s" % file), stdout = subprocess.PIPE) data = map(lambda s: string.rstrip(s), p.stdout.readlines()) returncode = p.wait() except TimedOutError: os.kill(p.pid, signal.SIGTERM) raise TimedOutError("Timed out reading from %s." % file) signal.alarm(0) if p.returncode != 0: raise wmiiError("Error reading %s." % file) return(data) def write(self, file, str, timeout = 5): """Writes str to file using wmiir. """ signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: p = subprocess.Popen(("wmiir", "write", "%s" % file), stdin = subprocess.PIPE) p.stdin.write(str) p.stdin.close() returncode = p.wait() except TimedOutError: os.kill(p.pid, signal.SIGTERM) raise TimedOutError("Timed out writing to %s." % file) signal.alarm(0) if returncode != 0: raise wmiiError("Error writing %s." % file) def ls(self, dir): """Returns a list of files in dir. """ return map(lambda s: string.split(s)[-1], self.read(dir)) def match_class(self, xclass_expr): """Returns a list of clients with matching classes. """ exp = re.compile(xclass_expr) l = [] for client in self.ls("/client"): xclass = self.read("/client/%s/class" % client)[0] if exp.search(xclass): l.append(client) return(l) def match_title(self, title_expr): """Returns a list of clients with matching titles. """ exp = re.compile(title_expr) l = [] for client in self.ls("/client"): title = self.read("/client/%s/name" % client)[0] if exp.search(title): l.append(client) return(l) def get_focus(self): """Returns the client with focus. """ return self.read("/view/sel/sel/index")[0] def find_client(self, client): """Returns a list of (view, column, window, client) quads where client can be found. """ numeric = re.compile(r"^\d+$") l = [] for view in self.read("/tags"): for col in filter(lambda s: numeric.match(s), self.ls("/%s" % view)): for win in filter(lambda s: numeric.match(s), self.ls("/%s/%s" % (view, col))): if client == self.read("/%s/%s/%s/index" % (view, col, win))[0]: l.append((view, col, win, client)) break else: # No need to continue looking for a match in this # view if one has already been found, so continue if # no match, otherwise break. continue break return(l) def show_client(self, clients): """Give focus to a client in clients. Priority is given to switching to a client other than the current one, if possible. If there are still multiple options preference is given to options in the current view. After that an option is chosen essentially at random. """ choices = [] for c in clients: choices = choices + self.find_client(c) num = len(choices) path = None if num == 0: return(False) # Filter out the currently displayed client curclient = self.get_focus() choices = filter(lambda t: t[3] != curclient, choices) num = len(choices) if num == 0: # The only option is the current client, so this is a null-op. return(True) elif num == 1: path = choices[0] else: # Still multiple options; check for any in the current view. curview = self.read("/view/name")[0] choices_view = filter(lambda t: t[0] == curview, choices) num_view = len(choices_view) if num_view == 1: path = choices_view[0] else: if num_view > 1: choices = choices_view num = num_view path = choices[0] # Now that a path to a window has been chosen, switch to it. # Selecting in order from most to least specific will eliminate # the possibility that some other window will have focus # temporarily, and maximize responsiveness. # Select the window: self.write("/%s/%s/ctl" % (path[0], path[1]), "select %s" % path[2]) # Select the column: self.write("/%s/ctl" % path[0], "select %s" % path[1]) # Select the view: self.write("/ctl", "view %s" % path[0]) return(True) def show_class(self, class_expr, cmd): """If a client with a matching class can be found, switch to it. Otherwise, run cmd. """ if not self.show_client(self.match_class(class_expr)): self.bg(cmd) def show_title(self, title_expr, cmd): """If a client with a matching title can be found, switch to it. Otherwise, run cmd. """ if not self.show_client(self.match_title(title_expr)): self.bg(cmd) def proglist(self, path = None): """ Generate a listing of executables in path (a list of directories). """ # I really should figure out how to cache this. if not path: path = string.split(os.environ['PATH'], ':') l = [] for d in path: if os.path.isdir(d): for f in dircache.listdir(d): full = os.path.join(d, f) try: mode = os.stat(full)[stat.ST_MODE] isexe = mode & 0111 isdir = stat.S_ISDIR(mode) if isexe and not isdir: l.append(f) except OSError, e: # Pass on file-not-found (broken link, probably), # otherwise re-raise the error. if e.errno != 2: raise e # Now to sort the list & weed out dups: n = len(l) if n > 0: l.sort() last = l[0] lasti = i = 1 while i < n: if l[i] != last: l[lasti] = last = l[i] lasti += 1 i += 1 return l[:lasti] def bg(self, cmd, path = None): """Run a command via a shell, backgrounded. Optionally, set the PATH env variable to path first. """ if path: environment = os.environ.copy() environment['PATH'] = path subprocess.call(cmd + " &", shell = True, env = environment) else: subprocess.call(cmd + " &", shell = True) def run_menu(self, menu, path = None): cmd = self.menu(menu) if cmd: self.bg(cmd, path) if __name__ == '__main__': cli = OptionParser(usage='%prog [options]\nTest access to wmii', version='%prog 0.1') options, args = cli.parse_args() wm = wmii() #print(repr(wm.read("/tags"))) #wm.write("/event", "Write test successful.\n") #wm.run_menu(wm.proglist(os.environ["PATH"].split(":"))) #wm.show_class(r"^Firefox-bin:", "firefox") #print "Menu: %s" % repr(wm.menu(["a", "b", "c"])) #events = wm.event() #print(repr(events.read())) #events.close()