# RubarBE.py -- Pamhyr RubarBE checkers # Copyright (C) 2026 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 -*- """RubarBE checkers. Each one runs a step of the checks of the RubarBE export (Solver/RubarBE.py), without writing any file, so that they always give the messages of the export: an error where the export would be cancelled, a warning for the data it ignores (empty boundary conditions, lateral contributions not read, solid lateral contributions without liquid one, structures without RubarBE equivalent...).""" import logging from PyQt5.QtCore import QCoreApplication from Modules import Modules from tools import logger_exception from Checker.Checker import AbstractModelChecker, STATUS _translate = QCoreApplication.translate logger = logging.getLogger() class RubarBEChecker(AbstractModelChecker): """Base class: _check(solver, study, context, warnings) runs the export step, raising RubarExportError (error) or appending to warnings (warning). Steps that need the reach are not run when the study has not a single valid reach: RubarBEReachChecker reports it""" _needs_reach = True def __init__(self, name, description, modules, version="rubarbe"): super(RubarBEChecker, self).__init__() self._name = name self._description = description self._solver = version self._modules = modules @staticmethod def _context(solver, study): # Read the parameters without creating them (get_params would # add default ones to the study) river = study.river if solver.type in river._parameters: values = solver._parameter_values(study) else: values = {k: str(v).strip() for k, v in solver.default_parameters()} return { "values": values, "trased": values.get("rubarbe_trased", "n").lower() in ("y", "o"), } def _check(self, solver, study, context, warnings): raise NotImplementedError def run(self, study): from Solver.RubarBE import RubarBE, RubarExportError if not self.basic_check(study): return False solver = RubarBE("checker") warnings = [] try: context = self._context(solver, study) if self._needs_reach: try: edge, profiles = solver._single_reach(study.river) except RubarExportError: self._status = STATUS.OK self._summary = "ok" return True context["hydraulics"] = { "edge": edge, "profiles": profiles, "tinit": solver._seconds( context["values"].get("rubarbe_tinit", "0")), "warnings": warnings, } self._check(solver, study, context, warnings) except (RubarExportError, ValueError) as error: self._status = STATUS.ERROR self._summary = str(error) return False except Exception as error: # Never stop the checker thread logger_exception(error) self._status = STATUS.UNKNOWN self._summary = str(error) return False if warnings: self._status = STATUS.WARNING self._summary = " | ".join(warnings) return False self._status = STATUS.OK self._summary = "ok" return True class RubarBEReachChecker(RubarBEChecker): _needs_reach = False def __init__(self, version="rubarbe"): super(RubarBEReachChecker, self).__init__( _translate("Checker", "RubarBE reach checker"), _translate( "Checker", "Check that the study has a single enabled reach, with at " "least two profiles of increasing rk" ), Modules.GEOMETRY, version, ) def _check(self, solver, study, context, warnings): solver._single_reach(study.river) class RubarBEGeometryChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBEGeometryChecker, self).__init__( _translate("Checker", "RubarBE geometry and sediment checker"), _translate( "Checker", "Check the main channel and active bed limits of the " "profiles and, with sediment transport, their sediment " "layers" ), # Rerun on changes of the geometry, which impact the sediment # layers Modules.SEDIMENT_LAYER, version, ) def _check(self, solver, study, context, warnings): solver._geometry_data(context["hydraulics"]["edge"], context["trased"]) class RubarBEBoundaryChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBEBoundaryChecker, self).__init__( _translate("Checker", "RubarBE boundary conditions checker"), _translate( "Checker", "Check the liquid and solid boundary conditions used by " "the boundary settings of the solver parameters" ), Modules.BOUNDARY_CONDITION, version, ) def _check(self, solver, study, context, warnings): values = context["values"] edge = context["hydraulics"]["edge"] solver._check_regime(values) solver._check_boundaries(study.river, edge, values, warnings) solver._solid_boundary(study.river, edge, values, context["trased"], warnings) class RubarBEFrictionChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBEFrictionChecker, self).__init__( _translate("Checker", "RubarBE friction checker"), _translate("Checker", "Check that the reach has frictions"), Modules.FRICTION, version, ) def _check(self, solver, study, context, warnings): solver._interpolated_frictions(context["hydraulics"]["edge"]) class RubarBEInitialConditionsChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBEInitialConditionsChecker, self).__init__( _translate("Checker", "RubarBE initial conditions checker"), _translate( "Checker", "Check the initial conditions of the reach"), Modules.INITIAL_CONDITION, version, ) def _check(self, solver, study, context, warnings): solver._initial_conditions(study.river, context["hydraulics"]["edge"]) class RubarBELateralChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBELateralChecker, self).__init__( _translate("Checker", "RubarBE lateral contributions checker"), _translate( "Checker", "Check the lateral contributions: read only with lateral " "overflows, solid ones attached to the liquid contribution " "of the same sections" ), Modules.LATERAL_CONTRIBUTION, version, ) def _check(self, solver, study, context, warnings): solver._lateral_data(study.river, context["hydraulics"], context["values"], context["trased"], warnings) class RubarBEStructureChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBEStructureChecker, self).__init__( _translate("Checker", "RubarBE hydraulic structures checker"), _translate( "Checker", "Check the hydraulic structures: weirs and rectangular and " "circular orifices only" ), Modules.HYDRAULIC_STRUCTURES, version, ) def _check(self, solver, study, context, warnings): solver._structure_data(study.river, context["hydraulics"], warnings) class RubarBESavesChecker(RubarBEChecker): def __init__(self, version="rubarbe"): super(RubarBESavesChecker, self).__init__( _translate("Checker", "RubarBE saves checker"), _translate( "Checker", "Check the save abscissas (ABSHYD) and times (TNPROF) of " "the solver parameters" ), # The solver parameters have no module: rerun on any change Modules.NONE, version, ) def _check(self, solver, study, context, warnings): solver._save_data(context["values"], context["hydraulics"])