import errno import threading import time import pandas as pd import wx from gui.config import * from gui.design_autogenerated import MeasuringDialog from modules.parsers.nbus_stream_parser import DataParser from nbus_api.nbus_module_slave import NBusSlaveModule from nbus_hal.nbus_serial.serial_config import NBusSerialConfig, NBusBaudrate, NBusParity from nbus_hal.nbus_serial.serial_port import NBusSerialPort from nbus_types.nbus_exceptions.nbus_network_exception import NBusErrorNetwork, NBusErrorNetworkType from serial import SerialException class MeasuringDialogFinal(MeasuringDialog): def __init__(self, parent): super().__init__(parent) self._df = pd.DataFrame() self._in_acquisition = False self._series_num = 0 self._nbus_port = NBusSerialPort(NBusSerialConfig(**config)) self._nbus_module = NBusSlaveModule(self._nbus_port, MODULE_ADDR, BRIDGE_ACQUIRE_WAIT) self._update_measurement_info(MeasuringPhase.IDLE.value) # start nbus logic after initialization wx.CallAfter(self._start_nbus) def GetDataFrame(self): return self._df def MainWindowOnClose(self, event): self._cleanup_and_close(save_data=False) def m_button_saveOnButtonClick(self, event): self._cleanup_and_close(save_data=True) def m_button_cancelOnButtonClick(self, event): self._cleanup_and_close(save_data=False) def _start_nbus(self): try: # download configuration from module self._nbus_module.init() # start data stream in listening mode self._nbus_module.start_streaming(send_start=False) # init loop variables last_phase = 0 self._in_acquisition = True while self._in_acquisition: # 1. Spracujeme kliknutia na tlačidlá a prekreslenie UI wx.GetApp().Yield(onlyIfNeeded=True) if not self._in_acquisition: break # 2. Krátke čakanie time.sleep(TASK_NBUS_DELAY) # 3. Vyčítanie dát stream_chunk = self._nbus_module.fetch_stream_chunk() if not stream_chunk.empty: current_phase = int(stream_chunk[PHASE_SID].iloc[-1]) if current_phase != last_phase: self._update_measurement_info(current_phase) last_phase = current_phase except NBusErrorNetwork as ex: if ex.args[0] in (NBusErrorNetworkType.DAMAGED_MESSAGE, NBusErrorNetworkType.MESSAGE_NOT_COMPLETE): self._show_error( "Ručný snímač nie je pripravený na meranie\n\n" "Prepnite ovládač do uzamknutého režimu a skúste to znova." ) elif ex.args[0] == NBusErrorNetworkType.EMPTY_MESSAGE: self._show_error( "Ručný snímač nie je pripojený\n\n" "Skontrolujte či je zariadenie zapnuté a má nabitú batériu." ) self._cleanup_and_close(save_data=False) except SerialException as ex: if ex.errno == errno.ENOENT or "FileNotFoundError" in str(ex): self._show_error( "Komunikačný port nebol nájdený\n\n" "Skontrolujte či je USB prijímač pripojený k PC a je zvolený správny komunikačný port." ) elif ex.errno in (errno.EACCES, errno.EBUSY) or "PermissionError" in str(ex): self._show_error( "Komunikačný port je obsadený\n\n" "Zatvorte ostatné programy, ktoré port môžu používať, a skúste to znova." ) else: self._show_error( "Chyba komunikácie so zariadením\n\n" "Skontrolujte pripojenie USB kábla a skúste akciu zopakovať." ) self._cleanup_and_close(save_data=False) except Exception as ex: self._show_error( f"Nastala neočakávaná chyba pri meraní\n\n" f"Skontrolujte pripojenie zariadenia a skúste akciu zopakovať.\n" f"Detail chyby: {ex}" ) self._cleanup_and_close(save_data=False) def _parse_data(self): if self._df is None or self._df.empty: self._show_error("nnnn") parser = DataParser(self._df) parser.remap_columns(NBUS_DATA_COLUMN_MAPPING) parser.remap_nans(NBUS_DATA_NAN_MAPPING) parser.validate_and_filter_series_by_zeros(CN_SER, CN_PHA, len(MeasuringPhase) - 1) parser.remove_specific_phases(CN_PHA, NBUS_DATA_PHASES_TO_REMOVE) parser.remap_phases(CN_PHA, NBUS_DATA_PHASE_MAPPING) parser.drop_rows_completely_null(NBUS_DATA_EMPTY_MAPPING) parser.clean_precision_noise(PRECISION) print("sasadassa", self._df) if self._df.empty: if self._df is None or self._df.empty: self._show_error("nnnn") self._df.to_csv("dataX.csv", index=False) def _update_measurement_info(self, new_phase): if new_phase not in MeasuringPhase: return if new_phase == MeasuringPhase.ACK.value: self._series_num += 1 if new_phase == MeasuringPhase.IDLE.value: self.m_staticText_seriesVal.SetLabel(str(self._series_num)) self.m_staticText_phaseVal.SetLabel(PHASES[new_phase]) self.m_gauge_phase.SetValue(new_phase) def _show_error(self, message): wx.MessageBox(message, "Chyba", wx.OK | wx.ICON_ERROR) def _cleanup_and_close(self, save_data): # 1. Zastavíme slučku while self._in_acquisition = False # Zablokujeme okno, aby používateľ nemohol kliknúť znova self.Disable() def worker(): try: # Pri force quit (save_data=False) sa neoplatí čakať na odpoveď zariadenia if save_data: self._nbus_module.stop_streaming(send_stop=False) time.sleep(0.2) self._df = self._nbus_module.fetch_full_stream() self._parse_data() else: self._df = None except Exception: self._df = None finally: try: # Ak port podporuje cancel_read / cancel_write, odblokuje to zamrznuté čítanie if hasattr(self._nbus_port, "cancel_read"): self._nbus_port.cancel_read() self._nbus_port.close() except Exception: pass # Bezpečné zatvorenie GUI okna z hlavného vlákna wx.CallAfter(self._finish_close) # Spustíme ako daemon vlákno, aby neblokovalo ukončenie programu threading.Thread(target=worker, daemon=True).start() def _finish_close(self): # Použi Destroy() namiesto Close(), aby sa znova nespustil EVT_CLOSE a nevznikla nekonečná slučka if self: self.Destroy()