From 4d9c4cb5d0f0ab24543919a7974ccccbfd03af6f Mon Sep 17 00:00:00 2001 From: Dylan Jeannin Date: Thu, 22 Oct 2026 17:53:29 +0200 Subject: [PATCH] RubarBE: add checkers running the export checks, with its errors and warnings (boundary conditions, lateral contributions, structures, saves...) --- src/Checker/RubarBE.py | 255 +++++++++++++++++++++++++++++++++++++++++ src/Solver/RubarBE.py | 54 ++++++--- 2 files changed, 291 insertions(+), 18 deletions(-) create mode 100644 src/Checker/RubarBE.py diff --git a/src/Checker/RubarBE.py b/src/Checker/RubarBE.py new file mode 100644 index 00000000..0f0e39da --- /dev/null +++ b/src/Checker/RubarBE.py @@ -0,0 +1,255 @@ +# 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"]) diff --git a/src/Solver/RubarBE.py b/src/Solver/RubarBE.py index d7a3f0f3..b69381e5 100644 --- a/src/Solver/RubarBE.py +++ b/src/Solver/RubarBE.py @@ -123,10 +123,21 @@ class Rubar3(CommandLineSolver): @classmethod def checkers(cls): - lst = [ - ] + from Checker.RubarBE import ( + RubarBEReachChecker, RubarBEGeometryChecker, + RubarBEBoundaryChecker, RubarBEFrictionChecker, + RubarBEInitialConditionsChecker, RubarBELateralChecker, + RubarBEStructureChecker, RubarBESavesChecker, + ) - return lst + return [ + checker(version=cls._type) for checker in ( + RubarBEReachChecker, RubarBEGeometryChecker, + RubarBEBoundaryChecker, RubarBEFrictionChecker, + RubarBEInitialConditionsChecker, RubarBELateralChecker, + RubarBEStructureChecker, RubarBESavesChecker, + ) + ] ########## # Export # @@ -154,19 +165,11 @@ class Rubar3(CommandLineSolver): name = study.name.replace(" ", "_") values = self._parameter_values(study) - if (values.get("rubarbe_regime") == "1" and - values.get("rubarbe_condam") == "0"): - trad = RunSolverTranslate() - message = trad["incompatible_upstream_boundary"] - logger.warning(message) - if qlog is not None: - qlog.put(trad["warning"].format(message=message)) - return False - trased = self._parameter_flag(study, "trased") == "O" # Check the hydraulic data before writing any file try: + self._check_regime(values) hydraulics = self._hydraulic_data(study, values) geometry = self._geometry_data(hydraulics["edge"], trased) solid = self._solid_boundary( @@ -232,13 +235,20 @@ class Rubar3(CommandLineSolver): # NCLMAX (variables.f90): points of all the solid hydrographs _max_solid_points = 1000 - def _hydraulic_data(self, study, values): - """Check the single reach data needed by DONNEE, MAIL, CONDIN, - FROT, HYDRO and CONDAV, and return them. Raise - RubarExportError with a user message""" - trad = RunSolverTranslate() - river = study.river + @staticmethod + def _check_regime(values): + """REGIME=1 imposes the water depth upstream, which a reflecting + upstream boundary (CONDAM=0) cannot do""" + if (values.get("rubarbe_regime") == "1" and + values.get("rubarbe_condam") == "0"): + raise RubarExportError( + RunSolverTranslate()["incompatible_upstream_boundary"]) + @staticmethod + def _single_reach(river): + """The single enabled reach and its profiles, with at least two + profiles of increasing rk. Raise RubarExportError""" + trad = RunSolverTranslate() edges = river.enable_edges() if len(edges) != 1: raise RubarExportError(trad["multireach_unsupported"].format( @@ -252,6 +262,14 @@ class Rubar3(CommandLineSolver): if any(b <= a for a, b in zip(rks, rks[1:])): raise RubarExportError(trad["rk_not_increasing"].format( reach=edge.name)) + return edge, profiles + + def _hydraulic_data(self, study, values): + """Check the single reach data needed by DONNEE, MAIL, CONDIN, + FROT, HYDRO and CONDAV, and return them. Raise + RubarExportError with a user message""" + river = study.river + edge, profiles = self._single_reach(river) warnings = [] boundaries = self._check_boundaries(river, edge, values, warnings)