Система сбора и визуализации данных с MEMS-гироскопа MPU3050 на базе Raspberry Pi Pico

Технические детали

Особенности реализации:

• Правильная обработка знаковых значений: Все данные с датчика интерпретируются как int16_t

• Точная формула температуры: Взята напрямую из официальной документации

• Калибровка гироскопа: Усреднение 200 отсчетов для определения смещения нуля

• CSV-протокол: Температура, Gx, Gy, Gz, Δt для простоты парсинга

Производительность:

• Частота опроса датчика: ~100 Гц

• Задержка визуализации: < 50 мс

• Потребление памяти Python: ~150 МБ

• Нагрузка на процессор: < 10% (на современном ПК)

MPU3050_PYTHON.ino

#include <Wire.h>

void setup() {
Wire.begin();
Serial.begin(115200);
while (!Serial); // Ждем подключения монитора порта
Serial.println("\nI2C Scanner");
}

void loop() {
byte error, address;
int nDevices = 0;

Serial.println("Scanning...");

for(address = 1; address < 127; address++ ) {
Wire.beginTransmission(address);
error = Wire.endTransmission();

if (error == 0) {
Serial.print("I2C device found at address 0x");
if (address < 16) Serial.print("0");
Serial.print(address, HEX);
Serial.println(" !");
nDevices++;
}
else if (error == 4) {
Serial.print("Unknown error at address 0x");
if (address < 16) Serial.print("0");
Serial.println(address, HEX);
}
}

if (nDevices == 0)
Serial.println("No I2C devices found\n");
else
Serial.println("done\n");

delay(5000);
}

mpu3050_visualizer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
MPU3050 Visualizer для Raspberry Pi Pico (RP2040) с выбором COM порта
"""

import sys
import platform

# Настройка бэкенда matplotlib до импорта
import matplotlib
try:
matplotlib.use('TkAgg')
except:
pass

# Импорт библиотек
try:
import matplotlib.pyplot as plt
from matplotlib.widgets import Button
import matplotlib.animation as animation
print("✅ matplotlib загружен успешно")
except ImportError as e:
print(f"❌ Ошибка загрузки matplotlib: {e}")
sys.exit(1)

try:
import numpy as np
print("✅ numpy загружен успешно")
except ImportError as e:
print(f"❌ Ошибка загрузки numpy: {e}")
sys.exit(1)

try:
import serial
import serial.tools.list_ports
print("✅ pyserial загружен успешно")
except ImportError as e:
print(f"❌ Ошибка загрузки pyserial: {e}")
sys.exit(1)

from collections import deque
import time
import tkinter as tk
from tkinter import ttk, messagebox
import threading

class PortSelector:
"""Класс для выбора COM порта через графический интерфейс"""

def __init__(self):
self.selected_port = None
self.root = None

def select_port_gui(self):
"""Открывает окно для выбора COM порта"""
self.root = tk.Tk()
self.root.title("Выбор COM порта для MPU3050")
self.root.geometry("500x400")
self.root.resizable(True, True)

# Центрируем окно
self.root.update_idletasks()
x = (self.root.winfo_screenwidth() // 2) - (self.root.winfo_width() // 2)
y = (self.root.winfo_screenheight() // 2) - (self.root.winfo_height() // 2)
self.root.geometry(f'+{x}+{y}')

# Заголовок
title_label = tk.Label(self.root, text="Выбор COM порта для Raspberry Pi Pico",
font=('Arial', 14, 'bold'))
title_label.pack(pady=10)

# Информация
info_label = tk.Label(self.root,
text="Найдены следующие последовательные порты:\n"
"Выберите порт, к которому подключена Raspberry Pi Pico",
font=('Arial', 10))
info_label.pack(pady=5)

# Создаем фрейм для списка портов
frame = ttk.Frame(self.root)
frame.pack(fill=tk.BOTH, expand=True, padx=20, pady=10)

# Создаем Treeview для отображения портов
columns = ('port', 'description', 'hardware_id')
self.tree = ttk.Treeview(frame, columns=columns, show='headings', height=8)

# Настройка колонок
self.tree.heading('port', text='Порт')
self.tree.heading('description', text='Описание')
self.tree.heading('hardware_id', text='Hardware ID')

self.tree.column('port', width=80)
self.tree.column('description', width=250)
self.tree.column('hardware_id', width=150)

# Добавляем скроллбар
scrollbar = ttk.Scrollbar(frame, orient=tk.VERTICAL, command=self.tree.yview)
self.tree.configure(yscrollcommand=scrollbar.set)

# Размещаем treeview и скроллбар
self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)

# Кнопки
button_frame = ttk.Frame(self.root)
button_frame.pack(fill=tk.X, padx=20, pady=10)

refresh_btn = ttk.Button(button_frame, text="???? Обновить список",
command=self.refresh_ports)
refresh_btn.pack(side=tk.LEFT, padx=5)

select_btn = ttk.Button(button_frame, text="✅ Выбрать порт",
command=self.select_port, style='Accent.TButton')
select_btn.pack(side=tk.LEFT, padx=5)

cancel_btn = ttk.Button(button_frame, text="❌ Отмена",
command=self.cancel)
cancel_btn.pack(side=tk.LEFT, padx=5)

auto_btn = ttk.Button(button_frame, text="???? Автоопределение",
command=self.auto_detect)
auto_btn.pack(side=tk.LEFT, padx=5)

# Статус бар
self.status_var = tk.StringVar()
self.status_var.set("Готов к работе")
status_bar = ttk.Label(self.root, textvariable=self.status_var,
relief=tk.SUNKEN, anchor=tk.W)
status_bar.pack(fill=tk.X, side=tk.BOTTOM, padx=5, pady=2)

# Загружаем порты
self.refresh_ports()

# Настройка стиля для акцентной кнопки
style = ttk.Style()
style.configure('Accent.TButton', font=('Arial', 10, 'bold'))

# Обработка закрытия окна
self.root.protocol("WM_DELETE_WINDOW", self.cancel)

# Запускаем главный цикл
self.root.mainloop()

return self.selected_port

def refresh_ports(self):
"""Обновление списка портов"""
# Очищаем текущий список
for item in self.tree.get_children():
self.tree.delete(item)

# Получаем список портов
ports = list(serial.tools.list_ports.comports())

if not ports:
self.tree.insert('', tk.END, values=('Порты не найдены', '', ''))
self.status_var.set("❌ Порты не найдены")
else:
for port in ports:
# Определяем, похоже ли на Pico
tags = ()
if self.is_pico_device(port):
tags = ('pico',)

self.tree.insert('', tk.END, values=(
port.device,
port.description,
f"{port.vid}:{port.pid}" if port.vid else "N/A"
), tags=tags)

# Выделяем устройства Pico цветом
self.tree.tag_configure('pico', background='#2a5c2a', foreground='white')

self.status_var.set(f"✅ Найдено {len(ports)} портов")

def is_pico_device(self, port):
"""Проверка, является ли устройство Raspberry Pi Pico"""
# Проверка по VID/PID
if port.vid == 0x2E8A: # Raspberry Pi VID
return True
if port.pid == 0x0005: # RP2040 PID
return True

# Проверка по описанию
desc_lower = port.description.lower()
pico_keywords = ['pico', 'rp2040', 'raspberry', 'usb serial', 'последовательный']
for keyword in pico_keywords:
if keyword in desc_lower:
return True

return False

def select_port(self):
"""Выбор выделенного порта"""
selection = self.tree.selection()
if not selection:
messagebox.showwarning("Нет выбора", "Пожалуйста, выберите порт из списка")
return

# Получаем выбранный порт
item = self.tree.item(selection[0])
self.selected_port = item['values'][0]

self.status_var.set(f"✅ Выбран порт: {self.selected_port}")

# Закрываем окно после короткой задержки
self.root.after(500, self.root.destroy)

def auto_detect(self):
"""Автоматическое определение Pico"""
ports = list(serial.tools.list_ports.comports())

for port in ports:
if self.is_pico_device(port):
self.selected_port = port.device
messagebox.showinfo("Устройство найдено",
f"Найдена Raspberry Pi Pico на порту {port.device}")
self.root.after(500, self.root.destroy)
return

messagebox.showwarning("Устройство не найдено",
"Raspberry Pi Pico не найдена.\n"
"Проверьте подключение и нажмите 'Обновить список'")

def cancel(self):
"""Отмена выбора"""
self.selected_port = None
self.root.destroy()

class ConsolePortSelector:
"""Класс для выбора COM порта через консоль"""

@staticmethod
def select_port():
"""Выбор порта через консоль"""
print("\n" + "=" * 60)
print("???? ВЫБОР COM ПОРТА ДЛЯ RASPBERRY PI PICO")
print("=" * 60)

# Получаем список портов
ports = list(serial.tools.list_ports.comports())

if not ports:
print("❌ Не найдено ни одного COM порта!")
print("\nПроверьте:")
print(" 1. Подключена ли Raspberry Pi Pico к USB")
print(" 2. Установлены ли драйверы")
print(" 3. Не используется ли порт другим приложением")
return None

print(f"\nНайдено {len(ports)} портов:\n")

# Создаем список для выбора
pico_ports = []
other_ports = []

for i, port in enumerate(ports, 1):
is_pico = ConsolePortSelector.is_pico_device(port)
port_info = {
'index': i,
'device': port.device,
'description': port.description,
'hwid': f"{port.vid}:{port.pid}" if port.vid else "N/A",
'is_pico': is_pico
}

if is_pico:
pico_ports.append(port_info)
else:
other_ports.append(port_info)

# Сначала показываем Pico устройства
if pico_ports:
print("???? УСТРОЙСТВА RASPBERRY PI PICO:")
print("-" * 60)
for port in pico_ports:
print(f" [{port['index']}] {port['device']}")
print(f" ???? {port['description']}")
print(f" ???? {port['hwid']}")
print()

# Потом остальные
if other_ports:
print("???? ДРУГИЕ УСТРОЙСТВА:")
print("-" * 60)
for port in other_ports:
print(f" [{port['index']}] {port['device']}")
print(f" ???? {port['description']}")
print(f" ???? {port['hwid']}")
print()

# Выбор порта
while True:
try:
choice = input("\n???? Введите номер порта (или 'auto' для автоопределения, 'q' для выхода): ").strip()

if choice.lower() == 'q':
return None

if choice.lower() == 'auto':
if pico_ports:
selected = pico_ports[0]['device']
print(f"✅ Автоматически выбран порт: {selected}")
return selected
else:
print("❌ Устройства Pico не найдены для автоопределения")
continue

choice = int(choice)
if 1 <= choice <= len(ports):
return ports[choice-1].device
else:
print(f"❌ Введите число от 1 до {len(ports)}")
except ValueError:
print("❌ Введите корректный номер")
except KeyboardInterrupt:
print("\n❌ Отмена выбора")
return None

@staticmethod
def is_pico_device(port):
"""Проверка, является ли устройство Raspberry Pi Pico"""
if port.vid == 0x2E8A:
return True
if port.pid == 0x0005:
return True

desc_lower = port.description.lower()
pico_keywords = ['pico', 'rp2040', 'raspberry', 'usb serial', 'последовательный']
for keyword in pico_keywords:
if keyword in desc_lower:
return True

return False

class MPU3050Visualizer:
def __init__(self, port=None, baudrate=115200, max_points=200):
"""
Инициализация визуализатора MPU3050

Args:
port: COM порт (если None - будет вызван диалог выбора)
baudrate: скорость соединения
max_points: количество отображаемых точек на графике
"""
self.baudrate = baudrate
self.max_points = max_points
self.running = True
self.data_received = False
self.selected_port = port

# Буферы данных
self.time_data = deque(maxlen=max_points)
self.temp_data = deque(maxlen=max_points)
self.gyro_x_data = deque(maxlen=max_points)
self.gyro_y_data = deque(maxlen=max_points)
self.gyro_z_data = deque(maxlen=max_points)

# Для интеграции углов
self.angle_x = 0
self.angle_y = 0
self.angle_z = 0
self.angle_x_data = deque(maxlen=max_points)
self.angle_y_data = deque(maxlen=max_points)
self.angle_z_data = deque(maxlen=max_points)

# Выбор порта
if self.selected_port is None:
self.selected_port = self.choose_port()

if self.selected_port is None:
print("❌ Порт не выбран. Программа завершена.")
sys.exit(1)

# Подключение к RP2040
self.connect_to_pico()

# Настройка графиков
self.setup_plots()

def choose_port(self):
"""Выбор порта через GUI или консоль"""
print("\n" + "=" * 60)
print("???? MPU3050 Visualizer для Raspberry Pi Pico")
print("=" * 60)

# Проверяем, доступен ли tkinter
try:
# Пробуем создать окно tkinter
root = tk.Tk()
root.withdraw() # Скрываем основное окно

# Спрашиваем пользователя, какой интерфейс использовать
print("\n???? Выберите способ выбора порта:")
print(" 1. Графический интерфейс")
print(" 2. Консольный интерфейс")
print(" 3. Автоматическое определение")

choice = input("\nВаш выбор (1-3, Enter для граф. интерфейса): ").strip()

if choice == '2':
# Консольный выбор
root.destroy()
selector = ConsolePortSelector()
return selector.select_port()
elif choice == '3':
# Автоматическое определение
root.destroy()
ports = list(serial.tools.list_ports.comports())
for port in ports:
if ConsolePortSelector.is_pico_device(port):
print(f"✅ Автоматически найден порт: {port.device}")
return port.device
print("❌ Автоматическое определение не удалось")
return None
else:
# Графический интерфейс
root.destroy()
selector = PortSelector()
return selector.select_port_gui()

except Exception as e:
print(f"⚠️ Не удалось создать графический интерфейс: {e}")
print("???? Используется консольный выбор порта")
selector = ConsolePortSelector()
return selector.select_port()

def find_pico_port(self):
"""Поиск порта с Raspberry Pi Pico (запасной метод)"""
print("\n???? Поиск подключенных устройств...")
ports = list(serial.tools.list_ports.comports())

if not ports:
print("❌ Не найдено ни одного COM порта!")
return None

print(f"\nНайдено {len(ports)} устройств:")
print("-" * 60)

for port in ports:
print(f"???? {port.device}")
print(f" ???? {port.description}")
print(f" ???? VID:PID: {port.vid}:{port.pid}")
print()

# Поиск RP2040
for port in ports:
if any(keyword in port.description.lower() for keyword in
['pico', 'rp2040', 'usb serial', 'последовательный']):
print(f"✅ Найдено подходящее устройство: {port.device}")
return port.device

if port.vid == 0x2E8A or port.pid == 0x0005:
print(f"✅ Найдено устройство RP2040: {port.device}")
return port.device

# Если не нашли, спросим пользователя
print("\n⚠️ Не удалось автоматически определить устройство RP2040.")
try:
choice = input("Введите номер порта (например, COM4) или нажмите Enter для выхода: ").strip()
if choice:
return choice
except:
pass

return None

def connect_to_pico(self):
"""Подключение к Raspberry Pi Pico"""
port = self.selected_port

print(f"\n???? Подключение к {port}...")

try:
self.ser = serial.Serial(port, self.baudrate, timeout=2)
print(f"✅ Подключено к {port}")

# Ожидание инициализации
print("⏳ Ожидание инициализации MPU3050...")
time.sleep(2)
self.ser.reset_input_buffer()

# Проверка готовности
ready = False
start_time = time.time()
timeout = 10

while not ready and time.time() - start_time < timeout:
if self.ser.in_waiting:
try:
line = self.ser.readline().decode().strip()
print(f"???? Получено: {line}")

if line == "MPU3050_READY":
ready = True
print("✅ MPU3050 готов к работе!")
elif line == "MPU3050_INIT":
print("???? Инициализация MPU3050...")
elif line.startswith("MPU3050_OFFSETS"):
offsets = line.split(':')[1].split(',')
print(f"???? Калибровочные смещения: X={offsets[0]}, Y={offsets[1]}, Z={offsets[2]}")
elif line == "MPU3050_CALIBRATING":
print("⚙️ Калибровка гироскопа... (не двигайте датчик)")
except:
pass
time.sleep(0.1)

if not ready:
print("⚠️ Не получен сигнал готовности, но пробуем работать...")

print("✅ Соединение установлено")

except Exception as e:
print(f"❌ Ошибка подключения: {e}")
print("\n???? Возможные решения:")
print(" 1. Проверьте, что устройство не используется другим приложением")
print(" 2. Перезагрузите Raspberry Pi Pico (кнопкой RESET)")
print(" 3. Проверьте правильность загрузки скетча")
print(" 4. Попробуйте другой USB порт")
sys.exit(1)

def setup_plots(self):
"""Настройка интерфейса графиков"""
plt.style.use('dark_background')
self.fig = plt.figure(figsize=(14, 8))
self.fig.suptitle('MPU3050 на Raspberry Pi Pico - Визуализация данных', fontsize=14, color='white')

# График гироскопа
self.ax1 = self.fig.add_subplot(2, 2, 1)
self.line_gx, = self.ax1.plot([], [], 'r-', label='X', linewidth=1.5)
self.line_gy, = self.ax1.plot([], [], 'g-', label='Y', linewidth=1.5)
self.line_gz, = self.ax1.plot([], [], 'b-', label='Z', linewidth=1.5)
self.ax1.set_ylim(-250, 250)
self.ax1.set_ylabel('Угловая скорость (°/s)')
self.ax1.set_title('Гироскоп (сырые данные)')
self.ax1.legend(loc='upper right')
self.ax1.grid(True, alpha=0.3)

# График углов
self.ax2 = self.fig.add_subplot(2, 2, 2)
self.line_ax, = self.ax2.plot([], [], 'r-', label='Roll (X)', linewidth=1.5)
self.line_ay, = self.ax2.plot([], [], 'g-', label='Pitch (Y)', linewidth=1.5)
self.line_az, = self.ax2.plot([], [], 'b-', label='Yaw (Z)', linewidth=1.5)
self.ax2.set_ylim(-180, 180)
self.ax2.set_ylabel('Угол (°)')
self.ax2.set_title('Ориентация')
self.ax2.legend(loc='upper right')
self.ax2.grid(True, alpha=0.3)

# График температуры
self.ax3 = self.fig.add_subplot(2, 2, 3)
self.line_temp, = self.ax3.plot([], [], 'c-', linewidth=2)
self.ax3.set_ylim(20, 40)
self.ax3.set_ylabel('Температура (°C)')
self.ax3.set_xlabel('Время (отсчеты)')
self.ax3.set_title('Температура датчика')
self.ax3.grid(True, alpha=0.3)

# Текстовый вывод
self.ax4 = self.fig.add_subplot(2, 2, 4)
self.ax4.axis('off')
self.text_stats = self.ax4.text(0.1, 0.9, '', transform=self.ax4.transAxes,
fontsize=11, verticalalignment='top',
fontfamily='monospace', color='white')

# Информация о порте
port_info = f"Порт: {self.ser.port}\nСкорость: {self.baudrate}\nУстройство: Raspberry Pi Pico"
self.ax4.text(0.1, 0.1, port_info, transform=self.ax4.transAxes,
fontsize=9, verticalalignment='bottom', color='gray')

# Кнопки управления
self.reset_button_ax = plt.axes([0.45, 0.02, 0.1, 0.04])
self.reset_button = Button(self.reset_button_ax, 'Сброс углов', color='#333333', hovercolor='#444444')
self.reset_button.on_clicked(self.reset_angles)

plt.tight_layout()
plt.subplots_adjust(bottom=0.08)

def reset_angles(self, event):
"""Сброс интегрированных углов"""
self.angle_x = 0
self.angle_y = 0
self.angle_z = 0
print("???? Углы сброшены")

def update_plots(self, frame):
"""Обновление графиков"""
try:
while self.ser.in_waiting:
line = self.ser.readline().decode().strip()
if line and not line.startswith('MPU'):
values = line.split(',')
if len(values) == 5:
try:
temp = float(values[0])
gx = float(values[1])
gy = float(values[2])
gz = float(values[3])
dt = float(values[4])

# Интегрирование
self.angle_x += gx * dt
self.angle_y += gy * dt
self.angle_z += gz * dt

# Добавление в буферы
current_time = len(self.time_data)
self.time_data.append(current_time)
self.temp_data.append(temp)
self.gyro_x_data.append(gx)
self.gyro_y_data.append(gy)
self.gyro_z_data.append(gz)
self.angle_x_data.append(self.angle_x)
self.angle_y_data.append(self.angle_y)
self.angle_z_data.append(self.angle_z)

self.data_received = True
except ValueError:
pass
except Exception as e:
print(f"Ошибка чтения: {e}")

if not self.data_received:
return (self.line_gx, self.line_gy, self.line_gz,
self.line_ax, self.line_ay, self.line_az,
self.line_temp, self.text_stats)

# Обновление графиков
x_data = list(range(len(self.gyro_x_data)))

self.line_gx.set_data(x_data, list(self.gyro_x_data))
self.line_gy.set_data(x_data, list(self.gyro_y_data))
self.line_gz.set_data(x_data, list(self.gyro_z_data))

self.line_ax.set_data(x_data, list(self.angle_x_data))
self.line_ay.set_data(x_data, list(self.angle_y_data))
self.line_az.set_data(x_data, list(self.angle_z_data))

self.line_temp.set_data(x_data, list(self.temp_data))

# Обновление осей
if len(self.gyro_x_data) > 10:
all_gyro = list(self.gyro_x_data) + list(self.gyro_y_data) + list(self.gyro_z_data)
if all_gyro:
max_gyro = max(all_gyro)
min_gyro = min(all_gyro)
margin = max(abs(max_gyro), abs(min_gyro)) * 0.1 + 10
self.ax1.set_ylim(min_gyro - margin, max_gyro + margin)

if self.temp_data:
min_temp = min(self.temp_data)
max_temp = max(self.temp_data)
self.ax3.set_ylim(min_temp - 1, max_temp + 1)

if len(self.gyro_x_data) > 1:
self.ax1.set_xlim(0, len(self.gyro_x_data))
self.ax2.set_xlim(0, len(self.angle_x_data))
self.ax3.set_xlim(0, len(self.temp_data))

# Обновление текста
if self.temp_data:
last_temp = self.temp_data[-1]
last_gx = self.gyro_x_data[-1]
last_gy = self.gyro_y_data[-1]
last_gz = self.gyro_z_data[-1]
last_ax = self.angle_x_data[-1]
last_ay = self.angle_y_data[-1]
last_az = self.angle_z_data[-1]

stats_text = f"""
╔════════════════════════════╗
║ ТЕКУЩИЕ ДАННЫЕ ║
╠════════════════════════════╣
║ Температура: {last_temp:>6.2f} °C ║
╟────────────────────────────╢
║ ГИРОСКОП (°/с): ║
║ X: {last_gx:>8.2f} ║
║ Y: {last_gy:>8.2f} ║
║ Z: {last_gz:>8.2f} ║
╟────────────────────────────╢
║ УГЛЫ (°): ║
║ Roll (X): {last_ax:>7.2f} ║
║ Pitch (Y): {last_ay:>7.2f} ║
║ Yaw (Z): {last_az:>7.2f} ║
╚════════════════════════════╝
"""
self.text_stats.set_text(stats_text)

return (self.line_gx, self.line_gy, self.line_gz,
self.line_ax, self.line_ay, self.line_az,
self.line_temp, self.text_stats)

def run(self):
"""Запуск визуализации"""
ani = animation.FuncAnimation(self.fig, self.update_plots,
interval=50, blit=True, cache_frame_data=False)

print("\n" + "=" * 60)
print("???? ВИЗУАЛИЗАЦИЯ MPU3050 НА RASPBERRY PI PICO")
print("=" * 60)
print(f"???? Порт: {self.ser.port}")
print(f"???? Графики обновляются в реальном времени")
print(f"???? Нажмите кнопку 'Сброс углов' для калибровки")
print(f"❌ Закройте окно для выхода")
print("=" * 60 + "\n")

plt.show()
self.close()

def close(self):
"""Закрытие соединения"""
self.running = False
if hasattr(self, 'ser') and self.ser.is_open:
self.ser.close()
print("✅ Соединение закрыто")

def main():
# Параметры подключения
port = None # None = показать диалог выбора
baudrate = 115200

# Создание и запуск визуализатора
visualizer = MPU3050Visualizer(port, baudrate)

try:
visualizer.run()
except KeyboardInterrupt:
print("\n???? Программа остановлена пользователем")
except Exception as e:
print(f"\n❌ Ошибка: {e}")
import traceback
traceback.print_exc()
finally:
visualizer.close()

if __name__ == "__main__":
main()