Что нужно: Каждые 5 секунд, считываются координаты из БД и отображаются на графике
matplotlib в окне tkinter.
Что сделано: рисуются только первый набор данных и не рисуются второй набор
ВОПРОС: как исправить чтобы рисовался сначала первый набор данных , а через 5 секунд
второй?
Вот обновленный минимальный рабочий код:
from threading import Thread
from queue import Empty, Queue
import time
import tkinter as tk
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg#, NavigationToolbar2TkAgg
from matplotlib.figure import Figure
class tkChartGUI(tk.Frame):
def __init__(self, parent):
tk.Frame.__init__(self, parent)
self.parent = parent
self.initUI()
def get_latest_data(self, dataid):
x_array=[]
y_array=[]
if (dataid == 1):
x_array=[1,2,3,4,5,6,7,8,9,10,11,12,13,14];
y_array=[0.5,0.7,0.3,1.0,0.6,0.9,0.5,0.2,0.1,0.5,0.33,0.55,0.3,0.6]
if (dataid == 2):
x_array=[1,2,3,4,5,6,7,8,9,10,11,12,13,14];
y_array=[0.1,0.2,0.3,0.4,0.5,0.6,0.7,0.8,0.9,1.0,1.1,1.1,1.2,1.3]
return (x_array, y_array)
def initUI(self):
self.parent.title("Simple chart")
self.parent.geometry("800x600+300+100")
result_queue = Queue()
Thread(target=self.get_latest_data, args=[result_queue], daemon=True).start()
x_array, y_array = self.get_latest_data(1)
f = Figure(figsize=(5, 3), dpi=150)
a = f.add_subplot(111)
a.set_xlabel("Values_X")
a.set_ylabel("Values_Y")
a.yaxis.grid(True, which='major')
a.xaxis.grid(True, which='major')
a.plot(x_array, y_array)
canvas = FigureCanvasTkAgg(f, master=self.parent)
canvas.show()
canvas.get_tk_widget().grid(row=0,column=0)
def display_result(a, q):
x_array = []
y_array = []
try:
x_array = q.get(block=False) # get data
y_array = q.get(block=False)
a.plot(x_array, y_array)
canvas.draw()
except Empty:
#a.clear()
timeout_millis = round(100 - (5000 * time.time()) % 100)
self.parent.after(timeout_millis, display_result, a, q)
def get_result(q):
x_array, y_array = self.get_latest_data(2)
q.put(x_array) # put data in FIFO queue x coords array
q.put(y_array) # put data in FIFO queue y coords array
display_result(a, result_queue)
def onExit(self):
self.quit()
def main():
root = tk.Tk()
my_gui = tkChartGUI(root)
root.mainloop()
if __name__ == '__main__':
main()
Ответы
Ответ 1
Чтобы постоянно не считывать данные из базы данных (polling), можно
определить trigger, чтобы вызвать функцию (callback), когда в БД
интересное событие произойдёт. К примеру, когда в нужную таблицу новое
значение добавляется (Launch a
Python Script from a sqlite3
Trigger):
CREATE TRIGGER tt AFTER INSERT ON t BEGIN SELECT got_y(NEW.y); END
Обновления графика происходят в got_y() обратном вызове,
определённом в make_callback() ниже. Код похож на ответ c loop(), определённой с
помощью .after(), с той разницей,
что здесь функция вызывается, когда нужные данные готовы, вместо того
чтобы периодически непрерывно новые данные запрашивать без блокировки:
#!/usr/bin/env python3
import datetime as DT
import sqlite3
import random
import threading
import time
import tkinter as tk
from collections import deque
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.dates import date2num
def generate_data(db): # generate dummy data for the example
while True:
time.sleep(.1)
db.execute('insert into t values(?)', (random.randrange(100),))
def make_callback(root):
# plot something
# see @MaxU answer https://ru.stackoverflow.com/q/801923/23044
n = 300
def get_t(): return date2num(DT.datetime.now())
xx = deque([get_t()], maxlen=n)
yy = deque([0], maxlen=n)
fig, ax = plt.subplots()
ax.set_ylim(0, 100)
line, = ax.plot_date(xx, yy, marker='')
format_time = '{:%Y-%m-%d %H:%M:%S}'.format
time_text = ax.text(0.5, 0.9, '', transform=ax.transAxes)
# add to GUI
canvas = FigureCanvasTkAgg(fig, master=root) # A tk.DrawingArea.
canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
def got_y(y):
# update plot
xx.append(get_t())
yy.append(int(y))
line.set_data(xx, yy)
ax.fill_between(xx, 0, yy, color='lightgrey')
time_text.set_text(format_time(DT.datetime.now()))
ax.relim() # update axes limits
ax.autoscale_view(scaley=False)
canvas.draw()
return got_y
def main():
root = tk.Tk()
# a dummy db for the example
db = sqlite3.connect(':memory:', check_same_thread=False)
db.execute('create table t(y)')
db.create_function('got_y', 1, make_callback(root))
db.execute('CREATE TRIGGER tt AFTER INSERT ON t BEGIN SELECT got_y(NEW.y); END')
threading.Thread(target=generate_data, args=[db], daemon=True).start()
root.mainloop()
if __name__ == '__main__':
main()
Здесь db.create_function() определяет какая Питон-функция используется в качестве
got_y(y) обратного вызова. В примере, callback вызывается в дочернем потоке, что не
всегда желаемо. Поток с generate_data() используется только для примера, фактически
данные могут вставляться в базу данных из другого процесса или других машин. How to
receive automatic notifications about changes in tables?
Для сравнения вот вариант с polling, где база данных постоянно
опрашивается в фоновом потоке, используя простой цикл (про
достоинства и недостатки цикла подробно описано в Как правильно сделать
временный цикл?)
def poll_db(interval=5):
while True:
time.sleep(interval - time.time() % interval) # avoid drift
emit(get_y_from_db())
здесь get_y_from_db() делает запрос к базе данных, а emit()
генерирует событие для GUI:
#!/usr/bin/env python3
import random
import threading
import time
import tkinter as tk
def get_y_from_db(): # generate dummy data
time.sleep(random.random()) # emulate blocking function
return random.randrange(100)
def poll_db(emit, interval=5):
while True:
time.sleep(interval - time.time() % interval) # avoid drift
emit(get_y_from_db())
root = tk.Tk()
root.bind('<>', lambda e, f=make_callback(root): f(e.y)) # subscribe
threading.Thread(target=poll_db,
args=[lambda y: root.event_generate('<>', when='tail', y=y)],
daemon=True).start()
root.mainloop()
Пример кода (где этот метод также используется): вывод процесса показывается в Tkinter
GUI с помощью root.event_generate()
Не на всех реализациях можно root.event_generate() в фоновом потоке
вызвать. В таких случаях можно использовать queue, чтобы данные между
потоками передавать:
#!/usr/bin/env python3
import datetime as DT
import random
import threading
import time
import tkinter as tk
from collections import deque
from queue import Empty, Queue
from time import time as timer
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.dates import date2num
def get_y_from_db(): # generate dummy data
time.sleep(random.random()) # emulate blocking function
return random.randrange(100)
def poll_db(emit, interval=1): # NOTE: interval is independant from the delay
while True:
time.sleep(interval)
emit(get_y_from_db())
def start_polling_loop(root, q, delay):
# plot something
# see @MaxU answer https://ru.stackoverflow.com/q/801923/23044
n = 600
def get_t(): return date2num(DT.datetime.now())
xx = deque([get_t()], maxlen=n)
yy = deque([0], maxlen=n)
fig, ax = plt.subplots()
ax.set_ylim(0, 100)
line, = ax.plot_date(xx, yy, marker='')
format_time = '{:%Y-%m-%d %H:%M:%S}'.format
time_text = ax.text(0.5, 0.9, '', transform=ax.transAxes)
# add to GUI
canvas = FigureCanvasTkAgg(fig, master=root) # A tk.DrawingArea.
canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# update in a loop
def loop():
timeout_millis = round(delay - (1000 * timer()) % delay)
root.after(timeout_millis, loop) # avoid drift
try:
y = q.get(block=False)
except Empty:
return # no new data
# update plot
xx.append(get_t())
yy.append(y)
line.set_data(xx, yy)
ax.fill_between(xx, 0, yy, color='lightgrey')
time_text.set_text(format_time(DT.datetime.now()))
ax.relim() # update axes limits
ax.autoscale_view(scaley=False)
canvas.draw()
root.after_idle(loop) # start
root = tk.Tk()
q = Queue()
threading.Thread(target=poll_db,
args=[q.put],
daemon=True).start()
start_polling_loop(root, q, delay=40)
root.mainloop()
Упрощённый вариант (с одним обновлением данных из потока), см. в Как сделать постоянное
обновление окна Tkinter? Как избежать подвисания на время ожидания ответа от сервера.
Ещё пример кода (где этот метод используется): вывод процесса показывается в Tkinter
GUI, используя widget.after(), q.get()/q.put().
Ответ 2
Чтобы в tkinter периодически действие выполнять, можно его запускать с помощью after
tcl команды. См. Обновление Label из цикла в tkinter.
На каждой итерации, обновляйте желаемые элементы, к примеру, как показано в Python
(jupyter) анимированный график и вызывайте canvas.draw():
#!/usr/bin/env python3
import datetime as DT
import tkinter as tk
from collections import deque
from time import time as timer
import matplotlib.pyplot as plt
import psutil
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.dates import date2num
def start_loop(root, delay):
# plot something
# see @MaxU answer https://ru.stackoverflow.com/q/801923/23044
n = 600
def get_t(): return date2num(DT.datetime.now())
xx = deque([get_t()], maxlen=n)
yy = deque([0], maxlen=n)
fig, ax = plt.subplots()
ax.set_ylim(0, 100)
line, = ax.plot_date(xx, yy, marker='')
format_time = '{:%Y-%m-%d %H:%M:%S}'.format
time_text = ax.text(0.5, 0.9, '', transform=ax.transAxes)
# add to GUI
canvas = FigureCanvasTkAgg(fig, master=root) # A tk.DrawingArea.
canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# update in a loop
def loop():
timeout_millis = round(delay - (1000 * timer()) % delay)
root.after(timeout_millis, loop) # avoid drift
# update plot
# cpu sine https://stackoverflow.com/q/551494/4279
xx.append(get_t())
yy.append(psutil.cpu_percent())
line.set_data(xx, yy)
ax.fill_between(xx, 0, yy, color='lightgrey')
time_text.set_text(format_time(DT.datetime.now()))
ax.relim() # update axes limits
ax.autoscale_view(scaley=False)
canvas.draw()
root.after_idle(loop) # start
def quit():
root.quit()
root.destroy() # https://matplotlib.org/gallery/user_interfaces/embedding_in_tk_sgskip.html
root = tk.Tk()
start_loop(root, 100)
root.protocol("WM_DELETE_WINDOW", quit)
root.eval('tk::PlaceWindow %s center' % root.winfo_pathname(root.winfo_id()))
root.mainloop()
Сутью решения является loop() функция здесь.
