0

Tkinter GUI から長時間実行される Linux スクリプトを開始できるようにする必要がありますが、同時に停止ボタンを有効にしてプロセスを停止できるようにしたいと考えています。Tkinter も popen もスレッドセーフではありません。popen 関数をスレッドに配置するか、スレッドでボタンを有効にするだけでよいと考えました。私は現在、Python 2.4.3 を使用する Red Hat Linux 5.9 を使用していますが、それ以降のバージョンをオンラインで入手できます。以下のプログラムでは、開始ボタンを停止ボタンに再構成しますが、開始ボタン機能がアクティブであるため機能しませんが、それは私の意図を示しています。stop 関数は単に子に対して os.kill() を実行することに注意してください。

#!/usr/bin/python
import subprocess
import sys
import Tkinter
import tkMessageBox
import signal
import os
import time

class popentest:

    def __init__(self):
        self.mainWindow = Tkinter.Tk()

    def __createUI(self, mainWindow):
        mainWindow.protocol("WM_DELETE_WINDOW", self.OnExitWindow)
        ## Local variables. 
        sleep=5
        iter=5
        self.pid=0
        self.mainWindow.title("Test popen")
        self.start=Tkinter.Button(mainWindow, text=u"Start", command=self.onStart)
        self.start.grid()
        self.kwit = Tkinter.Button(mainWindow,text=u"Quit !",
                                command=self.onQuit)
        self.kwit.grid()
        self.lIter=Tkinter.Label(mainWindow, text="Iterations: ")
        self.iterE=Tkinter.Entry(mainWindow, width=2)
        self.lSleep = Tkinter.Label(mainWindow, text="Sleep time")
        self.sleepEntry = Tkinter.Entry(mainWindow, width=3)
        self.lIter.grid()
        self.iterE.grid()
        self.lSleep.grid()
        self.sleepEntry.grid()
        self.iterE.insert(0, str(iter))
        self.sleepEntry.insert(0,str(sleep))

    def startPopen(self):
        self.__createUI(self.mainWindow)
        self.mainWindow.mainloop()

    def execute(self, numIters, sleep):
        self.p = subprocess.Popen(['./testpopen.sh',str(numIters), str(sleep)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        self.pid=self.p.pid
        print str(self.p.pid)+" started"
        for line in iter(self.p.stdout.readline, ''):
            print line
        self.p.stdout.close()
        self.pid=0
        self.start.configure(text=u"Start", command=self.onStart)

    def onStart(self):
        numIters=self.iterE.get()
        sleep=self.sleepEntry.get()
        if not numIters.isdigit():
            tkMessageBox.showerror(
            "invalid entry",
            "Iteration (%s)must be numeric\n" % numIters)
            return
        elif not sleep.isdigit():
            tkMessageBox.showerror(
            "invalid entry",
            "Sleep(%s) is not numeric\n" % sleep)
            return
        numIters=int(numIters)
        sleep=int(sleep)
        if numIters <= 0 or sleep <=0 :
            tkMessageBox.showerror(
            "invalid entry",
            "Either iteration (%d) or sleep(%d) is <= 0\n" % (numIters, sleep))
        else:
            print "configuring start to stop"
            self.start.configure(text=u"Stop", command=self.onStop)
            time.sleep(1)
            self.execute(numIters, sleep)

    def onStop(self):
        print "configuring stop to start"

        os.kill(p.pid, signal.SIGTERM)
        self.start.configure(text=u"Start", command=self.onStart)

    def OnExitWindow(self):
        if self.pid != 0 :
            os.kill(self.pid, signal.SIGKILL)
        self.mainWindow.destroy()

    def onQuit(self):
        if self.pid != 0 :
            os.kill(self.pid, signal.SIGKILL)
        self.mainWindow.destroy()

if __name__ == "__main__":  
    remote = popentest()
    remote.startPopen()
4

1 に答える 1

1

Popen を使用してプロセスを開始し、ノンブロッキング パイプを使用してプロセスと通信することができます。このようにして、その出力を非同期で受け取ることができます。私はすでに Popen の拡張バージョンを使用しており、コードは ActiveState Python クックブック レシピからのものでした。ウェブ上でレシピを見つけることができなくなりましたが、コードがまだ残っているので、ここに貼り付けました。

https://gist.github.com/mguijarr/6874724

次に、Tkinter コードでタイマーを使用して、プロセスの状態 (終了したかどうか) を定期的にチェックし、出力を取得できます。

self.p = EnhancedPopen.Popen(['./testpopen.sh',str(numIters), str(sleep)],
                             stdin=subprocess.PIPE,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE,
                             shell=True,universal_newlines=True)
self.mainWindow.after(100, self.check_process)

def check_process(self):
  # get stdout output
  output = EnhancedPopen.recv_some(self.p, e=0, stderr=0)
  ...
  if self.p.poll() is not None:
    # process terminated
    ...
    return
  # set timer again (until process exits)
  self.mainWindow.after(100, self.check_process_output)
于 2013-10-07T21:02:35.347 に答える