ok here is another try, which reads all output and errors too, in a separate thread and communicates via Queue.
I know it is not perfect(e.g. command with delayed output will not work and there output will get into next commnd for example tryr sleep 1; date) and replicating whole bash not trivial but for few commands i tested it seems to work fine
Regarding API of wx.py.shell I just implemented those method which Shell class was calling for Interpreter, if you go thru source code of Shell you will understand.
basically
- push is where user entered command is sent to interpreter
- getAutoCompleteKeys returns keys
which user can user for
auto completing commands e.g. tab key -
getAutoCompleteList return list of
command matching given text -
getCallTip “Display argument spec and
docstring in a popup window. so for
bash we may show man page 🙂
here is the source code
import threading
import Queue
import time
import wx
import wx.py
from subprocess import Popen, PIPE
class BashProcessThread(threading.Thread):
def __init__(self, readlineFunc):
threading.Thread.__init__(self)
self.readlineFunc = readlineFunc
self.outputQueue = Queue.Queue()
self.setDaemon(True)
def run(self):
while True:
line = self.readlineFunc()
self.outputQueue.put(line)
def getOutput(self):
""" called from other thread """
lines = []
while True:
try:
line = self.outputQueue.get_nowait()
lines.append(line)
except Queue.Empty:
break
return ''.join(lines)
class MyInterpretor(object):
def __init__(self, locals, rawin, stdin, stdout, stderr):
self.introText = "Welcome to stackoverflow bash shell"
self.locals = locals
self.revision = 1.0
self.rawin = rawin
self.stdin = stdin
self.stdout = stdout
self.stderr = stderr
self.more = False
# bash process
self.bp = Popen('bash', shell=False, stdout=PIPE, stdin=PIPE, stderr=PIPE)
# start output grab thread
self.outputThread = BashProcessThread(self.bp.stdout.readline)
self.outputThread.start()
# start err grab thread
self.errorThread = BashProcessThread(self.bp.stderr.readline)
self.errorThread.start()
def getAutoCompleteKeys(self):
return [ord('\t')]
def getAutoCompleteList(self, *args, **kwargs):
return []
def getCallTip(self, command):
return ""
def push(self, command):
command = command.strip()
if not command: return
self.bp.stdin.write(command+"\n")
# wait a bit
time.sleep(.1)
# print output
self.stdout.write(self.outputThread.getOutput())
# print error
self.stderr.write(self.errorThread.getOutput())
app = wx.PySimpleApp()
frame = wx.py.shell.ShellFrame(InterpClass=MyInterpretor)
frame.Show()
app.SetTopWindow(frame)
app.MainLoop()