import sys import serial import json import os import time from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QComboBox, QPushButton, QTextEdit, QTabWidget, QFormLayout, QLineEdit, QMessageBox, QGroupBox, QCheckBox, QSpinBox) from PyQt5.QtCore import QTimer, Qt from PyQt5.QtGui import QFont # Konfigurace CONFIG_FILE = 'rs485_config.json' def load_config(): """Načte konfiguraci ze souboru""" if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, 'r') as f: return json.load(f) except: pass return {'port': 'COM1', 'baudrate': 9600, 'auto_refresh': False, 'refresh_interval': 1000} def save_config(config): """Uloží konfiguraci do souboru""" with open(CONFIG_FILE, 'w') as f: json.dump(config, f) class RS485Communication: def __init__(self, port='COM1', baudrate=9600): self.port = port self.baudrate = baudrate self.serial = None self.is_connected = False self.try_connect() def try_connect(self): """Pokus o připojení k RS485""" try: if self.serial: self.close() self.serial = serial.Serial( port=self.port, baudrate=self.baudrate, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=1 ) self.is_connected = True return True except serial.SerialException: self.is_connected = False self.serial = None return False def read_data(self): """Čtení dat s automatickým pokusem o znovupřipojení""" if not self.is_connected: if not self.try_connect(): return { 'error': f'RS485 není připojeno (port: {self.port})', 'connected': False } try: if not self.serial.is_open: self.serial.open() # Očištění bufferu před čtením self.serial.reset_input_buffer() # Čekáme na data time.sleep(0.1) if self.serial.in_waiting: data = self.serial.read(self.serial.in_waiting) return { 'connected': True, 'hex': data.hex(), 'raw': data.decode('utf-8', errors='ignore') } return { 'connected': True, 'message': 'Žádná data k dispozici' } except serial.SerialException as e: self.is_connected = False return { 'error': f'Chyba komunikace: {str(e)}', 'connected': False } except Exception as e: return { 'error': f'Neočekávaná chyba: {str(e)}', 'connected': self.is_connected } def update_settings(self, port=None, baudrate=None): """Aktualizuje nastavení sériového portu""" if port: self.port = port if baudrate: self.baudrate = int(baudrate) return self.try_connect() def close(self): """Bezpečné uzavření připojení""" try: if self.serial and self.serial.is_open: self.serial.close() except: pass self.is_connected = False def get_available_ports(self): """Získá seznam dostupných COM portů""" available_ports = [] for i in range(1, 20): # Kontrola COM1 až COM19 port_name = f'COM{i}' try: s = serial.Serial(port_name) s.close() available_ports.append(port_name) except (OSError, serial.SerialException): pass return available_ports class RS485App(QMainWindow): def __init__(self): super().__init__() # Načtení konfigurace self.config = load_config() # Vytvoření instance RS485 self.rs485 = RS485Communication(port=self.config['port'], baudrate=self.config['baudrate']) # Nastavení automatického obnovení self.auto_refresh = self.config.get('auto_refresh', False) self.refresh_interval = self.config.get('refresh_interval', 1000) self.initUI() # Timer pro aktualizaci stavu self.status_timer = QTimer() self.status_timer.timeout.connect(self.update_status_display) self.status_timer.start(2000) # Aktualizace každé 2 sekundy # Timer pro automatické načítání dat self.data_timer = QTimer() self.data_timer.timeout.connect(self.read_data) # Aktivace automatického obnovení podle konfigurace if self.auto_refresh: self.data_timer.start(self.refresh_interval) def initUI(self): self.setWindowTitle('RS485 Komunikační Rozhraní') self.setGeometry(100, 100, 800, 600) # Hlavní widget a layout central_widget = QWidget() main_layout = QVBoxLayout() central_widget.setLayout(main_layout) self.setCentralWidget(central_widget) # Status Bar self.status_label = QLabel() self.update_status_display() # Záložky tab_widget = QTabWidget() main_layout.addWidget(self.status_label) main_layout.addWidget(tab_widget) # Vytvoření záložek settings_tab = self.create_settings_tab() data_tab = self.create_data_tab() # Přidání záložek tab_widget.addTab(settings_tab, "Nastavení") tab_widget.addTab(data_tab, "Data") def create_settings_tab(self): settings_widget = QWidget() layout = QVBoxLayout() # Sekce nastavení připojení settings_group = QGroupBox("Nastavení připojení RS485") form_layout = QFormLayout() # Port self.port_combo = QComboBox() ports = self.rs485.get_available_ports() if ports: self.port_combo.addItems(ports) else: # Pokud nejsou nalezeny žádné porty, přidáme alespoň aktuální port self.port_combo.addItem(self.rs485.port) # Nastavíme aktuální port index = self.port_combo.findText(self.rs485.port) if index >= 0: self.port_combo.setCurrentIndex(index) # Tlačítko pro obnovení seznamu portů refresh_button = QPushButton("Obnovit seznam portů") refresh_button.clicked.connect(self.refresh_ports) port_layout = QHBoxLayout() port_layout.addWidget(self.port_combo) port_layout.addWidget(refresh_button) form_layout.addRow("Sériový port:", port_layout) # Baud rate self.baudrate_combo = QComboBox() baud_rates = ['1200', '2400', '4800', '9600', '19200', '38400', '57600', '115200'] self.baudrate_combo.addItems(baud_rates) # Nastavíme aktuální baudrate index = self.baudrate_combo.findText(str(self.rs485.baudrate)) if index >= 0: self.baudrate_combo.setCurrentIndex(index) form_layout.addRow("Přenosová rychlost:", self.baudrate_combo) settings_group.setLayout(form_layout) layout.addWidget(settings_group) # Sekce automatického obnovení auto_refresh_group = QGroupBox("Automatické načítání dat") auto_refresh_layout = QFormLayout() # Checkbox pro aktivaci self.auto_refresh_checkbox = QCheckBox("Povolit automatické načítání") self.auto_refresh_checkbox.setChecked(self.auto_refresh) self.auto_refresh_checkbox.stateChanged.connect(self.toggle_auto_refresh) auto_refresh_layout.addRow(self.auto_refresh_checkbox) # Nastavení intervalu self.interval_spinner = QSpinBox() self.interval_spinner.setRange(100, 10000) # Od 100ms do 10 sekund self.interval_spinner.setSingleStep(100) self.interval_spinner.setSuffix(" ms") self.interval_spinner.setValue(self.refresh_interval) self.interval_spinner.valueChanged.connect(self.update_refresh_interval) auto_refresh_layout.addRow("Interval obnovení:", self.interval_spinner) auto_refresh_group.setLayout(auto_refresh_layout) layout.addWidget(auto_refresh_group) # Tlačítko pro uložení nastavení save_button = QPushButton("Uložit a připojit") save_button.clicked.connect(self.save_settings) layout.addWidget(save_button) # Přidání vertikálního mezery layout.addStretch() settings_widget.setLayout(layout) return settings_widget def create_data_tab(self): data_widget = QWidget() layout = QVBoxLayout() # Tlačítko pro načtení dat refresh_button = QPushButton("Načíst data") refresh_button.clicked.connect(self.read_data) layout.addWidget(refresh_button) # Indikátor automatického načítání self.auto_refresh_indicator = QLabel() self.update_auto_refresh_indicator() layout.addWidget(self.auto_refresh_indicator) # Oblast pro zobrazení dat self.data_display = QTextEdit() self.data_display.setReadOnly(True) self.data_display.setFont(QFont("Courier New", 10)) layout.addWidget(self.data_display) # Zaškrtávací pole pro zobrazení pouze nových dat self.show_only_new_data = QCheckBox("Zobrazovat pouze nová data (odstraňovat stará)") layout.addWidget(self.show_only_new_data) # Tlačítko pro vymazání clear_button = QPushButton("Vymazat zobrazená data") clear_button.clicked.connect(self.clear_data_display) layout.addWidget(clear_button) data_widget.setLayout(layout) return data_widget def refresh_ports(self): """Obnoví seznam dostupných portů""" self.port_combo.clear() ports = self.rs485.get_available_ports() if ports: self.port_combo.addItems(ports) else: self.port_combo.addItem(self.rs485.port) # Nastavíme zpět aktuální port index = self.port_combo.findText(self.rs485.port) if index >= 0: self.port_combo.setCurrentIndex(index) def save_settings(self): """Uloží nastavení a aktualizuje připojení""" port = self.port_combo.currentText() baudrate = int(self.baudrate_combo.currentText()) if self.rs485.update_settings(port=port, baudrate=baudrate): # Aktualizace konfigurace self.config = { 'port': port, 'baudrate': baudrate, 'auto_refresh': self.auto_refresh, 'refresh_interval': self.refresh_interval } save_config(self.config) QMessageBox.information(self, "Úspěch", "Nastavení bylo úspěšně aktualizováno a připojení navázáno.") self.update_status_display() else: QMessageBox.warning(self, "Chyba", f"Nepodařilo se připojit k portu {port} s rychlostí {baudrate}.") def toggle_auto_refresh(self, state): """Přepne stav automatického načítání dat""" self.auto_refresh = (state == Qt.Checked) if self.auto_refresh: self.data_timer.start(self.refresh_interval) else: self.data_timer.stop() self.update_auto_refresh_indicator() # Aktualizace konfigurace self.config['auto_refresh'] = self.auto_refresh save_config(self.config) def update_refresh_interval(self, value): """Aktualizuje interval obnovení""" self.refresh_interval = value # Pokud je aktivní automatické obnovení, restartujeme timer s novým intervalem if self.auto_refresh: self.data_timer.stop() self.data_timer.start(self.refresh_interval) # Aktualizace konfigurace self.config['refresh_interval'] = self.refresh_interval save_config(self.config) self.update_auto_refresh_indicator() def read_data(self): """Čte data z RS485 a zobrazuje je""" data = self.rs485.read_data() if 'error' in data: self.data_display.setTextColor(Qt.red) # Určí, zda vymazat stávající text nebo přidat k existujícímu if self.show_only_new_data.isChecked(): self.data_display.setText(f"CHYBA: {data['error']}") else: current_text = self.data_display.toPlainText() if current_text: self.data_display.append("\n\n" + "=" * 40 + "\n\n") self.data_display.append(f"CHYBA: {data['error']}") self.update_status_display() return # Zobrazení dat if 'message' in data: # Žádná data if self.show_only_new_data.isChecked(): self.data_display.setTextColor(Qt.black) self.data_display.setText(data['message']) else: timestamp = time.strftime("%H:%M:%S") self.data_display.setTextColor(Qt.gray) current_text = self.data_display.toPlainText() if current_text: self.data_display.append("\n\n" + "=" * 40 + "\n\n") self.data_display.append(f"[{timestamp}] {data['message']}") else: # Máme data self.data_display.setTextColor(Qt.black) timestamp = time.strftime("%H:%M:%S") display_text = f"[{timestamp}] Načtena data z portu {self.rs485.port}\n\n" if 'hex' in data: display_text += f"HEX DATA:\n{data['hex']}\n\n" if 'raw' in data: display_text += f"RAW DATA:\n{data['raw']}" # Určí, zda vymazat stávající text nebo přidat k existujícímu if self.show_only_new_data.isChecked(): self.data_display.setText(display_text) else: current_text = self.data_display.toPlainText() if current_text: self.data_display.append("\n\n" + "=" * 40 + "\n\n") self.data_display.append(display_text) def clear_data_display(self): """Vymaže obsah zobrazení dat""" self.data_display.clear() def update_status_display(self): """Aktualizuje zobrazení stavu připojení""" if self.rs485.is_connected: self.status_label.setText(f"Stav: PŘIPOJENO k portu {self.rs485.port} s rychlostí {self.rs485.baudrate} baud") self.status_label.setStyleSheet("background-color: #d4edda; color: #155724; padding: 8px; border-radius: 4px;") else: self.status_label.setText(f"Stav: ODPOJENO (poslední port: {self.rs485.port})") self.status_label.setStyleSheet("background-color: #f8d7da; color: #721c24; padding: 8px; border-radius: 4px;") def update_auto_refresh_indicator(self): """Aktualizuje zobrazení stavu automatického obnovení""" if self.auto_refresh: self.auto_refresh_indicator.setText(f"Automatické obnovení: AKTIVNÍ (každých {self.refresh_interval} ms)") self.auto_refresh_indicator.setStyleSheet("background-color: #d4edda; color: #155724; padding: 4px; border-radius: 4px;") else: self.auto_refresh_indicator.setText("Automatické obnovení: NEAKTIVNÍ") self.auto_refresh_indicator.setStyleSheet("background-color: #e2e3e5; color: #383d41; padding: 4px; border-radius: 4px;") def closeEvent(self, event): """Událost zavření aplikace""" self.rs485.close() event.accept() if __name__ == '__main__': app = QApplication(sys.argv) app.setStyle('Fusion') # Pro konzistentní vzhled napříč platformami main_window = RS485App() main_window.show() sys.exit(app.exec_())