diff --git a/tests/ea8b/conftest.py b/tests/ea8b/conftest.py index 8fee5be..10611c7 100644 --- a/tests/ea8b/conftest.py +++ b/tests/ea8b/conftest.py @@ -33,8 +33,8 @@ # ruff: noqa: E402 from itzi_core.const import TemporalType from itzi_core.data_containers import SimulationConfig, SurfaceFlowParameters -from itzi_core.providers.csv_mass_balance_output import CSVMassBalanceOutputProvider from itzi_core.providers.memory_output import ( + MemoryMassBalanceOutputProvider, MemoryRasterOutputProvider, MemoryVectorOutputProvider, ) @@ -175,13 +175,14 @@ def ea8b_simulation(ea8b_data, test_data_path, ea8b_temp_path): ) raster_output_provider = MemoryRasterOutputProvider() vector_output_provider = MemoryVectorOutputProvider() + mass_balance_output_provider = MemoryMassBalanceOutputProvider() simulation = ( SimulationBuilder(sim_config, arr_mask) .with_input_provider(raster_input_provider) .with_raster_output_provider(raster_output_provider) .with_vector_output_provider(vector_output_provider) - .with_mass_balance_output_provider(CSVMassBalanceOutputProvider(file_name="ea8b.csv")) + .with_mass_balance_output_provider(mass_balance_output_provider) .build() ) @@ -214,6 +215,7 @@ def ea8b_simulation(ea8b_data, test_data_path, ea8b_temp_path): return { "raster_output": raster_output_provider, "vector_output": vector_output_provider, + "mass_balance_output": mass_balance_output_provider, "hotstart_split_path": hotstart_split_path, "hotstart_end_path": hotstart_end_path, "final_state_path": final_state_path, diff --git a/tests/ea8b/test_scenario.py b/tests/ea8b/test_scenario.py index 90f6749..206ead1 100644 --- a/tests/ea8b/test_scenario.py +++ b/tests/ea8b/test_scenario.py @@ -18,10 +18,8 @@ import hashlib import json import zipfile -from pathlib import Path import numpy as np -import pandas as pd import pytest from tests.ea8b.helpers import EA8B_REFERENCE_MAX_RSR, EA8B_REFERENCE_MIN_NSE @@ -35,7 +33,6 @@ def test_ea8b_scenario( ea8b_drainage_results, ea8b_simulation, helpers, - ea8b_temp_path, ): ds_itzi_results = ea8b_drainage_results ds_ref = ea8b_reference @@ -74,28 +71,28 @@ def test_ea8b_scenario( assert spatial_shape[0] == expected_rows assert spatial_shape[1] == expected_cols - stat_file_path = Path(ea8b_temp_path) / "ea8b.csv" - if stat_file_path.exists(): - df_stats = pd.read_csv(stat_file_path, sep=",") - df_stats["err_ref"] = np.where( - df_stats["volume_change"] == 0, - 0.0, - df_stats["created_volume"] / df_stats["volume_change"], - ) - assert np.allclose(df_stats["created_volume_ratio"], df_stats["err_ref"], atol=0.0005) - - df_stats["vol_change_ref"] = ( - df_stats["boundary_volume"] - + df_stats["rainfall_volume"] - + df_stats["infiltration_volume"] - + df_stats["inflow_volume"] - + df_stats["losses_volume"] - + df_stats["drainage_network_volume"] - + df_stats["created_volume"] - ) - assert np.allclose( - df_stats["vol_change_ref"], df_stats["volume_change"], atol=1, rtol=0.01 - ) + reports = ea8b_simulation["mass_balance_output"].reports + assert reports + volume_change = np.array([report.volume_change for report in reports]) + created_volume = np.array([report.created_volume for report in reports]) + created_volume_ratio = np.array([report.created_volume_ratio for report in reports]) + expected_ratio = np.zeros_like(volume_change) + np.divide(created_volume, volume_change, out=expected_ratio, where=volume_change != 0) + assert np.allclose(created_volume_ratio, expected_ratio, atol=0.0005) + + volume_change_ref = np.array( + [ + report.boundary_volume + + report.rainfall_volume + + report.infiltration_volume + + report.inflow_volume + + report.losses_volume + + report.drainage_network_volume + + report.created_volume + for report in reports + ] + ) + assert np.allclose(volume_change_ref, volume_change, atol=1, rtol=0.01) hotstart_end_path = ea8b_simulation["hotstart_end_path"] with ( diff --git a/tests/test_analytic.py b/tests/test_analytic.py index 173a989..fd8359c 100644 --- a/tests/test_analytic.py +++ b/tests/test_analytic.py @@ -12,8 +12,6 @@ GNU Lesser General Public License for more details. """ -import csv -import os from collections import namedtuple from datetime import datetime, timedelta from pathlib import Path @@ -25,8 +23,8 @@ from itzi_core import DomainData from itzi_core.const import TemporalType from itzi_core.data_containers import SimulationConfig, SurfaceFlowParameters -from itzi_core.providers.csv_mass_balance_output import CSVMassBalanceOutputProvider from itzi_core.providers.memory_output import ( + MemoryMassBalanceOutputProvider, MemoryRasterOutputProvider, MemoryVectorOutputProvider, ) @@ -37,39 +35,6 @@ ) -def identify_temporal_string(s): - """Returns 'datetime', 'timedelta', or None""" - try: - pd.to_timedelta(s) - return "timedelta" - except ValueError: - try: - # Pandas has a small acceptable range - np.datetime64(s) - return "datetime" - except ValueError: - return None - - -def test_identify_temporal_string(): - # Usage - test_strings = [ - ("2023-12-25 10:30:00", "datetime"), - ("0001-01-01T01:30:00", "datetime"), - ("0001-01-01 01:30:00", "datetime"), - ("1 day 02:30:45", "timedelta"), - ("32 days 00:30:45", "timedelta"), - ("02:30:45", "timedelta"), - ("66:01:34", "timedelta"), - ("not a time string", None), - ("0001-01-01 31:30:00", None), - ] - for s, t in test_strings: - result = identify_temporal_string(s) - print(s, t) - assert result == t - - def read_ascii_grid(filepath): filepath = Path(filepath) with filepath.open("r") as f: @@ -108,7 +73,7 @@ def output_map_names(prefix, keys): @pytest.fixture(scope="module") -def mcdo_norain_sim(test_data_path, test_data_temp_path): +def mcdo_norain_sim(test_data_path): """Run a simulation for MacDonald 1D solution long channel without rain. Delestre, O., Lucas, C., Ksinant, P.-A., Darboux, F., Laguerre, C., Vo, T.-N.-T., … Cordier, S. (2013). SWASHES: a compilation of shallow water analytic solutions for hydraulic and environmental studies. @@ -136,8 +101,6 @@ def mcdo_norain_sim(test_data_path, test_data_temp_path): # No mask. Whole domain. array_mask = np.full(shape=arr_dem.shape, fill_value=False, dtype=np.bool_) - # Run the simulation in the temp dir - os.chdir(test_data_temp_path) config = SimulationConfig( start_time=datetime(2000, 1, 1), end_time=datetime(2000, 1, 1, 0, 20), @@ -156,14 +119,13 @@ def mcdo_norain_sim(test_data_path, test_data_temp_path): surface_flow_parameters=SurfaceFlowParameters(dtmax=2, cfl=0.5), ) raster_output = MemoryRasterOutputProvider() + mass_balance_output = MemoryMassBalanceOutputProvider() simulation = ( SimulationBuilder(config, array_mask, np.float32) .with_domain_data(domain_data) .with_raster_output_provider(raster_output) .with_vector_output_provider(MemoryVectorOutputProvider()) - .with_mass_balance_output_provider( - CSVMassBalanceOutputProvider(file_name="stats_mcdo_norain.csv") - ) + .with_mass_balance_output_provider(mass_balance_output) .build() ) # Set the input arrays @@ -176,14 +138,14 @@ def mcdo_norain_sim(test_data_path, test_data_temp_path): while simulation.sim_time < simulation.end_time: simulation.update() simulation.finalize() - return simulation, reference + return simulation, reference, mass_balance_output.reports class TestMcdo_norain: def test_mcdo_norain(self, mcdo_norain_sim): - simulation, reference = mcdo_norain_sim + simulation, reference, _ = mcdo_norain_sim raster_results = simulation.report.raster_provider.output_maps_dict - wse_time, wse_array = raster_results["water_surface_elevation"][-1] + _, wse_array = raster_results["water_surface_elevation"][-1] wse_centerline = pd.DataFrame({"wse_model": wse_array[1, :]}) df_results = reference.join(wse_centerline) df_results["abs_error"] = np.abs(df_results["wse_model"] - df_results["wse"]) @@ -191,7 +153,7 @@ def test_mcdo_norain(self, mcdo_norain_sim): assert mae < 0.03 def test_flow_is_unidimensional(self, mcdo_norain_sim): - simulation, _ = mcdo_norain_sim + simulation, _, _ = mcdo_norain_sim """In the MacDonald 1D test, flow should be unidimensional in the X dimension""" flow_rate_y_arrays = simulation.report.raster_provider.output_maps_dict["flow_rate_y"] for _, flow_rate_y_array in flow_rate_y_arrays: @@ -200,42 +162,35 @@ def test_flow_is_unidimensional(self, mcdo_norain_sim): assert np.min(flow_rate_y_array) == 0 assert np.max(flow_rate_y_array) == 0 - def test_stat_file_is_coherent(self, test_data_temp_path): - stat_file_path = Path(test_data_temp_path) / Path("stats_mcdo_norain.csv") - # Test time format - with open(stat_file_path, "r") as csvfile: - reader = csv.DictReader(csvfile) - for row in reader: - assert identify_temporal_string(row["simulation_time"]) == "timedelta" - - df_stats = pd.read_csv(stat_file_path, sep=",") - # Compute the reference error, preventing NaN - df_stats["err_ref"] = np.where( - df_stats["volume_change"] == 0, - 0.0, - df_stats["created_volume"] / df_stats["volume_change"], - ) - # Check if the created-volume ratio computation is correct - assert np.allclose(df_stats["created_volume_ratio"], df_stats["err_ref"], atol=0.0001) - - # Check if the volume change is coherent with the rest of the volumes - df_stats["vol_change_ref"] = ( - df_stats["boundary_volume"] - + df_stats["rainfall_volume"] - + df_stats["infiltration_volume"] - + df_stats["inflow_volume"] - + df_stats["losses_volume"] - + df_stats["drainage_network_volume"] - + df_stats["created_volume"] - ) - print(df_stats.to_string()) - assert np.allclose( - df_stats["vol_change_ref"], df_stats["volume_change"], atol=1, rtol=0.01 + def test_mass_balance_is_coherent(self, mcdo_norain_sim): + _, _, reports = mcdo_norain_sim + assert reports + assert all(isinstance(report.simulation_time, timedelta) for report in reports) + + volume_change = np.array([report.volume_change for report in reports]) + created_volume = np.array([report.created_volume for report in reports]) + created_volume_ratio = np.array([report.created_volume_ratio for report in reports]) + expected_ratio = np.zeros_like(volume_change) + np.divide(created_volume, volume_change, out=expected_ratio, where=volume_change != 0) + assert np.allclose(created_volume_ratio, expected_ratio, atol=0.0001) + + volume_change_ref = np.array( + [ + report.boundary_volume + + report.rainfall_volume + + report.infiltration_volume + + report.inflow_volume + + report.losses_volume + + report.drainage_network_volume + + report.created_volume + for report in reports + ] ) + assert np.allclose(volume_change_ref, volume_change, atol=1, rtol=0.01) @pytest.fixture(scope="module") -def mcdo_rain_sim(test_data_path, test_data_temp_path): +def mcdo_rain_sim(test_data_path): """Run a simulation for MacDonald 1D solution long channel with rain. Delestre, O., Lucas, C., Ksinant, P.-A., Darboux, F., Laguerre, C., Vo, T.-N.-T., … Cordier, S. (2013). SWASHES: a compilation of shallow water analytic solutions for hydraulic and environmental studies. @@ -265,8 +220,6 @@ def mcdo_rain_sim(test_data_path, test_data_temp_path): # No mask. Whole domain. array_mask = np.full(shape=arr_dem.shape, fill_value=False, dtype=np.bool_) - # Run the simulation in the temp dir - os.chdir(test_data_temp_path) config = SimulationConfig( start_time=datetime(2020, 2, 1, 0, 1), end_time=datetime(2020, 2, 1, 0, 41), @@ -286,14 +239,13 @@ def mcdo_rain_sim(test_data_path, test_data_temp_path): dtinf=1, ) raster_output = MemoryRasterOutputProvider() + mass_balance_output = MemoryMassBalanceOutputProvider() simulation = ( SimulationBuilder(config, array_mask, np.float32) .with_domain_data(domain_data) .with_raster_output_provider(raster_output) .with_vector_output_provider(MemoryVectorOutputProvider()) - .with_mass_balance_output_provider( - CSVMassBalanceOutputProvider(file_name="stats_mcdo_rain.csv") - ) + .with_mass_balance_output_provider(mass_balance_output) .build() ) # Set the input arrays @@ -307,11 +259,11 @@ def mcdo_rain_sim(test_data_path, test_data_temp_path): while simulation.sim_time < simulation.end_time: simulation.update() simulation.finalize() - return simulation, reference + return simulation, reference, mass_balance_output.reports -def test_mcdo_rain(mcdo_rain_sim, test_data_temp_path): - simulation, reference = mcdo_rain_sim +def test_mcdo_rain(mcdo_rain_sim): + simulation, reference, reports = mcdo_rain_sim raster_results = simulation.report.raster_provider.output_maps_dict _, wse_array = raster_results["water_surface_elevation"][-1] wse_centerline = pd.DataFrame({"wse_model": wse_array[1, :]}) @@ -321,9 +273,5 @@ def test_mcdo_rain(mcdo_rain_sim, test_data_temp_path): print(mae) assert mae < 0.035 - stat_file_path = Path(test_data_temp_path) / Path("stats_mcdo_rain.csv") - # Test time format - with open(stat_file_path, "r") as csvfile: - reader = csv.DictReader(csvfile) - for row in reader: - assert identify_temporal_string(row["simulation_time"]) == "datetime" + assert reports + assert all(isinstance(report.simulation_time, datetime) for report in reports) diff --git a/tests/test_csv_mass_balance_output.py b/tests/test_csv_mass_balance_output.py index 10ca0c5..48649e7 100644 --- a/tests/test_csv_mass_balance_output.py +++ b/tests/test_csv_mass_balance_output.py @@ -13,7 +13,7 @@ """ import csv -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest @@ -43,11 +43,18 @@ def test_init_with_default_filename(tmp_path, monkeypatch): assert provider.file_name.endswith("_stats.csv") -def test_log_absolute_time(provider_fixture): +@pytest.mark.parametrize( + "test_time", + [ + datetime(2000, 1, 1, tzinfo=UTC), + timedelta(hours=1, minutes=23, seconds=45), + ], + ids=["absolute", "relative"], +) +def test_log_temporal_value(provider_fixture, test_time): provider = CSVMassBalanceOutputProvider( file_name=provider_fixture["file_name"], ) - test_time = datetime.now(UTC) test_data = MassBalanceData( simulation_time=test_time, average_timestep=12.42345, diff --git a/tests/test_stats.py b/tests/test_stats.py index af4312f..72ed38c 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -20,11 +20,14 @@ import pandas as pd import pytest +from itzi_core import Simulation from itzi_core.const import InfiltrationModelType, TemporalType -from itzi_core.data_containers import SimulationConfig, SurfaceFlowParameters +from itzi_core.data_containers import MassBalanceData, SimulationConfig, SurfaceFlowParameters +from itzi_core.providers.base import MassBalanceOutputProvider from itzi_core.providers.csv_mass_balance_output import CSVMassBalanceOutputProvider from itzi_core.providers.memory_input import MemoryRasterInputProvider, TimedRasterSlice from itzi_core.providers.memory_output import ( + MemoryMassBalanceOutputProvider, MemoryRasterOutputProvider, MemoryVectorOutputProvider, ) @@ -61,17 +64,38 @@ def test_calculate_closure_uses_absolute_tolerance_for_negligible_flow(): @pytest.fixture(scope="module") -def sim_5by5_stats(domain_5by5, helpers, tmp_path_factory): +def sim_5by5_stats(domain_5by5, helpers): + mass_balance_output = MemoryMassBalanceOutputProvider() + simulation = _run_5by5_stats_simulation( + domain_5by5, + helpers, + mass_balance_output_provider=mass_balance_output, + ) + return simulation, domain_5by5.domain_data, mass_balance_output.reports + + +@pytest.fixture(scope="module") +def sim_5by5_stats_file(domain_5by5, helpers, tmp_path_factory): + stats_file = tmp_path_factory.mktemp("stats_test") / "5by5_stats.csv" + _run_5by5_stats_simulation( + domain_5by5, + helpers, + mass_balance_output_provider=CSVMassBalanceOutputProvider(file_name=str(stats_file)), + ) + return stats_file + + +def _run_5by5_stats_simulation( + domain_5by5, + helpers, + *, + mass_balance_output_provider: MassBalanceOutputProvider | None = None, +) -> Simulation: """Run a 5x5 simulation for 5s with 1s record step. Outputs: water_depth, mean_infiltration, mean_rainfall, mean_inflow, mean_losses, created_volume Uses CONSTANT infiltration model with rain, infiltration, losses, and inflow. """ - # Create temp directory for stats file - temp_dir = tmp_path_factory.mktemp("stats_test") - stats_file = temp_dir / "5by5_stats.csv" - - # Build SimulationConfig sim_config = SimulationConfig( start_time=datetime(2000, 1, 1, 0, 0, 0), end_time=datetime(2000, 1, 1, 0, 0, 5), # 5 seconds @@ -103,20 +127,17 @@ def sim_5by5_stats(domain_5by5, helpers, tmp_path_factory): infiltration_model=InfiltrationModelType.CONSTANT, ) - # Create output provider raster_output = MemoryRasterOutputProvider() - - # Build simulation - simulation = ( + builder = ( SimulationBuilder(sim_config, domain_5by5.arr_mask, np.float32) .with_domain_data(domain_5by5.domain_data) .with_raster_output_provider(raster_output) .with_vector_output_provider(MemoryVectorOutputProvider()) - .with_mass_balance_output_provider(CSVMassBalanceOutputProvider(file_name=str(stats_file))) - .build() ) + if mass_balance_output_provider is not None: + builder.with_mass_balance_output_provider(mass_balance_output_provider) + simulation = builder.build() - # Set input arrays simulation.set_array("ground_elevation", domain_5by5.arr_dem_flat) simulation.set_array("friction", domain_5by5.arr_n) simulation.set_array("water_depth", domain_5by5.arr_start_h) @@ -125,29 +146,24 @@ def sim_5by5_stats(domain_5by5, helpers, tmp_path_factory): simulation.set_array("losses", domain_5by5.arr_loss) simulation.set_array("inflow", domain_5by5.arr_inflow) - # Run simulation simulation.initialize() while simulation.sim_time < simulation.end_time: simulation.update() simulation.finalize() - return simulation, domain_5by5.domain_data, stats_file + return simulation class TestStatsFile: - """Test that the statistics CSV file has correct columns and volume values.""" + """Test statistics CSV artifacts and mass-balance volume values.""" - def test_stats_file_exists(self, sim_5by5_stats): + def test_stats_file_exists(self, sim_5by5_stats_file): """The stats CSV file should be created.""" - _, _, stats_file = sim_5by5_stats - assert stats_file.exists() + assert sim_5by5_stats_file.exists() - def test_stats_file_columns(self, sim_5by5_stats): + def test_stats_file_columns(self, sim_5by5_stats_file): """CSV columns should match MassBalanceData fields.""" - from itzi_core.data_containers import MassBalanceData - - _, _, stats_file = sim_5by5_stats - df = pd.read_csv(stats_file) + df = pd.read_csv(sim_5by5_stats_file) expected_cols = list(MassBalanceData.model_fields.keys()) assert df.columns.to_list() == expected_cols @@ -156,8 +172,7 @@ def test_stats_file_columns(self, sim_5by5_stats): def test_stats_volume_values(self, sim_5by5_stats): """Volume values should match expected rates x area.""" - _, domain_data, stats_file = sim_5by5_stats - df = pd.read_csv(stats_file) + _, domain_data, reports = sim_5by5_stats # Domain area: 5x5 cells at 10m resolution = 2500 m² area = domain_data.cell_area * domain_data.cells @@ -174,36 +189,53 @@ def test_stats_volume_values(self, sim_5by5_stats): # Inflow: 0.1 m/s expected_inflow_vol = 0.1 * area - # Ignore first row (initial state, before time-stepping) - assert np.all(np.isclose(df["rainfall_volume"][1:], expected_rain_vol, atol=0.001)) - assert np.isclose(df["infiltration_volume"].iloc[1], expected_inf_vol_first, atol=0.001) - assert np.all(np.isclose(df["infiltration_volume"][2:], expected_inf_vol, atol=0.001)) - assert np.all(np.isclose(df["inflow_volume"][1:], expected_inflow_vol, atol=0.001)) - assert np.isclose(df["losses_volume"].iloc[1], expected_losses_vol_first, atol=0.001) - assert np.all(np.isclose(df["losses_volume"][2:], expected_losses_vol, atol=0.001)) + # Ignore the initial report, which precedes time-stepping. + assert np.all( + np.isclose( + [report.rainfall_volume for report in reports[1:]], expected_rain_vol, atol=0.001 + ) + ) + assert np.isclose(reports[1].infiltration_volume, expected_inf_vol_first, atol=0.001) + assert np.all( + np.isclose( + [report.infiltration_volume for report in reports[2:]], + expected_inf_vol, + atol=0.001, + ) + ) + assert np.all( + np.isclose( + [report.inflow_volume for report in reports[1:]], expected_inflow_vol, atol=0.001 + ) + ) + assert np.isclose(reports[1].losses_volume, expected_losses_vol_first, atol=0.001) + assert np.all( + np.isclose( + [report.losses_volume for report in reports[2:]], expected_losses_vol, atol=0.001 + ) + ) def test_volume_change_coherence(self, sim_5by5_stats): """Volume change should be coherent with other volume components.""" - _, _, stats_file = sim_5by5_stats - df = pd.read_csv(stats_file) - - # Check if the volume change is coherent with the rest of the volumes - df["vol_change_ref"] = ( - df["boundary_volume"] - + df["rainfall_volume"] - + df["infiltration_volume"] - + df["inflow_volume"] - + df["losses_volume"] - + df["drainage_network_volume"] - + df["created_volume"] - ) - assert np.allclose(df["vol_change_ref"], df["volume_change"], atol=1, rtol=0.01) - assert np.allclose( - df["closure_residual"], - df["volume_change"] - df["vol_change_ref"], - equal_nan=True, + _, _, reports = sim_5by5_stats + volume_change_ref = np.array( + [ + report.boundary_volume + + report.rainfall_volume + + report.infiltration_volume + + report.inflow_volume + + report.losses_volume + + report.drainage_network_volume + + report.created_volume + for report in reports + ] ) + volume_change = np.array([report.volume_change for report in reports]) + closure_residual = np.array([report.closure_residual for report in reports]) + + assert np.allclose(volume_change_ref, volume_change, atol=1, rtol=0.01) + assert np.allclose(closure_residual, volume_change - volume_change_ref, equal_nan=True) def _make_timed_forcing_slices( @@ -243,16 +275,14 @@ def _make_timed_forcing_slices( def _run_timed_stats_simulation( domain_5by5, helpers, - tmp_path, *, forcing_key: str, forcing_values_by_second: dict[int, float], infiltration_model: InfiltrationModelType = InfiltrationModelType.NULL, initial_water_depth: np.ndarray | None = None, -) -> pd.DataFrame: +) -> list[MassBalanceData]: start_time = datetime(2000, 1, 1, 0, 0, 0) end_time = start_time + timedelta(seconds=20) - stats_file = tmp_path / f"timed_stats_{forcing_key}.csv" forcing_slices = _make_timed_forcing_slices( domain_5by5, start_time, @@ -294,12 +324,13 @@ def _run_timed_stats_simulation( "timed_arrays": {forcing_key: forcing_slices}, } ) + mass_balance_output = MemoryMassBalanceOutputProvider() simulation = ( SimulationBuilder(sim_config, domain_5by5.arr_mask, np.float32) .with_input_provider(input_provider) .with_raster_output_provider(MemoryRasterOutputProvider()) .with_vector_output_provider(MemoryVectorOutputProvider()) - .with_mass_balance_output_provider(CSVMassBalanceOutputProvider(file_name=str(stats_file))) + .with_mass_balance_output_provider(mass_balance_output) .build() ) @@ -307,7 +338,7 @@ def _run_timed_stats_simulation( while simulation.sim_time < simulation.end_time: simulation.update() simulation.finalize() - return pd.read_csv(stats_file) + return mass_balance_output.reports @pytest.mark.parametrize( @@ -357,7 +388,6 @@ def _run_timed_stats_simulation( def test_timed_input_stats_first_report_row_stays_interval_coherent( domain_5by5, helpers, - tmp_path, forcing_key: str, forcing_values_by_second: dict[int, float], volume_column: str, @@ -365,29 +395,28 @@ def test_timed_input_stats_first_report_row_stays_interval_coherent( infiltration_model: InfiltrationModelType, initial_water_depth: np.ndarray | None, ) -> None: - df = _run_timed_stats_simulation( + reports = _run_timed_stats_simulation( domain_5by5, helpers, - tmp_path, forcing_key=forcing_key, forcing_values_by_second=forcing_values_by_second, infiltration_model=infiltration_model, initial_water_depth=initial_water_depth, ) - first_report_row = df.iloc[1] + first_report = reports[1] volume_change_ref = ( - first_report_row["boundary_volume"] - + first_report_row["rainfall_volume"] - + first_report_row["infiltration_volume"] - + first_report_row["inflow_volume"] - + first_report_row["losses_volume"] - + first_report_row["drainage_network_volume"] - + first_report_row["created_volume"] + first_report.boundary_volume + + first_report.rainfall_volume + + first_report.infiltration_volume + + first_report.inflow_volume + + first_report.losses_volume + + first_report.drainage_network_volume + + first_report.created_volume ) - assert np.isclose(first_report_row[volume_column], expected_volume, atol=1e-6, rtol=1e-6) - assert np.isclose(first_report_row["volume_change"], volume_change_ref, atol=5e-4, rtol=1e-6) + assert np.isclose(getattr(first_report, volume_column), expected_volume, atol=1e-6, rtol=1e-6) + assert np.isclose(first_report.volume_change, volume_change_ref, atol=5e-4, rtol=1e-6) class TestStatsMaps: diff --git a/tests/test_steep_plane.py b/tests/test_steep_plane.py index 42c0fbb..878e240 100644 --- a/tests/test_steep_plane.py +++ b/tests/test_steep_plane.py @@ -14,18 +14,16 @@ from __future__ import annotations -import csv from datetime import UTC, datetime, timedelta -from pathlib import Path import numpy as np import pytest -from itzi_core import DomainData +from itzi_core import DomainData, Simulation from itzi_core.const import InfiltrationModelType, TemporalType from itzi_core.data_containers import SimulationConfig, SurfaceFlowParameters -from itzi_core.providers.csv_mass_balance_output import CSVMassBalanceOutputProvider from itzi_core.providers.memory_output import ( + MemoryMassBalanceOutputProvider, MemoryRasterOutputProvider, MemoryVectorOutputProvider, ) @@ -38,7 +36,7 @@ REGIME_SWITCH_DT = 0.1 -def _run_steep_plane(max_slope: float, stats_file: Path): +def _run_steep_plane(max_slope: float) -> tuple[Simulation, MemoryMassBalanceOutputProvider]: rows = cols = 9 cell_size = 1.0 start_time = datetime(2000, 1, 1, tzinfo=UTC) @@ -71,12 +69,13 @@ def _run_steep_plane(max_slope: float, stats_file: Path): ) array_mask = np.zeros((rows, cols), dtype=np.bool_) raster_output = MemoryRasterOutputProvider() + mass_balance_output = MemoryMassBalanceOutputProvider() simulation = ( SimulationBuilder(config, array_mask, np.float32) .with_domain_data(domain_data) .with_raster_output_provider(raster_output) .with_vector_output_provider(MemoryVectorOutputProvider()) - .with_mass_balance_output_provider(CSVMassBalanceOutputProvider(file_name=str(stats_file))) + .with_mass_balance_output_provider(mass_balance_output) .build() ) @@ -91,12 +90,7 @@ def _run_steep_plane(max_slope: float, stats_file: Path): while simulation.sim_time < simulation.end_time: simulation.update() simulation.finalize() - return simulation - - -def _last_rainfall_volume(stats_file: Path) -> float: - with stats_file.open(newline="") as file: - return float(list(csv.DictReader(file))[-1]["rainfall_volume"]) + return simulation, mass_balance_output def _regime_switch_parameters(**overrides) -> SurfaceFlowParameters: @@ -248,15 +242,15 @@ def _assert_hotstart_regime_switch_matches_continuous( ) -def test_rain_on_steep_plane_uses_capped_downhill_flow(tmp_path): +def test_rain_on_steep_plane_uses_capped_downhill_flow(): """Rainfall activates the high-slope GMS branch without creating invalid state.""" - low_cap = _run_steep_plane(max_slope=0.2, stats_file=tmp_path / "low_cap.csv") - high_cap = _run_steep_plane(max_slope=0.8, stats_file=tmp_path / "high_cap.csv") + low_cap, low_cap_mass_balance = _run_steep_plane(max_slope=0.2) + high_cap, high_cap_mass_balance = _run_steep_plane(max_slope=0.8) expected_rainfall_volume = RAIN_RATE * 9 * 9 * SIMULATION_DURATION.total_seconds() - for simulation, stats_file in [ - (low_cap, tmp_path / "low_cap.csv"), - (high_cap, tmp_path / "high_cap.csv"), + for simulation, mass_balance_output in [ + (low_cap, low_cap_mass_balance), + (high_cap, high_cap_mass_balance), ]: water_depth = simulation.get_array("water_depth") eastward_flow = simulation.get_array("discharge_east") @@ -268,7 +262,9 @@ def test_rain_on_steep_plane_uses_capped_downhill_flow(tmp_path): assert np.all(np.isfinite(southward_flow)) assert np.min(water_depth) >= 0 assert np.sum(water_depth) == pytest.approx(expected_rainfall_volume, rel=1e-5) - assert _last_rainfall_volume(stats_file) == pytest.approx(expected_rainfall_volume) + assert mass_balance_output.reports[-1].rainfall_volume == pytest.approx( + expected_rainfall_volume + ) # Both runs solve the same first wet face, so the GMS flux differs only by sqrt(max_slope). center = (4, 4)