#!/usr/bin/env python3 """ osc_matplotlib_simple.py - Осциллограф для отображения сигналов из TCP-сервера Версия с индивидуальными настройками смещения для каждого типа меток """ 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.animation import FuncAnimation 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 CSV_DELIMITER = ';' # Константы оптимизации MAX_MARKERS_DISPLAY = 100 # Максимум меток для отображения REDRAW_DELAY_MS = 100 # Задержка для накопления изменений (мс) 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) header = [ "index", "time_seconds", "vector_a", "vector_b", "stLA", "stLB", "stUA", "stUB", "stPA", "stPB", "packet_number", "sample_index" ] writer.writerow(header) # Пакетная запись 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: break except: self.error_callback("Ошибка подключения") 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): self.server_ip = server_ip self.server_port = server_port self.graph_duration = graph_duration self.data_saver = DataSaver() self.max_buffer_size = int(graph_duration * 50) # Настройки аннотаций для каждого типа (по умолчанию 20, 40, 60 пикселей) self.annotation_distance_la = 20 # stLA (красные) self.annotation_distance_ua = 40 # stUA (зеленые) self.annotation_distance_pa = 60 # stPA (синие) self.annotation_offset_percent = 15 # Общий отступ от краев в процентах # Буферы данных 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.last_stLA_value = None self.last_stUA_value = None self.last_stPA_value = None # Метки self.stLA_marker_times = [] self.stLA_marker_values = [] self.stLA_marker_numbers = [] self.stUA_marker_times = [] self.stUA_marker_values = [] self.stUA_marker_numbers = [] self.stPA_marker_times = [] self.stPA_marker_values = [] self.stPA_marker_numbers = [] # Аннотации self.stLA_annotations = [] self.stUA_annotations = [] self.stPA_annotations = [] self.packet_number_buffer = [] self.sample_index_buffer = [] # Флаги отображения self.show_avect = True self.show_bvect = True self.show_stLA_markers = False self.show_stUA_markers = False self.show_stPA_markers = False 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 self._cache_time = None # Флаг необходимости перерисовки self._need_redraw = False self._redraw_timer = None self.init_gui() self.client = OscilloscopeClient( server_ip, server_port, self.on_data_received, self.on_connection_status, self.on_error ) def init_gui(self): self.root = tk.Tk() self.root.title(f"Осциллограф - {self.server_ip}:{self.server_port}") self.root.geometry("1400x900") 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.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) self.record_indicator = ttk.Label(control_grid, text="●", font=('TkDefaultFont', 16, 'bold'), foreground="gray") self.record_indicator.grid(row=0, column=3, padx=5, pady=5, sticky=tk.W) ttk.Label(control_grid, text="Пакеты:").grid(row=0, column=4, 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=5, padx=5, pady=5, sticky=tk.W) ttk.Label(control_grid, text="Сэмплы:").grid(row=0, column=6, 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=7, padx=5, pady=5, sticky=tk.W) 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.stLA_markers, = self.ax.plot([], [], 'ro', markersize=4, visible=False) self.stUA_markers, = self.ax.plot([], [], 'go', markersize=4, visible=False) self.stPA_markers, = self.ax.plot([], [], 'bo', markersize=4, visible=False) 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) 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=1, 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=1, 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=1, column=2, padx=5, pady=2, sticky=tk.W) # Метки markers_frame = ttk.LabelFrame(scrollable_frame, text="Метки со стрелками", padding="10") markers_frame.pack(fill=tk.X, padx=10, pady=5) self.show_stLA_markers_var = tk.BooleanVar(value=self.show_stLA_markers) ttk.Checkbutton(markers_frame, text="stLA (красные)", variable=self.show_stLA_markers_var, command=self.toggle_stLA_markers).grid(row=0, column=0, padx=5, pady=2, sticky=tk.W) self.show_stUA_markers_var = tk.BooleanVar(value=self.show_stUA_markers) ttk.Checkbutton(markers_frame, text="stUA (зеленые)", variable=self.show_stUA_markers_var, command=self.toggle_stUA_markers).grid(row=0, column=1, padx=5, pady=2, sticky=tk.W) self.show_stPA_markers_var = tk.BooleanVar(value=self.show_stPA_markers) ttk.Checkbutton(markers_frame, text="stPA (синие)", variable=self.show_stPA_markers_var, command=self.toggle_stPA_markers).grid(row=0, column=2, padx=5, pady=2, sticky=tk.W) ttk.Button(markers_frame, text="Очистить метки", command=self.clear_all_markers).grid(row=1, column=0, padx=5, pady=5, columnspan=3, sticky=tk.W) # Настройки смещения для каждого типа offset_frame = ttk.LabelFrame(scrollable_frame, text="Смещение аннотаций (пиксели)", padding="10") offset_frame.pack(fill=tk.X, padx=10, pady=5) # stLA (красные) ttk.Label(offset_frame, text="stLA (красные):", foreground='red').grid(row=0, column=0, padx=5, pady=5, sticky=tk.W) self.annotation_distance_la_var = tk.StringVar(value=str(self.annotation_distance_la)) self.annotation_distance_la_entry = ttk.Entry(offset_frame, textvariable=self.annotation_distance_la_var, width=8) self.annotation_distance_la_entry.grid(row=0, column=1, padx=5, pady=5, sticky=tk.W) # stUA (зеленые) ttk.Label(offset_frame, text="stUA (зеленые):", foreground='green').grid(row=1, column=0, padx=5, pady=5, sticky=tk.W) self.annotation_distance_ua_var = tk.StringVar(value=str(self.annotation_distance_ua)) self.annotation_distance_ua_entry = ttk.Entry(offset_frame, textvariable=self.annotation_distance_ua_var, width=8) self.annotation_distance_ua_entry.grid(row=1, column=1, padx=5, pady=5, sticky=tk.W) # stPA (синие) ttk.Label(offset_frame, text="stPA (синие):", foreground='blue').grid(row=2, column=0, padx=5, pady=5, sticky=tk.W) self.annotation_distance_pa_var = tk.StringVar(value=str(self.annotation_distance_pa)) self.annotation_distance_pa_entry = ttk.Entry(offset_frame, textvariable=self.annotation_distance_pa_var, width=8) self.annotation_distance_pa_entry.grid(row=2, column=1, padx=5, pady=5, sticky=tk.W) # Общий отступ от краев ttk.Label(offset_frame, text="Отступ от краев (%):").grid(row=3, column=0, padx=5, pady=5, sticky=tk.W) self.annotation_offset_var = tk.StringVar(value=str(self.annotation_offset_percent)) self.annotation_offset_entry = ttk.Entry(offset_frame, textvariable=self.annotation_offset_var, width=8) self.annotation_offset_entry.grid(row=3, column=1, padx=5, pady=5, sticky=tk.W) # Кнопка применения ttk.Button(offset_frame, text="Применить все настройки", command=self.apply_annotation_settings, width=20).grid(row=4, column=0, padx=5, pady=10, columnspan=2) # Запись 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_indicator_settings = ttk.Label(rec_frame, text="●", font=('TkDefaultFont', 16, 'bold'), foreground="gray") self.record_indicator_settings.grid(row=0, column=1, padx=5, pady=5) self.record_counter_var_settings = tk.StringVar(value="0") ttk.Label(rec_frame, text="Сэмплов:").grid(row=0, column=2, padx=(10, 5), pady=5) ttk.Label(rec_frame, textvariable=self.record_counter_var_settings).grid(row=0, column=3, 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) # Информация 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 apply_annotation_settings(self): """Применение настроек аннотаций для всех типов""" try: # stLA distance_la = int(self.annotation_distance_la_var.get()) if 5 <= distance_la <= 200: self.annotation_distance_la = distance_la else: messagebox.showerror("Ошибка", "Расстояние для stLA должно быть от 5 до 200 пикселей") return # stUA distance_ua = int(self.annotation_distance_ua_var.get()) if 5 <= distance_ua <= 200: self.annotation_distance_ua = distance_ua else: messagebox.showerror("Ошибка", "Расстояние для stUA должно быть от 5 до 200 пикселей") return # stPA distance_pa = int(self.annotation_distance_pa_var.get()) if 5 <= distance_pa <= 200: self.annotation_distance_pa = distance_pa else: messagebox.showerror("Ошибка", "Расстояние для stPA должно быть от 5 до 200 пикселей") return # Общий отступ offset = float(self.annotation_offset_var.get()) if 5 <= offset <= 30: self.annotation_offset_percent = offset else: messagebox.showerror("Ошибка", "Отступ должен быть от 5% до 30%") return self.update_info(f"Настройки: LA={distance_la}px, UA={distance_ua}px, PA={distance_pa}px, отступ={offset}%") self.schedule_redraw() 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.current_record_file = filepath self.record_btn.config(text="Остановить запись") self.record_btn_settings.config(text="Остановить запись") self.record_indicator.config(foreground="red") self.record_indicator_settings.config(foreground="red") 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_indicator.config(foreground="gray") self.record_indicator_settings.config(foreground="gray") self.record_counter_var_settings.set("0") self.update_info(f"Записано {result['samples']} сэмплов") self.current_record_file = None def toggle_avect_display(self): self.show_avect = self.show_avect_var.get() self.schedule_redraw() def toggle_bvect_display(self): self.show_bvect = self.show_bvect_var.get() self.schedule_redraw() def toggle_stLA_markers(self): self.show_stLA_markers = self.show_stLA_markers_var.get() self.schedule_redraw() def toggle_stUA_markers(self): self.show_stUA_markers = self.show_stUA_markers_var.get() self.schedule_redraw() def toggle_stPA_markers(self): self.show_stPA_markers = self.show_stPA_markers_var.get() self.schedule_redraw() 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 * 50) 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.root.title(f"Осциллограф - {self.server_ip}:{self.server_port} (буфер {duration:.0f} сек)") self.update_info(f"Длительность: {duration:.1f} сек") self.schedule_redraw() 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") default_filename = f"oscilloscope_buffer_{timestamp}.csv" filepath = filedialog.asksaveasfilename( title="Сохранить буфер", defaultextension=".csv", initialfile=default_filename, 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.clear_all_markers() 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.last_stLA_value = None self.last_stUA_value = None self.last_stPA_value = None self._times_np = None self._avect_np = None self._bvect_np = None self._cache_time = None self.ax.set_xlim(0, self.graph_duration) self.ax.set_ylim(0, 16000000) self.canvas.draw_idle() def clear_all_markers(self): """Очистка меток и удаление аннотаций""" self.stLA_marker_times.clear() self.stLA_marker_values.clear() self.stLA_marker_numbers.clear() self.stUA_marker_times.clear() self.stUA_marker_values.clear() self.stUA_marker_numbers.clear() self.stPA_marker_times.clear() self.stPA_marker_values.clear() self.stPA_marker_numbers.clear() # Удаляем аннотации for ann in self.stLA_annotations + self.stUA_annotations + self.stPA_annotations: ann.remove() self.stLA_annotations.clear() self.stUA_annotations.clear() self.stPA_annotations.clear() self.schedule_redraw() 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"Меток: LA={len(self.stLA_marker_times)}, UA={len(self.stUA_marker_times)}, PA={len(self.stPA_marker_times)}", f"Смещения: LA={self.annotation_distance_la}px, UA={self.annotation_distance_ua}px, PA={self.annotation_distance_pa}px", f"Отступ от краев: {self.annotation_offset_percent}%", 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'] avect_data = data['Avect'] bvect_data = data['Bvect'] stLA_data = data['stLA'] stLB_data = data['stLB'] stUA_data = data['stUA'] stUB_data = data['stUB'] stPA_data = data['stPA'] stPB_data = data['stPB'] 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 self.last_stLA_value = stLA_data[0] self.last_stUA_value = stUA_data[0] self.last_stPA_value = stPA_data[0] packet_start_time = current_time - 1.0 time_step = 1.0 / actual_length # Пакетное добавление new_times = [packet_start_time + i * time_step for i in range(actual_length)] self.time_buffer.extend(new_times) self.avect_buffer.extend(avect_data[:actual_length]) self.bvect_buffer.extend(bvect_data[:actual_length]) self.stLA_buffer.extend(stLA_data[:actual_length]) self.stLB_buffer.extend(stLB_data[:actual_length]) self.stUA_buffer.extend(stUA_data[:actual_length]) self.stUB_buffer.extend(stUB_data[:actual_length]) self.stPA_buffer.extend(stPA_data[:actual_length]) self.stPB_buffer.extend(stPB_data[: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:] # Сбрасываем кэш numpy массивов self._times_np = None self._avect_np = None self._bvect_np = None # Проверяем изменения для всех сэмплов в пакете for i in range(actual_length): sample_time = packet_start_time + i * time_step if self.last_stLA_value != stLA_data[i]: self.stLA_marker_times.append(sample_time) self.stLA_marker_values.append(avect_data[i]) self.stLA_marker_numbers.append(stLA_data[i]) self.last_stLA_value = stLA_data[i] if self.last_stUA_value != stUA_data[i]: self.stUA_marker_times.append(sample_time) self.stUA_marker_values.append(avect_data[i]) self.stUA_marker_numbers.append(stUA_data[i]) self.last_stUA_value = stUA_data[i] if self.last_stPA_value != stPA_data[i]: self.stPA_marker_times.append(sample_time) self.stPA_marker_values.append(avect_data[i]) self.stPA_marker_numbers.append(stPA_data[i]) self.last_stPA_value = stPA_data[i] self.samples_received += actual_length self.samples_var.set(f"{self.samples_received:,}") # Запись в CSV (пакетная) if self.data_saver.is_recording: timestamp_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] for i in range(actual_length): sample_time = packet_start_time + i * time_step self.data_saver.save_data( timestamp_str, sample_time, avect_data[i], bvect_data[i], stLA_data[i], stLB_data[i], stUA_data[i], stUB_data[i], stPA_data[i], stPB_data[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 # Обновляем numpy массивы при необходимости 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 = self._times_np avect = self._avect_np bvect = 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) # Удаляем старые аннотации for ann in self.stLA_annotations + self.stUA_annotations + self.stPA_annotations: ann.remove() self.stLA_annotations.clear() self.stUA_annotations.clear() self.stPA_annotations.clear() # Определяем видимый диапазон данных 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 = np.min(all_visible) max_val = np.max(all_visible) # Расширяем диапазон для размещения аннотаций data_range = max_val - min_val if data_range > 0: margin_percent = self.annotation_offset_percent / 100.0 margin = data_range * margin_percent y_min = max(0, min_val - margin) y_max = max_val + margin self.ax.set_ylim(y_min, y_max) # Функция для создания аннотации со стрелкой def add_annotation(t, v, num, color, annotations_list, index, distance_px): # Конвертируем пиксели в points # 1 point = 1/72 дюйма, dpi определяет соотношение с пикселями points_per_pixel = 72 / self.fig.dpi point_offset = distance_px * points_per_pixel # Определяем направление: чередуем вверх/вниз if index % 2 == 0: y_offset = point_offset va = 'bottom' arrow_dir = 1 else: y_offset = -point_offset va = 'top' arrow_dir = -1 # Проверяем, не выйдет ли подпись за пределы y_min, y_max = self.ax.get_ylim() # Оценка высоты подписи в координатах данных label_height_points = 20 # Примерная высота подписи в points label_height_data = (y_max - y_min) * (label_height_points / (self.fig.get_figheight() * 72)) # Корректируем если выходит за границы if y_offset > 0 and v + label_height_data > y_max: y_offset = -point_offset * 0.5 va = 'top' arrow_dir = -1 elif y_offset < 0 and v - label_height_data < y_min: y_offset = point_offset * 0.5 va = 'bottom' arrow_dir = 1 # Создаем аннотацию ann = self.ax.annotate( str(num), xy=(t, v), xytext=(15, y_offset), textcoords='offset points', color=color, fontsize=8, fontweight='bold', verticalalignment=va, horizontalalignment='left', bbox=dict(boxstyle='round,pad=0.2', fc='white', alpha=0.9, ec=color, linewidth=1), arrowprops=dict( arrowstyle='-', color=color, alpha=0.7, linewidth=1, connectionstyle=f'arc3,rad={0.1 * arrow_dir}' ), zorder=10 ) annotations_list.append(ann) # Добавляем аннотации для видимых меток с индивидуальными расстояниями if self.show_stLA_markers and self.stLA_marker_times: visible_indices = [] for i, t in enumerate(self.stLA_marker_times[-MAX_MARKERS_DISPLAY:]): if start_time <= t <= current_time: visible_indices.append(i) for idx_in_list, marker_idx in enumerate(visible_indices): t = self.stLA_marker_times[-MAX_MARKERS_DISPLAY:][marker_idx] v = self.stLA_marker_values[-MAX_MARKERS_DISPLAY:][marker_idx] n = self.stLA_marker_numbers[-MAX_MARKERS_DISPLAY:][marker_idx] add_annotation(t, v, n, 'red', self.stLA_annotations, idx_in_list, self.annotation_distance_la) if self.show_stUA_markers and self.stUA_marker_times: visible_indices = [] for i, t in enumerate(self.stUA_marker_times[-MAX_MARKERS_DISPLAY:]): if start_time <= t <= current_time: visible_indices.append(i) for idx_in_list, marker_idx in enumerate(visible_indices): t = self.stUA_marker_times[-MAX_MARKERS_DISPLAY:][marker_idx] v = self.stUA_marker_values[-MAX_MARKERS_DISPLAY:][marker_idx] n = self.stUA_marker_numbers[-MAX_MARKERS_DISPLAY:][marker_idx] add_annotation(t, v, n, 'green', self.stUA_annotations, idx_in_list + 100, self.annotation_distance_ua) if self.show_stPA_markers and self.stPA_marker_times: visible_indices = [] for i, t in enumerate(self.stPA_marker_times[-MAX_MARKERS_DISPLAY:]): if start_time <= t <= current_time: visible_indices.append(i) for idx_in_list, marker_idx in enumerate(visible_indices): t = self.stPA_marker_times[-MAX_MARKERS_DISPLAY:][marker_idx] v = self.stPA_marker_values[-MAX_MARKERS_DISPLAY:][marker_idx] n = self.stPA_marker_numbers[-MAX_MARKERS_DISPLAY:][marker_idx] add_annotation(t, v, n, 'blue', self.stPA_annotations, idx_in_list + 200, self.annotation_distance_pa) # Устанавливаем пределы X self.ax.set_xlim(start_time, current_time) # Обновляем заголовок points_in_view = np.sum(visible_mask) if np.any(visible_mask) else 0 self.ax.set_title(f'Осциллограф: {points_in_view} точек, ' f'LA:{len(self.stLA_annotations)} UA:{len(self.stUA_annotations)} PA:{len(self.stPA_annotations)} | ' f'LA:{self.annotation_distance_la} UA:{self.annotation_distance_ua} PA:{self.annotation_distance_pa}px', fontsize=12) # Обновляем канвас self.canvas.draw_idle() except Exception as e: print(f"Ошибка при перерисовке: {e}") import traceback traceback.print_exc() def run(self): try: self.root.mainloop() except KeyboardInterrupt: if self.data_saver.is_recording: self.data_saver.stop_recording() self.client.stop() self.root.quit() def main(): parser = argparse.ArgumentParser(description='Осциллограф') 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)') args = parser.parse_args() if args.duration < 1.0 or args.duration > 3600.0: print("Ошибка: Длительность должна быть от 1 до 3600 секунд") sys.exit(1) app = OscilloscopeApp(args.ip, args.port, args.duration) app.run() if __name__ == "__main__": main()