2026, Jan 14 09:03
Как строить потоковый график в Matplotlib в реальном времени
Пошаговый разбор живых графиков в Matplotlib: одна линия Line2D, обновление через set_data и plt.pause, корректные временные метки через datetime и date2num.
Передавать потоковые данные в график Matplotlib кажется простым, пока не пытаешься сохранить график «живым», быстрым и корректно синхронизированным. Почти сразу всплывают две ловушки: неверное обновление фигуры внутри плотного цикла и подача на ось временных меток, которые Matplotlib не понимает. Ниже — практический разбор, что идёт не так и как аккуратно это исправить.
Тестовая установка, которая выявляет проблему
Фрагмент ниже имитирует сетевой источник, передающий значения, и пытается строить их по текущему времени в окне фиксированной ширины. Он подчёркивает обе проблемы: стратегию обновления внутри цикла и способ преобразования времени в формат дат Matplotlib.
import threading
import random
import time
import signal
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as md
class Buf2D:
def __init__(self, capacity):
self.grid = np.zeros((capacity, 2)) # строки x 2 столбца
self.capacity = capacity
self.idx = 0
def __len__(self):
return len(self.grid)
def __str__(self):
return str(self.grid)
def append_row(self, row):
self.grid[self.idx] = row
self.idx = (self.idx + 1) % self.capacity
def span_x(self):
return (self.grid.min(axis=0)[0], self.grid.max(axis=0)[0])
class Feeder(threading.Thread):
def __init__(self):
super().__init__()
random.seed()
self.alive = True
self.store = Buf2D(100)
def payload(self):
return self.store.grid
def shutdown(self):
self.alive = False
def run(self):
while self.alive:
now_ms_num = md.date2num(int(time.time() * 1000)) # мс
point = np.array([now_ms_num, np.random.randint(0, 999)])
self.store.append_row(point)
time.sleep(0.1)
app_active = True
def on_signal(sig, frame):
global app_active
app_active = False
def launch():
signal.signal(signal.SIGINT, on_signal)
prod = Feeder()
prod.start()
fig, axis = plt.subplots()
formatter = md.DateFormatter('%H:%M:%S.%f')
axis.xaxis.set_major_formatter(formatter)
plt.show(block=False)
while app_active:
xr = prod.store.span_x()
axis.set_xlim(xr)
print(prod.payload())
plt.draw()
plt.pause(0.05)
prod.shutdown()
Что на самом деле не так
Первая ловушка — стратегия обновления. Постоянные вызовы функций построения, таких как plot, внутри цикла добавляют новых графических объектов каждый раз, что медленно и быстро превращает фигуру в кашу из наложенных линий. Эффективный приём — создать один объект Line2D и в цикле менять только его данные через set_data. Для интерактивного обновления экрана используйте fig.canvas.draw_idle() или просто plt.pause() в цикле; второй вариант — самый прямой путь к живым обновлениям.
Вторая ловушка — время. Даты в Matplotlib — это числа с плавающей точкой, измеряемые в «днях с 0001-01-01 UTC». Передача сырых целочисленных миллисекунд или их преобразование через date2num в таком виде не работает как задумано и приводит к ошибкам форматтера вроде OverflowError: int too big to convert. Замечание: если в массив явно кладутся только int, NumPy сохранит целочисленный dtype, и форматтер по‑прежнему увидит гигантские числа. Если нужен миллисекундный масштаб в виде float, умножение на 1e3 создаёт число с плавающей точкой. Однако самый надёжный подход здесь — использовать datetime.datetime.now() и md.date2num(), чтобы получить корректный «дневной» float, которого ждёт Matplotlib. Подписи с миллисекундами «из коробки» не поддерживаются, но можно форматировать деления оси с долями секунды, применив форматтер вроде %H:%M:%S.%f.
Аккуратное решение для плавных обновлений в реальном времени
Исправленная версия ниже хранит один объект линии, обновляет его свежими данными, подгоняет окно по времени к фиксированному горизонту, заканчивающемуся «сейчас», и обновляет холст в интерактивном цикле. Временные значения сохраняются как даты Matplotlib, полученные из datetime.
import threading
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as md
import datetime
import time
import signal
class RingStore:
def __init__(self, limit):
self.buf = np.zeros((limit, 2))
self.limit = limit
self.pos = 0
self.wrapped = False
def push(self, row):
self.buf[self.pos] = row
self.pos = (self.pos + 1) % self.limit
if self.pos == 0:
self.wrapped = True
def view(self):
if self.wrapped:
return np.vstack((self.buf[self.pos:], self.buf[:self.pos]))
return self.buf[:self.pos]
def x_window(self, window_seconds=10):
current = self.view()
if len(current) == 0:
now = md.date2num(datetime.datetime.now())
return (now - window_seconds / 86400.0, now)
last = current[-1, 0]
return (last - window_seconds / 86400.0, last)
class Streamer(threading.Thread):
def __init__(self):
super().__init__()
self.on = True
self.store = RingStore(1000)
def stop(self):
self.on = False
def run(self):
while self.on:
ts = md.date2num(datetime.datetime.now())
val = np.random.randint(0, 999)
self.store.push(np.array([ts, val]))
time.sleep(0.1)
running = True
def trap(sig, frame):
global running
running = False
def start():
signal.signal(signal.SIGINT, trap)
src = Streamer()
src.start()
fig, ax = plt.subplots()
fmt = md.DateFormatter('%H:%M:%S.%f')
ax.xaxis.set_major_formatter(fmt)
trace, = ax.plot([], [], 'b-')
ax.set_ylim(0, 999)
plt.show(block=False)
while running:
current = src.store.view()
if len(current) > 0:
trace.set_data(current[:, 0], current[:, 1])
ax.set_xlim(src.store.x_window(window_seconds=10))
ax.figure.canvas.draw_idle()
plt.pause(0.05)
src.stop()
Почему это важно
Визуализация в реальном времени не прощает неэффективных перерисовок и некорректных временных данных. Создание новых объектов графика в цикле незаметно ухудшает производительность и раздувает потребление памяти, а неправильно масштабированные временные метки ломают форматтеры и логику осей. Один обновляемый объект линии и перевод времени в родной «дневной» float Matplotlib обеспечивают отзывчивость графика и точную ось времени. Форматирование делений с долями секунды даёт читаемость на уровне миллисекунд без борьбы с моделью дат под капотом.
Выводы
Когда вы строите живой график Matplotlib с фоновым источником данных, держите путь отрисовки простым: создайте одну линию, обновляйте её данные и позвольте plt.pause() или триггеру перерисовки холста поддерживать отзывчивость интерфейса. Аккуратно работайте со временем: генерируйте метки через datetime.datetime.now() и md.date2num(), а окно просмотра держите фиксированным и «скользящим» назад. Такая дисциплина избавляет от дублирующихся объектов, предотвращает переполнения в форматтерах и даёт плавный, стабильный график, готовый к производственной нагрузке.