Hotstart config

docs/conf.py

@@ -51,17 +51,17 @@
 
 # General information about the project.
 project = "Itzï"
-copyright = "2017-2025, Laurent Courty"
+copyright = "2016-2026, Laurent Courty"
 author = "Laurent Courty"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
 # built documents.
 #
 # The short X.Y version.
-version = "25.8"
+version = "26.6"
 # The full version, including alpha/beta/rc tags.
-release = "25.8"
+release = "26.6"
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

docs/conf_file.rst

@@ -33,6 +33,29 @@ Valid combinations:
 -  *start_time* and *duration*
 -  *duration* only
 
+[hotstart]
+----------
+
+.. versionadded:: 26.6
+    The hotstart feature is added.
+
+A hotstart file saves the current state of the simulation at a give point in time, allowing to restart a simulation from a given point.
+
+.. list-table::
+   :header-rows: 1
+   :widths: 25 60 15
+
+   * - Keyword
+     - Description
+     - Format
+   * - wallclock_step
+     - Wall clock duration between records.
+     - HH:MM:SS
+   * - save_file
+     - File name for the hotstart file. Each new file overwrites the anterior. Only the last created file is kept.
+     - Text string
+
+
 [input]
 -------
 

src/itzi/configreader.py

@@ -23,7 +23,12 @@
 import itzi.messenger as msgr
 from itzi.array_definitions import ARRAY_DEFINITIONS, ArrayCategory
 from itzi.const import InfiltrationModelType, TemporalType
-from itzi.data_containers import GrassParams, SimulationConfig, SurfaceFlowParameters
+from itzi.data_containers import (
+    GrassParams,
+    SimulationConfig,
+    SurfaceFlowParameters,
+    HotstartRunConfig,
+)
 
 DEPRECATED_INPUT_ALIASES: list[tuple[str, str]] = [
     # (old, new)
@@ -54,6 +59,7 @@
 )
 
 TIME_OPTION_KEYS = ("start_time", "end_time", "duration", "record_step")
+HOTSTART_OPTION_KEYS = ("wallclock_step", "save_file")
 GREEN_AMPT_KEYS = (
     "effective_porosity",
     "capillary_pressure",
@@ -142,6 +148,24 @@ def _read_time_values(params: ConfigParser) -> dict[str, str | None]:
     return time_values
 
 
+def _read_hotstart_values(params: ConfigParser) -> HotstartRunConfig | None:
+    """Read optional hotstart settings from the config file."""
+    if not params.has_section("hotstart"):
+        return None
+
+    hotstart_values = {
+        "wallclock_step": _read_optional_value(params, "hotstart", "wallclock_step", params.get),
+        "save_file_name": _read_optional_value(params, "hotstart", "save_file", params.get),
+    }
+    if all(value is None for value in hotstart_values.values()):
+        return None
+
+    try:
+        return HotstartRunConfig(**hotstart_values)
+    except ValidationError as error:
+        _fatal_validation_error(error)
+
+
 def _read_input_map_names(params: ConfigParser) -> dict[str, str | None]:
     """Read input map names and normalize deprecated aliases."""
     map_names: dict[str, str | None] = dict.fromkeys(INPUT_MAP_KEYS)
@@ -236,17 +260,6 @@ def _build_simulation_config(**kwargs: Any) -> SimulationConfig:
         _fatal_validation_error(error)
 
 
-def _legacy_drainage_params(sim_config: SimulationConfig) -> dict[str, str | float | None]:
-    """Expose drainage settings in the legacy dictionary shape."""
-    return {
-        "swmm_inp": sim_config.swmm_inp,
-        "output": sim_config.drainage_output,
-        "orifice_coeff": sim_config.orifice_coeff,
-        "free_weir_coeff": sim_config.free_weir_coeff,
-        "submerged_weir_coeff": sim_config.submerged_weir_coeff,
-    }
-
-
 class SimulationTimes(BaseModel):
     """Parsed and validated simulation time settings."""
 
@@ -350,6 +363,7 @@ def __init__(self, filename: str | None) -> None:
         self.grass_mandatory = list(GRASS_MANDATORY_KEYS)
 
         params = _read_parser(filename)
+        self.hotstart_config = _read_hotstart_values(params)
         self.raw_input_times = _read_time_values(params)
         self.input_map_names = _read_input_map_names(params)
         self.out_prefix, self.out_values, self.output_map_names = _read_output_config(params)
@@ -374,6 +388,9 @@ def __init__(self, filename: str | None) -> None:
             "infiltration_model": infiltration_model,
         }
 
+        if self.hotstart_config is not None:
+            simulation_kwargs["hotstart_config"] = self.hotstart_config
+
         stats_file = _read_optional_value(params, "statistics", "stats_file", params.get)
         if stats_file is not None:
             simulation_kwargs["stats_file"] = stats_file
@@ -382,14 +399,6 @@ def __init__(self, filename: str | None) -> None:
         simulation_kwargs.update(_read_simulation_drainage_values(params))
 
         self.sim_config = _build_simulation_config(**simulation_kwargs)
-        self.stats_file = self.sim_config.stats_file
-        self.drainage_params = _legacy_drainage_params(self.sim_config)
-        self._grass_params = self.grass_params.model_dump()
-        self.sim_param = {
-            **self.sim_config.surface_flow_parameters.model_dump(),
-            "dtinf": self.sim_config.dtinf,
-            "inf_model": self.sim_config.infiltration_model,
-        }
 
     def _check_grass_params(self, grass_params: GrassParams) -> None:
         """Ensure mandatory GRASS settings are provided together."""

src/itzi/data/example.ini

@@ -8,6 +8,12 @@ duration =
 # Duration between two records. Format HH:MM:SS
 record_step =
 
+[hotstart]
+# Wallclock duration between records. Format HH:MM:SS
+wallclock_step = 00:15:00
+# File name for the hotstart file
+save_file =
+
 [input]
 # elevation (raster map/stds)
 dem =

src/itzi/data_containers.py

@@ -192,6 +192,13 @@ class SurfaceFlowParameters(BaseModel):
     max_error: PositiveFloat = DefaultValues.MAX_ERROR
 
 
+class HotstartRunConfig(BaseModel):
+    """Configuration to restart a simulation from a hotstart file."""
+
+    wallclock_step: timedelta
+    save_file_name: str | Path
+
+
 class SimulationConfig(BaseModel):
     """Configuration data for a simulation run."""
 
@@ -202,6 +209,8 @@ class SimulationConfig(BaseModel):
     end_time: datetime
     record_step: timedelta
     temporal_type: TemporalType
+    # Hotstart config
+    hotstart_config: HotstartRunConfig | None = None
     # Input and output raster maps
     input_map_names: Dict[str, str | None]
     output_map_names: Dict[str, str | None]

src/itzi/simulation.py

@@ -100,6 +100,12 @@ def __init__(
                 for n in self.drainage_nodes_list
                 if n.node_object.is_coupled()
             }
+
+        if self.sim_config.hotstart_config:
+            self.hotstart_step: timedelta = self.sim_config.hotstart_config.wallclock_step
+            self.hotstart_filename: str = self.sim_config.hotstart_config.save_file_name
+            self.last_hotstart: datetime = datetime.now()
+
         # Grid spacing (for BMI)
         self.spacing = (self.raster_domain.dy, self.raster_domain.dx)
         # time step counter
@@ -273,6 +279,17 @@ def update(self) -> Self:
 
         # Update input arrays()
         self.update_input_arrays()
+
+        # Save hotstart
+        if self.sim_config.hotstart_config:
+            wall_time_now: datetime = datetime.now()
+            elapsed: timedelta = wall_time_now - self.last_hotstart
+            if elapsed >= self.hotstart_step:
+                hotstart_bytes: io.BytesIO = self.create_hotstart()
+                with open(self.hotstart_filename, "wb") as f:
+                    f.write(hotstart_bytes.getbuffer())
+                self.last_hotstart = wall_time_now
+
         return self
 
     def update_until(self, then: timedelta):

tests/core/test_configreader.py

@@ -7,7 +7,8 @@
 import pytest
 
 from itzi.configreader import ConfigReader
-from itzi.const import InfiltrationModelType, TemporalType
+from itzi.const import DefaultValues, InfiltrationModelType, TemporalType
+from itzi.data_containers import SurfaceFlowParameters
 from itzi.itzi_error import ItziFatal
 
 
@@ -26,6 +27,7 @@ def make_config_dict(
     time: dict[str, str] | None = None,
     input_maps: dict[str, str] | None = None,
     output: dict[str, str] | None = None,
+    hotstart: dict[str, str] | None = None,
     options: dict[str, str] | None = None,
     drainage: dict[str, str] | None = None,
     statistics: dict[str, str] | None = None,
@@ -39,6 +41,8 @@ def make_config_dict(
         "output": output or {"prefix": "out", "values": "water_depth"},
     }
 
+    if hotstart:
+        config_dict["hotstart"] = hotstart
     if options:
         config_dict["options"] = options
     if drainage:
@@ -50,6 +54,32 @@ def make_config_dict(
     return config_dict
 
 
+def test_reader_uses_defaults_when_optional_sections_are_missing(tmp_path):
+    config_file = write_config_file(tmp_path, make_config_dict())
+
+    reader = ConfigReader(config_file)
+    sim_config = reader.get_sim_params()
+    grass_params = reader.get_grass_params()
+
+    assert sim_config.hotstart_config is None
+    assert sim_config.surface_flow_parameters == SurfaceFlowParameters()
+    assert sim_config.stats_file is None
+    assert sim_config.dtinf == DefaultValues.DTINF
+    assert sim_config.swmm_inp is None
+    assert sim_config.drainage_output is None
+    assert sim_config.orifice_coeff == DefaultValues.ORIFICE_COEFF
+    assert sim_config.free_weir_coeff == DefaultValues.FREE_WEIR_COEFF
+    assert sim_config.submerged_weir_coeff == DefaultValues.SUBMERGED_WEIR_COEFF
+    assert grass_params.model_dump() == {
+        "grassdata": None,
+        "location": None,
+        "mapset": None,
+        "region": None,
+        "mask": None,
+        "grass_bin": None,
+    }
+
+
 def test_reader_normalizes_deprecated_aliases(tmp_path, caplog):
     config_file = write_config_file(
         tmp_path,

tests/core/test_hotstart_integration.py

@@ -1,8 +1,8 @@
 """Integration tests for hotstart round-trip and scheduler state restoration.
 
-This file implements phases 4 and 7 from the hotstart testing plan:
-- Phase 4: Round-Trip Resume Tests
-- Phase 7: Scheduler/Runtime State Tests
+This file implements the following:
+- Round-Trip Resume Tests
+- Scheduler/Runtime State Tests
 """
 
 from __future__ import annotations