====== Catturare output comando in tempo reale in una gui Python (GTK) ======
Autore: **//Fabio Di Matteo//** \\ Ultima revisione: **// 21/12/2018 - 10:48 //** // //
{{:programmazione:python:python_get_output.png?400|}}
#!/usr/bin/env python
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import GLib, Gtk
import sys, subprocess, shlex, os
import threading
class MainWindow (Gtk.Builder):
textView=None
textbuffer = None
def cmdThread(self,cmd):
print("Comando: %s" % cmd)
txtBuffer=""
out=""
fileOutErr = open("errors.txt","w")
#Lo stdout lo leggo dalla pipe, gli errori li salvo in un file che leggero' dopo
proc = subprocess.Popen(shlex.split(cmd), bufsize=0 ,stdout=subprocess.PIPE, stderr=fileOutErr)
#leggo la pipe finchè non finisce(proc.poll() is None)
while proc.poll() is None :
out = proc.stdout.readline()
txtBuffer=txtBuffer+str(out.decode("utf-8"))
GLib.idle_add(MainWindow.textbuffer.set_text,txtBuffer)
print(str(out.decode("utf-8")))
fileOutErr.close()
#Stampo gli errori dopo la fine della pipe
if (os.stat("errors.txt").st_size != 0):
fileOutErr= open("errors.txt","r")
listErr=fileOutErr.readlines()
txtBuffer=txtBuffer+"\n\nErrori:\n------------------\n\n"
for e in listErr:
txtBuffer=txtBuffer+e
#Aggiungo al text buffer gli errori presi dal file e chiudo il file
GLib.idle_add(MainWindow.textbuffer.set_text,txtBuffer)
fileOutErr.close()
def on_button_clicked(self, textView,entry):
cmdt = threading.Thread(target=MainWindow.cmdThread, args=(self,entry.get_text()),)
cmdt.daemon = True
cmdt.start()
def quit(self,win):
print ("Exit!")
Gtk.main_quit(win)
def __init__(self):
self = Gtk.Builder()
self.add_from_file("builder.ui")
window = self.get_object("window")
entry = self.get_object("entry")
MainWindow.textView = self.get_object("textView")
button = self.get_object("button")
#Init
MainWindow.textbuffer = MainWindow.textView.get_buffer()
MainWindow.textbuffer.set_text("This is some text inside of a Gtk.TextView. "
+ "Select text and click one of the buttons 'bold', 'italic', "
+ "or 'underline' to modify the text accordingly.")
if (sys.platform == "win32"):
entry.set_text("./rsync-win/rsync.exe -avz /cygdrive/c/msys64/mingw64/bin /cygdrive/c/users/admin/desktop/tt/")
window.show_all()
window.connect("delete-event", MainWindow.quit)
button.connect("clicked", MainWindow.on_button_clicked, MainWindow.textView,entry)
myWindow = MainWindow()
Gtk.main()
**builder.ui**
===== Catturare progresso totale di rsync in una GtkProgressbar =====
{{:programmazione:python:python_get_output.png?400|}}
Per questo esperimento useremo l'opzione di rsync **--info=progress2 **
**get-rsync-progress.py**
#!/usr/bin/env python
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import GLib, Gtk
import sys, subprocess, shlex, os
import threading
class MainWindow (Gtk.Builder):
textView=None
textbuffer = None
progressBar=None
def getPercent(line):
# Example 11,131,595 1% 10.47MB/s 0:00:00 (xfr#16, to-chk=2761/4538)
# Splitto la corrente riga dello stdout su una lista con separatore lo spazio
line=line.split(" ")
# Se uno degli elementi della lista contiene un simbolo di percentuale (%)
# allora ritorno solo l'intero (sotto forma di stringa) ,sia esso formato
# da 1,2 o tre cifre
for l in line:
if ('%' in l):
if (len(l)==2): return l[0]
if (len(l)==3): return l[0]+l[1]
if (len(l)==4): return l[0]+l[1]+l[2]
def cmdThread(self,cmd):
print("Comando: %s" % cmd)
txtBuffer=""
out=""
fileOutErr = open("errors.txt","w")
# Lo stdout lo leggo dalla pipe, gli errori li salvo in un file che leggero' dopo
proc = subprocess.Popen(shlex.split(cmd), bufsize=0 ,stdout=subprocess.PIPE, stderr=fileOutErr)
# leggo la pipe finchè non finisce(proc.poll() is None)
while proc.poll() is None :
out = proc.stdout.readline()
txtBuffer=txtBuffer+str(out.decode("utf-8"))
GLib.idle_add(MainWindow.textbuffer.set_text,txtBuffer)
# Faccio il parsing della percentuale di progresso e aggiorno la gtkProgressbar
# dal metodo MainWindow.getPercent ricevo una stringa del genere "50","10". etc...
if (MainWindow.getPercent(str(out.decode("utf-8"))) is not None):
p=float(MainWindow.getPercent(str(out.decode("utf-8"))))/100
GLib.idle_add(MainWindow.progressBar.set_fraction,p)
fileOutErr.close()
GLib.idle_add(MainWindow.progressBar.set_fraction,1.0)
#Stampo gli errori dopo la fine della pipe
if (os.stat("errors.txt").st_size != 0):
fileOutErr= open("errors.txt","r")
listErr=fileOutErr.readlines()
txtBuffer=txtBuffer+"\n\nErrori:\n------------------\n\n"
for e in listErr:
txtBuffer=txtBuffer+e
#Aggiungo al text buffer gli errori presi dal file e chiudo il file
GLib.idle_add(MainWindow.textbuffer.set_text,txtBuffer)
fileOutErr.close()
def on_button_clicked(self, textView,entry):
cmdt = threading.Thread(target=MainWindow.cmdThread, args=(self,entry.get_text()),)
cmdt.daemon = True
cmdt.start()
def quit(self,win):
print ("Exit!")
Gtk.main_quit(win)
def __init__(self):
self = Gtk.Builder()
self.add_from_file("builder.ui")
window = self.get_object("window")
entry = self.get_object("entry")
MainWindow.textView = self.get_object("textView")
button = self.get_object("button")
MainWindow.progressBar=self.get_object("progressBar")
#Init
MainWindow.textbuffer = MainWindow.textView.get_buffer()
MainWindow.textbuffer.set_text("testo libero")
if (sys.platform == "win32"):
entry.set_text("./rsync-win/rsync.exe --info=progress2 -avz /cygdrive/c/msys64/mingw64/bin /cygdrive/c/users/admin/desktop/tt/")
window.show_all()
window.connect("delete-event", MainWindow.quit)
button.connect("clicked", MainWindow.on_button_clicked, MainWindow.textView,entry)
myWindow = MainWindow()
Gtk.main()
**builder.ui**