wxPython/demo/wxProcess.py

   1
   2 from wxPython.wx import *
   3
   4 #----------------------------------------------------------------------
   5
   6 class TestPanel(wxPanel):
   7     def __init__(self, parent, ID, log):
   8         wxPanel.__init__(self, parent, ID)
   9         self.log = log
  10
  11         self.process = None
  12         EVT_IDLE(self, self.OnIdle)
  13
  14         # We can either derive from wxProcess and override OnTerminate
  15         # or we can let wxProcess send this window an event that is
  16         # caught in the normal way...
  17         EVT_END_PROCESS(self, -1, self.OnProcessEnded)
  18
  19
  20         # Make the controls
  21         prompt = wxStaticText(self, -1, 'Command line:')
  22         self.cmd = wxTextCtrl(self, -1, 'python -u data/echo.py')
  23         self.exBtn = wxButton(self, -1, 'Execute')
  24
  25         self.out = wxTextCtrl(self, -1, '', style=wxTE_MULTILINE|wxTE_READONLY)
  26
  27         self.inp = wxTextCtrl(self, -1, '', style=wxTE_PROCESS_ENTER)
  28         self.sndBtn = wxButton(self, -1, 'Send')
  29         self.termBtn = wxButton(self, -1, 'Close Stream')
  30         self.inp.Enable(False)
  31         self.sndBtn.Enable(False)
  32         self.termBtn.Enable(False)
  33
  34         # Hook up the events
  35         EVT_BUTTON(self, self.exBtn.GetId(), self.OnExecuteBtn)
  36         EVT_BUTTON(self, self.sndBtn.GetId(), self.OnSendText)
  37         EVT_BUTTON(self, self.termBtn.GetId(), self.OnCloseStream)
  38         EVT_TEXT_ENTER(self, self.inp.GetId(), self.OnSendText)
  39
  40
  41         # Do the layout
  42         box1 = wxBoxSizer(wxHORIZONTAL)
  43         box1.Add(prompt, 0, wxALIGN_CENTER)
  44         box1.Add(self.cmd, 1, wxALIGN_CENTER|wxLEFT|wxRIGHT, 5)
  45         box1.Add(self.exBtn, 0)
  46
  47         box2 = wxBoxSizer(wxHORIZONTAL)
  48         box2.Add(self.inp, 1, wxALIGN_CENTER)
  49         box2.Add(self.sndBtn, 0, wxLEFT, 5)
  50         box2.Add(self.termBtn, 0, wxLEFT, 5)
  51
  52         sizer = wxBoxSizer(wxVERTICAL)
  53         sizer.Add(box1, 0, wxEXPAND|wxALL, 10)
  54         sizer.Add(self.out, 1, wxEXPAND|wxALL, 10)
  55         sizer.Add(box2, 0, wxEXPAND|wxALL, 10)
  56
  57         self.SetSizer(sizer)
  58         self.SetAutoLayout(True)
  59
  60
  61     def __del__(self):
  62         if self.process is not None:
  63             self.process.Detach()
  64             self.process.CloseOutput()
  65             self.process = None
  66
  67
  68     def OnExecuteBtn(self, evt):
  69         cmd = self.cmd.GetValue()
  70
  71         self.process = wxProcess(self)
  72         self.process.Redirect();
  73         pid = wxExecute(cmd, wxEXEC_ASYNC, self.process)
  74         self.log.write('OnExecuteBtn: "%s" pid: %s\n' % (cmd, pid))
  75
  76         self.inp.Enable(True)
  77         self.sndBtn.Enable(True)
  78         self.termBtn.Enable(True)
  79         self.cmd.Enable(False)
  80         self.exBtn.Enable(False)
  81         self.inp.SetFocus()
  82
  83
  84     def OnSendText(self, evt):
  85         text = self.inp.GetValue()
  86         self.inp.SetValue('')
  87         self.log.write('OnSendText: "%s"\n' % text)
  88         self.process.GetOutputStream().write(text + '\n')
  89         self.inp.SetFocus()
  90
  91
  92     def OnCloseStream(self, evt):
  93         self.log.write('OnCloseStream\n')
  94         #print "b4 CloseOutput"
  95         self.process.CloseOutput()
  96         #print "after CloseOutput"
  97
  98     def OnIdle(self, evt):
  99         if self.process is not None:
 100             stream = self.process.GetInputStream()
 101
 102             if stream.CanRead():
 103                 text = stream.read()
 104                 self.out.AppendText(text)
 105
 106
 107     def OnProcessEnded(self, evt):
 108         self.log.write('OnProcessEnded, pid:%s,  exitCode: %s\n' %
 109                        (evt.GetPid(), evt.GetExitCode()))
 110
 111         stream = self.process.GetInputStream()
 112         if stream.CanRead():
 113             text = stream.read()
 114             self.out.AppendText(text)
 115
 116         self.process.Destroy()
 117         self.process = None
 118         self.inp.Enable(False)
 119         self.sndBtn.Enable(False)
 120         self.termBtn.Enable(False)
 121         self.cmd.Enable(True)
 122         self.exBtn.Enable(True)
 123
 124
 125 #----------------------------------------------------------------------
 126
 127 def runTest(frame, nb, log):
 128     win = TestPanel(nb, -1, log)
 129     return win
 130
 131
 132 #----------------------------------------------------------------------
 133
 134
 135 overview = """\
 136 <html><body>
 137 <h2>wxProcess</h2>
 138
 139 wxProcess lets you get notified when an asyncronous child process
 140 started by wxExecute terminates, and also to get input/output streams
 141 for the child process's stdout, stderr and stdin.
 142
 143 <p>
 144 This demo launches a simple python script that echos back on stdout
 145 lines that it reads from stdin.  You can send text to the echo
 146 process' stdin by typing in the lower textctrl and clicking Send.
 147
 148 <p>
 149 Clicking the Close Stream button will close the demo's end of the
 150 stdin pipe to the child process.  In our case that will cause the
 151 child process to exit its main loop.
 152
 153 </body><html>
 154 """
 155
 156
 157
 158
 159
 160 if __name__ == '__main__':
 161     import sys,os
 162     import run
 163     run.main(['', os.path.basename(sys.argv[0])])