From 5d46401b850a73353653d960c68d08b10d8bd9ea Mon Sep 17 00:00:00 2001 From: Pierre-Antoine Rouby Date: Thu, 24 Sep 2026 10:30:16 +0200 Subject: [PATCH] Results: Refactoring '_export_rk' methode. --- src/Model/Geometry/ProfileXYZ.py | 7 + src/View/Results/Window.py | 395 ++++++++++++------------------- 2 files changed, 155 insertions(+), 247 deletions(-) diff --git a/src/Model/Geometry/ProfileXYZ.py b/src/Model/Geometry/ProfileXYZ.py index 905b2701..aac30727 100644 --- a/src/Model/Geometry/ProfileXYZ.py +++ b/src/Model/Geometry/ProfileXYZ.py @@ -826,6 +826,13 @@ class ProfileXYZ(Profile, SQLSubModel): return n_zones + def froude(self, v, z): + return v / np.sqrt( + 9.81 * ( + self.wet_area(z) / self.wet_width(z) + ) + ) + @timer def get_all_water_limits_ac(self, z): """ diff --git a/src/View/Results/Window.py b/src/View/Results/Window.py index 6937d3b9..26f57d88 100644 --- a/src/View/Results/Window.py +++ b/src/View/Results/Window.py @@ -876,266 +876,167 @@ class ResultsWindow(PamhyrWindow): dict_x = self._trad.get_dict("values_x") dict_y = self._trad.get_dict("values_y") - if self._current_results == 2: - envelop = False - reach1 = results[0].river.reachs[self._get_current_reach()] - reach2 = results[1].river.reachs[self._get_current_reach()] - if envelop: dict_y.update(self._trad.get_dict("values_y_envelop")) - my_dict = {} - my_dict[dict_x["rk"]] = reach.geometry.get_rk() - table = results.get("table") id_ts = results.get_timestamp_id(timestamp) - z = table["Z"] - q = table["Q"] - v = table["V"] - zfd = table["zfd"] + ind = reach.global_index - if "bed_elevation" in y: - if reach.has_bedload(): - zmin = list(map( - lambda p: zfd[id_ts, p.global_index], - reach.profiles - )) - else: - zmin = reach.geometry.get_z_min() - my_dict[dict_y["bed_elevation"]] = zmin - # if envelop and reach.has_sediment(): - if "discharge" in y: - my_dict[dict_y["discharge"]] = list( - map( - lambda p: q[id_ts, p.global_index], - reach.profiles - ) + my_dict = {} + data = { + "z": table["Z"][:, ind[0]:ind[1]+1], + "q": table["Q"][:, ind[0]:ind[1]+1], + "v": table["V"][:, ind[0]:ind[1]+1], + "zfd": table["zfd"][:, ind[0]:ind[1]+1], + + "reach": reach, + "id_ts": id_ts, "envelop": envelop, + "my_dict": my_dict, + } + + func = { + "bed_elevation": self._export_rk_compute_elevation, + "discharge": self._export_rk_compute_discharge, + "water_elevation": self._export_rk_compute_water_elevation, + "velocity": self._export_rk_compute_velocity, + "depth": self._export_rk_compute_depth, + "mean_depth": self._export_rk_compute_mean_depth, + "froude": self._export_rk_compute_froude, + "wet_area": self._export_rk_compute_wet_area, + } + + # Compute values + for key in y: + func[key](data) + + # Translate dictionary keys + translated_dict = {} + for key in my_dict: + translated_dict[dict_y[key]] = my_dict[key] + + translated_dict[dict_x["rk"]] = reach.geometry.get_rk() + + return translated_dict + + def _export_rk_compute_elevation(self, data): + my_dict = data["my_dict"] + reach = data["reach"] + id_ts = data["id_ts"] + + if reach.has_bedload(): + zfd = data["zfd"] + zmin = zfd[id_ts, :] + else: + zmin = reach.geometry.get_z_min() + + my_dict["bed_elevation"] = zmin + + def _export_rk_compute_discharge(self, data): + my_dict = data["my_dict"] + envelop = data["envelop"] + id_ts = data["id_ts"] + q = data["q"] + + my_dict["discharge"] = q[id_ts, :] + + if envelop: + my_dict["min_discharge"] = np.min(q, axis=0) + my_dict["max_discharge"] = np.max(q, axis=0) + + def _export_rk_compute_water_elevation(self, data): + my_dict = data["my_dict"] + envelop = data["envelop"] + id_ts = data["id_ts"] + z = data["z"] + + my_dict["water_elevation"] = z[id_ts, :] + + if envelop: + my_dict["min_water_elevation"] = np.min(z, axis=0) + my_dict["max_water_elevation"] = np.max(z, axis=0) + + def _export_rk_compute_velocity(self, data): + my_dict = data["my_dict"] + envelop = data["envelop"] + id_ts = data["id_ts"] + v = data["v"] + + my_dict["velocity"] = v[id_ts, :] + + if envelop: + my_dict["min_velocity"] = np.min(v, axis=0) + my_dict["max_velocity"] = np.max(v, axis=0) + + def _export_rk_compute_depth(self, data): + my_dict = data["my_dict"] + envelop = data["envelop"] + reach = data["reach"] + id_ts = data["id_ts"] + z = data["z"] + + my_dict["depth"] = list( + map( + lambda p: p.geometry.max_water_depth( + z[id_ts, p.global_index] + ), + reach.profiles ) - if envelop: - my_dict[dict_y["min_discharge"]] = list( - map( - lambda p: np.min(q[:, p.global_index]), - reach.profiles - ) - ) - my_dict[dict_y["max_discharge"]] = list( - map( - lambda p: np.max(q[:, p.global_index]), - reach.profiles - ) - ) - if "water_elevation" in y: - my_dict[dict_y["water_elevation"]] = list( - map( - lambda p: z[id_ts, p.global_index], - reach.profiles - ) + ) + + if envelop: + z_min = reach.geometry.get_z_min() + + my_dict["min_depth"] = np.min(z, axis=0) - z_min + my_dict["max_depth"] = np.max(z, axis=0) - z_min + + def _export_rk_compute_mean_depth(self, data): + my_dict = data["my_dict"] + reach = data["reach"] + id_ts = data["id_ts"] + z = data["z"] + + my_dict["mean_depth"] = list( + map( + lambda p: p.geometry.mean_water_depth( + z[id_ts, p.global_index] + ), + reach.profiles ) - if envelop: - my_dict[dict_y["min_water_elevation"]] = list( - map( - lambda p: np.min(z[:, p.global_index]), - reach.profiles - ) - ) - my_dict[dict_y["max_water_elevation"]] = list( - map( - lambda p: np.max(z[:, p.global_index]), - reach.profiles - ) - ) + ) - if "velocity" in y: - my_dict[dict_y["velocity"]] = list( - map( - lambda p: v[id_ts, p.global_index], - reach.profiles - ) + def _export_rk_compute_froude(self, data): + my_dict = data["my_dict"] + reach = data["reach"] + id_ts = data["id_ts"] + z = data["z"] + v = data["v"] + + my_dict["froude"] = list( + map( + lambda p: p.geometry.froude( + v[id_ts, p.global_index], + z[id_ts, p.global_index] + ), + reach.profiles ) - if envelop: - velocities = list(map( - lambda p: v[:, p.global_index], - reach.profiles - )) - my_dict[dict_y["min_velocity"]] = [min(v) for v in velocities] - my_dict[dict_y["max_velocity"]] = [max(v) for v in velocities] + ) - if "depth" in y: - if self._current_results != 2: - d = my_dict[dict_y["depth"]] = list( - map( - lambda p: p.geometry.max_water_depth( - z[id_ts, p.global_index] - ), - reach.profiles - ) - ) - else: - d1 = list( - map( - lambda p: p.geometry.max_water_depth( - z[id_ts, p.global_index] - ), - reach1.profiles - ) - ) - d2 = list( - map( - lambda p: p.geometry.max_water_depth( - z[id_ts, p.global_index] - ), - reach2.profiles - ) - ) - d = list( - map( - lambda x, y: x - y, - d1, d2 - ) - ) - my_dict[dict_y["depth"]] = d - if envelop: - my_dict[dict_y["min_depth"]] = list(map( - lambda p1, p2: p1 - p2, map( - lambda p: np.min(z[:, p.global_index]), - reach.profiles - ), reach.geometry.get_z_min() - ) - ) - my_dict[dict_y["max_depth"]] = list(map( - lambda p1, p2: p1 - p2, map( - lambda p: np.max(z[:, p.global_index]), - reach.profiles - ), reach.geometry.get_z_min() - ) - ) + def _export_rk_compute_wet_area(self, data): + my_dict = data["my_dict"] + reach = data["reach"] + id_ts = data["id_ts"] + z = data["z"] - if "mean_depth" in y: - if self._current_results != 2: - d = list( - map( - lambda p: p.geometry.mean_water_depth( - z[id_ts, p.global_index] - ), - reach.profiles - ) - ) - else: - d1 = list( - map( - lambda p: p.geometry.mean_water_depth( - z[id_ts, p.global_index] - ), - reach1.profiles - ) - ) - d2 = list( - map( - lambda p: p.geometry.mean_water_depth( - z[id_ts, p.global_index] - ), - reach2.profiles - ) - ) - d = list( - map( - lambda x, y: x - y, - d1, d2 - ) - ) - my_dict[dict_y["mean_depth"]] = d - if "froude" in y: - if self._current_results != 2: - fr = my_dict[dict_y["froude"]] = list( - map( - lambda p: ( - v[id_ts, p.global_index] / - sqrt(9.81 * ( - p.geometry.wet_area( - z[id_ts, p.global_index] - ) / - p.geometry.wet_width( - z[id_ts, p.global_index] - ) - )) - ), - reach.profiles - ) - ) - else: - fr1 = list( - map( - lambda p: ( - v[id_ts, p.global_index] / - sqrt(9.81 * ( - p.geometry.wet_area( - z[id_ts, p.global_index] - ) / - p.geometry.wet_width( - z[id_ts, p.global_index] - ) - )) - ), - reach1.profiles - ) - ) - fr2 = list( - map( - lambda p: ( - v[id_ts, p.global_index] / - sqrt(9.81 * ( - p.geometry.wet_area( - z[id_ts, p.global_index] - ) / - p.geometry.wet_width( - z[id_ts, p.global_index] - ) - )) - ), - reach2.profiles - ) - ) - fr = list(map( - lambda x, y: x - y, - fr1, fr2 - )) - my_dict[dict_y["froude"]] = fr - if "wet_area" in y: - if self._current_results != 2: - wa = list( - map( - lambda p: p.geometry.wet_area( - z[id_ts, p.global_index] - ), - reach.profiles - ) - ) - else: - wa1 = list( - map( - lambda p: p.geometry.wet_area( - z[id_ts, p.global_index] - ), - reach1.profiles - ) - ) - wa2 = list( - map( - lambda p: p.geometry.wet_area( - z[id_ts, p.global_index] - ), - reach2.profiles - ) - ) - wa = list(map( - lambda x, y: x - y, - wa1, wa2 - )) - - my_dict[dict_y["wet_area"]] = wa - - return my_dict + my_dict["wet_area"] = list( + map( + lambda p: p.geometry.wet_area( + z[id_ts, p.global_index] + ), + reach.profiles + ) + ) def _export_time(self, profile, y, solver_id): results = self._results[solver_id]