#!/usr/bin/env python
#
# $Id: PipeOutputWindow.py,v 1.8 2001/11/03 11:05:22 doughellmann Exp $
#
# Copyright 2001 Doug Hellmann.
#
#
# All Rights Reserved
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose and without fee is hereby
# granted, provided that the above copyright notice appear in all
# copies and that both that copyright notice and this permission
# notice appear in supporting documentation, and that the name of Doug
# Hellmann not be used in advertising or publicity pertaining to
# distribution of the software without specific, written prior
# permission.
#
# DOUG HELLMANN DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
# NO EVENT SHALL DOUG HELLMANN BE LIABLE FOR ANY SPECIAL, INDIRECT OR
# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
"""Display output of a piped command with optional colorization.
Use this widget to run shell processes in the background and display
their output. The window automatically scrolls to follow the
output, and each line is compared against various regular
expressions to control the presentation options.
Refer to the example application in the module for references as to
how to add tags and patterns.
"""
__rcs_info__ = {
#
# Creation Information
#
'module_name' : '$RCSfile: PipeOutputWindow.py,v $',
'rcs_id' : '$Id: PipeOutputWindow.py,v 1.8 2001/11/03 11:05:22 doughellmann Exp $',
'creator' : 'Doug Hellmann <doug@hellfly.net>',
'project' : 'PmwContribD',
'created' : 'Sun, 01-Apr-2001 13:11:52 EDT',
#
# Current Information
#
'author' : '$Author: doughellmann $',
'version' : '$Revision: 1.8 $',
'date' : '$Date: 2001/11/03 11:05:22 $',
}
#
# Import system modules
#
import Tkinter
import Pmw
import re
import os
import types
#
# Import Local modules
#
#
# Module
#
#
# Functions which return status information about
# how a child process died or exited.
#
def WIFEXITED(status):
"Did the program exit with no errors?"
return ((status & 0xff) == 0)
def WIFSIGNALED(status):
"Did the program exit as a result of a signal?"
return (((status & 0xff) != 0) and ((status & 0xff) != 0x7f))
def WEXITSTATUS(status):
"What was the exit code of the progrm?"
return ((status & 0xff00) >> 8)
def WTERMSIG(status):
"Did the program exit as a result of the TERM signal?"
return (status & 0x7f)
class GuiAppDServicesForPipeOutputWindow:
"""Class which defines the GuiAppD methods needed by PipeOutputWindow.
The PipeOutputWindow expects to have certain methods available in
the GuiAppD argument to its constructor. This class enumerates
those methods, and provides a default implementation which does
nothing when called that is used as a default parameter to the
PipeOutputWindow constructor.
"""
def showMessage(self, type, msgTxt=None):
return
def busyStart(self, newcursor=None):
return
def busyEnd(self):
return
class PipeOutputWindow(Pmw.ScrolledText):
"""Text window which controls background processes and their output.
A ScrolledText window which can control background processes.
Each child process is run and the output is captured to the text
window. Patterns can be applied. to the output so that color
highlighting can be used on a line by line basis.
To create one, a GuiAppD instance must be passed in addition to
the parent widget. The following services provided by the GuiAppD
are used by this widget:
- 'showMessage()'
- 'busyStart()'
- 'busyEnd()'
Any object which can provide these same services can be used in
place of an actual GuiAppD.
Options
'errorhandler' -- a function to call when an error occurs, takes
a single string as an argument
'labelpos' -- Position of label.
'logfile' -- a file where the output of child processes should
be written (in addition to being displayed on the screen)
'state' -- the text widget is usually created disabled
"""
def __init__(self, parent, guiapp=GuiAppDServicesForPipeOutputWindow(), **kw):
"""Initialize a PipeOutputWindow instance.
In addition to the constructor *options* listed above in the
class documentation, this method takes several arguments.
Arguments
parent -- Parent widget.
guiapp -- Reference to a GuiAppD container.
"""
self.guiapp = guiapp
INITOPT = Pmw.INITOPT
optiondefs = (
('state', 'disabled', self.changeState),
('logfile', None, INITOPT),
('errorhandler', None, INITOPT),
('labelpos', 'nw', INITOPT),
('text_wrap', 'none', INITOPT),
)
self.defineoptions(kw, optiondefs)
Pmw.ScrolledText.__init__(self, parent)
self.parent = parent
# pipes from running commands
self.cmdPipes = []
# callbacks associated with commands, to be called
# when the command ends
self.pendingCB = None
# styles associated with output text from commands
self.cmdStyles = []
# dict keyed on regular expressions to match against output
# to set the style of each line of text added to the display
self.cmdReObjs = {}
# the most recently applied tag, to be used for 'continue'
# styles
self.lastTag = None
# list of jobs which have been submitted but not started
self.pendingJobs = []
if self['logfile'] and self['labelpos']:
self.configure(label_text='%s (%s)' % (self['label_text'], self['logfile']))
self.initialiseoptions(self.__class__)
return
def addTagStyle(self, tagName, pattern):
"""Add a tag style recognition pattern.
Arguments
tagName -- Name of the tag (must already be defined) to use.
pattern -- Regular expression (re module) which matches
lines that should use the tag 'tagName'.
"""
self.cmdStyles.append( (pattern, tagName) )
self.cmdReObjs[pattern] = re.compile(pattern)
return
def changeState(self):
self.component('text').configure(state=self['state'])
return
def showCmdOutput(self, outputText, writeToLog=1):
"""
Given some output text, this method
will parse it and tag it to be displayed
in the cmdoutput component.
"""
# get the text widget, so we don't have to
# keep calling this over and over
widget = self.component('text')
# figure out what tag to use based on
# the command style definitions
tag = None
for pattern, style in self.cmdStyles:
matchObj = self.cmdReObjs[pattern].search(outputText)
if matchObj:
tag = style
break
# handle the special case of the 'continue'
# tag, which tells us to keep using the previous
# style
if tag == 'continue':
tag = self.lastTag
elif tag == None:
tag = 'default'
else:
self.lastTag = tag
# set the state so we can insert text into the widget
self.configure(state='normal')
# insert the text
widget.insert('end', outputText, tag)
# write the text to the log file
if writeToLog:
self.logText(outputText)
# scroll down to see the new text
self.component('text').yview_pickplace('end')
# set the widget so that it is not editable
self.configure(text_state='disabled')
return
def showCmdPreview(self, command):
"""Display the command being executed.
Arguments
'command' -- String command to be executed.
"""
self.showCmdCall(command, preview=1)
return
def logText(self, text):
"""Write text to an output log file.
Arguments
'text' -- Content to be written to log file.
"""
if self['logfile']:
logFileHandle = open(self['logfile'], 'a')
logFileHandle.write(text)
logFileHandle.close()
return
def showCmdCall(self, command, preview=0):
"""Add a command call to the command output.
Arguments
'command' -- Shell command to be executed.
'preview=0' -- Boolean flag indicating whether the command
string should be displayed before the output.
"""
self.configure(text_state='normal')
if preview:
self.component('text').insert('end', 'PREVIEW : ',
'commandpreview')
self.component('text').insert('end', ' ')
else:
self.component('text').insert('end', '\n')
self.component('text').insert('end',
' %%%% %s\n' % command,
'commandline')
self.logText('\n')
self.logText('%%%% %s\n' % command)
if not preview:
self.component('text').insert('end', '\n')
self.configure(text_state='disabled')
# scroll down to see the new text
self.component('text').yview_pickplace('end')
return
def closeCmdPipe(self, cmdPipe):
"""Close a command we're following.
Close a command pipe which we've been processing in the
background. We also need to remove the pipe from the list of
pipes to watch, and display a 'finished' message if we were
given one when the pipe was started.
"""
# get rid of the handler
self.stopFileHandler(cmdPipe)
# close the pipe
exitStatus = cmdPipe.close()
messageTemplate = 'The external command: "%s"\n%%s\n\nOther queued jobs will be canceled.\n' % cmdPipe.name
if exitStatus is None:
exitStatus = 0
else:
if WIFSIGNALED(exitStatus) or WTERMSIG(exitStatus):
# Signal
message = messageTemplate % 'was terminated by a signal.'
self.showCmdOutput(message)
if self['errorhandler']:
self['errorhandler'](message)
self.pendingJobs = []
else:
# Exited with status code
exitStatus = WEXITSTATUS(exitStatus)
if exitStatus != 0:
message = messageTemplate % ('exited with error code %d.' % exitStatus)
self.showCmdOutput(message)
if self['errorhandler']:
self['errorhandler'](message)
self.pendingJobs = []
# take the pipe out of the
# list of pipes we're watching
self.cmdPipes.remove(cmdPipe)
# look for a "finished" callback
if self.pendingCB and not self.pendingJobs:
self.pendingCB()
# get rid of the finished message
self.pendingCB = None
# reset the busy state of the app
self.guiapp.showMessage('busy', '')
self.guiapp.busyEnd()
# start the next job, if there is one
self.startBackgroundPipe()
return exitStatus
def readCmdOutput(self, cmdPipe):
"""Read output from the command pipe.
"""
text = cmdPipe.readline()
if not text:
self.showCmdOutput('\n')
self.showCmdOutput('-- Done (%s) --\n' % cmdPipe.name)
self.closeCmdPipe(cmdPipe)
else:
self.showCmdOutput(text)
return
def fileHandler(self, pipe, mode):
"Tkinter file handle processing callback."
if mode == Tkinter.READABLE:
self.readCmdOutput(pipe)
else:
raise IOError(pipe)
return
def stopFileHandler(self, openPipe):
"Stop listening to a file handle."
#print 'Stopping file handler on (%s)' % openPipe.name
Tkinter._tkinter.deletefilehandler(openPipe)
return
def startFileHandler(self, openPipe):
"Start listening to a file handle."
#print 'Starting file handler on (%s)' % openPipe.name
Tkinter._tkinter.createfilehandler(
openPipe,
Tkinter.READABLE,
self.fileHandler)
return
def startBackgroundPipe(self, commands=None, completedCallback=None):
"""Start a command in the background and capture the output.
"""
if type(commands) == types.StringType:
commands = (commands,)
if commands:
for cmd in commands:
self.pendingJobs.append(cmd)
self.pendingCB = completedCallback
if ( not self.cmdPipes ) and ( self.pendingJobs ):
# get the next command and update the queue
command = self.pendingJobs[0]
self.pendingJobs = self.pendingJobs[1:]
# show what we're going to do
self.showCmdCall(command)
# start the command
pipe = os.popen(command, 'r', 1)
self.cmdPipes.append(pipe)
# set the busy state
self.guiapp.busyStart()
self.guiapp.showMessage('busy', command)
self.startFileHandler(pipe)
else:
self._hull.bell()
return
class PipeOutputDialog(Pmw.MegaToplevel):
"""Dialog containing a PipeOutputWindow.
Components
'errorhandler' -- A function to call when an error occurs, takes
a single string as an argument
'jobmanager' -- PipeOutputWindow instance.
'logfile' -- File to which output should be written, in addition
to being displayed on the screen.
"""
def __init__(self, parent, guiapp, **kw):
self.guiapp = guiapp
INITOPT = Pmw.INITOPT
optiondefs = (
('logfile', None, INITOPT),
('errorhandler', None, INITOPT),
('jobmanager_hscrollmode', 'static', INITOPT),
('jobmanager_vscrollmode', 'static', INITOPT),
('jobmanager_text_wrap', 'none', INITOPT),
('jobmanager_text_background', 'white', INITOPT),
('jobmanager_text_height', 40, INITOPT),
('jobmanager_text_width', 80, INITOPT),
('jobmanager_labelpos', None, INITOPT),
)
self.defineoptions(kw, optiondefs)
Pmw.MegaToplevel.__init__(self, parent)
#
# Don't delete the window when they hit the
# X, just take it off of the screen.
#
self.protocol('WM_DELETE_WINDOW', self.withdraw)
self.createinterior()
self.initialiseoptions(self.__class__)
if self['logfile']:
self.configure(title='%s (%s)' % (self['title'], self['logfile']))
#
# Don't show ourself yet.
#
self.withdraw()
return
def showCmdPreview(self, command):
"Open the dialog and show the command in preview mode."
self.deiconify()
self.component('jobmanager').showCmdPreview(command)
return
def startBackgroundPipe(self, commands=None, completedCallback=None):
"""Open the dialog and start executing the command.
Arguments
'commands' -- Sequence of commands to execute.
'completedCallback' -- Callback to execute when all commands
are complete.
"""
self.deiconify()
self.component('jobmanager').startBackgroundPipe(commands, completedCallback)
return
def findtarget(self):
return self.component('jobmanager')
def createinterior(self):
self.__cmdout = self.createcomponent(
'jobmanager',
(), None,
PipeOutputWindow,
(self.interior(), self.guiapp,),
errorhandler=self['errorhandler'],
logfile=self['logfile'],
)
self.__cmdout.pack(
side=Tkinter.TOP,
expand=Tkinter.YES,
fill=Tkinter.BOTH,
)
return
#
# The PipeOutputDialog should behave just like the
# PipeOutputWindow, so here we use the Pmw utility
# for inheriting the methods.
#
Pmw.forwardmethods(PipeOutputDialog, PipeOutputWindow,
PipeOutputDialog.findtarget)
if __name__ == '__main__':
from GuiAppD import GuiAppD
import Tkinter
class TestGui(GuiAppD):
appname='PipeOutputWindow Test App'
def loadTestLog(self):
self.busyStart()
from tkFileDialog import askopenfilename
filename = askopenfilename()
if not filename:
self.busyEnd()
return
f = open(filename, 'r')
for line in f.readlines():
self.component('jobmanagerdlg').showCmdOutput(line, 0)
self.component('jobmanagerdlg').update_idletasks()
f.close()
self.busyEnd()
def exitZero(self, component):
jobs = (
'echo this should exit zero',
'echo you should see this line too',
'ls -lR . 2>&1',
)
self.component(component).startBackgroundPipe(jobs)
return
def exitOne(self, component):
jobs = (
'echo this should exit one; exit 1',
'echo you should not see this',
)
self.component(component).startBackgroundPipe(jobs)
return
def exitThree(self, component):
jobs = (
'echo this should exit three; exit 3',
'echo you should not see this',
)
self.component(component).startBackgroundPipe(jobs)
return
def showDialog(self):
self.component('jobmanagerdlg').deiconify()
return
def createInterface(self):
self.createcomponent(
'Exit 0 (Main)',
(), None,
Tkinter.Button,
(self.interior(),),
command=lambda s=self: s.exitZero('jobmanager'),
text='Exit 0 (Main)',
).pack()
self.createcomponent(
'Exit 0 (Dialog)',
(), None,
Tkinter.Button,
(self.interior(),),
command=lambda s=self: s.exitZero('jobmanagerdlg'),
text='Exit 0 (Dialog)',
).pack()
self.createcomponent(
'Exit 1 (Main)',
(), None,
Tkinter.Button,
(self.interior(),),
command=lambda s=self: s.exitOne('jobmanager'),
text='Exit 1 (Main)',
).pack()
self.createcomponent(
'Exit 1 (Dialog)',
(), None,
Tkinter.Button,
(self.interior(),),
command=lambda s=self: s.exitOne('jobmanagerdlg'),
text='Exit 1 (Dialog)',
).pack()
self.createcomponent(
'Exit 3 (Main)',
(), None,
Tkinter.Button,
(self.interior(),),
command=lambda s=self: s.exitThree('jobmanager'),
text='Exit 3 (Main)',
).pack()
self.createcomponent(
'Exit 3 (Dialog)',
(), None,
Tkinter.Button,
(self.interior(),),
command=lambda s=self: s.exitThree('jobmanagerdlg'),
text='Exit 3 (Dialog)',
).pack()
self.createcomponent(
'Show Dialog',
(), None,
Tkinter.Button,
(self.interior(),),
command=self.showDialog,
text='Show Dialog',
).pack()
self.createcomponent(
'Load Test Log',
(), None,
Tkinter.Button,
(self.interior(),),
command=self.loadTestLog,
text='Load Test Log',
).pack()
cmdout = self.createcomponent(
'jobmanager',
(), None,
PipeOutputWindow,
(self.interior(), self,),
#text_height=15,
text_height=10,
hscrollmode='static',
vscrollmode='static',
text_wrap='none',
text_background='white',
labelpos='nw',
label_text='Command Log',
errorhandler=self.showError,
)
cmdout.pack(side=Tkinter.BOTTOM,
expand=Tkinter.YES,
fill=Tkinter.BOTH)
cmdoutdlg = self.createcomponent(
'jobmanagerdlg',
(), None,
PipeOutputDialog,
(self.interior(), self,),
#text_height=15,
jobmanager_text_height=10,
#hscrollmode='static',
#vscrollmode='static',
errorhandler=self.showError,
)
self.computedFonts = {}
self.computedFonts['helvetica'] = Pmw.logicalfont('Helvetica', -1, weight='bold')
self.computedFonts['times'] = Pmw.logicalfont('Times', 0)
self.computedFonts['courier'] = Pmw.logicalfont('Courier', -1)
self.computedFonts['courierbold'] = Pmw.logicalfont('Courier', -1, weight='bold')
#
# Special tag name for showing commands
#
cmdout.tag_configure('commandline',
font=self.computedFonts['courier'],
relief=Tkinter.RAISED,
background='#cccccc',
borderwidth=2,
)
#
# Define the faces for the widget.
#
cmdout.tag_configure('default',
font=self.computedFonts['courier'])
cmdout.tag_configure('command_complete',
font=self.computedFonts['courier'],
relief=Tkinter.RAISED,
background='#cccccc',
borderwidth=2,
)
cmdout.tag_configure('directory',
font=self.computedFonts['courier'],
background='yellow',
)
cmdout.tag_configure('subdirectory',
font=self.computedFonts['courier'],
foreground='#cccccc',
)
cmdout.tag_configure('executable',
font=self.computedFonts['courier'],
foreground='#00ff00',
)
cmdout.tag_configure('error',
font=self.computedFonts['courier'],
background='red',
)
#
# Map the output patterns to faces
#
cmdout.addTagStyle('command_complete', '^--.+--$')
cmdout.addTagStyle('directory', '^[^:]+:$')
cmdout.addTagStyle('subdirectory', '^d')
cmdout.addTagStyle('executable', '^[^d]..x..[x-]..[x-]')
cmdout.addTagStyle('error', '^ls: [^:]*: .*')
cmdout.addTagStyle('default', '.*')
return
TestGui().run()
|