# ReadingResultsDialog.py -- Pamhyr # Copyright (C) 2023-2025 INRAE # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # -*- coding: utf-8 -*- import logging from tools import logger_exception from View.Tools.PamhyrWindow import PamhyrDialog from View.Results.translate import ResultsTranslate from PyQt5.QtGui import ( QKeySequence, ) from PyQt5.QtCore import ( Qt, QVariant, QAbstractTableModel, QTimer, QObject, pyqtSlot, pyqtSignal, QThread, ) from PyQt5.QtWidgets import ( QLabel, ) logger = logging.getLogger() class Worker(QObject): signalStatus = pyqtSignal(str) def __init__(self, reading_fn, parent=None): super(self.__class__, self).__init__(parent) self._reading_fn = reading_fn def process(self): self._reading_fn() self.signalStatus.emit('end') class ReadingResultsDialog(PamhyrDialog): _pamhyr_ui = "WaitingDialog" _pamhyr_name = "Reading results" _spin = ["-", "\\", "|", "/"] def __init__(self, reading_fn, parent=None): trad = ResultsTranslate() super(ReadingResultsDialog, self).__init__( title=trad[self._pamhyr_name], trad=trad, options=[], parent=parent ) self._reading_fn = reading_fn self._init_default_values() self.setup_timer() self.setup_worker() self._timer.start(150) self._worker_thread.start() def _init_default_values(self): self._spinner_step = 0 self.set_label_text("label_spinner", self._spin[0]) def setup_worker(self): self._worker = Worker(self._reading_fn) self._worker_thread = QThread(parent=self) self._worker.moveToThread(self._worker_thread) self._worker.signalStatus.connect(self.close) self._worker_thread.started.connect(self._worker.process) def setup_timer(self): self._timer = QTimer(self) self._timer.timeout.connect(self.update_spinner) def update_spinner(self): self._spinner_step += 1 self._spinner_step %= len(self._spin) self.set_label_text( "label_spinner", self._spin[self._spinner_step] ) def end_worker(self): self._worker_thread.terminate() self._worker_thread.wait() def close(self): try: self._timer.stop() self.end_worker() except Exception as e: logger_exception(e) super().close()