在以非阻塞方式更新 GUI tkinter 的同时继续读取数据
Keep reading data while updating GUI tkinter in a non blocking way
我是异步、线程、子进程的新手，我正在尝试构建一个持续从串行读取数据的应用程序，将它们放入另一个 process/thread/asyncio 函数使用的队列中以使用它们并显示到 tkinter GUI 中。
我能够在使用以下代码继续读取数据的同时使 GUI 成为非阻塞状态。
import tkinter as tk
import time
import queue
import logging
import serial
import sys
class SampleApp(tk.Tk):
def __init__(self, *args, **kwargs):
tk.Tk.__init__(self, *args, **kwargs)
self.serial_text_label = tk.Label(self, text="String")
self.serial_text_label.pack()
self.serial_text = tk.Text(self, height=1, width=21)
self.serial_text.pack()
self.port = 'COM3'
self.baud = 38400
self.ser = serial.Serial(self.port, self.baud, timeout=0)
if self.ser.isOpen():
self.ser.close()
self.ser.open()
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
logging.info("created serial port")
# start the serial_text_label "ticking"
self.update_screen()
def update_screen(self):
self.serial_text.delete('1.0', tk.END)
data = ""
data_raw = self.ser.read(1)
if data_raw == b'\x02':
data_raw = self.ser.read(6)
data = "02-" + str(data_raw.hex('-'))
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
self.serial_text.insert(tk.END, data)
# self.serial_text_label.configure(text=data)
# call this function again when want to refresh
self.after(500, self.update_screen)
if __name__== "__main__":
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)', )
app = SampleApp()
app.mainloop()
我的代码的唯一问题是来自串行端口的数据的所有读取和详细说明都在更新屏幕的刷新周期内。我想将该功能分离为某种 thread/subprocess，它与 GUI 的刷新同时工作。
我尝试在 class SampleApp(tk.Tk) 中创建一个 async def do_serial() 函数，如下所示：
async def do_serial():
logging.debug("do serial")
data = ""
data_raw = ser.read(1)
if data_raw == b'\x02':
data_raw = ser.read(6)
data = "02-" + str(data_raw.hex('-'))
ser.reset_input_buffer()
ser.reset_output_buffer()
# add data to queue
if data != "":
logging.debug('put:' + str(data))
incoming_serial_queue.put(data)
await asyncio.sleep(1)
在 update_screen 函数中我调用 asyncio.run(do_serial())
if not incoming_serial_queue.empty():
data = incoming_serial_queue.get()
不幸的是它不起作用，代码甚至不显示 GUI
有没有办法以 asyncronus/parallel 的方式处理串行数据，而不必在刷新 GUI 函数中编写所有函数？
通常，您会使用 root.mainloop() 作为函数，但对于异步，您将不得不使用其他东西，但是如果您想要 window 更新每一帧。异步函数只是 root.update().
希望对您有所帮助！
尝试在单独的线程中进行阻塞调用。在 update_screen 内，您应该足够快地进行调用，以免冻结 GUI。这意味着，你不应该在那里阅读输入。
import tkinter as tk
import time
import queue
import logging
import serial
import sys
import threading
from concurrent.futures import ThreadPoolExecutor
class SampleApp(tk.Tk):
def __init__(self, *args, **kwargs):
tk.Tk.__init__(self, *args, **kwargs)
self.serial_text_label = tk.Label(self, text="String")
self.serial_text_label.pack()
self.serial_text = tk.Text(self, height=1, width=21)
self.serial_text.pack()
self.port = 'COM3'
self.baud = 38400
self.ser = serial.Serial(self.port, self.baud, timeout=0)
if self.ser.isOpen():
self.ser.close()
self.ser.open()
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
logging.info("created serial port")
# start the serial_text_label "ticking"
self._update_scheduled = threading.Condition()
self._terminating = threading.Event()
self.update_screen()
def mainloop(self):
with ThreadPoolExecutor() as executor:
future = executor.submit(self._do_update_screen_loop)
try:
return super().mainloop()
finally:
# letting the thread to know we're done
self._terminating.set()
with self._update_scheduled:
self._update_scheduled.notify_all()
def update_screen(self):
with self._update_scheduled:
self._update_scheduled.notify_all()
self.after(500, self.update_screen)
def _do_update_screen_loop(self):
while True:
with self._update_scheduled:
self._update_scheduled.wait()
if self._terminating.is_set():
return
self._do_update_screen()
def _do_update_screen(self):
self.serial_text.delete('1.0', tk.END)
data = ""
data_raw = self.ser.read(1)
if data_raw == b'\x02':
data_raw = self.ser.read(6)
data = "02-" + str(data_raw.hex('-'))
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
self.serial_text.insert(tk.END, data)
# self.serial_text_label.configure(text=data)
if __name__== "__main__":
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)', )
app = SampleApp()
app.mainloop()
我是异步、线程、子进程的新手，我正在尝试构建一个持续从串行读取数据的应用程序，将它们放入另一个 process/thread/asyncio 函数使用的队列中以使用它们并显示到 tkinter GUI 中。
我能够在使用以下代码继续读取数据的同时使 GUI 成为非阻塞状态。
import tkinter as tk
import time
import queue
import logging
import serial
import sys
class SampleApp(tk.Tk):
def __init__(self, *args, **kwargs):
tk.Tk.__init__(self, *args, **kwargs)
self.serial_text_label = tk.Label(self, text="String")
self.serial_text_label.pack()
self.serial_text = tk.Text(self, height=1, width=21)
self.serial_text.pack()
self.port = 'COM3'
self.baud = 38400
self.ser = serial.Serial(self.port, self.baud, timeout=0)
if self.ser.isOpen():
self.ser.close()
self.ser.open()
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
logging.info("created serial port")
# start the serial_text_label "ticking"
self.update_screen()
def update_screen(self):
self.serial_text.delete('1.0', tk.END)
data = ""
data_raw = self.ser.read(1)
if data_raw == b'\x02':
data_raw = self.ser.read(6)
data = "02-" + str(data_raw.hex('-'))
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
self.serial_text.insert(tk.END, data)
# self.serial_text_label.configure(text=data)
# call this function again when want to refresh
self.after(500, self.update_screen)
if __name__== "__main__":
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)', )
app = SampleApp()
app.mainloop()
我的代码的唯一问题是来自串行端口的数据的所有读取和详细说明都在更新屏幕的刷新周期内。我想将该功能分离为某种 thread/subprocess，它与 GUI 的刷新同时工作。
我尝试在 class SampleApp(tk.Tk) 中创建一个 async def do_serial() 函数，如下所示：
async def do_serial():
logging.debug("do serial")
data = ""
data_raw = ser.read(1)
if data_raw == b'\x02':
data_raw = ser.read(6)
data = "02-" + str(data_raw.hex('-'))
ser.reset_input_buffer()
ser.reset_output_buffer()
# add data to queue
if data != "":
logging.debug('put:' + str(data))
incoming_serial_queue.put(data)
await asyncio.sleep(1)
在 update_screen 函数中我调用 asyncio.run(do_serial())
if not incoming_serial_queue.empty():
data = incoming_serial_queue.get()
不幸的是它不起作用，代码甚至不显示 GUI
有没有办法以 asyncronus/parallel 的方式处理串行数据，而不必在刷新 GUI 函数中编写所有函数？
通常，您会使用 root.mainloop() 作为函数，但对于异步，您将不得不使用其他东西，但是如果您想要 window 更新每一帧。异步函数只是 root.update().
希望对您有所帮助！
尝试在单独的线程中进行阻塞调用。在 update_screen 内，您应该足够快地进行调用，以免冻结 GUI。这意味着，你不应该在那里阅读输入。
import tkinter as tk
import time
import queue
import logging
import serial
import sys
import threading
from concurrent.futures import ThreadPoolExecutor
class SampleApp(tk.Tk):
def __init__(self, *args, **kwargs):
tk.Tk.__init__(self, *args, **kwargs)
self.serial_text_label = tk.Label(self, text="String")
self.serial_text_label.pack()
self.serial_text = tk.Text(self, height=1, width=21)
self.serial_text.pack()
self.port = 'COM3'
self.baud = 38400
self.ser = serial.Serial(self.port, self.baud, timeout=0)
if self.ser.isOpen():
self.ser.close()
self.ser.open()
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
logging.info("created serial port")
# start the serial_text_label "ticking"
self._update_scheduled = threading.Condition()
self._terminating = threading.Event()
self.update_screen()
def mainloop(self):
with ThreadPoolExecutor() as executor:
future = executor.submit(self._do_update_screen_loop)
try:
return super().mainloop()
finally:
# letting the thread to know we're done
self._terminating.set()
with self._update_scheduled:
self._update_scheduled.notify_all()
def update_screen(self):
with self._update_scheduled:
self._update_scheduled.notify_all()
self.after(500, self.update_screen)
def _do_update_screen_loop(self):
while True:
with self._update_scheduled:
self._update_scheduled.wait()
if self._terminating.is_set():
return
self._do_update_screen()
def _do_update_screen(self):
self.serial_text.delete('1.0', tk.END)
data = ""
data_raw = self.ser.read(1)
if data_raw == b'\x02':
data_raw = self.ser.read(6)
data = "02-" + str(data_raw.hex('-'))
self.ser.reset_input_buffer()
self.ser.reset_output_buffer()
self.serial_text.insert(tk.END, data)
# self.serial_text_label.configure(text=data)
if __name__== "__main__":
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)', )
app = SampleApp()
app.mainloop()