Window.py

# Window.py -- Pamhyr
# Copyright (C) 2023  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 <https://www.gnu.org/licenses/>.

# -*- coding: utf-8 -*-

import os
import logging
import tempfile

from queue import Queue
from tools import trace, timer

from View.ASubWindow import ASubWindow, ASubMainWindow
from View.ListedSubWindow import ListedSubWindow

from PyQt5.QtGui import (
    QKeySequence,
)

from PyQt5.QtCore import (
    Qt, QVariant, QAbstractTableModel,
    QCoreApplication, QModelIndex, pyqtSlot,
    QRect, QTimer, QProcess,
)

from PyQt5.QtWidgets import (
    QDialogButtonBox, QPushButton, QLineEdit,
    QFileDialog, QTableView, QAbstractItemView,
    QUndoStack, QShortcut, QAction, QItemDelegate,
    QComboBox, QVBoxLayout, QHeaderView, QTabWidget,
    QTextEdit,
)

from View.RunSolver.Log.Window import SolverLogFileWindow

try:
    from signal import SIGTERM, SIGSTOP, SIGCONT
    _signal = True
except:
    _signal = False

_translate = QCoreApplication.translate

logger = logging.getLogger()

class SelectSolverWindow(ASubWindow, ListedSubWindow):
    def __init__(self, title="Select solver",
                 study=None, config=None,
                 parent=None):
        self._title = title

        self._study = study
        self._config = config
        self._solver = None

        super(SelectSolverWindow, self).__init__(
            name=self._title, ui="SelectSolver", parent=parent
        )
        self.ui.setWindowTitle(self._title)

        self.setup_combobox()
        self.setup_connections()

    def setup_combobox(self):
        solvers = self._config.solvers
        solvers_name = list(map(lambda s: s.name + f" - ({s._type})", solvers))

        self.combobox_add_items("comboBox", solvers_name)

    def setup_connections(self):
        self.find(QPushButton, "pushButton_run").clicked.connect(self.accept)
        self.find(QPushButton, "pushButton_cancel").clicked.connect(self.reject)

    @property
    def solver(self):
        return self._solver

    def accept(self):
        solver_name = self.get_combobox_text("comboBox")
        solver_name = solver_name.rsplit(" - ", 1)[0]

        self._solver = next(
            filter(
                lambda s: s.name == solver_name,
                self._config.solvers
            )
        )

        super(SelectSolverWindow, self).accept()


class SolverLogWindow(ASubMainWindow, ListedSubWindow):
    def __init__(self, title="Solver logs",
                 study=None, config=None,
                 solver=None, parent=None):
        self._title = title
        self._parent = parent

        self._study = study
        self._config = config
        self._solver = solver

        self._results = None

        super(SolverLogWindow, self).__init__(
            name=self._title, ui="SolverLog", parent=parent
        )
        self.ui.setWindowTitle(self._title)

        self.setup_action()
        self.setup_alarm()
        self.setup_connections()

        self._workdir = ""
        if self._study.filename == "":
            self._workdir = tempfile.TemporaryDirectory()
        else:
            self._workdir = os.path.join(
                os.path.dirname(self._study.filename),
                "_PAMHYR_",
                self._study.name.replace(" ", "_"),
                self._solver.name.replace(" ", "_"),
            )
            os.makedirs(self._workdir, exist_ok = True)

        self._alarm.start(500)
        self._output = Queue()
        self._process = self.new_process(parent)

        self._log(f" *** Export study {self._solver.name}", color="blue")
        self._solver.export(self._study, self._workdir, qlog = self._output)

        self.update()

        self._log(f" *** Run solver {self._solver.name}", color="blue")
        self._solver.run(
            process = self._process,
            output_queue = self._output
        )

    def new_process(self, parent):
        new = QProcess(parent)
        new.setWorkingDirectory(self._workdir)
        return new

    def setup_action(self):
        self.find(QAction, "action_start").setEnabled(False)
        if _signal:
            self.find(QAction, "action_pause").setEnabled(True)
        else:
            self.find(QAction, "action_pause").setEnabled(False)

        self.find(QAction, "action_stop").setEnabled(True)
        self.find(QAction, "action_log_file").setEnabled(False)

    def setup_alarm(self):
        self._alarm = QTimer()

    def setup_connections(self):
        self.find(QAction, "action_start").triggered.connect(self.start)
        self.find(QAction, "action_pause").triggered.connect(self.pause)
        self.find(QAction, "action_stop").triggered.connect(self.stop)
        self.find(QAction, "action_log_file").triggered.connect(self.log_file)
        self.find(QAction, "action_results").triggered.connect(self.results)

        self._alarm.timeout.connect(self.update)

    def _log(self, msg, color=None):
        if type(msg) == str:
            msg = msg.rsplit('\n')[0]

            if color is not None:
                msg = f"<font color=\"{color}\">" + msg + "</font>"

            self.find(QTextEdit, "textEdit").append(msg)

        elif type(msg) == int:
            color = "blue" if msg == 0 else "red"
            self.find(QTextEdit, "textEdit")\
                .append(f"<font color=\"{color}\">" +
                        f" *** Finished with code {msg}" +
                        "</font>")

    def update(self):
        if self._solver.is_stoped():
            self.find(QAction, "action_start").setEnabled(True)
            self.find(QAction, "action_pause").setEnabled(False)
            self.find(QAction, "action_stop").setEnabled(False)
            self.find(QAction, "action_results").setEnabled(True)
            if self._solver.log_file() != "":
                self.find(QAction, "action_log_file").setEnabled(True)

        while self._output.qsize() != 0:
            s = self._output.get()
            if type(s) is str and "[ERROR]" in s:
                self._log(s, color="red")
            else:
                self._log(s)

    def start(self):
        if self._solver.is_stoped():
            self._log(f" *** Export study {self._solver.name}", color="blue")
            self._solver.export(self._study, self._workdir, qlog = self._output)
            self.update()

            self._process = self.new_process(self._parent)

        self._log(" *** Start", color="blue")
        self._results = None
        self._solver.start(self._process)

        self.find(QAction, "action_start").setEnabled(False)
        if _signal:
            self.find(QAction, "action_pause").setEnabled(True)
        else:
            self.find(QAction, "action_pause").setEnabled(False)
        self.find(QAction, "action_stop").setEnabled(True)
        self.find(QAction, "action_log_file").setEnabled(False)
        self.find(QAction, "action_results").setEnabled(False)

    def pause(self):
        self._log(" *** Pause", color="blue")
        self._solver.pause()

        self.find(QAction, "action_start").setEnabled(True)
        self.find(QAction, "action_pause").setEnabled(False)
        self.find(QAction, "action_stop").setEnabled(True)
        self.find(QAction, "action_results").setEnabled(False)

    def stop(self):
        self._log(" *** Stop", color="blue")
        self._solver.kill()

        self.find(QAction, "action_start").setEnabled(True)
        self.find(QAction, "action_pause").setEnabled(False)
        self.find(QAction, "action_stop").setEnabled(False)
        self.find(QAction, "action_results").setEnabled(True)
        if self._solver.log_file() != "":
            self.find(QAction, "action_log_file").setEnabled(True)

    def results(self):
        if self._results is None:
            self._results = self._solver.results(self._study, self._workdir, qlog = self._output)
        self._parent.open_solver_results(self._solver, self._results)

    def log_file(self):
        file_name = os.path.join(self._workdir, self._solver.log_file())
        log = SolverLogFileWindow(
            file_name = file_name,
            study = self._study,
            config = self._config,
            solver = self._solver,
            parent = self,
        )
        log.show()