import socket import time import re import tkinter as tk from tkinter import ttk, messagebox import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import numpy as np from threading import Thread, Event import collections from scipy import signal from matplotlib.figure import Figure class SimpleUDPClient: def __init__(self, host='192.168.0.50', port=5009): self.host = host self.port = port # Создаем UDP сокет self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.settimeout(0.1) # Таймаут 2 секунды print(f"UDP client initialized for {host}:{port}") def send_command(self, command="GET_P2"): """Отправка команды по UDP""" try: message = command + '\r\n' self.sock.sendto(message.encode('utf-8'), (self.host, self.port)) print(f"Sent: {command}") return True except Exception as e: print(f"Send error: {e}") return False def receive_data(self): """Получение данных по UDP""" try: data, addr = self.sock.recvfrom(2048) # Буфер 1024 байта decoded_data = data.decode('utf-8', errors='ignore').strip() print(f"Received from {addr}: {decoded_data}") return decoded_data except socket.timeout: print("Timeout: no data received") return None except Exception as e: print(f"Receive error: {e}") return None def get_data(self): """Отправка команды и получение ответа""" if self.send_command("GET_P2"): return self.receive_data() return None def extract_lca_value(self, response): """Извлечение значения из ответа устройства""" if response is None: return None, None # Парсинг значения из ответа # PRT2TSM255509PRL10PRU10PRP92STA1STB0LCA7871512LCB8385993WSA5000WSB0LRA7871660LRB8385888WRA3512WRB0 # Ищем паттерн <число> match = re.search(r'LCA(-?\d+)', response) match2 = re.search(r'LRA(-?\d+)', response) lca_value = None lra_value = None if match: try: lra_value = float(match.group(1)) except ValueError: print(f"Error converting LRA value: {match.group(1)}") if match2: try: lca_value = float(match2.group(1)) except ValueError: print(f"Error converting LCA value: {match2.group(1)}") return lra_value, lca_value # Возвращаем оба значения def close(self): """Закрытие сокета""" self.sock.close() print("UDP socket closed") class DataAcquisitionApp: def __init__(self, root): self.root = root self.root.title("Сбор данных и FFT анализ") self.root.geometry("1200x900") # UDP клиент self.udp_client = None # Переменные для данных self.lca_data = collections.deque(maxlen=5000) self.lra_data = collections.deque(maxlen=5000) # Добавляем буфер для LRA данных self.sampling_interval = 0.02 # 20 мс по умолчанию self.samples_per_second = int(1 / self.sampling_interval) # Переменные для FFT self.fft_data = collections.deque(maxlen=50000) self.fft_sampling_rate = 50 # 50 Гц по умолчанию self.fft_is_running = False self.log_scale = False # Флаги управления self.is_running = False self.acquisition_thread = None self.stop_event = Event() self._is_destroyed = False # Статистика self.data_count = 0 self.success_count = 0 # Создание интерфейса self.setup_ui() def setup_ui(self): """Создание единого интерфейса""" # Основной фрейм main_frame = ttk.Frame(self.root, padding="10") main_frame.pack(fill=tk.BOTH, expand=True) # === ФРЕЙМ НАСТРОЕК === settings_frame = ttk.LabelFrame(main_frame, text="Настройки", padding="5") settings_frame.pack(fill=tk.X, pady=5) # Первая строка настроек row1_frame = ttk.Frame(settings_frame) row1_frame.pack(fill=tk.X, pady=2) # Настройки подключения ttk.Label(row1_frame, text="IP:").grid(row=0, column=0, padx=5, sticky=tk.W) self.host_entry = ttk.Entry(row1_frame, width=15) self.host_entry.insert(0, "192.168.0.50") self.host_entry.grid(row=0, column=1, padx=5) ttk.Label(row1_frame, text="Порт:").grid(row=0, column=2, padx=5, sticky=tk.W) self.port_entry = ttk.Entry(row1_frame, width=10) self.port_entry.insert(0, "5009") self.port_entry.grid(row=0, column=3, padx=5) # Настройки частоты опроса ttk.Label(row1_frame, text="Циклов/сек:").grid(row=0, column=4, padx=5, sticky=tk.W) self.samples_entry = ttk.Entry(row1_frame, width=8) self.samples_entry.insert(0, "50") self.samples_entry.grid(row=0, column=5, padx=5) # Вторая строка настроек row2_frame = ttk.Frame(settings_frame) row2_frame.pack(fill=tk.X, pady=2) # Настройки FFT ttk.Label(row2_frame, text="Мин. частота FFT (Гц):").grid(row=0, column=0, padx=5, sticky=tk.W) self.min_freq_entry = ttk.Entry(row2_frame, width=8) self.min_freq_entry.insert(0, "0.01") self.min_freq_entry.grid(row=0, column=1, padx=5) ttk.Label(row2_frame, text="Макс. частота FFT (Гц):").grid(row=0, column=2, padx=5, sticky=tk.W) self.max_freq_entry = ttk.Entry(row2_frame, width=8) self.max_freq_entry.insert(0, "25") self.max_freq_entry.grid(row=0, column=3, padx=5) # Настройки времени отображения ttk.Label(row2_frame, text="Время графика (сек):").grid(row=0, column=4, padx=5, sticky=tk.W) self.time_window_entry = ttk.Entry(row2_frame, width=8) self.time_window_entry.insert(0, "10") self.time_window_entry.grid(row=0, column=5, padx=5) # Переключение шкалы частот self.log_scale_var = tk.BooleanVar() self.log_scale_check = ttk.Checkbutton( row2_frame, text="Логарифмическая шкала частот", variable=self.log_scale_var, command=self.toggle_log_scale ) self.log_scale_check.grid(row=0, column=6, padx=10) # === ФРЕЙМ УПРАВЛЕНИЯ === control_frame = ttk.LabelFrame(main_frame, text="Управление", padding="5") control_frame.pack(fill=tk.X, pady=5) # Кнопки управления self.start_button = ttk.Button(control_frame, text="Старт сбор", command=self.start_acquisition) self.start_button.grid(row=0, column=0, padx=5) self.stop_button = ttk.Button(control_frame, text="Стоп сбор", command=self.stop_acquisition, state=tk.DISABLED) self.stop_button.grid(row=0, column=1, padx=5) self.test_button = ttk.Button(control_frame, text="Тестовый запрос", command=self.test_connection) self.test_button.grid(row=0, column=2, padx=5) self.start_fft_button = ttk.Button(control_frame, text="Запуск FFT", command=self.start_fft) self.start_fft_button.grid(row=0, column=3, padx=5) self.stop_fft_button = ttk.Button(control_frame, text="Остановка FFT", command=self.stop_fft, state=tk.DISABLED) self.stop_fft_button.grid(row=0, column=4, padx=5) # Кнопка для низких частот self.low_freq_button = ttk.Button( control_frame, text="Диапазон 0.001-1 Гц", command=self.set_low_frequency_range ) self.low_freq_button.grid(row=0, column=5, padx=5) # === ФРЕЙМ ИНФОРМАЦИИ === info_frame = ttk.LabelFrame(main_frame, text="Информация", padding="5") info_frame.pack(fill=tk.X, pady=5) info_subframe = ttk.Frame(info_frame) info_subframe.pack(fill=tk.X) # Статус сбора данных self.status_label = ttk.Label(info_subframe, text="Готов к работе") self.status_label.grid(row=0, column=0, padx=10, sticky=tk.W) self.data_count_label = ttk.Label(info_subframe, text="Запросов: 0") self.data_count_label.grid(row=0, column=1, padx=10, sticky=tk.W) self.success_rate_label = ttk.Label(info_subframe, text="Успешных: 0/0 (0%)") self.success_rate_label.grid(row=0, column=2, padx=10, sticky=tk.W) self.current_value_label = ttk.Label(info_subframe, text="Текущее: -") self.current_value_label.grid(row=0, column=3, padx=10, sticky=tk.W) # Статус FFT self.fft_status_label = ttk.Label(info_subframe, text="FFT: остановлен") self.fft_status_label.grid(row=0, column=4, padx=10, sticky=tk.W) self.fft_data_label = ttk.Label(info_subframe, text="Точек FFT: 0") self.fft_data_label.grid(row=0, column=5, padx=10, sticky=tk.W) self.dominant_freq_label = ttk.Label(info_subframe, text="Доминирующая: -") self.dominant_freq_label.grid(row=0, column=6, padx=10, sticky=tk.W) # === ГРАФИКИ === plots_frame = ttk.Frame(main_frame) plots_frame.pack(fill=tk.BOTH, expand=True, pady=5) # Создаем фигуру с двумя подграфиками self.fig = Figure(figsize=(12, 8)) self.signal_ax = self.fig.add_subplot(211) # Верхний график - сигнал self.fft_ax = self.fig.add_subplot(212) # Нижний график - FFT # Настройка графика сигнала self.signal_ax.set_title("Сигнал LRA и LCA") self.signal_ax.set_xlabel("Время (секунды)") self.signal_ax.set_ylabel("Значение") self.signal_ax.grid(True, alpha=0.3) # Настройка графика FFT self.fft_ax.set_title("FFT анализ") self.fft_ax.set_xlabel("Частота (Гц)") self.fft_ax.set_ylabel("Амплитуда (дБ)") self.fft_ax.grid(True, alpha=0.3) self.fft_ax.set_xscale('linear') # По умолчанию линейная шкала # Инициализация линий графиков self.signal_line, = self.signal_ax.plot([], [], 'b-', linewidth=1, label='LСA', alpha=0.6) self.signal_line2, = self.signal_ax.plot([], [], 'r-', linewidth=1, label='LRA', alpha=0.2) # Добавляем красную линию для LCA # Добавляем легенду self.signal_ax.legend() # Две линии для FFT: красная (линейная амплитуда) и синяя (логарифмическая амплитуда) self.fft_line_linear, = self.fft_ax.plot([], [], 'r-', linewidth=1.0, label='LIN амплитуда', alpha=0.6) self.fft_line_log, = self.fft_ax.plot([], [], 'b-', linewidth=0.8, label='LOG амплитуда (дБ)', alpha=0.2) self.fft_ax.legend() # Добавляем вторую ось Y для линейной амплитуды self.fft_ax_linear = self.fft_ax.twinx() self.fft_ax_linear.set_ylabel('Амплитуда LIN') # Встраивание в Tkinter self.canvas = FigureCanvasTkAgg(self.fig, plots_frame) self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) self.fig.tight_layout() def toggle_log_scale(self): """Переключение между линейной и логарифмической шкалой частот""" self.log_scale = self.log_scale_var.get() if self.log_scale: # Логарифмическая шкала self.fft_ax.set_xscale('log') self.fft_ax.set_xlabel("Частота LOG (Гц)") # Улучшаем отображение сетки для логарифмической шкалы self.fft_ax.grid(True, which='major', alpha=0.5) self.fft_ax.grid(True, which='minor', alpha=0.2) # Устанавливаем разумные пределы для логарифмической шкалы try: min_freq = max(0.001, float(self.min_freq_entry.get())) max_freq = float(self.max_freq_entry.get()) self.fft_ax.set_xlim(min_freq, max_freq) except: pass else: # Линейная шкала self.fft_ax.set_xscale('linear') self.fft_ax.set_xlabel("Частота LIN (Гц)") self.fft_ax.grid(True, alpha=0.3) if hasattr(self, 'canvas'): self.canvas.draw_idle() def set_low_frequency_range(self): """Установка диапазона для детального анализа низких частот""" self.min_freq_entry.delete(0, tk.END) self.min_freq_entry.insert(0, "0.001") self.max_freq_entry.delete(0, tk.END) self.max_freq_entry.insert(0, "1.0") # Включаем логарифмическую шкалу для лучшего отображения self.log_scale_var.set(True) self.toggle_log_scale() def start_fft(self): """Запуск FFT анализа""" if not self.fft_is_running: self.fft_is_running = True self.start_fft_button.config(state=tk.DISABLED) self.stop_fft_button.config(state=tk.NORMAL) self.fft_status_label.config(text="FFT: запущен") self.update_fft() def stop_fft(self): """Остановка FFT анализа""" self.fft_is_running = False self.start_fft_button.config(state=tk.NORMAL) self.stop_fft_button.config(state=tk.DISABLED) self.fft_status_label.config(text="FFT: остановлен") def safe_update_ui(self, update_func): """Безопасное обновление UI с проверкой существования окна""" if not self._is_destroyed and self.root.winfo_exists(): try: update_func() except (tk.TclError, RuntimeError): self._is_destroyed = True def update_fft(self): """Обновление FFT анализа""" if not self.fft_is_running or self._is_destroyed: return try: # Используем данные из основного буфера if len(self.lra_data) >= 100: # Минимум 100 точек для качественного FFT # Обновляем данные для FFT self.fft_data.extend(self.lra_data) # Выполняем FFT анализ self.calculate_fft() # Обновляем информацию def update_info(): self.fft_data_label.config(text=f"Точек FFT: {len(self.fft_data)}") # Расчет разрешения по частоте if len(self.fft_data) >= 2: freq_resolution = self.fft_sampling_rate / len(self.fft_data) # self.resolution_label.config(text=f"Разрешение: {freq_resolution:.4f} Гц") self.safe_update_ui(update_info) except Exception as e: print(f"FFT error: {e}") # Планируем следующее обновление if self.fft_is_running and not self._is_destroyed: self.root.after(1000, self.update_fft) # Обновление каждую секунду def calculate_fft(self): """Вычисление FFT""" if len(self.fft_data) < 100: return data = np.array(list(self.fft_data)) n = len(data) # Обновляем частоту дискретизации на основе текущих настроек try: self.fft_sampling_rate = float(self.samples_entry.get()) except: self.fft_sampling_rate = 50 # Удаляем постоянную составляющую (DC offset) data = data - np.mean(data) # Применяем окно Ханна для уменьшения утечки спектра window = np.hanning(n) windowed_data = data * window # Вычисляем FFT fft_result = np.fft.fft(windowed_data) fft_freq = np.fft.fftfreq(n, 1/self.fft_sampling_rate) # Берем только положительные частоты positive_freq_idx = fft_freq > 0 fft_freq_positive = fft_freq[positive_freq_idx] fft_amplitude = np.abs(fft_result[positive_freq_idx]) / (n/2) # Нормализация # Преобразуем в дБ для логарифмического отображения fft_db = 20 * np.log10(fft_amplitude + 1e-12) # Получаем настройки диапазона частот try: min_freq = max(0.001, float(self.min_freq_entry.get())) max_freq = float(self.max_freq_entry.get()) except: min_freq = 0.01 max_freq = 25 # Фильтруем по заданному диапазону частот freq_mask = (fft_freq_positive >= min_freq) & (fft_freq_positive <= max_freq) fft_freq_filtered = fft_freq_positive[freq_mask] fft_amplitude_filtered = fft_amplitude[freq_mask] fft_db_filtered = fft_db[freq_mask] # Находим доминирующую частоту if len(fft_amplitude_filtered) > 0: dominant_idx = np.argmax(fft_amplitude_filtered) dominant_freq = fft_freq_filtered[dominant_idx] dominant_amp = fft_amplitude_filtered[dominant_idx] dominant_db = fft_db_filtered[dominant_idx] def update_dominant(): self.dominant_freq_label.config( text=f"Доминирующая: {dominant_freq:.4f} Гц" ) self.safe_update_ui(update_dominant) # Обновляем графики FFT self.fft_line_linear.set_data(fft_freq_filtered, fft_amplitude_filtered) self.fft_line_log.set_data(fft_freq_filtered, fft_db_filtered) # Масштабируем оси self.fft_ax.relim() self.fft_ax.autoscale_view() # Для логарифмической шкалы устанавливаем специальные деления if self.log_scale: from matplotlib.ticker import LogLocator self.fft_ax.xaxis.set_major_locator(LogLocator(base=10.0, numticks=12)) self.fft_ax.xaxis.set_minor_locator(LogLocator(base=10.0, subs=np.arange(2,10)*0.1, numticks=12)) # Обновляем canvas if not self._is_destroyed and self.root.winfo_exists(): try: self.canvas.draw_idle() except (tk.TclError, RuntimeError): self._is_destroyed = True def connect_udp(self): """Подключение к UDP устройству""" try: host = self.host_entry.get() port = int(self.port_entry.get()) self.udp_client = SimpleUDPClient(host, port) return True except Exception as e: messagebox.showerror("Ошибка подключения", f"Не удалось подключиться: {e}") return False def test_connection(self): """Тестовый запрос для проверки подключения и парсинга""" if not self.udp_client and not self.connect_udp(): return response = self.udp_client.get_data() if response: lra_value, lca_value = self.udp_client.extract_lca_value(response) if lra_value is not None or lca_value is not None: messagebox.showinfo("Тестовый запрос", f"Успешно!\nОтвет: {response}\nLRA значение: {lra_value}\nLCA значение: {lca_value}") else: messagebox.showwarning("Тестовый запрос", f"Получен ответ, но значения не распознаны:\n{response}") else: messagebox.showerror("Тестовый запрос", "Не получен ответ от устройства") def start_acquisition(self): if not self.is_running: # Обновляем настройки из полей ввода try: samples_per_sec = float(self.samples_entry.get()) self.sampling_interval = 1.0 / samples_per_sec self.samples_per_second = int(samples_per_sec) except: messagebox.showerror("Ошибка", "Некорректное значение циклов в секунду") return if not self.udp_client and not self.connect_udp(): return self.is_running = True self.stop_event.clear() self.start_button.config(state=tk.DISABLED) self.stop_button.config(state=tk.NORMAL) self.status_label.config(text="Сбор данных...") # Сброс статистики self.data_count = 0 self.success_count = 0 self.update_stats() self.acquisition_thread = Thread(target=self.data_acquisition_loop) self.acquisition_thread.daemon = True self.acquisition_thread.start() def stop_acquisition(self): if self.is_running: self.is_running = False self.stop_event.set() self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) self.status_label.config(text="Остановлено") def update_stats(self): """Обновление статистики в UI""" def update(): success_rate = (self.success_count / self.data_count * 100) if self.data_count > 0 else 0 self.data_count_label.config(text=f"Запросов: {self.data_count}") self.success_rate_label.config(text=f"Успешных: {self.success_count}/{self.data_count} ({success_rate:.1f}%)") self.safe_update_ui(update) def data_acquisition_loop(self): """Цикл сбора данных""" while self.is_running and not self.stop_event.is_set() and not self._is_destroyed: # Накопление данных за 1 секунду second_data = [] second_data2 = [] # Для LCA данных start_time = time.time() for i in range(self.samples_per_second): if not self.is_running or self.stop_event.is_set() or self._is_destroyed: break # Получение данных response = self.udp_client.get_data() self.data_count += 1 lra_value, lca_value = self.udp_client.extract_lca_value(response) # Теперь получаем два значения if lra_value is not None: second_data.append(lra_value) self.success_count += 1 # Обновление текущего значения def update_current(v=lra_value): self.current_value_label.config(text=f"Текущее: {v:.2f}") self.safe_update_ui(update_current) else: # Если данные не получены, добавляем последнее значение или 0 if self.lra_data: second_data.append(self.lra_data[-1] if self.lra_data else 0) else: second_data.append(0) # Обработка LCA данных (match2) if lca_value is not None: second_data2.append(lca_value) else: if self.lca_data: second_data2.append(self.lca_data[-1] if self.lca_data else 0) else: second_data2.append(0) # Обновление статистики в UI self.update_stats() # Точное ожидание до следующего отсчета elapsed = time.time() - start_time sleep_time = (i + 1) * self.sampling_interval - elapsed if sleep_time > 0: time.sleep(sleep_time) # Обновление графиков в основном потоке if second_data and self.is_running and not self._is_destroyed: self.root.after(0, self.update_plot, second_data, second_data2) def update_plot(self, new_data, new_data2): """Обновление графика сигнала новыми данными""" if self._is_destroyed or not self.root.winfo_exists(): return # Добавление новых данных self.lra_data.extend(new_data) # LRA данные (синяя линия) self.lca_data.extend(new_data2) # LCA данные (красная линия) # Получаем настройку времени отображения try: time_window = float(self.time_window_entry.get()) max_points = int(time_window / self.sampling_interval) # Ограничиваем данные по времени if len(self.lra_data) > max_points: data_to_show = list(self.lra_data)[-max_points:] data_to_show2 = list(self.lca_data)[-max_points:] x_data = np.linspace(-time_window, 0, len(data_to_show)) else: data_to_show = list(self.lra_data) data_to_show2 = list(self.lca_data) x_data = np.arange(len(data_to_show)) * self.sampling_interval - (len(data_to_show) * self.sampling_interval) except: # В случае ошибки показываем все данные data_to_show = list(self.lra_data) data_to_show2 = list(self.lca_data) x_data = np.arange(len(data_to_show)) * self.sampling_interval # Обновление графиков сигнала self.signal_line.set_data(x_data, data_to_show) # LRA (синий) self.signal_line2.set_data(x_data, data_to_show2) # LCA (красный) # Автоматическое масштабирование осей self.signal_ax.relim() self.signal_ax.autoscale_view() # Обновление canvas try: self.canvas.draw_idle() except (tk.TclError, RuntimeError): self._is_destroyed = True # Обновление статуса def update_status(): self.status_label.config(text=f"Данные обновлены. Точек: {len(self.lra_data)}") self.safe_update_ui(update_status) def on_closing(self): """Обработчик закрытия окна""" self._is_destroyed = True self.stop_acquisition() self.stop_fft() if self.udp_client: self.udp_client.close() self.root.destroy() def main(): """Запуск графического приложения""" root = tk.Tk() app = DataAcquisitionApp(root) # Обработчик закрытия окна root.protocol("WM_DELETE_WINDOW", app.on_closing) root.mainloop() if __name__ == "__main__": main()