import socket import matplotlib.pyplot as plt import matplotlib.animation as animation from collections import deque from datetime import datetime, timedelta import re import time import numpy as np import matplotlib.dates as mdates from matplotlib.gridspec import GridSpec class ColorChangingUDPPlotter: def __init__(self, host='192.168.0.50', port=5009, max_points=2000, time_window_minutes=5): self.host = host self.port = port self.max_points = max_points self.time_window_minutes = time_window_minutes self.update_interval = 20 # 100 ms = 10 Hz 20 ms = 50 Hz # Режим отображения (1 - стандартный, 2 - гистограмма) self.display_mode = 1 # Создаем UDP сокет self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.settimeout(0.05) # Инициализация данных self.timestamps = deque(maxlen=max_points) self.brutto_g0_data = deque(maxlen=max_points) self.stat_proc_data = deque(maxlen=max_points) self.stbf0_data = deque(maxlen=max_points) # Для отслеживания stbf0 # Для гистограммы (конфигурация 2) self.histogram_data = deque(maxlen=100) # Отдельный буфер для гистограммы # Для отслеживания изменений stat_proc self.last_stat_proc = None self.last_stbf0 = None # Для фиксированной шкалы времени self.x_min = None self.x_max = None # Цветовые сегменты self.color_segments = [] # (start_index, end_index, color) self.current_color = 'blue' # Начальный цвет # Статистика self.packet_count = 0 self.error_count = 0 self.last_update = time.time() self.start_time = time.time() # Статистика шума для конфигурации 2 self.noise_stats = { 'std_dev': 0, 'variance': 0, 'max_noise': 0, 'mean': 0, 'min': 0, 'max': 0 } # Настройка графиков self.fig = plt.figure(figsize=(14, 10)) self.gs = GridSpec(2, 1, figure=self.fig, height_ratios=[3, 1]) self.ax_main = self.fig.add_subplot(self.gs[0]) self.ax_stats = self.fig.add_subplot(self.gs[1]) self.ax_stats.axis('off') # Будем менять в зависимости от режима self.fig.suptitle(f'ТехАвтоматика TALENTUM MODIUS - Weight Monitoring ({host}:{port})', fontsize=16) # Инициализация основного графика self.setup_configuration_1() # Регулярное выражение для парсинга self.pattern = re.compile( r'PRT2TSM(\d+)PRL(\d+)PRU(\d+)PRP(\d+)STA(\d+)STB(\d+)LCA(\d+)LCB(\d+)WSA(-?\d+)WSB(-?\d+)' ) # Подключаем обработчики клавиш self.fig.canvas.mpl_connect('key_press_event', self.on_key_press) print(f"Color-changing plotter initialized for {host}:{port}") print(f"Time window: {time_window_minutes} minutes") print(f"Update frequency: 10 Hz") print("Press 'm' to switch between configurations") print("Press +/- to change time window, Ctrl+C to stop") def setup_configuration_1(self): """Настройка конфигурации 1 (стандартный график)""" self.display_mode = 1 self.ax_stats.axis('off') # Скрываем нижнюю панель для статистики текста # Очищаем и настраиваем основной график self.ax_main.clear() self.ax_main.set_title(f'Brutto Weight ({self.time_window_minutes}min window) - Blue: STBF0=1, Red: STBF0=0 | +/- to zoom') self.ax_main.grid(True, alpha=0.3) self.ax_main.set_ylabel('Weight (grams)') self.ax_main.set_xlabel('Time') # Текущее значение self.current_value_line = self.ax_main.axhline(y=0, color='green', linestyle='--', alpha=0.7, label='Current') # Легенда для цветов from matplotlib.lines import Line2D legend_elements = [ Line2D([0], [0], color='blue', lw=2, label='STBF0 = 1 (Stable)'), Line2D([0], [0], color='red', lw=2, label='STBF0 = 0 (Unstable)'), Line2D([0], [0], color='green', linestyle='--', lw=1, label='Current Value') ] self.ax_main.legend(handles=legend_elements) # Инициализация сегментов и аннотаций self.line_segments = [] self.proc_annotations = [] self.proc_markers = [] # Текст статистики self.stats_text = self.ax_stats.text(0.02, 0.9, '', transform=self.ax_stats.transAxes, fontsize=10, verticalalignment='top', bbox=dict(boxstyle='round', facecolor='lightgray', alpha=0.8)) def setup_configuration_2(self): """Настройка конфигурации 2 (гистограмма распределения)""" self.display_mode = 2 # Очищаем и настраиваем основной график для гистограммы self.ax_main.clear() self.ax_main.set_title('Weight Distribution Histogram - Noise Analysis') self.ax_main.grid(True, alpha=0.3) self.ax_main.set_ylabel('Frequency') self.ax_main.set_xlabel('Weight Value (grams)') # Настраиваем нижнюю панель для дополнительной статистики self.ax_stats.clear() self.ax_stats.axis('off') # Инициализация гистограммы self.histogram_plot = None self.mean_line = None self.std_dev_area = None # Текст статистики шума self.noise_stats_text = self.ax_stats.text(0.02, 0.9, '', transform=self.ax_stats.transAxes, fontsize=9, verticalalignment='top', bbox=dict(boxstyle='round', facecolor='lightblue', alpha=0.8)) def on_key_press(self, event): """Обработка нажатий клавиш для изменения временного окна и переключения режимов""" if event.key == '+': self.time_window_minutes += 1 self.update_display_title() elif event.key == '-': self.time_window_minutes = max(1, self.time_window_minutes - 1) # Минимум 1 минута self.update_display_title() elif event.key == 'r': self.time_window_minutes = 2 # Сброс к 2 минутам self.update_display_title() elif event.key == 'm': # Переключение между конфигурациями if self.display_mode == 1: self.setup_configuration_2() print("Switched to Configuration 2: Histogram Mode") else: self.setup_configuration_1() print("Switched to Configuration 1: Standard Mode") self.fig.canvas.draw() def update_display_title(self): """Обновление заголовка при изменении временного окна""" if self.display_mode == 1: self.ax_main.set_title(f'Brutto Weight ({self.time_window_minutes}min window) - Blue: STBF0=1, Red: STBF0=0 | +/- to zoom') self.fig.canvas.draw() def update_time_window(self, current_time): """Обновление фиксированного временного окна (только для конфигурации 1)""" if self.display_mode != 1: return window_end = current_time window_start = window_end - timedelta(minutes=self.time_window_minutes) # Устанавливаем фиксированные пределы по X self.ax_main.set_xlim(window_start, window_end) self.x_min = mdates.date2num(window_start) self.x_max = mdates.date2num(window_end) def get_color_for_stbf0(self, stbf0_value): """Возвращает цвет в зависимости от значения STBF0""" return 'blue' if stbf0_value == 1 else 'red' def update_line_segments(self): """Обновление сегментов линии с разными цветами (конфигурация 1)""" # Очищаем старые сегменты for segment in self.line_segments: segment.remove() self.line_segments = [] if len(self.timestamps) < 2: return # Конвертируем временные метки time_numeric = [mdates.date2num(ts) for ts in self.timestamps] # Создаем сегменты для каждого непрерывного участка с одинаковым stbf0 start_idx = 0 current_color = self.get_color_for_stbf0(self.stbf0_data[0]) if self.stbf0_data else 'blue' for i in range(1, len(self.stbf0_data)): if self.stbf0_data[i] != self.stbf0_data[i-1]: # Изменение состояния - создаем сегмент if i - start_idx > 1: # Нужно хотя бы 2 точки для сегмента segment, = self.ax_main.plot( time_numeric[start_idx:i], list(self.brutto_g0_data)[start_idx:i], color=current_color, linewidth=2, zorder=5 ) self.line_segments.append(segment) start_idx = i current_color = self.get_color_for_stbf0(self.stbf0_data[i]) # Добавляем последний сегмент if len(self.stbf0_data) - start_idx > 1: segment, = self.ax_main.plot( time_numeric[start_idx:], list(self.brutto_g0_data)[start_idx:], color=current_color, linewidth=2, zorder=5 ) self.line_segments.append(segment) def update_histogram(self): """Обновление гистограммы (конфигурация 2)""" if not self.histogram_data: return weights = list(self.histogram_data) # Вычисляем статистику шума self.calculate_noise_statistics(weights) # Очищаем предыдущую гистограмму if self.histogram_plot: for patch in self.histogram_plot: patch.remove() if self.mean_line: self.mean_line.remove() if self.std_dev_area: self.std_dev_area.remove() # Создаем гистограмму n, bins, patches = self.ax_main.hist(weights, bins=50, alpha=0.7, color='skyblue', edgecolor='black') self.histogram_plot = patches # Добавляем линию среднего значения self.mean_line = self.ax_main.axvline(self.noise_stats['mean'], color='red', linestyle='--', linewidth=2, label=f'Mean: {self.noise_stats["mean"]:.2f}g') # Добавляем область стандартного отклонения std_min = self.noise_stats['mean'] - self.noise_stats['std_dev'] std_max = self.noise_stats['mean'] + self.noise_stats['std_dev'] self.std_dev_area = self.ax_main.axvspan(std_min, std_max, alpha=0.2, color='green', label=f'±1σ: {self.noise_stats["std_dev"]:.2f}g') # Обновляем легенду self.ax_main.legend() # Обновляем текст статистики self.update_noise_statistics_text() def calculate_noise_statistics(self, weights): """Вычисление статистики шума""" if not weights: return weights_array = np.array(weights) self.noise_stats['mean'] = np.mean(weights_array) self.noise_stats['std_dev'] = np.std(weights_array) self.noise_stats['variance'] = np.var(weights_array) self.noise_stats['max_noise'] = np.max(np.abs(weights_array - self.noise_stats['mean'])) self.noise_stats['min'] = np.min(weights_array) self.noise_stats['max'] = np.max(weights_array) def update_noise_statistics_text(self): """Обновление текста статистики шума""" stats_text = ( f"Sample Size: {len(self.histogram_data)}\n" f"Mean: {self.noise_stats['mean']:.2f} g\n" f"Std Dev: {self.noise_stats['std_dev']:.2f} g\n" f"Variance: {self.noise_stats['variance']:.2f} g²\n" f"Max Noise: ±{self.noise_stats['max_noise']:.2f} g\n" f"Range: {self.noise_stats['min']:.0f} - {self.noise_stats['max']:.0f} g\n" f"Dynamic Range: {self.noise_stats['max'] - self.noise_stats['min']:.2f} g" ) self.noise_stats_text.set_text(stats_text) def send_command(self): """Отправка команды GET_P2""" try: self.sock.sendto(b"GET_P2\r\n", (self.host, self.port)) return True except Exception as e: self.error_count += 1 return False def receive_data(self): """Получение данных по UDP""" try: data, addr = self.sock.recvfrom(1024) return data.decode('utf-8', errors='ignore').strip() except socket.timeout: return None except Exception as e: self.error_count += 1 return None def parse_data(self, data_string): """Парсинг строки данных""" if not data_string: return None, None, None match = self.pattern.match(data_string) if match: try: brutto_g0 = int(match.group(9)) stat_proc = int(match.group(4)) stbf0 = int(match.group(5)) # STB0 значение return brutto_g0, stat_proc, stbf0 except (ValueError, IndexError): self.error_count += 1 return None, None, None return None, None, None def add_proc_annotation(self, timestamp, brutto_value, proc_value): """Добавление аннотации для изменения stat_proc""" timestamp_num = mdates.date2num(timestamp) if self.x_min <= timestamp_num <= self.x_max: marker, = self.ax_main.plot([timestamp_num], [brutto_value], 'ro', markersize=6, alpha=0.7, zorder=10) self.proc_markers.append(marker) annotation = self.ax_main.annotate( f'{proc_value}', xy=(timestamp_num, brutto_value), xytext=(8, 15), textcoords='offset points', fontsize=8, fontweight='bold', color='darkred', bbox=dict(boxstyle='round,pad=0.2', facecolor='yellow', alpha=0.7), zorder=10 ) self.proc_annotations.append(annotation) def clear_old_annotations(self): """Очистка аннотаций, вышедших за пределы временного окна""" annotations_to_keep = [] markers_to_keep = [] for i, (ann, marker) in enumerate(zip(self.proc_annotations, self.proc_markers)): ann_pos = ann.xy[0] if self.x_min <= ann_pos <= self.x_max: annotations_to_keep.append(ann) markers_to_keep.append(marker) else: ann.remove() marker.remove() self.proc_annotations = annotations_to_keep self.proc_markers = markers_to_keep def update_plot(self, frame): """Обновление графика""" current_time = datetime.now() # Отправляем запрос и получаем данные if self.send_command(): raw_data = self.receive_data() if raw_data: brutto_value, stat_proc_value, stbf0_value = self.parse_data(raw_data) if brutto_value is not None and stat_proc_value is not None and stbf0_value is not None: self.packet_count += 1 # Добавляем данные в соответствующие буферы self.timestamps.append(current_time) self.brutto_g0_data.append(brutto_value) self.stat_proc_data.append(stat_proc_value) self.stbf0_data.append(stbf0_value) self.histogram_data.append(brutto_value) # Для гистограммы # Обновляем в зависимости от режима отображения if self.display_mode == 1: self.update_configuration_1(current_time, brutto_value, stat_proc_value, stbf0_value) else: self.update_configuration_2() # Обновляем статистику if time.time() - self.last_update > 0.5: self.update_statistics(brutto_value, stat_proc_value, stbf0_value) self.last_update = time.time() # Собираем все artists для возврата artists = self.get_current_artists() return artists def update_configuration_1(self, current_time, brutto_value, stat_proc_value, stbf0_value): """Обновление конфигурации 1 (стандартный график)""" # Обновляем временное окно self.update_time_window(current_time) # Проверяем изменение stat_proc if self.last_stat_proc is not None and self.last_stat_proc != stat_proc_value: self.add_proc_annotation(current_time, brutto_value, stat_proc_value) self.last_stat_proc = stat_proc_value # Обновляем график если есть данные if len(self.timestamps) > 1: # Обновляем сегменты линии с правильными цветами self.update_line_segments() # Обновляем линию текущего значения self.current_value_line.set_ydata([brutto_value, brutto_value]) # Автомасштабирование только по Y if self.brutto_g0_data: y_min = min(self.brutto_g0_data) y_max = max(self.brutto_g0_data) y_range = y_max - y_min if y_range == 0: y_range = 100 self.ax_main.set_ylim(y_min - y_range * 0.1, y_max + y_range * 0.1) # Форматирование времени self.ax_main.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) self.ax_main.xaxis.set_major_locator(mdates.SecondLocator(interval=5)) plt.setp(self.ax_main.xaxis.get_majorticklabels(), rotation=90) # Очищаем старые аннотации self.clear_old_annotations() def update_configuration_2(self): """Обновление конфигурации 2 (гистограмма)""" # Обновляем гистограмму self.update_histogram() def get_current_artists(self): """Возвращает текущие artists в зависимости от режима""" if self.display_mode == 1: artists = [self.current_value_line, self.stats_text] artists.extend(self.line_segments) artists.extend(self.proc_annotations) artists.extend(self.proc_markers) else: artists = [] if self.histogram_plot: artists.extend(self.histogram_plot) if self.mean_line: artists.append(self.mean_line) if self.std_dev_area: artists.append(self.std_dev_area) artists.append(self.noise_stats_text) return artists def update_statistics(self, current_weight, current_proc, current_stbf0): """Обновление статистики""" elapsed_time = time.time() - self.start_time packets_per_second = self.packet_count / elapsed_time if elapsed_time > 0 else 0 if self.display_mode == 1: # Статус STBF0 stbf0_status = "NORMAL (1)" if current_stbf0 == 1 else "ALERT (0)" stbf0_color = "blue" if current_stbf0 == 1 else "red" stats_text = ( f"Weight: {current_weight} g | Process: {current_proc}\n" f"STBF0: {stbf0_status} | Color: {stbf0_color.upper()}\n" f"Packets: {self.packet_count} | Errors: {self.error_count}\n" f"Success rate: {self.packet_count/(self.packet_count + self.error_count + 1e-10)*100:.1f}%\n" f"Data rate: {packets_per_second:.1f} Hz\n" f"Window: {self.time_window_minutes} min | Running: {elapsed_time:.0f}s" ) self.stats_text.set_text(stats_text) def start(self): """Запуск визуализации""" try: if self.display_mode == 1: self.update_time_window(datetime.now()) ani = animation.FuncAnimation( self.fig, self.update_plot, interval=self.update_interval, blit=False, cache_frame_data=False ) plt.tight_layout() plt.subplots_adjust(top=0.92, hspace=0.3) plt.show() except Exception as e: print(f"Visualization error: {e}") finally: self.close() def close(self): """Закрытие сокета""" self.sock.close() elapsed = time.time() - self.start_time # Подсчитываем время в каждом состоянии blue_time = sum(1 for stbf0 in self.stbf0_data if stbf0 == 1) * (self.update_interval / 1000) red_time = sum(1 for stbf0 in self.stbf0_data if stbf0 == 0) * (self.update_interval / 1000) print(f"\nConnection closed after {elapsed:.1f} seconds") print(f"Total packets: {self.packet_count}, Errors: {self.error_count}") print(f"STBF0=1 (Blue): {blue_time:.1f}s ({blue_time/elapsed*100:.1f}%)") print(f"STBF0=0 (Red): {red_time:.1f}s ({red_time/elapsed*100:.1f}%)") print(f"Final time window: {self.time_window_minutes} minutes") print(f"Histogram samples: {len(self.histogram_data)}") # Упрощенная версия с одной линией (меняет цвет всей линии) class SimpleColorPlotter(ColorChangingUDPPlotter): def __init__(self, host='192.168.0.50', port=5009, max_points=2000, time_window_minutes=2): super().__init__(host, port, max_points, time_window_minutes) # Одна линия вместо сегментов self.main_line, = self.ax_main.plot([], [], linewidth=2, zorder=5) self.current_color = 'blue' def update_configuration_1(self, current_time, brutto_value, stat_proc_value, stbf0_value): """Упрощенное обновление с одной линией""" self.update_time_window(current_time) # Обновляем цвет линии на основе текущего stbf0 new_color = self.get_color_for_stbf0(stbf0_value) if new_color != self.current_color: self.current_color = new_color self.main_line.set_color(new_color) if len(self.timestamps) > 1: time_numeric = [mdates.date2num(ts) for ts in self.timestamps] self.main_line.set_data(time_numeric, list(self.brutto_g0_data)) self.current_value_line.set_ydata([brutto_value, brutto_value]) if self.brutto_g0_data: y_min = min(self.brutto_g0_data) y_max = max(self.brutto_g0_data) y_range = y_max - y_min if y_range == 0: y_range = 100 self.ax_main.set_ylim(y_min - y_range * 0.1, y_max + y_range * 0.1) def get_current_artists(self): """Возвращает artists для упрощенной версии""" if self.display_mode == 1: return [self.main_line, self.current_value_line, self.stats_text] else: return super().get_current_artists() # Запуск программы if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description='Color-changing Weight Monitor by STBF0') parser.add_argument('--host', default='192.168.0.50', help='Device IP address') parser.add_argument('--port', type=int, default=5009, help='UDP port') parser.add_argument('--minutes', type=int, default=2, help='Time window in minutes') parser.add_argument('--simple', action='store_true', help='Use simple single-line version') args = parser.parse_args() print(f"Starting color-changing monitor for {args.host}:{args.port}") print(f"Time window: {args.minutes} minutes") print("Press 'm' to switch between configurations") print("Configuration 1: Standard time-series plot") print("Configuration 2: Histogram with noise analysis") print("Press +/- to change time window, 'r' to reset to 2 minutes") print("Press Ctrl+C to stop...") try: if args.simple: plotter = SimpleColorPlotter(args.host, args.port, time_window_minutes=args.minutes) print("Using simple single-line version") else: plotter = ColorChangingUDPPlotter(args.host, args.port, time_window_minutes=args.minutes) print("Using multi-segment version with historical colors") plotter.start() except KeyboardInterrupt: print("\nMonitoring stopped by user") except Exception as e: print(f"Error: {e}")