#!/usr/bin/env python3 """ talentum_PRT3_FFT.py - Осциллограф с FFT анализом Добавлена настройка частоты дискретизации Добавлены отдельные настройки масштаба для X и Y Исправлена ошибка инициализации data_saver """ import sys import socket import struct import zlib import time import argparse import csv import os from datetime import datetime from collections import deque from threading import Thread, Lock, Event import numpy as np import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import tkinter as tk from tkinter import ttk, scrolledtext, messagebox, filedialog # Константы структур SEND_SIGNATURE = 0x9B22BD09 RECV_SIGNATURE = 0x6971CCAB VERSION = 0x01 FLAGS = 0x00 RESERVED = 0x00 CMD = 0x01 PACKET_SIZE = 1848 CSV_DELIMITER = ';' MAX_MARKERS_DISPLAY = 100 REDRAW_DELAY_MS = 100 FFT_AUTO_UPDATE_INTERVAL_MS = 1000 # Автообновление FFT каждую секунду class DataSaver: """Класс для сохранения данных в файл""" def __init__(self): self.csv_file = None self.csv_writer = None self.is_recording = False self.record_start_time = None self.samples_recorded = 0 self.output_directory = "recordings" if not os.path.exists(self.output_directory): os.makedirs(self.output_directory) def start_recording(self): if self.is_recording: return False timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"oscilloscope_data_{timestamp}.csv" filepath = os.path.join(self.output_directory, filename) try: self.csv_file = open(filepath, 'w', newline='', encoding='utf-8') self.csv_writer = csv.writer(self.csv_file, delimiter=CSV_DELIMITER) header = ["timestamp", "time_seconds", "vector_a", "vector_b", "stLA", "stLB", "stUA", "stUB", "stPA", "stPB", "packet_number", "sample_index"] self.csv_writer.writerow(header) self.is_recording = True self.record_start_time = time.time() self.samples_recorded = 0 return filepath except Exception as e: print(f"Ошибка при создании файла: {e}") return None def save_data(self, timestamp, time_seconds, vector_a, vector_b, stLA, stLB, stUA, stUB, stPA, stPB, packet_number, sample_index): if not self.is_recording or self.csv_writer is None: return False try: self.csv_writer.writerow([timestamp, f"{time_seconds:.6f}", vector_a, vector_b, stLA, stLB, stUA, stUB, stPA, stPB, packet_number, sample_index]) self.samples_recorded += 1 return True except Exception as e: print(f"Ошибка при записи данных: {e}") return False def stop_recording(self): if not self.is_recording: return False try: if self.csv_file: self.csv_file.close() self.is_recording = False record_duration = time.time() - self.record_start_time if self.record_start_time else 0 return {'samples': self.samples_recorded, 'duration': record_duration, 'start_time': self.record_start_time} except Exception as e: print(f"Ошибка при закрытии файла: {e}") return None def save_buffer_to_csv(self, filename, time_buffer, avect_data, bvect_data, stLA_buffer, stLB_buffer, stUA_buffer, stUB_buffer, stPA_buffer, stPB_buffer, packet_numbers, sample_indices): try: with open(filename, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.writer(csvfile, delimiter=CSV_DELIMITER) writer.writerow(["index", "time_seconds", "vector_a", "vector_b", "stLA", "stLB", "stUA", "stUB", "stPA", "stPB", "packet_number", "sample_index"]) rows = [] min_len = min(len(time_buffer), len(avect_data), len(bvect_data)) for i in range(min_len): rows.append([i + 1, f"{time_buffer[i]:.6f}", avect_data[i], bvect_data[i], stLA_buffer[i] if i < len(stLA_buffer) else 0, stLB_buffer[i] if i < len(stLB_buffer) else 0, stUA_buffer[i] if i < len(stUA_buffer) else 0, stUB_buffer[i] if i < len(stUB_buffer) else 0, stPA_buffer[i] if i < len(stPA_buffer) else 0, stPB_buffer[i] if i < len(stPB_buffer) else 0, packet_numbers[i] if i < len(packet_numbers) else 0, sample_indices[i] if i < len(sample_indices) else 0]) writer.writerows(rows) return True except Exception as e: print(f"Ошибка при сохранении буфера: {e}") return False class OscilloscopeClient: """Клиент для сбора данных с сервера""" def __init__(self, server_ip, server_port, data_callback, status_callback, error_callback): self.server_ip = server_ip self.server_port = server_port self.data_callback = data_callback self.status_callback = status_callback self.error_callback = error_callback self.running = False self.sock = None self.thread = None self.lock = Lock() self.stop_event = Event() self.request_interval = 1.0 self._request_packet = None def create_request_packet(self): if self._request_packet is None: packet_without_crc = struct.pack('= self.request_interval: self.sock.sendall(request_packet) data_buffer.clear() bytes_received = 0 while bytes_received < PACKET_SIZE: chunk = self.sock.recv(min(4096, PACKET_SIZE - bytes_received)) if not chunk: break data_buffer.extend(chunk) bytes_received += len(chunk) if bytes_received == PACKET_SIZE: response = self.parse_response(bytes(data_buffer)) if response: self.data_callback(response) last_request_time = current_time self.stop_event.wait(0.01) except socket.timeout: continue except Exception as e: print(f"Ошибка в цикле: {e}") break except Exception as e: self.error_callback(f"Ошибка подключения: {e}") self.status_callback(False, "Ошибка") finally: self.stop() def start(self): if not self.running: self.stop_event.clear() self.thread = Thread(target=self.client_thread, daemon=True) self.thread.start() def stop(self): self.running = False self.stop_event.set() if self.sock: try: self.sock.close() except: pass self.status_callback(False, "Отключено") class OscilloscopeApp: """Главное приложение осциллографа""" def __init__(self, server_ip, server_port, graph_duration=600.0, sampling_freq=50.0): self.server_ip = server_ip self.server_port = server_port self.graph_duration = graph_duration self.sampling_freq = sampling_freq # Частота дискретизации в Гц # СНАЧАЛА создаем DataSaver self.data_saver = DataSaver() self.max_buffer_size = int(graph_duration * sampling_freq) # Буферы данных self.avect_buffer = [] self.bvect_buffer = [] self.time_buffer = [] self.stLA_buffer = [] self.stLB_buffer = [] self.stUA_buffer = [] self.stUB_buffer = [] self.stPA_buffer = [] self.stPB_buffer = [] self.packet_number_buffer = [] self.sample_index_buffer = [] # Флаги отображения self.show_avect = True self.show_bvect = True self.overlay_mode = False # Статистика self.packet_count = 0 self.samples_received = 0 self.data_received = False self.start_time = None self.last_packet_time = None # Кэш numpy self._times_np = None self._avect_np = None self._bvect_np = None # FFT настройки self.fft_range_map = {"1-25Гц": (1.0, 25.0), "0,01-1Гц": (0.01, 1.0), "0,01-25Гц": (0.01, 25.0), "Полный": (0.01, sampling_freq/2)} # Создаем корневое окно self.root = tk.Tk() self.root.title(f"Осциллограф - {self.server_ip}:{self.server_port}") self.root.geometry("1400x900") # Tkinter переменные self.fft_range_var = tk.StringVar(value="1-25Гц") self.fft_points_var = tk.StringVar(value="32768") self.fft_scale_x_var = tk.StringVar(value="Линейный") self.fft_scale_y_var = tk.StringVar(value="Линейный") self.fft_auto_update_var = tk.BooleanVar(value=True) self.sampling_freq_var = tk.StringVar(value=str(sampling_freq)) self._need_redraw = False self._redraw_timer = None self._fft_update_timer = None # ТЕПЕРЬ создаем GUI self.init_gui() # Создаем клиента self.client = OscilloscopeClient(server_ip, server_port, self.on_data_received, self.on_connection_status, self.on_error) # Запускаем автообновление FFT self.start_fft_auto_update() def init_gui(self): self.tab_control = ttk.Notebook(self.root) self.tab_control.pack(expand=True, fill=tk.BOTH, padx=10, pady=10) self.tab_plot = ttk.Frame(self.tab_control) self.tab_control.add(self.tab_plot, text='График') self.create_plot_tab() self.tab_settings = ttk.Frame(self.tab_control) self.tab_control.add(self.tab_settings, text='Настройки') self.create_settings_tab() self.tab_fft = ttk.Frame(self.tab_control) self.tab_control.add(self.tab_fft, text='FFT') self.create_fft_tab() self.tab_control.bind("<>", self.on_tab_change) self.status_var = tk.StringVar(value="Статус: Не подключено") status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W, padding=5) status_bar.pack(side=tk.BOTTOM, fill=tk.X) def create_plot_tab(self): main_plot_frame = ttk.Frame(self.tab_plot, padding="10") main_plot_frame.pack(fill=tk.BOTH, expand=True) plot_control_frame = ttk.LabelFrame(main_plot_frame, text="Управление", padding="10") plot_control_frame.pack(fill=tk.X, pady=(0, 10)) control_grid = ttk.Frame(plot_control_frame) control_grid.pack(fill=tk.X) self.connect_btn = ttk.Button(control_grid, text="Подключиться", command=self.toggle_connection, width=15) self.connect_btn.grid(row=0, column=0, padx=5, pady=5, sticky=tk.W) self.clear_btn = ttk.Button(control_grid, text="Очистить", command=self.clear_graphs, width=15) self.clear_btn.grid(row=0, column=1, padx=5, pady=5, sticky=tk.W) self.record_btn = ttk.Button(control_grid, text="Начать запись", command=self.toggle_recording, width=15) self.record_btn.grid(row=0, column=2, padx=5, pady=5, sticky=tk.W) ttk.Label(control_grid, text="Пакеты:").grid(row=0, column=3, padx=(20, 5), pady=5, sticky=tk.W) self.packet_var = tk.StringVar(value="0") ttk.Label(control_grid, textvariable=self.packet_var, font=('TkDefaultFont', 10, 'bold')).grid(row=0, column=4, padx=5, pady=5) ttk.Label(control_grid, text="Сэмплы:").grid(row=0, column=5, padx=(10, 5), pady=5, sticky=tk.W) self.samples_var = tk.StringVar(value="0") ttk.Label(control_grid, textvariable=self.samples_var, font=('TkDefaultFont', 10, 'bold')).grid(row=0, column=6, padx=5, pady=5) plot_frame = ttk.Frame(main_plot_frame) plot_frame.pack(fill=tk.BOTH, expand=True) self.fig, self.ax = plt.subplots(figsize=(14, 6), dpi=100) self.ax.set_facecolor('#f5f5f5') self.fig.patch.set_facecolor('#f0f0f0') self.line1, = self.ax.plot([], [], 'b-', linewidth=1.5, label='Вектор A') self.line2, = self.ax.plot([], [], 'r-', linewidth=1.5, label='Вектор B') self.ax.set_xlabel('Время (сек)', fontsize=12) self.ax.set_ylabel('Значение (24-бит)', fontsize=12) self.ax.set_title(f'Осциллограф - буфер {self.graph_duration} сек', fontsize=14, fontweight='bold') self.ax.legend(loc='upper right', fontsize=10) self.ax.grid(True, linestyle='--', alpha=0.7) self.ax.set_xlim(0, self.graph_duration) self.ax.set_ylim(0, 16000000) self.canvas = FigureCanvasTkAgg(self.fig, plot_frame) self.canvas.draw() self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) def create_settings_tab(self): main_settings_frame = ttk.Frame(self.tab_settings) main_settings_frame.pack(fill=tk.BOTH, expand=True) canvas = tk.Canvas(main_settings_frame) scrollbar = ttk.Scrollbar(main_settings_frame, orient="vertical", command=canvas.yview) scrollable_frame = ttk.Frame(canvas) scrollable_frame.bind("", lambda e: canvas.configure(scrollregion=canvas.bbox("all"))) canvas.create_window((0, 0), window=scrollable_frame, anchor="nw") canvas.configure(yscrollcommand=scrollbar.set) canvas.pack(side="left", fill="both", expand=True) scrollbar.pack(side="right", fill="y") conn_frame = ttk.LabelFrame(scrollable_frame, text="Подключение", padding="10") conn_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(conn_frame, text="IP:").grid(row=0, column=0, padx=5, pady=5, sticky=tk.W) self.ip_var = tk.StringVar(value=self.server_ip) ttk.Entry(conn_frame, textvariable=self.ip_var, width=15).grid(row=0, column=1, padx=5, pady=5) ttk.Label(conn_frame, text="Порт:").grid(row=0, column=2, padx=(10, 5), pady=5, sticky=tk.W) self.port_var = tk.StringVar(value=str(self.server_port)) ttk.Entry(conn_frame, textvariable=self.port_var, width=8).grid(row=0, column=3, padx=5, pady=5) self.connection_status_var = tk.StringVar(value="Не подключено") ttk.Label(conn_frame, text="Статус:").grid(row=1, column=0, padx=5, pady=5, sticky=tk.W) ttk.Label(conn_frame, textvariable=self.connection_status_var, font=('TkDefaultFont', 10, 'bold')).grid(row=1, column=1, padx=5, pady=5, columnspan=3, sticky=tk.W) disp_frame = ttk.LabelFrame(scrollable_frame, text="Отображение", padding="10") disp_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(disp_frame, text="Длительность (сек):").grid(row=0, column=0, padx=5, pady=5, sticky=tk.W) self.duration_var = tk.StringVar(value=str(self.graph_duration)) ttk.Entry(disp_frame, textvariable=self.duration_var, width=8).grid(row=0, column=1, padx=5, pady=5) ttk.Button(disp_frame, text="Применить", command=self.apply_duration, width=10).grid(row=0, column=2, padx=5, pady=5) ttk.Label(disp_frame, text="Частота дискретизации (Гц):").grid(row=1, column=0, padx=5, pady=5, sticky=tk.W) ttk.Entry(disp_frame, textvariable=self.sampling_freq_var, width=8).grid(row=1, column=1, padx=5, pady=5) ttk.Button(disp_frame, text="Применить", command=self.apply_sampling_freq, width=10).grid(row=1, column=2, padx=5, pady=5) self.show_avect_var = tk.BooleanVar(value=self.show_avect) ttk.Checkbutton(disp_frame, text="Вектор A", variable=self.show_avect_var, command=self.toggle_avect_display).grid(row=2, column=0, padx=5, pady=2, sticky=tk.W) self.show_bvect_var = tk.BooleanVar(value=self.show_bvect) ttk.Checkbutton(disp_frame, text="Вектор B", variable=self.show_bvect_var, command=self.toggle_bvect_display).grid(row=2, column=1, padx=5, pady=2, sticky=tk.W) self.overlay_var = tk.BooleanVar(value=self.overlay_mode) ttk.Checkbutton(disp_frame, text="Наложение", variable=self.overlay_var, command=self.toggle_overlay_mode).grid(row=2, column=2, padx=5, pady=2, sticky=tk.W) rec_frame = ttk.LabelFrame(scrollable_frame, text="Запись", padding="10") rec_frame.pack(fill=tk.X, padx=10, pady=5) self.record_btn_settings = ttk.Button(rec_frame, text="Начать запись", command=self.toggle_recording, width=15) self.record_btn_settings.grid(row=0, column=0, padx=5, pady=5) self.record_counter_var_settings = tk.StringVar(value="0") ttk.Label(rec_frame, text="Сэмплов:").grid(row=0, column=1, padx=(10, 5), pady=5) ttk.Label(rec_frame, textvariable=self.record_counter_var_settings).grid(row=0, column=2, padx=5, pady=5) ttk.Button(rec_frame, text="Сохранить буфер", command=self.save_buffer_to_file, width=15).grid(row=1, column=0, padx=5, pady=5, columnspan=2) ttk.Button(rec_frame, text="Папка записей", command=self.open_recordings_folder, width=15).grid(row=1, column=2, padx=5, pady=5, columnspan=2) fft_settings_frame = ttk.LabelFrame(scrollable_frame, text="Настройка FFT", padding="10") fft_settings_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(fft_settings_frame, text="Диапазон частот:").grid(row=0, column=0, padx=5, pady=5, sticky=tk.W) fft_range_combo = ttk.Combobox(fft_settings_frame, textvariable=self.fft_range_var, values=list(self.fft_range_map.keys()), state="readonly", width=12) fft_range_combo.grid(row=0, column=1, padx=5, pady=5) ttk.Label(fft_settings_frame, text="Точек накопления:").grid(row=1, column=0, padx=5, pady=5, sticky=tk.W) ttk.Entry(fft_settings_frame, textvariable=self.fft_points_var, width=8).grid(row=1, column=1, padx=5, pady=5) ttk.Label(fft_settings_frame, text="Масштаб X (частота):").grid(row=2, column=0, padx=5, pady=5, sticky=tk.W) ttk.Combobox(fft_settings_frame, textvariable=self.fft_scale_x_var, values=["Линейный", "Логарифмический"], state="readonly", width=15).grid(row=2, column=1, padx=5, pady=5) ttk.Label(fft_settings_frame, text="Масштаб Y (амплитуда):").grid(row=3, column=0, padx=5, pady=5, sticky=tk.W) ttk.Combobox(fft_settings_frame, textvariable=self.fft_scale_y_var, values=["Линейный", "Логарифмический"], state="readonly", width=15).grid(row=3, column=1, padx=5, pady=5) ttk.Checkbutton(fft_settings_frame, text="Автообновление (1 сек)", variable=self.fft_auto_update_var, command=self.toggle_fft_auto_update).grid(row=4, column=0, columnspan=2, padx=5, pady=5, sticky=tk.W) ttk.Button(fft_settings_frame, text="Применить настройки FFT", command=self.apply_fft_settings, width=20).grid(row=5, column=0, padx=5, pady=10, columnspan=2) info_frame = ttk.LabelFrame(scrollable_frame, text="Информация", padding="10") info_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) self.info_text = scrolledtext.ScrolledText(info_frame, height=8, width=80, font=('Courier', 9), wrap=tk.WORD) self.info_text.pack(fill=tk.BOTH, expand=True) self.info_text.insert('1.0', "Ожидание данных...\n") self.info_text.config(state='disabled') def create_fft_tab(self): main_fft_frame = ttk.Frame(self.tab_fft, padding="10") main_fft_frame.pack(fill=tk.BOTH, expand=True) fft_control_frame = ttk.Frame(main_fft_frame) fft_control_frame.pack(fill=tk.X, pady=(0, 10)) self.fft_refresh_btn = ttk.Button(fft_control_frame, text="Обновить FFT", command=self.update_fft_plot, width=15) self.fft_refresh_btn.pack(side=tk.LEFT, padx=5) self.fft_status_var = tk.StringVar(value="Готово к анализу") ttk.Label(fft_control_frame, textvariable=self.fft_status_var).pack(side=tk.LEFT, padx=10) fft_plot_frame = ttk.Frame(main_fft_frame) fft_plot_frame.pack(fill=tk.BOTH, expand=True) self.fft_fig, self.fft_ax = plt.subplots(figsize=(14, 6), dpi=100) self.fft_ax.set_facecolor('#f5f5f5') self.fft_fig.patch.set_facecolor('#f0f0f0') self.fft_ax.set_xlabel('Частота (Гц)', fontsize=12) self.fft_ax.set_ylabel('Амплитуда', fontsize=12) self.fft_ax.set_title('Спектральный анализ (FFT)', fontsize=14, fontweight='bold') self.fft_ax.grid(True, linestyle='--', alpha=0.7) self.fft_canvas = FigureCanvasTkAgg(self.fft_fig, fft_plot_frame) self.fft_canvas.draw() self.fft_canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) def start_fft_auto_update(self): """Запускает автоматическое обновление FFT""" if self.fft_auto_update_var.get(): self.update_fft_plot() self._fft_update_timer = self.root.after(FFT_AUTO_UPDATE_INTERVAL_MS, self.start_fft_auto_update) def toggle_fft_auto_update(self): """Включает/выключает автообновление FFT""" if self.fft_auto_update_var.get(): self.start_fft_auto_update() self.fft_status_var.set("Автообновление ВКЛ") else: if self._fft_update_timer: self.root.after_cancel(self._fft_update_timer) self._fft_update_timer = None self.fft_status_var.set("Автообновление ВЫКЛ") def on_tab_change(self, event): if self.tab_control.tab(self.tab_control.select(), "text") == "FFT": self.update_fft_plot() # ==================== МЕТОДЫ УПРАВЛЕНИЯ ==================== def apply_sampling_freq(self): """Применяет новую частоту дискретизации""" try: new_freq = float(self.sampling_freq_var.get()) if new_freq <= 0: messagebox.showerror("Ошибка", "Частота должна быть положительной") return if new_freq > 1000: if not messagebox.askyesno("Подтверждение", f"Частота {new_freq} Гц очень высокая. Это может замедлить работу. Продолжить?"): return self.sampling_freq = new_freq self.max_buffer_size = int(self.graph_duration * self.sampling_freq) # Обновляем диапазоны FFT self.fft_range_map["Полный"] = (0.01, self.sampling_freq/2) # Обрезаем буферы если они стали слишком большими if len(self.time_buffer) > self.max_buffer_size: excess = len(self.time_buffer) - self.max_buffer_size self.time_buffer = self.time_buffer[excess:] self.avect_buffer = self.avect_buffer[excess:] self.bvect_buffer = self.bvect_buffer[excess:] self._times_np = None self._avect_np = None self._bvect_np = None self.update_info(f"Частота дискретизации: {self.sampling_freq} Гц") self.schedule_redraw() if self.fft_auto_update_var.get(): self.update_fft_plot() except ValueError: messagebox.showerror("Ошибка", "Введите корректное число") def apply_fft_settings(self): try: points = int(self.fft_points_var.get()) available_points = len(self.avect_buffer) if not (1024 <= points <= 65536): messagebox.showerror("Ошибка", "Количество точек должно быть от 1024 до 65536") return if available_points < 1024: messagebox.showwarning("Предупреждение", f"В буфере только {available_points} точек. FFT будет использовать доступные данные.") self.update_info(f"Настройки FFT применены (запрошено {points} точек)") if self.fft_auto_update_var.get(): self.update_fft_plot() except ValueError: messagebox.showerror("Ошибка", "Введите корректное число точек") def toggle_connection(self): if not self.client.running: try: self.server_ip = self.ip_var.get() self.server_port = int(self.port_var.get()) self.client.server_ip = self.server_ip self.client.server_port = self.server_port self.start_time = None self.last_packet_time = None self.connect_btn.config(text="Отключиться") self.status_var.set("Статус: Подключение...") self.connection_status_var.set("Подключение...") self.update_info("Подключение...") self.client.start() except: messagebox.showerror("Ошибка", "Некорректные параметры") else: self.disconnect() def disconnect(self): self.connect_btn.config(text="Подключиться") self.client.stop() def toggle_recording(self): if not self.data_saver.is_recording: filepath = self.data_saver.start_recording() if filepath: self.record_btn.config(text="Остановить запись") self.record_btn_settings.config(text="Остановить запись") self.update_info(f"Запись: {os.path.basename(filepath)}") else: messagebox.showerror("Ошибка", "Не удалось начать запись") else: result = self.data_saver.stop_recording() if result: self.record_btn.config(text="Начать запись") self.record_btn_settings.config(text="Начать запись") self.record_counter_var_settings.set("0") self.update_info(f"Записано {result['samples']} сэмплов") def toggle_avect_display(self): self.show_avect = self.show_avect_var.get() self.schedule_redraw() if self.fft_auto_update_var.get(): self.update_fft_plot() def toggle_bvect_display(self): self.show_bvect = self.show_bvect_var.get() self.schedule_redraw() if self.fft_auto_update_var.get(): self.update_fft_plot() def toggle_overlay_mode(self): self.overlay_mode = self.overlay_var.get() self.schedule_redraw() def apply_duration(self): try: duration = float(self.duration_var.get()) if 1.0 <= duration <= 3600.0: self.graph_duration = duration self.max_buffer_size = int(duration * self.sampling_freq) if len(self.time_buffer) > self.max_buffer_size: excess = len(self.time_buffer) - self.max_buffer_size self.time_buffer = self.time_buffer[excess:] self.avect_buffer = self.avect_buffer[excess:] self.bvect_buffer = self.bvect_buffer[excess:] self.root.title(f"Осциллограф - {self.server_ip}:{self.server_port} (буфер {duration:.0f} сек)") self.update_info(f"Длительность: {duration:.1f} сек") self.schedule_redraw() else: messagebox.showerror("Ошибка", "Длительность должна быть от 1 до 3600 секунд") except: messagebox.showerror("Ошибка", "Введите число") def save_buffer_to_file(self): if len(self.time_buffer) == 0: messagebox.showwarning("Предупреждение", "Нет данных") return timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filepath = filedialog.asksaveasfilename(title="Сохранить буфер", defaultextension=".csv", initialfile=f"oscilloscope_buffer_{timestamp}.csv", filetypes=[("CSV files", "*.csv")]) if filepath: success = self.data_saver.save_buffer_to_csv(filepath, self.time_buffer, self.avect_buffer, self.bvect_buffer, self.stLA_buffer, self.stLB_buffer, self.stUA_buffer, self.stUB_buffer, self.stPA_buffer, self.stPB_buffer, self.packet_number_buffer, self.sample_index_buffer) if success: self.update_info(f"Сохранено: {os.path.basename(filepath)}") messagebox.showinfo("Успех", f"Сохранено в {filepath}") def open_recordings_folder(self): try: if sys.platform == "win32": os.startfile(self.data_saver.output_directory) elif sys.platform == "darwin": os.system(f'open "{self.data_saver.output_directory}"') else: os.system(f'xdg-open "{self.data_saver.output_directory}"') except: messagebox.showerror("Ошибка", "Не удалось открыть папку") def clear_graphs(self): self.avect_buffer.clear() self.bvect_buffer.clear() self.time_buffer.clear() self.stLA_buffer.clear() self.stLB_buffer.clear() self.stUA_buffer.clear() self.stUB_buffer.clear() self.stPA_buffer.clear() self.stPB_buffer.clear() self.packet_number_buffer.clear() self.sample_index_buffer.clear() self.packet_count = 0 self.samples_received = 0 self.packet_var.set("0") self.samples_var.set("0") self.data_received = False self.start_time = None self.last_packet_time = None self._times_np = None self._avect_np = None self._bvect_np = None self.ax.set_xlim(0, self.graph_duration) self.ax.set_ylim(0, 16000000) self.canvas.draw_idle() self.fft_status_var.set("Буфер очищен") def update_info(self, message): self.info_text.config(state='normal') self.info_text.delete('1.0', tk.END) lines = [ f"[{datetime.now().strftime('%H:%M:%S')}] {message}", "-" * 50, f"Пакетов: {self.packet_count}", f"Сэмплов: {self.samples_received:,}", f"В буфере: {len(self.time_buffer):,} / {self.max_buffer_size:,}", f"Частота дискретизации: {self.sampling_freq} Гц", f"Запись: {'ВКЛ' if self.data_saver.is_recording else 'ВЫКЛ'}" ] if self.data_saver.is_recording: self.record_counter_var_settings.set(str(self.data_saver.samples_recorded)) self.info_text.insert('1.0', '\n'.join(lines)) self.info_text.config(state='disabled') # ==================== ОБРАБОТКА ДАННЫХ ==================== def on_data_received(self, data): try: self.packet_count += 1 self.packet_var.set(str(self.packet_count)) current_time = time.time() actual_length = data['actual_length'] if actual_length > 0: self.data_received = True if self.start_time is None: self.start_time = current_time self.last_packet_time = current_time # Используем реальную частоту дискретизации time_step = 1.0 / self.sampling_freq packet_start_time = current_time - (actual_length * time_step) new_times = [packet_start_time + i * time_step for i in range(actual_length)] self.time_buffer.extend(new_times) self.avect_buffer.extend(data['Avect'][:actual_length]) self.bvect_buffer.extend(data['Bvect'][:actual_length]) self.stLA_buffer.extend(data['stLA'][:actual_length]) self.stLB_buffer.extend(data['stLB'][:actual_length]) self.stUA_buffer.extend(data['stUA'][:actual_length]) self.stUB_buffer.extend(data['stUB'][:actual_length]) self.stPA_buffer.extend(data['stPA'][:actual_length]) self.stPB_buffer.extend(data['stPB'][:actual_length]) self.packet_number_buffer.extend([self.packet_count] * actual_length) self.sample_index_buffer.extend(range(1, actual_length + 1)) if len(self.time_buffer) > self.max_buffer_size: excess = len(self.time_buffer) - self.max_buffer_size self.time_buffer = self.time_buffer[excess:] self.avect_buffer = self.avect_buffer[excess:] self.bvect_buffer = self.bvect_buffer[excess:] self.stLA_buffer = self.stLA_buffer[excess:] self.stLB_buffer = self.stLB_buffer[excess:] self.stUA_buffer = self.stUA_buffer[excess:] self.stUB_buffer = self.stUB_buffer[excess:] self.stPA_buffer = self.stPA_buffer[excess:] self.stPB_buffer = self.stPB_buffer[excess:] self.packet_number_buffer = self.packet_number_buffer[excess:] self.sample_index_buffer = self.sample_index_buffer[excess:] self._times_np = None self._avect_np = None self._bvect_np = None self.samples_received += actual_length self.samples_var.set(f"{self.samples_received:,}") if self.data_saver.is_recording: ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] for i in range(actual_length): self.data_saver.save_data(ts, packet_start_time + i*time_step, data['Avect'][i], data['Bvect'][i], data['stLA'][i], data['stLB'][i], data['stUA'][i], data['stUB'][i], data['stPA'][i], data['stPB'][i], self.packet_count, i + 1) if self.packet_count % 10 == 0: self.update_info(f"Пакет {self.packet_count}") self.schedule_redraw() except Exception as e: print(f"Ошибка: {e}") def on_connection_status(self, connected, message): if connected: self.status_var.set("Статус: Подключено") self.connection_status_var.set("Подключено") self.update_info("Подключено") else: self.status_var.set(f"Статус: {message}") self.connection_status_var.set(message) if self.data_saver.is_recording: self.toggle_recording() if self.client.running: self.connect_btn.config(text="Подключиться") def on_error(self, error_msg): self.update_info(f"Ошибка: {error_msg}") # ==================== ПЕРЕРИСОВКА ПО СОБЫТИЮ ==================== def schedule_redraw(self): self._need_redraw = True if self._redraw_timer is None: self._redraw_timer = self.root.after(REDRAW_DELAY_MS, self.do_redraw) def do_redraw(self): if not self._need_redraw: return self._need_redraw = False self._redraw_timer = None try: if not self.data_received or len(self.time_buffer) == 0: return if self._times_np is None: self._times_np = np.array(self.time_buffer) self._avect_np = np.array(self.avect_buffer) self._bvect_np = np.array(self.bvect_buffer) times, avect, bvect = self._times_np, self._avect_np, self._bvect_np current_time = times[-1] start_time = max(0, current_time - self.graph_duration) if self.show_avect: self.line1.set_data(times, avect) self.line1.set_visible(True) else: self.line1.set_visible(False) if self.show_bvect: self.line2.set_data(times, bvect) self.line2.set_visible(True) else: self.line2.set_visible(False) visible_mask = (times >= start_time) & (times <= current_time) if np.any(visible_mask): visible_data = [] if self.show_avect: visible_data.append(avect[visible_mask]) if self.show_bvect: visible_data.append(bvect[visible_mask]) if visible_data: all_visible = np.concatenate(visible_data) if len(all_visible) > 0: min_val, max_val = np.min(all_visible), np.max(all_visible) margin = (max_val - min_val) * 0.1 self.ax.set_ylim(max(0, min_val - margin), max_val + margin) self.ax.set_xlim(start_time, current_time) self.ax.set_title(f'Осциллограф: {np.sum(visible_mask)} точек', fontsize=12) self.canvas.draw_idle() except Exception as e: print(f"Ошибка при перерисовке: {e}") def update_fft_plot(self): """Обновляет FFT график с раздельными масштабами X и Y""" try: available_points = len(self.avect_buffer) if available_points < 1024: self.fft_status_var.set(f"Недостаточно данных для FFT (есть {available_points}, мин. 1024)") self.fft_ax.clear() self.fft_ax.set_title("Ожидание данных...") self.fft_ax.set_xlabel('Частота (Гц)', fontsize=12) self.fft_ax.set_ylabel('Амплитуда', fontsize=12) self.fft_canvas.draw_idle() return # Получаем запрошенное количество точек try: requested_points = int(self.fft_points_var.get()) except: requested_points = 32768 # Используем минимальное из доступного и запрошенного points = min(requested_points, available_points) # Берем последние 'points' точек data_a = np.array(self.avect_buffer[-points:], dtype=np.float64) data_b = np.array(self.bvect_buffer[-points:], dtype=np.float64) # Используем заданную частоту дискретизации fs = self.sampling_freq # Вычисляем FFT freqs = np.fft.rfftfreq(points, d=1.0/fs) fft_a = np.abs(np.fft.rfft(data_a)) / points fft_b = np.abs(np.fft.rfft(data_b)) / points # Применяем масштаб Y if self.fft_scale_y_var.get() == "Логарифмический": fft_a = 20 * np.log10(fft_a + 1e-12) fft_b = 20 * np.log10(fft_b + 1e-12) y_label = "Амплитуда (dB)" else: y_label = "Амплитуда" # Получаем диапазон частот f_min, f_max = self.fft_range_map.get(self.fft_range_var.get(), (1.0, 25.0)) # Ограничиваем максимальную частоту частотой Найквиста f_max = min(f_max, fs/2) # Создаем маску для фильтрации частот mask = (freqs >= f_min) & (freqs <= f_max) # Применяем маску freqs_masked = freqs[mask] fft_a_masked = fft_a[mask] fft_b_masked = fft_b[mask] # Применяем масштаб X (логарифмический требует положительных частот) use_log_x = (self.fft_scale_x_var.get() == "Логарифмический") if use_log_x and len(freqs_masked) > 0: # Для логарифмического масштаба X нужно убедиться, что нет нулевых частот positive_mask = freqs_masked > 0 freqs_masked = freqs_masked[positive_mask] fft_a_masked = fft_a_masked[positive_mask] fft_b_masked = fft_b_masked[positive_mask] # Очищаем и перерисовываем график self.fft_ax.clear() plot_anything = False if self.show_avect and len(fft_a_masked) > 0: if use_log_x: self.fft_ax.semilogx(freqs_masked, fft_a_masked, 'b-', linewidth=1.5, label='FFT Вектор A') else: self.fft_ax.plot(freqs_masked, fft_a_masked, 'b-', linewidth=1.5, label='FFT Вектор A') plot_anything = True if self.show_bvect and len(fft_b_masked) > 0: if use_log_x: self.fft_ax.semilogx(freqs_masked, fft_b_masked, 'r-', linewidth=1.5, label='FFT Вектор B') else: self.fft_ax.plot(freqs_masked, fft_b_masked, 'r-', linewidth=1.5, label='FFT Вектор B') plot_anything = True if not plot_anything: self.fft_ax.text(0.5, 0.5, 'Нет данных для отображения\n(выберите вектор A или B)', transform=self.fft_ax.transAxes, ha='center', va='center') x_label = 'Частота (Гц)' if use_log_x: x_label = 'Частота (Гц) - логарифмическая шкала' self.fft_ax.set_xlabel(x_label, fontsize=12) self.fft_ax.set_ylabel(y_label, fontsize=12) self.fft_ax.set_title(f'FFT: Диапазон {f_min:.2f}-{f_max:.2f} Гц, Точек: {points}, fs={fs:.1f}Гц', fontsize=14, fontweight='bold') if plot_anything: self.fft_ax.legend(loc='upper right', fontsize=10) self.fft_ax.grid(True, linestyle='--', alpha=0.7) # Автоматическая настройка пределов по Y all_values = [] if self.show_avect and len(fft_a_masked) > 0: all_values.extend(fft_a_masked) if self.show_bvect and len(fft_b_masked) > 0: all_values.extend(fft_b_masked) if all_values: # Отфильтровываем inf и nan all_values = [v for v in all_values if np.isfinite(v)] if all_values: y_min = np.min(all_values) y_max = np.max(all_values) if y_min != y_max: margin = (y_max - y_min) * 0.1 self.fft_ax.set_ylim(y_min - margin, y_max + margin) self.fft_canvas.draw_idle() # Обновляем статус self.fft_status_var.set(f"Обновлено: {points} точек из {available_points}, fs={fs:.1f}Гц") except Exception as e: print(f"Ошибка FFT: {e}") import traceback traceback.print_exc() self.fft_status_var.set(f"Ошибка: {str(e)[:50]}") def run(self): try: self.root.mainloop() except KeyboardInterrupt: if self.data_saver.is_recording: self.data_saver.stop_recording() if self.fft_auto_update_var.get() and self._fft_update_timer: self.root.after_cancel(self._fft_update_timer) self.client.stop() self.root.quit() def main(): parser = argparse.ArgumentParser(description='Осциллограф с FFT анализом') parser.add_argument('ip', help='IP адрес сервера') parser.add_argument('port', type=int, help='Порт') parser.add_argument('duration', type=float, nargs='?', default=600.0, help='Длительность графика (1-3600)') parser.add_argument('--sampling_freq', type=float, default=50.0, help='Частота дискретизации в Гц (по умолчанию 50)') args = parser.parse_args() if not (1.0 <= args.duration <= 3600.0): print("Ошибка: Длительность должна быть от 1 до 3600 секунд") sys.exit(1) if args.sampling_freq <= 0: print("Ошибка: Частота дискретизации должна быть положительной") sys.exit(1) app = OscilloscopeApp(args.ip, args.port, args.duration, args.sampling_freq) app.run() if __name__ == "__main__": main()