Pamhyr2/src/View/MainWindowTabChecker.py

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

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

import logging

from queue import Queue
from tools import timer, trace, flatten

from PyQt5.QtCore import (
    QThread,
)

from PyQt5.QtWidgets import (
    QTableView, QListView,
)

from Modules import Modules
from Model.Study import Study
from Solver.Solvers import solver_type_list, solver_long_name

from View.Tools.PamhyrWidget import PamhyrWidget
from View.CheckList.Table import TabTableModel
from View.CheckList.List import TabListModel
from View.CheckList.Worker import TabWorker

logger = logging.getLogger()


class WidgetChecker(PamhyrWidget):
    _pamhyr_ui = "MainWindowTabCheckers"

    def __init__(self, study=None, config=None, parent=None):
        self._study = study
        self._config = config

        super(WidgetChecker, self).__init__(
            parent=parent
        )

        self.setup_worker()
        self.setup_solver_list()
        self.setup_checker_list()
        self.setup_table()
        self.setup_list()
        self.setup_connections()

    def setup_worker(self):
        self._worker_q = Queue()

        self._worker = TabWorker(self._study, self._worker_q)
        self._worker_thread = QThread()
        self._worker.moveToThread(self._worker_thread)

        self._worker.signalStatus.connect(self.update_thread)
        self._worker_thread.started.connect(self._worker.process)

        self._worker_thread.start()

    def setup_solver_list(self):
        self._solvers = list(
            map(
                lambda solver: solver_type_list[solver],
                solver_type_list
            )
        )

    def setup_checker_list(self):
        self._checkers = flatten(
            map(
                lambda solver: solver.checkers(),
                [Study] + self._solvers
            )
        )

    def setup_table(self):
        header = {
            'type': "Type", "study": "Study",
            **solver_long_name
        }

        table = self.find(QTableView, f"tableView_checker")
        self._table = TabTableModel(
            table_view=table,
            table_headers=header,
            data=self._checkers,
            opt_data=Modules.modelling_list(),
            options=[],
        )

    def setup_list(self):
        lst = self.find(QListView, f"listView_current")
        self._list = TabListModel(
            list_view=lst,
            data=[],
        )

        lst = self.find(QListView, f"listView_errors")
        self._list_error = TabListModel(
            list_view=lst,
            data=[],
        )

    def setup_connections(self):
        table = self.find(QTableView, f"tableView_checker")
        table.selectionModel()\
             .selectionChanged\
             .connect(self.update_selection)

    @property
    def study(self):
        return self._study

    @study.setter
    def study(self, study):
        self._study = study

    def _checkers_filtered(self, modules):
        impacted = modules | modules.impact()
        return filter(
            lambda c: c._modules in impacted,
            self._checkers
        )

    def update_selection(self):
        table = self.find(QTableView, f"tableView_checker")
        indexes = table.selectedIndexes()

        checkers = self._table.get_checkers_from_indexes(indexes)
        self._list._data = checkers
        self._list.update()

    def update_thread(self, key):
        if key == "progress":
            self._table.update()
            self.update_errors()

    def update_errors(self):
        checkers = list(
            filter(
                lambda c: c.is_error() or c.is_warning(),
                self._checkers
            )
        )

        self._list_error._data = checkers
        self._list_error.update()

    def update(self, modules=Modules.NONE):
        if modules is Modules.NONE:
            return

        if Modules.STUDY in modules:
            self._worker.study = self._study

            for checker in self._checkers:
                if self._study is None:
                    checker.reset()

                self._worker_q.put(checker)

            self._table.update()
            return

        for checker in self._checkers_filtered(modules):
            checker.reset()
            self._worker_q.put(checker)