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 import csv from datetime import datetime 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) # Буфер 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_all_values(self, response): """Извлечение всех значений из ответа устройства""" if response is None: return {} # Парсинг всех значений из ответа # PRT2TSM255509PRL10PRU10PRP92STA1STB0LCA7871512LCB8385993WSA5000WSB0LRA7871660LRB8385888WRA3512WRB0 patterns = { 'PRT': r'PRT(\d+)', 'TSM': r'TSM(\d+)', 'PRL': r'PRL(\d+)', 'PRU': r'PRU(\d+)', 'PRP': r'PRP(\d+)', 'STA': r'STA(\d+)', 'STB': r'STB(\d+)', 'LCA': r'LCA(\d+)', 'LCB': r'LCB(\d+)', 'WSA': r'WSA(-?\d+)', 'WSB': r'WSB(-?\d+)', 'LRA': r'LRA(\d+)', 'LRB': r'LRB(\d+)', 'WRA': r'WRA(-?\d+)', 'WRB': r'WRB(-?\d+)' } extracted_values = {} for key, pattern in patterns.items(): match = re.search(pattern, response) if match: try: extracted_values[key] = float(match.group(1)) except ValueError: print(f"Error converting {key} value: {match.group(1)}") extracted_values[key] = None else: extracted_values[key] = None return extracted_values def extract_values(self, response): """Извлечение значений WSA и WRA из ответа устройства (для обратной совместимости)""" if response is None: return None, None all_values = self.extract_all_values(response) return all_values.get('WSA'), all_values.get('WRA') def close(self): """Закрытие сокета""" self.sock.close() print("UDP socket closed") class DataAcquisitionApp: def __init__(self, root): self.root = root self.root.title("Сбор данных WSA и WRA") self.root.geometry("800x600") # UDP клиент self.udp_client = None # Переменные для данных self.wsa_data = collections.deque(maxlen=5000) # Увеличили для хранения большего количества данных self.wra_data = collections.deque(maxlen=5000) self.time_data = collections.deque(maxlen=5000) # Храним временные метки self.sampling_interval = 0.02 # 20 мс self.samples_per_second = int(1 / self.sampling_interval) # 50 отсчетов в секунду self.display_window = 10 # Длина отображаемого окна по умолчанию (секунд) self.start_time = None # Время начала сбора данных # Флаги управления 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) # Фрейм настроек подключения connection_frame = ttk.LabelFrame(main_frame, text="Настройки подключения", padding="5") connection_frame.pack(fill=tk.X, pady=5) ttk.Label(connection_frame, text="Host:").grid(row=0, column=0, padx=5, pady=2) self.host_entry = ttk.Entry(connection_frame, width=15) self.host_entry.insert(0, "192.168.0.50") self.host_entry.grid(row=0, column=1, padx=5, pady=2) ttk.Label(connection_frame, text="Port:").grid(row=0, column=2, padx=5, pady=2) self.port_entry = ttk.Entry(connection_frame, width=10) self.port_entry.insert(0, "5009") self.port_entry.grid(row=0, column=3, padx=5, pady=2) # Фрейм управления 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) # Настройка длины отображаемого окна ttk.Label(control_frame, text="Длина окна (сек):").grid(row=0, column=3, padx=5) self.window_length_var = tk.StringVar(value="10") window_length_combo = ttk.Combobox(control_frame, textvariable=self.window_length_var, values=[str(i) for i in range(1, 101)], width=5) window_length_combo.grid(row=0, column=4, padx=5) window_length_combo.bind('<>', self.update_window_length) # Фрейм информации info_frame = ttk.LabelFrame(main_frame, text="Информация", padding="5") info_frame.pack(fill=tk.X, pady=5) self.status_label = ttk.Label(info_frame, text="Готов к работе") self.status_label.pack(anchor=tk.W) self.data_count_label = ttk.Label(info_frame, text="Получено данных: 0") self.data_count_label.pack(anchor=tk.W) self.success_rate_label = ttk.Label(info_frame, text="Успешных запросов: 0/0 (0%)") self.success_rate_label.pack(anchor=tk.W) # Текущие значения self.current_wsa_label = ttk.Label(info_frame, text="Текущее WSA: -") self.current_wsa_label.pack(anchor=tk.W) self.current_wra_label = ttk.Label(info_frame, text="Текущее WRA: -") self.current_wra_label.pack(anchor=tk.W) # Статистика WSA wsa_stats_frame = ttk.Frame(info_frame) wsa_stats_frame.pack(fill=tk.X, pady=2) ttk.Label(wsa_stats_frame, text="WSA:", font=('Arial', 9, 'bold')).pack(side=tk.LEFT, padx=5) self.wsa_min_label = ttk.Label(wsa_stats_frame, text="Мин: -") self.wsa_min_label.pack(side=tk.LEFT, padx=5) self.wsa_max_label = ttk.Label(wsa_stats_frame, text="Макс: -") self.wsa_max_label.pack(side=tk.LEFT, padx=5) self.wsa_rms_label = ttk.Label(wsa_stats_frame, text="RMS: -") self.wsa_rms_label.pack(side=tk.LEFT, padx=5) # Статистика WRA wra_stats_frame = ttk.Frame(info_frame) wra_stats_frame.pack(fill=tk.X, pady=2) ttk.Label(wra_stats_frame, text="WRA:", font=('Arial', 9, 'bold')).pack(side=tk.LEFT, padx=5) self.wra_min_label = ttk.Label(wra_stats_frame, text="Мин: -") self.wra_min_label.pack(side=tk.LEFT, padx=5) self.wra_max_label = ttk.Label(wra_stats_frame, text="Макс: -") self.wra_max_label.pack(side=tk.LEFT, padx=5) self.wra_rms_label = ttk.Label(wra_stats_frame, text="RMS: -") self.wra_rms_label.pack(side=tk.LEFT, padx=5) # График self.setup_plot(main_frame) def setup_plot(self, parent): # Создание фигуры matplotlib plot_frame = ttk.LabelFrame(parent, text="График сигналов WSA и WRA", padding="5") plot_frame.pack(fill=tk.BOTH, expand=True, pady=5) self.fig, self.ax = plt.subplots(figsize=(8, 4)) self.ax.set_title("Сигналы WSA и WRA") self.ax.set_xlabel("Время (секунды)") self.ax.set_ylabel("Значения") self.ax.grid(True) # Инициализация линий графика self.wsa_line, = self.ax.plot([], [], 'b-', linewidth=1, label='WSA') self.wra_line, = self.ax.plot([], [], 'r-', linewidth=1, label='WRA') self.ax.legend() # Встраивание в Tkinter self.canvas = FigureCanvasTkAgg(self.fig, plot_frame) self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) 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_window_length(self, event=None): """Обновление длины отображаемого окна""" try: new_length = int(self.window_length_var.get()) if 1 <= new_length <= 100: self.display_window = new_length # Обновляем максимальную длину данных для хранения max_samples = new_length * self.samples_per_second self.wsa_data = collections.deque(self.wsa_data, maxlen=max_samples) self.wra_data = collections.deque(self.wra_data, maxlen=max_samples) self.time_data = collections.deque(self.time_data, maxlen=max_samples) # Обновляем график if self.is_running: self.update_plot_after_acquisition() except ValueError: messagebox.showerror("Ошибка", "Некорректное значение длины окна") 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: wsa_value, wra_value = self.udp_client.extract_values(response) if wsa_value is not None and wra_value is not None: messagebox.showinfo("Тестовый запрос", f"Успешно!\nОтвет: {response}\nWSA значение: {wsa_value}\nWRA значение: {wra_value}") else: messagebox.showwarning("Тестовый запрос", f"Получен ответ, но значения не распознаны:\n{response}") else: messagebox.showerror("Тестовый запрос", "Не получен ответ от устройства") def start_acquisition(self): if not self.is_running: 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.start_time = time.time() self.wsa_data.clear() self.wra_data.clear() self.time_data.clear() 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}%)") # Обновление статистики сигналов if self.wsa_data: wsa_min = min(self.wsa_data) wsa_max = max(self.wsa_data) wsa_rms = np.sqrt(np.mean(np.array(self.wsa_data)**2)) self.wsa_min_label.config(text=f"Мин: {wsa_min:.2f}") self.wsa_max_label.config(text=f"Макс: {wsa_max:.2f}") self.wsa_rms_label.config(text=f"RMS: {wsa_rms:.2f}") if self.wra_data: wra_min = min(self.wra_data) wra_max = max(self.wra_data) wra_rms = np.sqrt(np.mean(np.array(self.wra_data)**2)) self.wra_min_label.config(text=f"Мин: {wra_min:.2f}") self.wra_max_label.config(text=f"Макс: {wra_max:.2f}") self.wra_rms_label.config(text=f"RMS: {wra_rms:.2f}") self.safe_update_ui(update) def data_acquisition_loop(self): """Цикл сбора данных с интервалом 20мс в течение 1 секунды (50 циклов)""" while self.is_running and not self.stop_event.is_set() and not self._is_destroyed: # Накопление данных за 1 секунду (50 отсчетов с интервалом 20мс) wsa_second_data = [] wra_second_data = [] time_second_data = [] 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 # Получение данных WSA и WRA response = self.udp_client.get_data() self.data_count += 1 wsa_value, wra_value = self.udp_client.extract_values(response) current_time = time.time() - self.start_time if wsa_value is not None and wra_value is not None: wsa_second_data.append(wsa_value) wra_second_data.append(wra_value) time_second_data.append(current_time) self.success_count += 1 # Обновление текущих значений def update_current(wsa=wsa_value, wra=wra_value): self.current_wsa_label.config(text=f"Текущее WSA: {wsa}") self.current_wra_label.config(text=f"Текущее WRA: {wra}") self.safe_update_ui(update_current) else: # Если данные не получены, добавляем последние значения или 0 if self.wsa_data and self.wra_data: wsa_second_data.append(self.wsa_data[-1] if self.wsa_data else 0) wra_second_data.append(self.wra_data[-1] if self.wra_data else 0) time_second_data.append(current_time) else: wsa_second_data.append(0) wra_second_data.append(0) time_second_data.append(current_time) # Обновление статистики в 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) # Обновление графика в основном потоке после сбора 50 точек if wsa_second_data and wra_second_data and time_second_data and self.is_running and not self._is_destroyed: self.root.after(0, self.update_plot_after_acquisition, wsa_second_data, wra_second_data, time_second_data) def update_plot_after_acquisition(self, wsa_new_data=None, wra_new_data=None, time_new_data=None): """Обновление графика новыми данными после сбора 50 точек""" if self._is_destroyed or not self.root.winfo_exists(): return # Добавление новых данных (если переданы) if wsa_new_data is not None and wra_new_data is not None and time_new_data is not None: self.wsa_data.extend(wsa_new_data) self.wra_data.extend(wra_new_data) self.time_data.extend(time_new_data) # Обновление графика current_time_data = list(self.time_data) current_wsa_data = list(self.wsa_data) current_wra_data = list(self.wra_data) # Ограничиваем отображаемые данные текущим окном if current_time_data: current_time = current_time_data[-1] min_display_time = current_time - self.display_window # Находим индекс первого элемента, который попадает в окно отображения display_start_idx = 0 for i, t in enumerate(current_time_data): if t >= min_display_time: display_start_idx = i break displayed_time = current_time_data[display_start_idx:] displayed_wsa = current_wsa_data[display_start_idx:] displayed_wra = current_wra_data[display_start_idx:] else: displayed_time = [] displayed_wsa = [] displayed_wra = [] self.wsa_line.set_data(displayed_time, displayed_wsa) self.wra_line.set_data(displayed_time, displayed_wra) # Устанавливаем правильные пределы по оси X if displayed_time: self.ax.set_xlim(displayed_time[0], displayed_time[-1]) # Автоматическое масштабирование по Y all_displayed_values = displayed_wsa + displayed_wra if all_displayed_values: y_min = min(all_displayed_values) y_max = max(all_displayed_values) y_range = y_max - y_min if y_range == 0: # Если все значения одинаковые y_range = 1 self.ax.set_ylim(y_min - 0.1 * y_range, y_max + 0.1 * y_range) # Обновление 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.wsa_data)}") self.safe_update_ui(update_status) def on_closing(self): """Обработчик закрытия окна""" self._is_destroyed = True self.stop_acquisition() if self.udp_client: self.udp_client.close() self.root.destroy() def continuous_polling(host='192.168.0.50', port=5009, interval=1.0, save_to_csv=False): """Непрерывный опрос устройства с возможностью сохранения в CSV""" print(f"Starting continuous polling every {interval} seconds...") if save_to_csv: print("CSV logging enabled - данные будут сохраняться в файл") print("Press Ctrl+C to stop") client = SimpleUDPClient(host, port) message_count = 0 # Подготовка CSV файла если нужно csv_file = None csv_writer = None if save_to_csv: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"device_data_{timestamp}.csv" csv_file = open(filename, 'w', newline='', encoding='utf-8') csv_writer = csv.writer(csv_file, delimiter=';') # Записываем заголовок с порядковым номером header = ['Request_Number', 'Timestamp', 'PRT', 'TSM', 'PRL', 'PRU', 'PRP', 'STA', 'STB', 'LCA', 'LCB', 'WSA', 'WSB', 'LRA', 'LRB', 'WRA', 'WRB'] csv_writer.writerow(header) print(f"Данные сохраняются в файл: {filename}") try: while True: message_count += 1 print(f"\n--- Message #{message_count} ---") response = client.get_data() if response: # Простой парсинг для отображения if 'PRT2TSM' in response: print("✓ Valid data format received") all_values = client.extract_all_values(response) # Сохраняем в CSV если нужно if save_to_csv and csv_writer: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] row = [message_count, timestamp] # Порядковый номер и timestamp for key in ['PRT', 'TSM', 'PRL', 'PRU', 'PRP', 'STA', 'STB', 'LCA', 'LCB', 'WSA', 'WSB', 'LRA', 'LRB', 'WRA', 'WRB']: value = all_values.get(key) row.append(value if value is not None else '') csv_writer.writerow(row) csv_file.flush() # Сбрасываем буфер для немедленной записи # Выводим основные значения в консоль wsa_value = all_values.get('WSA') wra_value = all_values.get('WRA') if wsa_value is not None and wra_value is not None: print(f"WSA value: {wsa_value}, WRA value: {wra_value}") else: print("WSA or WRA values not found") # Выводим все распарсенные значения для информации print("Все значения:", end=" ") for key, value in all_values.items(): if value is not None: print(f"{key}:{value}", end=" ") print() else: print("⚠ Unknown response format") time.sleep(interval) except KeyboardInterrupt: print("\nStopped by user") except Exception as e: print(f"Error: {e}") finally: if csv_file: csv_file.close() print(f"Данные сохранены в файл, всего запросов: {message_count}") client.close() def stress_test(host='192.168.0.50', port=5009, count=10, delay=0.1, save_to_csv=False): """Тестирование с множественными запросами с возможностью сохранения в CSV""" print(f"Stress test: {count} requests with {delay}s delay") if save_to_csv: print("CSV logging enabled") client = SimpleUDPClient(host, port) success_count = 0 # Подготовка CSV файла если нужно csv_file = None csv_writer = None if save_to_csv: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"stress_test_{timestamp}.csv" csv_file = open(filename, 'w', newline='', encoding='utf-8') csv_writer = csv.writer(csv_file, delimiter=';') header = ['Request_Number', 'Timestamp', 'PRT', 'TSM', 'PRL', 'PRU', 'PRP', 'STA', 'STB', 'LCA', 'LCB', 'WSA', 'WSB', 'LRA', 'LRB', 'WRA', 'WRB'] csv_writer.writerow(header) print(f"Данные сохраняются в файл: {filename}") for i in range(count): request_number = i + 1 print(f"\nRequest {request_number}/{count}:") response = client.get_data() if response: all_values = client.extract_all_values(response) wsa_value = all_values.get('WSA') wra_value = all_values.get('WRA') # Сохраняем в CSV если нужно if save_to_csv and csv_writer: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] row = [request_number, timestamp] # Порядковый номер и timestamp for key in ['PRT', 'TSM', 'PRL', 'PRU', 'PRP', 'STA', 'STB', 'LCA', 'LCB', 'WSA', 'WSB', 'LRA', 'LRB', 'WRA', 'WRB']: value = all_values.get(key) row.append(value if value is not None else '') csv_writer.writerow(row) csv_file.flush() if wsa_value is not None and wra_value is not None: success_count += 1 print(f"✓ Success, WSA: {wsa_value}, WRA: {wra_value}") else: print("✓ Success (no WSA or WRA values)") else: print("✗ Failed") time.sleep(delay) if csv_file: csv_file.close() print(f"Данные теста сохранены в файл, всего запросов: {count}") client.close() print(f"\nTest completed: {success_count}/{count} successful") return success_count def main(): """Запуск графического приложения""" root = tk.Tk() app = DataAcquisitionApp(root) # Обработчик закрытия окна root.protocol("WM_DELETE_WINDOW", app.on_closing) root.mainloop() if __name__ == "__main__": print("=" * 50) print("UDP Client for WSA and WRA Data Acquisition") print("=" * 50) # Выбор режима работы choice = input("Выберите режим (1 - Графический интерфейс, 2 - Консольный режим): ") if choice == "1": main() else: # Консольный режим HOST = '192.168.0.50' PORT = 5009 # Спрашиваем про сохранение в CSV save_csv = input("Сохранять данные в CSV файл? (y/n): ").lower().strip() == 'y' # Выбор типа теста test_type = input("Выберите тип теста (1 - Непрерывный опрос, 2 - Стресс-тест): ") if test_type == "2": try: count = int(input("Количество запросов: ") or "10") delay = float(input("Интервал между запросами (сек): ") or "0.1") stress_test(HOST, PORT, count, delay, save_csv) except ValueError as e: print(f"Ошибка ввода: {e}") else: try: interval = float(input("Интервал опроса (секунды): ") or "1.0") continuous_polling(HOST, PORT, interval, save_csv) except ValueError: print("Invalid interval, using 1.0 second") continuous_polling(HOST, PORT, 1.0, save_csv) except KeyboardInterrupt: print("\nПрограмма завершена")