import sys import serial import json import os import time import re from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QComboBox, QPushButton, QTextEdit, QTabWidget, QFormLayout, QLineEdit, QMessageBox, QGroupBox, QCheckBox, QSpinBox, QTableWidget, QTableWidgetItem) from PyQt5.QtCore import QTimer, Qt from PyQt5.QtGui import QFont # Konfigurace CONFIG_FILE = 'rs485_config.json' def parse_game_data(input_string, source_string): """Parsování herních dat podle zadaného formátování""" parsed_data = {} lines = input_string.strip().split("\n") for line in lines: parts = line.split(";") if len(parts) < 3: continue # Přeskočení neplatných řádků key, label, formula = parts[:3] value = "" ad_matches = re.findall(r'AD\((\d+),(\d+)\)', formula) ab_matches = re.findall(r'AB\((\d+),(\d+),7,([^)]+)\)', formula) for row, col in ad_matches: row, col = int(row), int(col) index = row * 10 + col if index < len(source_string): value += source_string[index] for row, col, suffix in ab_matches: row, col = int(row), int(col) index = row * 10 + col if index < len(source_string): value += source_string[index] + suffix if key == "GAME_TIMER" and len(value) >= 4: value = f"{value[:2]}:{value[2:]}" # Formátování času mm:ss parsed_data[key] = value return parsed_data 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) # Proměnná pro ukládání poslední přijatá RAW data pro parsování self.last_raw_data = "" # Načítání parsovacích pravidel self.parse_rules = """GAME_TIMER;Hrací čas;AD(0,5)+AD(0,6)+AB(0,6,7,:)+AB(0,7,7,SCS_GAME_TIMER_SS_D)+AD(0,7)+AD(0,8); PERIOD;Perioda; AD(0,13); SCORE_HOME;Skóre Domácí;AD(0,10)+AD(0,11)+AD(0,12); SCORE_VISITORS;Skóre Hosté;AD(0,14)+AD(0,15)+AD(0,16); FAUL_TEAM_HOME;Fauly týmu D;AD(0,4); FAUL_TEAM_VISITORS;Fauly týmu H;AD(0,9); TIMEOUT_TIME;Čas timeoutu;AD(0,2)+AD(0,3); TIMEOUT_DOTS_HOME;Timeout Domácí;AB(0,10,7,.)+ST( )+AB(0,11,7,.)+ST( )+AB(0,12,7,.); TIMEOUT_DOTS_VISITORS;Timeout Hosté;AB(0,14,7,.)+ST( )+AB(0,15,7,.)+ST( )+AB(0,16,7,.); PENALTY_TIME_HOME1;Čas vyloučení D hráč 1;AD(1,22)+AB(1,22,7,:)+AD(1,23)+AD(1,24); PENALTY_TIME_HOME2;Čas vyloučení D hráč 2;AD(1,27)+AB(1,27,7,:)+AD(1,28)+AD(1,29); PENALTY_TIME_VISITORS1;Čas vyloučení H hráč 1;AD(1,43)+AB(1,43,7,:)+AD(1,44)+AD(1,45); PENALTY_TIME_VISITORS2;Čas vyloučení H hráč 2;AD(1,48)+AB(1,48,7,:)+AD(1,49)+AD(1,50); PENALTY_PLAYERID_HOME1;Číslo vyloučení D hráč 1;AD(1,20)+AD(1,21); PENALTY_PLAYERID_HOME2;Číslo vyloučení D hráč 2;AD(1,25)+AD(1,26); PENALTY_PLAYERID_VISITORS1;Číslo vyloučení H hráč 1;AD(1,41)+AD(1,42); PENALTY_PLAYERID_VISITORS2;Číslo vyloučení H hráč 2;AD(1,46)+AD(1,47);""" 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, 900, 700) # 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() parsed_tab = self.create_parsed_data_tab() parsing_rules_tab = self.create_parsing_rules_tab() # Přidání záložek tab_widget.addTab(settings_tab, "Nastavení") tab_widget.addTab(data_tab, "Data") tab_widget.addTab(parsed_tab, "Parsovaná data") tab_widget.addTab(parsing_rules_tab, "Pravidla parsování") 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 create_parsed_data_tab(self): """Vytvoří záložku pro zobrazení parsovaných dat""" parsed_widget = QWidget() layout = QVBoxLayout() # Tabulka pro parsovaná data self.parsed_table = QTableWidget(0, 3) self.parsed_table.setHorizontalHeaderLabels(["Klíč", "Popis", "Hodnota"]) self.parsed_table.horizontalHeader().setStretchLastSection(True) layout.addWidget(self.parsed_table) # Tlačítko pro manuální parsování parse_button = QPushButton("Parsovat poslední přijatá data") parse_button.clicked.connect(self.parse_last_data) layout.addWidget(parse_button) # Checkbox pro automatické parsování self.auto_parse_checkbox = QCheckBox("Automaticky parsovat přijatá data") self.auto_parse_checkbox.setChecked(True) layout.addWidget(self.auto_parse_checkbox) parsed_widget.setLayout(layout) return parsed_widget def create_parsing_rules_tab(self): """Vytvoří záložku pro pravidla parsování""" rules_widget = QWidget() layout = QVBoxLayout() # Editor pravidel rules_label = QLabel("Pravidla parsování (formát: KLÍČ;POPIS;VZOREC)") layout.addWidget(rules_label) self.rules_editor = QTextEdit() self.rules_editor.setFont(QFont("Courier New", 10)) self.rules_editor.setPlainText(self.parse_rules) layout.addWidget(self.rules_editor) # Tlačítko pro uložení pravidel save_rules_button = QPushButton("Uložit pravidla") save_rules_button.clicked.connect(self.save_parsing_rules) layout.addWidget(save_rules_button) # Vzorová data pro testování example_label = QLabel("Příklad pravidla: GAME_TIMER;Hrací čas;AD(0,5)+AD(0,6)+AB(0,7,7,:)+AD(0,7)+AD(0,8)") layout.addWidget(example_label) rules_widget.setLayout(layout) return rules_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 save_parsing_rules(self): """Uloží pravidla parsování""" self.parse_rules = self.rules_editor.toPlainText() QMessageBox.information(self, "Uloženo", "Pravidla parsování byla uložena.") def parse_last_data(self): """Parsuje poslední přijatá RAW data""" if not self.last_raw_data: QMessageBox.warning(self, "Žádná data", "Nejsou k dispozici žádná data k parsování.") return try: parsed_data = parse_game_data(self.parse_rules, self.last_raw_data) self.display_parsed_data(parsed_data) except Exception as e: QMessageBox.warning(self, "Chyba parsování", f"Nastala chyba při parsování dat: {str(e)}") def display_parsed_data(self, parsed_data): """Zobrazí parsovaná data v tabulce""" # Získáme popisy z pravidel parsování descriptions = {} for line in self.parse_rules.strip().split("\n"): parts = line.split(";") if len(parts) >= 2: descriptions[parts[0]] = parts[1] # Vyčistíme tabulku self.parsed_table.setRowCount(0) # Přidáme data do tabulky for i, (key, value) in enumerate(parsed_data.items()): self.parsed_table.insertRow(i) self.parsed_table.setItem(i, 0, QTableWidgetItem(key)) self.parsed_table.setItem(i, 1, QTableWidgetItem(descriptions.get(key, ""))) self.parsed_table.setItem(i, 2, QTableWidgetItem(value)) # Přizpůsobíme šířku sloupců self.parsed_table.resizeColumnsToContents() 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 # Zpracování a zobrazení dat if 'raw' in data: # Uložíme poslední RAW data pro parsování self.last_raw_data = data['raw'] # Automatické parsování, pokud je povoleno if self.auto_parse_checkbox.isChecked(): try: parsed_data = parse_game_data(self.parse_rules, self.last_raw_data) self.display_parsed_data(parsed_data) except Exception as e: print(f"Chyba při automatickém parsování: {str(e)}") # 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']}\n\n" # Přidáme sekci s parsovanými daty try: parsed = parse_game_data(self.parse_rules, data['raw']) display_text += "PARSOVANÁ DATA:\n" for key, value in parsed.items(): display_text += f"{key}: {value}\n" except Exception as e: display_text += f"Chyba parsování: {str(e)}\n" # 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_())