SimulationConfig validates map_names against array definitions

src/itzi_core/__init__.py

@@ -1,6 +1,12 @@
 from itzi_core.array_definitions import (
     ARRAY_DEFINITIONS as ARRAY_DEFINITIONS,
 )
+from itzi_core.array_definitions import (
+    INPUT_ARRAY_KEYS as INPUT_ARRAY_KEYS,
+)
+from itzi_core.array_definitions import (
+    OUTPUT_ARRAY_KEYS as OUTPUT_ARRAY_KEYS,
+)
 from itzi_core.array_definitions import (
     ArrayCategory as ArrayCategory,
 )
@@ -31,6 +37,8 @@
 
 __all__ = [
     "ARRAY_DEFINITIONS",
+    "INPUT_ARRAY_KEYS",
+    "OUTPUT_ARRAY_KEYS",
     "ArrayCategory",
     "ArrayDefinition",
     "DomainData",

src/itzi_core/array_definitions.py

@@ -597,3 +597,10 @@ class ArrayDefinition:
     + _ACCUM_ARRAY_DEFINITIONS
     + _OUTPUT_ARRAY_DEFINITIONS
 )
+
+INPUT_ARRAY_KEYS = frozenset(
+    arr_def.key for arr_def in ARRAY_DEFINITIONS if ArrayCategory.INPUT in arr_def.category
+)
+OUTPUT_ARRAY_KEYS = frozenset(
+    arr_def.key for arr_def in ARRAY_DEFINITIONS if ArrayCategory.OUTPUT in arr_def.category
+)

src/itzi_core/data_containers.py

@@ -31,6 +31,7 @@
     field_validator,
 )
 
+from itzi_core.array_definitions import INPUT_ARRAY_KEYS, OUTPUT_ARRAY_KEYS
 from itzi_core.const import DefaultValues, InfiltrationModelType, TemporalType
 
 if TYPE_CHECKING:
@@ -250,12 +251,32 @@ class SimulationConfig(BaseModel):
     free_weir_coeff: NonNegativeFloat = Field(DefaultValues.FREE_WEIR_COEFF, ge=0, le=1)
     submerged_weir_coeff: NonNegativeFloat = Field(DefaultValues.SUBMERGED_WEIR_COEFF, ge=0, le=1)
 
-    @field_validator("input_map_names", "output_map_names", mode="before")
+    @field_validator("input_map_names", mode="before")
     @classmethod
-    def remove_inactive_map_names(cls, value: object) -> object:
-        """Normalize legacy null-valued map entries to omitted inactive entries."""
+    def validate_input_map_names(cls, value: object) -> object:
+        """Normalize inactive entries and validate canonical input keys."""
         if isinstance(value, dict):
-            return {key: map_name for key, map_name in value.items() if map_name is not None}
+            map_names = {key: map_name for key, map_name in value.items() if map_name is not None}
+            invalid_keys = sorted(set(map_names) - INPUT_ARRAY_KEYS)
+            if invalid_keys:
+                raise ValueError(
+                    f"input_map_names contain invalid input keys: {', '.join(invalid_keys)}"
+                )
+            return map_names
+        return value
+
+    @field_validator("output_map_names", mode="before")
+    @classmethod
+    def validate_output_map_names(cls, value: object) -> object:
+        """Normalize inactive entries and validate canonical output keys."""
+        if isinstance(value, dict):
+            map_names = {key: map_name for key, map_name in value.items() if map_name is not None}
+            invalid_keys = sorted(set(map_names) - OUTPUT_ARRAY_KEYS)
+            if invalid_keys:
+                raise ValueError(
+                    f"output_map_names contain invalid output keys: {', '.join(invalid_keys)}"
+                )
+            return map_names
         return value
 
     def as_str_dict(self) -> dict:

src/itzi_core/providers/memory_input.py

@@ -21,7 +21,7 @@
 
 import numpy as np
 
-from itzi_core.array_definitions import ARRAY_DEFINITIONS, ArrayCategory
+from itzi_core.array_definitions import INPUT_ARRAY_KEYS
 from itzi_core.domain_data import DomainData
 from itzi_core.providers.base import RasterInputProvider
 
@@ -41,11 +41,6 @@ class MemoryRasterInputConfig(TypedDict):
     timed_arrays: NotRequired[Mapping[str, Sequence[TimedRasterSlice]]]
 
 
-VALID_INPUT_KEYS: frozenset[str] = frozenset(
-    arr_def.key for arr_def in ARRAY_DEFINITIONS if ArrayCategory.INPUT in arr_def.category
-)
-
-
 class MemoryRasterInputProvider(RasterInputProvider):
     """Provide raster inputs directly from in-memory numpy arrays."""
 
@@ -78,7 +73,7 @@ def __init__(self, config: MemoryRasterInputConfig) -> None:
         }
 
     def _validate_input_keys(self, arrays: Mapping[str, object], *, config_name: str) -> None:
-        invalid_keys = sorted(set(arrays.keys()) - VALID_INPUT_KEYS)
+        invalid_keys = sorted(set(arrays.keys()) - INPUT_ARRAY_KEYS)
         if invalid_keys:
             raise ValueError(f"invalid array keys in {config_name}: {', '.join(invalid_keys)}")
 

src/itzi_core/providers/xarray_input.py

@@ -27,19 +27,14 @@
         "'uv add itzi-core[xarray]' or 'pip install itzi-core[xarray]'"
     )
 
-from itzi_core.array_definitions import ARRAY_DEFINITIONS, ArrayCategory
+from itzi_core.array_definitions import INPUT_ARRAY_KEYS
 from itzi_core.const import TemporalType
 from itzi_core.domain_data import DomainData
 from itzi_core.providers.base import RasterInputProvider
 
 __all__ = ["XarrayDimensions", "XarrayRasterInputConfig", "XarrayRasterInputProvider"]
 
 
-VALID_INPUT_KEYS: frozenset[str] = frozenset(
-    arr_def.key for arr_def in ARRAY_DEFINITIONS if ArrayCategory.INPUT in arr_def.category
-)
-
-
 class XarrayDimensions(BaseModel):
     """Names of the time and spatial dimensions for one data variable."""
 
@@ -66,7 +61,7 @@ def validate_configuration(self) -> XarrayRasterInputConfig:
         if self.simulation_start_time >= self.simulation_end_time:
             raise ValueError("simulation_start_time must be before simulation_end_time")
 
-        invalid_input_keys = sorted(set(self.input_map_names) - VALID_INPUT_KEYS)
+        invalid_input_keys = sorted(set(self.input_map_names) - INPUT_ARRAY_KEYS)
         if invalid_input_keys:
             raise ValueError(
                 f"input_map_names contain invalid input keys: {', '.join(invalid_input_keys)}"

tests/test_array_definitions.py

@@ -44,13 +44,6 @@ def test_array_definitions():
         assert not duplicates, f"Found duplicates in <{attr}>: {duplicates}"
 
 
-def test_maximum_arrays_are_internal_outputs():
-    definitions = {arr_def.key: arr_def for arr_def in ARRAY_DEFINITIONS}
-    for key in ("max_water_depth", "max_flow_speed"):
-        assert ArrayCategory.INTERNAL in definitions[key].category
-        assert ArrayCategory.OUTPUT in definitions[key].category
-
-
 def test_computed_from_arrays_are_defined():
     keys = {arr_def.key for arr_def in ARRAY_DEFINITIONS}
     for arr_def in ARRAY_DEFINITIONS:

tests/test_data_containers.py

@@ -1,9 +1,25 @@
 """Tests for validation in configuration data containers."""
 
+from datetime import UTC, datetime, timedelta
+
 import pytest
 from pydantic import ValidationError
 
-from itzi_core.data_containers import SurfaceFlowParameters
+from itzi_core.const import TemporalType
+from itzi_core.data_containers import SimulationConfig, SurfaceFlowParameters
+
+
+def _simulation_config_data() -> dict:
+    start_time = datetime(2000, 1, 1, tzinfo=UTC)
+    return {
+        "start_time": start_time,
+        "end_time": start_time + timedelta(seconds=1),
+        "record_step": timedelta(seconds=1),
+        "temporal_type": TemporalType.RELATIVE,
+        "input_map_names": {},
+        "output_map_names": {},
+        "surface_flow_parameters": SurfaceFlowParameters(),
+    }
 
 
 @pytest.mark.parametrize(
@@ -22,3 +38,59 @@ def test_surface_flow_parameters_require_max_slope_to_cover_slope_threshold(para
 def test_surface_flow_parameters_report_invalid_slope_threshold():
     with pytest.raises(ValidationError, match="slope_threshold"):
         SurfaceFlowParameters(slope_threshold=-0.1)
+
+
+def test_simulation_config_removes_null_map_entries():
+    config = SimulationConfig.model_validate(
+        _simulation_config_data()
+        | {
+            "input_map_names": {"ground_elevation": "dem", "rainfall_rate": None},
+            "output_map_names": {"water_depth": "depth", "flow_speed": None},
+        }
+    )
+
+    assert config.input_map_names == {"ground_elevation": "dem"}
+    assert config.output_map_names == {"water_depth": "depth"}
+
+
+def test_simulation_config_rejects_unknown_input_map_key():
+    with pytest.raises(
+        ValidationError, match="input_map_names contain invalid input keys: unknown"
+    ):
+        SimulationConfig.model_validate(
+            _simulation_config_data() | {"input_map_names": {"unknown": "source"}}
+        )
+
+
+def test_simulation_config_rejects_output_key_in_input_map():
+    with pytest.raises(
+        ValidationError, match="input_map_names contain invalid input keys: flow_rate_x"
+    ):
+        SimulationConfig.model_validate(
+            _simulation_config_data() | {"input_map_names": {"flow_rate_x": "source"}}
+        )
+
+
+def test_simulation_config_rejects_input_key_in_output_map():
+    with pytest.raises(
+        ValidationError, match="output_map_names contain invalid output keys: friction"
+    ):
+        SimulationConfig.model_validate(
+            _simulation_config_data() | {"output_map_names": {"friction": "result"}}
+        )
+
+
+def test_simulation_config_allows_empty_and_sparse_map_names():
+    empty_config = SimulationConfig.model_validate(_simulation_config_data())
+    sparse_config = SimulationConfig.model_validate(
+        _simulation_config_data()
+        | {
+            "input_map_names": {"ground_elevation": "dem"},
+            "output_map_names": {"flow_rate_x": "flux_east"},
+        }
+    )
+
+    assert empty_config.input_map_names == {}
+    assert empty_config.output_map_names == {}
+    assert sparse_config.input_map_names == {"ground_elevation": "dem"}
+    assert sparse_config.output_map_names == {"flow_rate_x": "flux_east"}