Skip to content

Configuration

The configuration module provides a schema-driven approach to loading and validating MOBIDIC model configurations using YAML.

Overview

Configuration files define all aspects of a MOBIDIC simulation, including:

  • Basin metadata: Optional basin ID, optional parameter set ID, baricenter coordinates
  • Input/output paths: Meteodata, GIS data, network, states, output directories
  • Raster and vector data sources: DTM, flow direction/accumulation, soil/energy parameters, river network
  • Model parameters: Organized into subsections:
    • soil: Hydraulic conductivity, water holding capacity, flow coefficients
    • energy: Thermal properties, turbulent exchange, albedo
    • routing: Channel routing method, wave celerity, Manning coefficient
    • groundwater: Model type, global loss
    • reservoirs: Reservoir shapefiles, stage-storage curves, regulation curves/schedules (optional)
    • multipliers: Multiplying factors for calibration
  • Initial conditions: Initial state (hillslope runoff, soil saturation, reservoir volumes)
  • Simulation settings: Time step, resampling, soil/energy balance schemes, precipitation interpolation
  • Output options:
    • output_states: Boolean flags for state variables to save
    • output_states_settings: State output format (NetCDF) and intervals
    • output_report: Report variables to save (discharge, lateral inflow)
    • output_report_settings: Report format (CSV/Parquet) and reach selection
    • output_forcing_data: Meteorological forcing data output options
  • Advanced settings: Logging level and log file

The configuration system ensures all required fields are present, validates ranges and consistency, and provides sensible defaults for optional parameters.

Functions

Load and validate MOBIDIC configuration from YAML file.

Parameters:

Name Type Description Default
config_path Union[str, Path]

Path to the YAML configuration file.

 required

Returns:

Name Type Description
MOBIDICConfig MOBIDICConfig

Validated configuration object.

Raises:

Type Description
FileNotFoundError

If the configuration file does not exist.
YAMLError

If the YAML file is invalid.
ValueError

If the configuration does not match the schema.

Examples:

>>> config = load_config("examples/sample_config.yaml")
>>> print(config.basin.id)
'Basin'
Source code in mobidic/config/parser.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def load_config(config_path: Union[str, Path]) -> MOBIDICConfig:
    """
    Load and validate MOBIDIC configuration from YAML file.

    Args:
        config_path: Path to the YAML configuration file.

    Returns:
        MOBIDICConfig: Validated configuration object.

    Raises:
        FileNotFoundError: If the configuration file does not exist.
        yaml.YAMLError: If the YAML file is invalid.
        ValueError: If the configuration does not match the schema.

    Examples:
        >>> config = load_config("examples/sample_config.yaml")
        >>> print(config.basin.id)
        'Basin'
    """
    config_path = Path(config_path)

    if not config_path.exists():
        logger.error(f"Configuration file not found: {config_path}")
        raise FileNotFoundError(f"Configuration file not found: {config_path}")

    logger.info(f"Loading configuration from: {config_path}")

    # Load YAML file
    with open(config_path, "r", encoding="utf-8") as f:
        try:
            config_dict = yaml.safe_load(f)
            logger.success(f"Successfully loaded YAML file: {config_path}")
        except yaml.YAMLError as e:
            logger.error(f"Error parsing YAML file {config_path}: {e}")
            raise yaml.YAMLError(f"Error parsing YAML file {config_path}: {e}") from e

    # Validate and parse with Pydantic
    try:
        config = MOBIDICConfig(**config_dict)
        logger.info("Configuration validated successfully")
    except ValidationError as e:
        # Add context to the error message and re-raise
        logger.error(f"Configuration validation failed for {config_path}")
        raise ValueError(f"Configuration validation failed for {config_path}:\n{e}") from e

    # Resolve all paths to absolute paths relative to the YAML file location
    config_dir = config_path.parent.resolve()
    logger.debug(f"Resolving paths relative to: {config_dir}")

    def resolve_path(path_str: Union[str, Path]) -> Path:
        """Convert path to absolute, resolving relative paths from config directory."""
        if not path_str:
            return Path(path_str)
        path = Path(path_str)
        if path.is_absolute():
            return path
        else:
            return (config_dir / path).resolve()

    def is_path_field(field_info) -> bool:
        """Check if a field is a PathField type."""
        annotation = field_info.annotation

        # Handle Optional[PathField] - strip Optional first
        origin = get_origin(annotation)
        if origin is Union:
            args = get_args(annotation)
            # Remove NoneType if present (for Optional fields)
            non_none_args = [arg for arg in args if arg is not type(None)]

            # If we have a single non-None arg, check if it's Annotated
            if len(non_none_args) == 1:
                inner_annotation = non_none_args[0]
                inner_origin = get_origin(inner_annotation)

                # Check if it's Annotated[Union[str, Path], ...]
                if inner_origin is Annotated:
                    inner_args = get_args(inner_annotation)
                    if len(inner_args) > 0:
                        base_type = inner_args[0]
                        base_origin = get_origin(base_type)
                        if base_origin is Union:
                            base_args = get_args(base_type)
                            if len(base_args) == 2 and str in base_args and Path in base_args:
                                return True

            # Check if it's exactly Union[str, Path] (non-Optional case)
            if len(non_none_args) == 2 and str in non_none_args and Path in non_none_args:
                return True

        # Handle non-Optional Annotated[Union[str, Path], ...]
        if origin is Annotated:
            args = get_args(annotation)
            if len(args) > 0:
                base_type = args[0]
                base_origin = get_origin(base_type)
                if base_origin is Union:
                    base_args = get_args(base_type)
                    if len(base_args) == 2 and str in base_args and Path in base_args:
                        return True

        return False

    def resolve_path_fields(obj):
        """Recursively resolve all PathField attributes in a Pydantic model."""
        if obj is None:
            return

        # Get model fields metadata from the class, not the instance
        obj_class = type(obj)
        if hasattr(obj_class, "model_fields"):
            for field_name, field_info in obj_class.model_fields.items():
                field_value = getattr(obj, field_name)

                # Skip None values
                if field_value is None:
                    continue

                # Check if this field is a PathField
                if is_path_field(field_info):
                    # Resolve the path
                    resolved = resolve_path(field_value)
                    setattr(obj, field_name, resolved)
                    logger.debug(f"Resolved {field_name}: {field_value} -> {resolved}")
                # If it's a nested model, recurse
                elif hasattr(type(field_value), "model_fields"):
                    resolve_path_fields(field_value)

    # Recursively resolve all path fields in the config
    resolve_path_fields(config)

    return config

Save MOBIDIC configuration to YAML file.

Parameters:

Name Type Description Default
config MOBIDICConfig

Configuration object to save.

 required
output_path Union[str, Path]

Path where the YAML file will be saved.

 required

Examples:

>>> save_config(config, "config/config_modified.yaml")
Source code in mobidic/config/parser.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def save_config(config: MOBIDICConfig, output_path: Union[str, Path]) -> None:
    """
    Save MOBIDIC configuration to YAML file.

    Args:
        config: Configuration object to save.
        output_path: Path where the YAML file will be saved.

    Examples:
        >>> save_config(config, "config/config_modified.yaml")
    """
    output_path = Path(output_path)

    logger.info(f"Saving configuration to: {output_path}")

    # Convert Pydantic model to dictionary with JSON-compatible types (converts Path to str)
    config_dict = config.model_dump(exclude_none=True, mode="json")

    # Save to YAML file
    with open(output_path, "w", encoding="utf-8") as f:
        yaml.dump(
            config_dict,
            f,
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
            indent=2,
        )

    logger.info(f"Configuration saved successfully to: {output_path}")

Classes

Main configuration

Bases: BaseModel

Complete MOBIDIC configuration.

Source code in mobidic/config/schema.py 
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
class MOBIDICConfig(BaseModel):
    """Complete MOBIDIC configuration."""

    model_config = ConfigDict(
        validate_assignment=True,  # Validate on assignment
    )

    basin: Basin
    paths: Paths
    vector_files: VectorFiles
    raster_files: RasterFiles
    raster_settings: RasterSettings
    parameters: Parameters
    initial_conditions: Optional[InitialConditions] = Field(default_factory=InitialConditions)
    simulation: Simulation
    output_states: Optional[OutputStates] = Field(default_factory=OutputStates)
    output_states_settings: Optional[OutputStatesSettings] = Field(default_factory=OutputStatesSettings)
    output_report: Optional[OutputReport] = Field(default_factory=OutputReport)
    output_report_settings: Optional[OutputReportSettings] = Field(default_factory=OutputReportSettings)
    output_forcing_data: Optional[OutputForcingData] = Field(default_factory=OutputForcingData)
    hyetograph: Optional[HyetographConfig] = Field(None, description="Hyetograph generation configuration (optional)")
    advanced: Optional[Advanced] = Field(default_factory=Advanced)

    @model_validator(mode="after")
    def check_hyetograph_config_consistency(self) -> "MOBIDICConfig":
        """Validate that hyetograph config section is provided when paths.hyetograph is specified."""
        if self.paths.hyetograph is not None and self.hyetograph is None:
            raise ValueError(
                "The 'hyetograph' configuration section must be provided when 'paths.hyetograph' is specified."
            )
        return self

basin instance-attribute

paths instance-attribute

vector_files instance-attribute

raster_files instance-attribute

raster_settings instance-attribute

parameters instance-attribute

initial_conditions = Field(default_factory=InitialConditions) class-attribute instance-attribute

simulation instance-attribute

output_states = Field(default_factory=OutputStates) class-attribute instance-attribute

output_states_settings = Field(default_factory=OutputStatesSettings) class-attribute instance-attribute

output_report = Field(default_factory=OutputReport) class-attribute instance-attribute

output_report_settings = Field(default_factory=OutputReportSettings) class-attribute instance-attribute

output_forcing_data = Field(default_factory=OutputForcingData) class-attribute instance-attribute

hyetograph = Field(None, description='Hyetograph generation configuration (optional)') class-attribute instance-attribute

advanced = Field(default_factory=Advanced) class-attribute instance-attribute

Nested models

The configuration is organized hierarchically (using nested Pydantic models):

Bases: BaseModel

Basin identification and metadata.

Optional fields
  • id: Descriptive name of basin (default: empty string)
  • paramset_id: Descriptive name of parameter set / scenario (default: empty string)
Source code in mobidic/config/schema.py 
19
20
21
22
23
24
25
26
27
28
29
class Basin(BaseModel):
    """Basin identification and metadata.

    Optional fields:
        - id: Descriptive name of basin (default: empty string)
        - paramset_id: Descriptive name of parameter set / scenario (default: empty string)
    """

    id: Optional[str] = Field("", description="Descriptive name of basin")
    paramset_id: Optional[str] = Field("", description="Descriptive name of parameter set / scenario")
    baricenter: BasinBaricenter

Bases: BaseModel

File paths for input and output data.

Source code in mobidic/config/schema.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class Paths(BaseModel):
    """File paths for input and output data."""

    meteodata: Optional[PathField] = Field(
        None, description="File where the meteo data files are stored (time-series format)"
    )
    meteoraster: Optional[PathField] = Field(None, description="File where the meteo data are stored (raster format)")
    hyetograph: Optional[PathField] = Field(
        None,
        description="NetCDF file where the design hyetograph is written. "
        "If provided, hyetograph generation is enabled and the 'hyetograph' configuration section must be provided.",
    )
    gisdata: PathField = Field(..., description="Consolidated dataset to be created by GIS preprocessing")
    network: PathField = Field(..., description="Consolidated hydrographic network to be created by GIS preprocessing")
    reservoirs: Optional[PathField] = Field(
        None, description="Consolidated reservoirs dataset to be created by GIS preprocessing (geoparquet format)"
    )
    states: PathField = Field(..., description="Directory where the model states will be stored")
    output: PathField = Field(..., description="Directory where output report files will be stored")

    @model_validator(mode="after")
    def check_meteo_paths(self) -> "Paths":
        """Validate that exactly one among meteodata, meteoraster, or hyetograph is provided."""
        provided = sum(
            [
                self.meteodata is not None,
                self.meteoraster is not None,
                self.hyetograph is not None,
            ]
        )

        if provided == 0:
            raise ValueError(
                "Exactly one among 'meteodata', 'meteoraster', or 'hyetograph' must be provided. None were provided."
            )
        elif provided > 1:
            raise ValueError(
                "Exactly one among 'meteodata', 'meteoraster', or 'hyetograph' must be provided. "
                f"Found {provided} provided."
            )

        return self

check_meteo_paths()

Validate that exactly one among meteodata, meteoraster, or hyetograph is provided.

Source code in mobidic/config/schema.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@model_validator(mode="after")
def check_meteo_paths(self) -> "Paths":
    """Validate that exactly one among meteodata, meteoraster, or hyetograph is provided."""
    provided = sum(
        [
            self.meteodata is not None,
            self.meteoraster is not None,
            self.hyetograph is not None,
        ]
    )

    if provided == 0:
        raise ValueError(
            "Exactly one among 'meteodata', 'meteoraster', or 'hyetograph' must be provided. None were provided."
        )
    elif provided > 1:
        raise ValueError(
            "Exactly one among 'meteodata', 'meteoraster', or 'hyetograph' must be provided. "
            f"Found {provided} provided."
        )

    return self

Bases: BaseModel

Vector file paths and settings.

Source code in mobidic/config/schema.py 
82
83
84
85
class VectorFiles(BaseModel):
    """Vector file paths and settings."""

    river_network: RiverNetworkVector

Bases: BaseModel

Raster file paths.

Source code in mobidic/config/schema.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class RasterFiles(BaseModel):
    """Raster file paths."""

    dtm: PathField = Field(..., description="Grid of basin elevation in meters above sea level")
    flow_dir: PathField = Field(..., description="Grid of flow directions")
    flow_acc: PathField = Field(..., description="Grid of flow accumulation, as number of upstream cells")
    Wc0: PathField = Field(
        ..., description="Grid of maximum water holding capacity in soil small pores, in millimiters"
    )
    Wg0: PathField = Field(
        ..., description="Grid of maximum water holding capacity in soil large pores, in millimiters"
    )
    ks: PathField = Field(..., description="Grids of soil hydraulic conductivity, in millimiters per hour")
    kf: Optional[PathField] = Field(
        None, description="Grid of (real or ideal) aquifer conductivity, in meters per second"
    )
    CH: Optional[PathField] = Field(None, description="Grid of turbulent exchange coeff. for heat, non dimensional")
    Alb: Optional[PathField] = Field(None, description="Grid of surface albedo, non dimensional")
    Ma: Optional[PathField] = Field(
        None, description="Grid of binary mask (0,1) defining the artesian aquifer extension"
    )
    Mf: Optional[PathField] = Field(
        None, description="Grid of binary mask (0,1) defining the freatic aquifer extension"
    )
    gamma: Optional[PathField] = Field(None, description="Grid of percolation coefficient, in one over seconds")
    kappa: Optional[PathField] = Field(None, description="Grid of adsorption coefficient, in one over seconds")
    beta: Optional[PathField] = Field(None, description="Grid of hypodermic flow coefficient, in one over seconds")
    alpha: Optional[PathField] = Field(None, description="Grid of hillslope flow coefficient, in one over seconds")

Bases: BaseModel

Raster processing settings.

Source code in mobidic/config/schema.py 
118
119
120
121
122
123
124
class RasterSettings(BaseModel):
    """Raster processing settings."""

    flow_dir_type: Literal["Grass", "Arc"] = Field(
        ...,
        description="Flow direction pointer type: 'Grass' for 1-8 notation or 'Arc' for 1-2-4-8-16-32-64-128 notation",
    )

Bases: BaseModel

Global land and model parameters.

Source code in mobidic/config/schema.py 
281
282
283
284
285
286
287
288
289
class Parameters(BaseModel):
    """Global land and model parameters."""

    soil: SoilParameters
    energy: Optional[EnergyParameters] = Field(default_factory=EnergyParameters)
    routing: RoutingParameters
    groundwater: GroundwaterParameters
    reservoirs: Optional[ReservoirParameters] = Field(default_factory=ReservoirParameters)
    multipliers: Optional[Multipliers] = Field(default_factory=Multipliers)

Bases: BaseModel

Soil-related parameters.

Optional fields with defaults
  • Wc0: Maximum water holding capacity in soil small pores (default: 0.0 mm)
  • Wg0: Maximum water holding capacity in soil large pores (default: 0.0 mm)
  • ks: Soil hydraulic conductivity (default: 1.0 mm/h)
  • kf: Aquifer conductivity (default: 1.0e-7 m/s)
Source code in mobidic/config/schema.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class SoilParameters(BaseModel):
    """Soil-related parameters.

    Optional fields with defaults:
        - Wc0: Maximum water holding capacity in soil small pores (default: 0.0 mm)
        - Wg0: Maximum water holding capacity in soil large pores (default: 0.0 mm)
        - ks: Soil hydraulic conductivity (default: 1.0 mm/h)
        - kf: Aquifer conductivity (default: 1.0e-7 m/s)
    """

    Wc0: float = Field(
        0.0, description="Default value of maximum water holding capacity in soil small pores, in millimiters"
    )
    Wg0: float = Field(
        0.0, description="Default value of maximum water holding capacity in soil large pores, in millimiters"
    )
    ks: float = Field(1.0, description="Default value of soil hydraulic conductivity, in mm/h")
    ks_min: Optional[float] = Field(None, description="Default value of minimum soil hydraulic conductivity, in mm/h")
    ks_max: Optional[float] = Field(None, description="Default value of maximum soil hydraulic conductivity, in mm/h")
    kf: float = Field(1.0e-7, description="Default value of (real or ideal) aquifer conductivity, in m/s")
    gamma: float = Field(..., description="Percolation coefficient, in 1/s")
    kappa: float = Field(..., description="Adsorption coefficient, in 1/s")
    beta: float = Field(..., description="Hypodermic flow coefficient, in 1/s")
    alpha: float = Field(..., description="Hillslope flow coefficient, in 1/s")

    @field_validator("Wc0", "Wg0", "ks", "kf", "gamma", "kappa", "beta", "alpha")
    @classmethod
    def check_positive(cls, v: float) -> float:
        """Validate that required parameters are non-negative."""
        if v < 0:
            raise ValueError("Value must be non-negative")
        return v

check_positive(v) classmethod

Validate that required parameters are non-negative.

Source code in mobidic/config/schema.py 
152
153
154
155
156
157
158
@field_validator("Wc0", "Wg0", "ks", "kf", "gamma", "kappa", "beta", "alpha")
@classmethod
def check_positive(cls, v: float) -> float:
    """Validate that required parameters are non-negative."""
    if v < 0:
        raise ValueError("Value must be non-negative")
    return v

Bases: BaseModel

Energy balance parameters.

This entire section is optional. If omitted from configuration, all defaults will be used.

All fields are optional with defaults
  • Tconst: Deep ground temperature (default: 290.0 K)
  • kaps: Soil thermal conductivity (default: 2.5 W/m/K)
  • nis: Soil thermal diffusivity (default: 0.8e-6 m²/s)
  • CH: Turbulent exchange coefficient for heat (default: 1e-3)
  • Alb: Surface albedo (default: 0.2)
Source code in mobidic/config/schema.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class EnergyParameters(BaseModel):
    """Energy balance parameters.

    This entire section is optional. If omitted from configuration, all defaults will be used.

    All fields are optional with defaults:
        - Tconst: Deep ground temperature (default: 290.0 K)
        - kaps: Soil thermal conductivity (default: 2.5 W/m/K)
        - nis: Soil thermal diffusivity (default: 0.8e-6 m²/s)
        - CH: Turbulent exchange coefficient for heat (default: 1e-3)
        - Alb: Surface albedo (default: 0.2)
    """

    Tconst: float = Field(290.0, description="Deep ground temperature, in deg. Kelvin")
    kaps: float = Field(2.5, description="Soil thermal conductivity, in W/m/K")
    nis: float = Field(0.8e-6, description="Soil thermal diffusivity, in m²/s")
    CH: float = Field(1e-3, description="Default value of turbulent exchange coeff. for heat, non dimensional")
    Alb: float = Field(0.2, description="Default value of surface albedo, non dimensional")

    @field_validator("Tconst", "kaps", "nis", "CH")
    @classmethod
    def check_positive(cls, v: float) -> float:
        """Validate that parameters are positive."""
        if v <= 0:
            raise ValueError("Value must be positive")
        return v

    @field_validator("Alb")
    @classmethod
    def check_albedo_range(cls, v: float) -> float:
        """Validate that albedo is in valid range."""
        if not 0 <= v <= 1:
            raise ValueError("Albedo must be between 0 and 1")
        return v

check_albedo_range(v) classmethod

Validate that albedo is in valid range.

Source code in mobidic/config/schema.py 
188
189
190
191
192
193
194
@field_validator("Alb")
@classmethod
def check_albedo_range(cls, v: float) -> float:
    """Validate that albedo is in valid range."""
    if not 0 <= v <= 1:
        raise ValueError("Albedo must be between 0 and 1")
    return v

check_positive(v) classmethod

Validate that parameters are positive.

Source code in mobidic/config/schema.py 
180
181
182
183
184
185
186
@field_validator("Tconst", "kaps", "nis", "CH")
@classmethod
def check_positive(cls, v: float) -> float:
    """Validate that parameters are positive."""
    if v <= 0:
        raise ValueError("Value must be positive")
    return v

Bases: BaseModel

Channel routing parameters.

Optional fields with defaults
  • Br0: Width of channels with first Strahler order (default: 1.0 m)
  • NBr: Exponent of equation W = Br0*O^NBr (default: 1.5)
  • n_Man: Manning roughness coefficient for channels (default: 0.03 s/m^(1/3))
Source code in mobidic/config/schema.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
class RoutingParameters(BaseModel):
    """Channel routing parameters.

    Optional fields with defaults:
        - Br0: Width of channels with first Strahler order (default: 1.0 m)
        - NBr: Exponent of equation W = Br0*O^NBr (default: 1.5)
        - n_Man: Manning roughness coefficient for channels (default: 0.03 s/m^(1/3))
    """

    method: Literal["Musk", "MuskCun", "Lag", "Linear"] = Field(..., description="Type of channel routing scheme")
    wcel: float = Field(..., description="Flood wave celerity in channels, in m/s")
    Br0: float = Field(1.0, description="Width of channels with first Strahler order, in meters")
    NBr: float = Field(
        1.5, description="Exponent of equation W = Br0*O^NBr, where W=Width of channels and O=Strahler order"
    )
    n_Man: float = Field(0.03, description="Manning roughness coefficient for channels, in s/m^(1/3)")

    @field_validator("wcel", "Br0", "n_Man")
    @classmethod
    def check_positive(cls, v: float) -> float:
        """Validate that parameters are positive."""
        if v <= 0:
            raise ValueError("Value must be positive")
        return v

    @field_validator("NBr")
    @classmethod
    def check_nbr_range(cls, v: float) -> float:
        """Validate that NBr is greater than 1."""
        if v <= 1:
            raise ValueError("NBr must be greater than 1")
        return v

check_nbr_range(v) classmethod

Validate that NBr is greater than 1.

Source code in mobidic/config/schema.py 
222
223
224
225
226
227
228
@field_validator("NBr")
@classmethod
def check_nbr_range(cls, v: float) -> float:
    """Validate that NBr is greater than 1."""
    if v <= 1:
        raise ValueError("NBr must be greater than 1")
    return v

check_positive(v) classmethod

Validate that parameters are positive.

Source code in mobidic/config/schema.py 
214
215
216
217
218
219
220
@field_validator("wcel", "Br0", "n_Man")
@classmethod
def check_positive(cls, v: float) -> float:
    """Validate that parameters are positive."""
    if v <= 0:
        raise ValueError("Value must be positive")
    return v

Bases: BaseModel

Groundwater model parameters.

Source code in mobidic/config/schema.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
class GroundwaterParameters(BaseModel):
    """Groundwater model parameters."""

    model: Literal["None", "Linear", "Linear_mult", "Dupuit", "MODFLOW"] = Field(
        ..., description="Groundwater model type"
    )
    global_loss: Optional[float] = Field(0.0, description="Global water loss from aquifers, in m³/s")

    @field_validator("global_loss")
    @classmethod
    def check_non_negative(cls, v: Optional[float]) -> float:
        """Validate that global_loss is non-negative."""
        if v is not None and v < 0:
            raise ValueError("global_loss must be non-negative")
        return v if v is not None else 0.0

check_non_negative(v) classmethod

Validate that global_loss is non-negative.

Source code in mobidic/config/schema.py 
239
240
241
242
243
244
245
@field_validator("global_loss")
@classmethod
def check_non_negative(cls, v: Optional[float]) -> float:
    """Validate that global_loss is non-negative."""
    if v is not None and v < 0:
        raise ValueError("global_loss must be non-negative")
    return v if v is not None else 0.0

Bases: BaseModel

Reservoir input data files.

Source code in mobidic/config/schema.py 
248
249
250
251
252
253
254
class ReservoirParameters(BaseModel):
    """Reservoir input data files."""

    res_shape: Optional[PathField] = Field(None, description="Shapefile of reservoirs and lakes (polygon features)")
    stage_storage: Optional[PathField] = Field(None, description="Reservoir stage-storage curves (CSV format)")
    regulation_curves: Optional[PathField] = Field(None, description="Reservoir regulation curves (CSV format)")
    regulation_schedule: Optional[PathField] = Field(None, description="Reservoir regulation schedules (CSV format)")

Bases: BaseModel

Parameter multipliers for calibration.

Source code in mobidic/config/schema.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
class Multipliers(BaseModel):
    """Parameter multipliers for calibration."""

    ks_factor: Optional[float] = Field(1.0, description="Multiplying factor of soil hydraulic conductivity")
    Wc_factor: Optional[float] = Field(
        1.0, description="Multiplying factor of maximum water holding capacity in soil small pores"
    )
    Wg_factor: Optional[float] = Field(
        1.0, description="Multiplying factor of maximum water holding capacity in soil large pores"
    )
    Wg_Wc_tr: Optional[float] = Field(1.0, description="Transition factor between gravitational and capillary storage")
    CH_factor: Optional[float] = Field(1.0, description="Multiplying factor of turbulent exchange coeff. for heat")
    cel_factor: Optional[float] = Field(1.0, description="Multiplying factor for flood wave celerity")
    chan_factor: Optional[float] = Field(0.0, description="Scale factor for fraction of channalized flow")

    @field_validator("ks_factor", "Wc_factor", "Wg_factor", "Wg_Wc_tr", "CH_factor", "cel_factor")
    @classmethod
    def check_positive(cls, v: Optional[float]) -> float:
        """Validate that multipliers are positive."""
        if v is not None and v <= 0:
            raise ValueError("Multiplier must be positive")
        return v if v is not None else 1.0

check_positive(v) classmethod

Validate that multipliers are positive.

Source code in mobidic/config/schema.py 
272
273
274
275
276
277
278
@field_validator("ks_factor", "Wc_factor", "Wg_factor", "Wg_Wc_tr", "CH_factor", "cel_factor")
@classmethod
def check_positive(cls, v: Optional[float]) -> float:
    """Validate that multipliers are positive."""
    if v is not None and v <= 0:
        raise ValueError("Multiplier must be positive")
    return v if v is not None else 1.0

Bases: BaseModel

Initial state conditions.

All fields are optional with defaults
  • Ws: Initial depth of hillslope runoff (default: 0.0 m)
  • Wcsat: Initial relative saturation of capillary soil (default: 0.3)
  • Wgsat: Initial relative saturation of gravitational soil (default: 0.01)
  • reservoir_volumes: Path to CSV file with initial reservoir volumes (default: None)
Source code in mobidic/config/schema.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
class InitialConditions(BaseModel):
    """Initial state conditions.

    All fields are optional with defaults:
        - Ws: Initial depth of hillslope runoff (default: 0.0 m)
        - Wcsat: Initial relative saturation of capillary soil (default: 0.3)
        - Wgsat: Initial relative saturation of gravitational soil (default: 0.01)
        - reservoir_volumes: Path to CSV file with initial reservoir volumes (default: None)
    """

    Ws: Optional[float] = Field(0.0, description="Initial depth of hillslope runoff, in meters")
    Wcsat: Optional[float] = Field(0.3, description="Initial relative saturation of capillary soil, non dimensional")
    Wgsat: Optional[float] = Field(
        0.01, description="Initial relative saturation of gravitational soil, non dimensional"
    )
    reservoir_volumes: Optional[PathField] = Field(
        None,
        description=(
            "Path to CSV file with initial reservoir volumes (columns: 'reservoir_id', 'volume_m3'). "
            "If not provided, initial volumes are auto-calculated as 100% capacity (volume interpolated at z_max)"
        ),
    )

    @field_validator("Ws")
    @classmethod
    def check_ws_non_negative(cls, v: Optional[float]) -> float:
        """Validate that Ws is non-negative."""
        if v is not None and v < 0:
            raise ValueError("Ws must be non-negative")
        return v if v is not None else 0.0

    @field_validator("Wcsat", "Wgsat")
    @classmethod
    def check_saturation_range(cls, v: Optional[float]) -> float:
        """Validate that saturation is in valid range [0, 1]."""
        if v is not None and not 0 <= v <= 1:
            raise ValueError("Saturation must be between 0 and 1")
        return v

check_saturation_range(v) classmethod

Validate that saturation is in valid range [0, 1].

Source code in mobidic/config/schema.py 
323
324
325
326
327
328
329
@field_validator("Wcsat", "Wgsat")
@classmethod
def check_saturation_range(cls, v: Optional[float]) -> float:
    """Validate that saturation is in valid range [0, 1]."""
    if v is not None and not 0 <= v <= 1:
        raise ValueError("Saturation must be between 0 and 1")
    return v

check_ws_non_negative(v) classmethod

Validate that Ws is non-negative.

Source code in mobidic/config/schema.py 
315
316
317
318
319
320
321
@field_validator("Ws")
@classmethod
def check_ws_non_negative(cls, v: Optional[float]) -> float:
    """Validate that Ws is non-negative."""
    if v is not None and v < 0:
        raise ValueError("Ws must be non-negative")
    return v if v is not None else 0.0

Bases: BaseModel

Simulation control parameters.

Optional fields with defaults
  • decimation: Decimation factor from grid data space resolution to model space resolution (default: 1)
Source code in mobidic/config/schema.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
class Simulation(BaseModel):
    """Simulation control parameters.

    Optional fields with defaults:
        - decimation: Decimation factor from grid data space resolution to model space resolution (default: 1)
    """

    timestep: float = Field(..., description="Data and model time step, in seconds")
    decimation: int = Field(
        1, description="Decimation factor from grid data space resolution to model space resolution"
    )
    soil_scheme: Literal["Bucket", "CN"] = Field(..., description="Type of soil hydrology scheme")
    energy_balance: Literal["None", "1L", "5L", "Snow"] = Field(
        ..., description="Type of surface energy balance scheme"
    )
    precipitation_interp: Optional[Literal["Nearest", "IDW"]] = Field(
        "IDW", description="Precipitation interpolation method"
    )

    @field_validator("timestep")
    @classmethod
    def check_timestep_positive(cls, v: float) -> float:
        """Validate that timestep is positive."""
        if v <= 0:
            raise ValueError("timestep must be positive")
        return v

    @field_validator("decimation")
    @classmethod
    def check_decimation_positive(cls, v: int) -> int:
        """Validate that decimation is a positive integer."""
        if v <= 0:
            raise ValueError("decimation must be a positive integer")
        return v

check_decimation_positive(v) classmethod

Validate that decimation is a positive integer.

Source code in mobidic/config/schema.py 
359
360
361
362
363
364
365
@field_validator("decimation")
@classmethod
def check_decimation_positive(cls, v: int) -> int:
    """Validate that decimation is a positive integer."""
    if v <= 0:
        raise ValueError("decimation must be a positive integer")
    return v

check_timestep_positive(v) classmethod

Validate that timestep is positive.

Source code in mobidic/config/schema.py 
351
352
353
354
355
356
357
@field_validator("timestep")
@classmethod
def check_timestep_positive(cls, v: float) -> float:
    """Validate that timestep is positive."""
    if v <= 0:
        raise ValueError("timestep must be positive")
    return v

Bases: BaseModel

Output state options.

Source code in mobidic/config/schema.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
class OutputStates(BaseModel):
    """Output state options."""

    discharge: Optional[bool] = Field(True, description="Option to save states of river network for results analysis")
    reservoir_states: Optional[bool] = Field(
        False, description="Option to save states of reservoirs and lakes for results analysis"
    )
    soil_capillary: Optional[bool] = Field(
        True, description="Option to save states of soil small pores for results analysis"
    )
    soil_gravitational: Optional[bool] = Field(
        True, description="Option to save states of soil large pores for results analysis"
    )
    soil_plant: Optional[bool] = Field(
        True, description="Option to save states of plant/canopy water for results analysis"
    )
    soil_surface: Optional[bool] = Field(
        True, description="Option to save states of surface water for results analysis"
    )
    surface_temperature: Optional[bool] = Field(
        False, description="Option to save states of land surface temperature for results analysis"
    )
    ground_temperature: Optional[bool] = Field(
        False, description="Option to save states of ground temperature for results analysis"
    )
    aquifer_head: Optional[bool] = Field(False, description="Option to save states of aquifers for results analysis")
    evapotranspiration: Optional[bool] = Field(
        False, description="Option to save states of evapotranspiration for results analysis"
    )

Bases: BaseModel

Output state file settings.

Source code in mobidic/config/schema.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
class OutputStatesSettings(BaseModel):
    """Output state file settings."""

    output_format: Optional[Literal["netCDF"]] = Field("netCDF", description="Format for state output files")
    output_states: Optional[Literal["all", "final", "list", "None"]] = Field(
        None, description="Which time steps to save: 'all', 'final', 'list', or 'None' (no states saved). Default: None"
    )
    output_interval: Optional[float] = Field(None, description="Time interval for state output, in seconds")
    output_list: Optional[list[str]] = Field(
        default_factory=list,
        description="List of datetime strings (YYYY-MM-DD HH:MM:SS) to save (used when output_states='list')",
    )
    flushing: Optional[int] = Field(
        -1,
        description="How often to flush states to disk during simulation. "
        "Positive integer N = flush every N timesteps, -1 = flush only at end",
    )
    max_file_size: Optional[float] = Field(
        500.0,
        description="Maximum file size for state output files, in megabytes (MB). "
        "When a file reaches this size, a new chunk file will be created. Default: 500 MB",
    )

    @field_validator("output_interval")
    @classmethod
    def check_interval_positive(cls, v: Optional[float]) -> Optional[float]:
        """Validate that output_interval is positive if provided."""
        if v is not None and v <= 0:
            raise ValueError("output_interval must be positive")
        return v

    @field_validator("output_list")
    @classmethod
    def validate_datetime_strings(cls, v: Optional[list[str]]) -> Optional[list[str]]:
        """Validate that output_list contains valid datetime strings."""
        if v is None:
            return v

        from datetime import datetime

        for i, date_str in enumerate(v):
            try:
                # Try to parse as datetime
                datetime.fromisoformat(date_str.replace(" ", "T"))
            except (ValueError, AttributeError) as e:
                raise ValueError(
                    f"output_list[{i}] = '{date_str}' is not a valid datetime. "
                    f"Expected format: 'YYYY-MM-DD HH:MM:SS'. Error: {e}"
                ) from e
        return v

    @field_validator("flushing")
    @classmethod
    def check_flushing_valid(cls, v: Optional[int]) -> Optional[int]:
        """Validate that flushing is either -1 or a positive integer."""
        if v is None:
            return -1
        if v != -1 and v <= 0:
            raise ValueError("flushing must be either -1 (flush at end) or a positive integer")
        return v

    @field_validator("max_file_size")
    @classmethod
    def check_max_file_size_positive(cls, v: Optional[float]) -> Optional[float]:
        """Validate that max_file_size is positive if provided."""
        if v is None:
            return 500.0
        if v <= 0:
            raise ValueError("max_file_size must be positive")
        return v

    @model_validator(mode="after")
    def check_output_settings_consistency(self) -> "OutputStatesSettings":
        """Validate that output settings are consistent."""
        if self.output_states == "list" and not self.output_list:
            raise ValueError("output_list must be provided when output_states='list'")
        return self

check_flushing_valid(v) classmethod

Validate that flushing is either -1 or a positive integer.

Source code in mobidic/config/schema.py 
450
451
452
453
454
455
456
457
458
@field_validator("flushing")
@classmethod
def check_flushing_valid(cls, v: Optional[int]) -> Optional[int]:
    """Validate that flushing is either -1 or a positive integer."""
    if v is None:
        return -1
    if v != -1 and v <= 0:
        raise ValueError("flushing must be either -1 (flush at end) or a positive integer")
    return v

check_interval_positive(v) classmethod

Validate that output_interval is positive if provided.

Source code in mobidic/config/schema.py 
422
423
424
425
426
427
428
@field_validator("output_interval")
@classmethod
def check_interval_positive(cls, v: Optional[float]) -> Optional[float]:
    """Validate that output_interval is positive if provided."""
    if v is not None and v <= 0:
        raise ValueError("output_interval must be positive")
    return v

check_max_file_size_positive(v) classmethod

Validate that max_file_size is positive if provided.

Source code in mobidic/config/schema.py 
460
461
462
463
464
465
466
467
468
@field_validator("max_file_size")
@classmethod
def check_max_file_size_positive(cls, v: Optional[float]) -> Optional[float]:
    """Validate that max_file_size is positive if provided."""
    if v is None:
        return 500.0
    if v <= 0:
        raise ValueError("max_file_size must be positive")
    return v

check_output_settings_consistency()

Validate that output settings are consistent.

Source code in mobidic/config/schema.py 
470
471
472
473
474
475
@model_validator(mode="after")
def check_output_settings_consistency(self) -> "OutputStatesSettings":
    """Validate that output settings are consistent."""
    if self.output_states == "list" and not self.output_list:
        raise ValueError("output_list must be provided when output_states='list'")
    return self

validate_datetime_strings(v) classmethod

Validate that output_list contains valid datetime strings.

Source code in mobidic/config/schema.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
@field_validator("output_list")
@classmethod
def validate_datetime_strings(cls, v: Optional[list[str]]) -> Optional[list[str]]:
    """Validate that output_list contains valid datetime strings."""
    if v is None:
        return v

    from datetime import datetime

    for i, date_str in enumerate(v):
        try:
            # Try to parse as datetime
            datetime.fromisoformat(date_str.replace(" ", "T"))
        except (ValueError, AttributeError) as e:
            raise ValueError(
                f"output_list[{i}] = '{date_str}' is not a valid datetime. "
                f"Expected format: 'YYYY-MM-DD HH:MM:SS'. Error: {e}"
            ) from e
    return v

Bases: BaseModel

Output report options.

Source code in mobidic/config/schema.py 
478
479
480
481
482
483
484
class OutputReport(BaseModel):
    """Output report options."""

    discharge: Optional[bool] = Field(True, description="Option to save discharge hydrograph at selected reach IDs")
    lateral_inflow: Optional[bool] = Field(
        False, description="Option to save lateral inflow hydrograph at selected reach IDs"
    )

Bases: BaseModel

Output report file settings.

Source code in mobidic/config/schema.py 
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
class OutputReportSettings(BaseModel):
    """Output report file settings."""

    output_format: Optional[Literal["csv", "Parquet"]] = Field("Parquet", description="Format for report output files")
    report_interval: Optional[float] = Field(None, description="Time interval for report output, in seconds")
    reach_selection: Optional[Literal["all", "file", "list"]] = Field(
        "all", description="Method for selecting reaches to output"
    )
    sel_file: Optional[PathField] = Field(None, description="Path to JSON file containing reach IDs to output")
    sel_list: Optional[list[int]] = Field(None, description="List of reach IDs to output")

    @field_validator("report_interval")
    @classmethod
    def check_interval_positive(cls, v: Optional[float]) -> Optional[float]:
        """Validate that report_interval is positive if provided."""
        if v is not None and v <= 0:
            raise ValueError("report_interval must be positive")
        return v

    @model_validator(mode="after")
    def check_selection_consistency(self) -> "OutputReportSettings":
        """Validate that selection method matches provided data."""
        if self.reach_selection == "file" and self.sel_file is None:
            raise ValueError("sel_file must be provided when reach_selection='file'")
        if self.reach_selection == "list" and not self.sel_list:
            raise ValueError("sel_list must be provided when reach_selection='list'")
        return self

check_interval_positive(v) classmethod

Validate that report_interval is positive if provided.

Source code in mobidic/config/schema.py 
498
499
500
501
502
503
504
@field_validator("report_interval")
@classmethod
def check_interval_positive(cls, v: Optional[float]) -> Optional[float]:
    """Validate that report_interval is positive if provided."""
    if v is not None and v <= 0:
        raise ValueError("report_interval must be positive")
    return v

check_selection_consistency()

Validate that selection method matches provided data.

Source code in mobidic/config/schema.py 
506
507
508
509
510
511
512
513
@model_validator(mode="after")
def check_selection_consistency(self) -> "OutputReportSettings":
    """Validate that selection method matches provided data."""
    if self.reach_selection == "file" and self.sel_file is None:
        raise ValueError("sel_file must be provided when reach_selection='file'")
    if self.reach_selection == "list" and not self.sel_list:
        raise ValueError("sel_list must be provided when reach_selection='list'")
    return self

Bases: BaseModel

Output forcing data options.

Source code in mobidic/config/schema.py 
516
517
518
519
class OutputForcingData(BaseModel):
    """Output forcing data options."""

    meteo_data: Optional[bool] = Field(False, description="Option to save meteorological forcing data to NetCDF")

Bases: BaseModel

Hyetograph construction configuration.

Configuration for generating synthetic hyetographs from IDF (Intensity-Duration-Frequency) parameters. Used to create design storm precipitation fields for simulation.

The IDF formula used is: h = ka * k * a * t^n where: - h is precipitation depth [mm] - ka is the areal reduction factor (ARF) coefficient - k is the return period factor (spatially distributed raster) - a is the IDF scale parameter (spatially distributed raster) - n is the IDF shape parameter (spatially distributed raster) - t is duration [hours]

Source code in mobidic/config/schema.py 
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
class HyetographConfig(BaseModel):
    """Hyetograph construction configuration.

    Configuration for generating synthetic hyetographs from IDF (Intensity-Duration-Frequency)
    parameters. Used to create design storm precipitation fields for simulation.

    The IDF formula used is: h = ka * k * a * t^n
    where:
        - h is precipitation depth [mm]
        - ka is the areal reduction factor (ARF) coefficient
        - k is the return period factor (spatially distributed raster)
        - a is the IDF scale parameter (spatially distributed raster)
        - n is the IDF shape parameter (spatially distributed raster)
        - t is duration [hours]
    """

    a_raster: PathField = Field(..., description="Path to GeoTIFF raster with IDF 'a' parameter (scale factor)")
    n_raster: PathField = Field(..., description="Path to GeoTIFF raster with IDF 'n' parameter (shape/exponent)")
    k_raster: PathField = Field(..., description="Path to GeoTIFF raster with IDF 'k' parameter (return period factor)")
    duration_hours: int = Field(..., description="Total duration of the hyetograph in hours")
    ka: float = Field(1.0, description="Areal reduction factor (ARF) coefficient")
    hyetograph_type: Literal["chicago_decreasing"] = Field(
        "chicago_decreasing",
        description="Hyetograph construction method. Currently only 'chicago_decreasing' is implemented.",
    )
    timestep_hours: int = Field(1, description="Time step for hyetograph in hours")

    @field_validator("duration_hours", "timestep_hours")
    @classmethod
    def check_positive_int(cls, v: int) -> int:
        """Validate that duration and timestep are positive."""
        if v <= 0:
            raise ValueError("Value must be a positive integer")
        return v

    @field_validator("ka")
    @classmethod
    def check_ka_range(cls, v: float) -> float:
        """Validate that ka is in valid range (0, 1]."""
        if v <= 0 or v > 1:
            raise ValueError("ka (areal reduction factor) must be in range (0, 1]")
        return v

check_ka_range(v) classmethod

Validate that ka is in valid range (0, 1].

Source code in mobidic/config/schema.py 
557
558
559
560
561
562
563
@field_validator("ka")
@classmethod
def check_ka_range(cls, v: float) -> float:
    """Validate that ka is in valid range (0, 1]."""
    if v <= 0 or v > 1:
        raise ValueError("ka (areal reduction factor) must be in range (0, 1]")
    return v

check_positive_int(v) classmethod

Validate that duration and timestep are positive.

Source code in mobidic/config/schema.py 
549
550
551
552
553
554
555
@field_validator("duration_hours", "timestep_hours")
@classmethod
def check_positive_int(cls, v: int) -> int:
    """Validate that duration and timestep are positive."""
    if v <= 0:
        raise ValueError("Value must be a positive integer")
    return v

Bases: BaseModel

Advanced settings.

Source code in mobidic/config/schema.py 
566
567
568
569
570
571
572
class Advanced(BaseModel):
    """Advanced settings."""

    log_level: Optional[Literal["DEBUG", "INFO", "SUCCESS", "WARNING", "ERROR"]] = Field(
        "INFO", description="Logging level"
    )
    log_file: Optional[PathField] = Field(None, description="Path to log file")

Configuration structure

See the sample configuration file for a complete example with all available options and their descriptions.

Validation

The configuration system performs validation of the .yaml files with the following checks:

  • Type checking: All fields are type-checked by Pydantic
  • Range validation: Numeric parameters are validated against physical constraints (e.g., albedo 0-1, saturation 0-1)
  • Required fields: Missing required fields raise validation errors
  • Consistency checks: Inter-field dependencies are validated (e.g., if reach_selection='file', then sel_file must be provided)
  • Path validation: File paths can optionally be validated for existence

Invalid configurations will raise pydantic.ValidationError with detailed error messages.

Auxiliary files

Reach selection file

When using reach_selection='file' in the output report settings, you need to provide a JSON file containing the reach IDs to include in the output reports. The file should contain a simple JSON array of integer reach IDs (corresponding to mobidic_id values in the processed network).

Example reach_ids.json:

[
  1000,
  1050,
  1100,
  1200,
  1234
]

The reach IDs correspond to the mobidic_id field in the processed river network.

Configuration example:

output_report_settings:
  output_format: Parquet
  reach_selection: file
  sel_file: data/reach_ids.json  # Path to JSON file

Reservoir CSV files

When using reservoirs (parameters.reservoirs.res_shape is set), you need to provide CSV files for stage-storage curves, regulation curves, and regulation schedules.

Stage-storage CSV (stage_storage.csv):

Defines the relationship between water stage (elevation) and reservoir volume.

reservoir_id,stage_m,volume_m3
1,219.9,0.0
1,230.0,5000000.0
1,240.0,15000000.0
1,250.0,30000000.0
1,254.9,45000000.0

Regulation curves CSV (regulation_curves.csv):

Defines stage-discharge relationships for different regulation periods (e.g., winter vs summer operations).

reservoir_id,regulation_name,stage_m,discharge_m3s
1,winter,219.9,0.0
1,winter,230.0,5.0
1,winter,240.0,20.0
1,winter,250.0,50.0
1,summer,219.9,0.0
1,summer,230.0,2.0
1,summer,240.0,10.0
1,summer,250.0,30.0

Regulation schedule CSV (regulation_schedule.csv):

Defines which regulation curve to use during different time periods.

reservoir_id,start_date,end_date,regulation_name
1,2000-01-01,2000-05-31,winter
1,2000-06-01,2000-09-30,summer
1,2000-10-01,2000-12-31,winter
1,2001-01-01,2001-05-31,winter
1,2001-06-01,2001-09-30,summer

Initial volumes CSV (initial_conditions.csv) (optional):

Defines initial reservoir volumes. If not provided, volumes are auto-calculated from z_max field in the shapefile.

reservoir_id,volume_m3
1,20000000.0

Configuration example:

parameters:
  reservoirs:
    res_shape: reservoirs/reservoirs.shp
    stage_storage: reservoirs/stage_storage.csv
    regulation_curves: reservoirs/regulation_curves.csv
    regulation_schedule: reservoirs/regulation_schedule.csv

initial_conditions:
  reservoir_volumes: reservoirs/initial_volumes.csv  # Optional

paths:
  reservoirs: output/reservoirs.parquet  # Consolidated output

output_states:
  reservoir_states: true  # Enable reservoir state output

Logging

MOBIDICpy uses loguru for logging throughout the package. Logging is automatically configured with INFO level when the package is imported, but can be customized using either programmatic configuration or YAML configuration.

Functions

Configure the logger for MOBIDIC package.

This function configures the loguru logger with a consistent format for use across the MOBIDIC package and example scripts.

Note

The MOBIDIC package automatically calls this function with default settings (INFO level) when imported. Call this function again to reconfigure logging behavior.

Parameters:

Name Type Description Default
level str

Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Default: INFO.

 'INFO'
format_string str | None

Custom format string for log messages. If None, uses default format.

 None
log_file str | Path | None

Optional path to log file. If provided, logs will be written to this file in addition to stdout.

 None
colorize bool

Whether to use colored output (default: True).

 True

Examples:

>>> from mobidic.utils import configure_logger
>>> configure_logger(level="DEBUG", log_file="mobidic.log")
Source code in mobidic/utils/logging.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def configure_logger(
    level: str = "INFO",
    format_string: str | None = None,
    colorize: bool = True,
    log_file: str | Path | None = None,
) -> None:
    """Configure the logger for MOBIDIC package.

    This function configures the loguru logger with a consistent format
    for use across the MOBIDIC package and example scripts.

    Note:
        The MOBIDIC package automatically calls this function with default
        settings (INFO level) when imported. Call this function again to
        reconfigure logging behavior.

    Args:
        level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Default: INFO.
        format_string: Custom format string for log messages. If None, uses default format.
        log_file: Optional path to log file. If provided, logs will be written to this file
                  in addition to stdout.
        colorize: Whether to use colored output (default: True).

    Examples:
        >>> from mobidic.utils import configure_logger
        >>> configure_logger(level="DEBUG", log_file="mobidic.log")
    """
    # Remove default handler
    logger.remove()

    # Default format string
    if format_string is None:
        if colorize:
            if level == "DEBUG":
                format_string = (
                    "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
                    "<level>{level: <8}</level> | "
                    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
                    "{message}"
                )
            else:
                format_string = "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | {message}"
        else:
            format_string = "{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {message}"

    # Add stdout handler
    logger.add(
        sys.stdout,
        format=format_string,
        level=level,
        colorize=colorize,
    )

    # Add file handler if specified
    if log_file is not None:
        log_file = Path(log_file)
        if level == "DEBUG":
            format_logfile = "{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} | {message}"
        else:
            format_logfile = "{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {message}"
        logger.add(
            log_file,
            format=format_logfile,
            level=level,
            rotation="10 MB",  # Rotate when file reaches 10 MB
            retention="30 days",  # Keep logs for 30 days
            compression="zip",  # Compress rotated logs
        )
        logger.info(f"Logging to file: {log_file}")

Configure the logger from a MOBIDICConfig object.

This function reads the logging settings from the config’s advanced section and configures the logger accordingly.

Parameters:

Name Type Description Default
config MOBIDICConfig

MOBIDICConfig object containing the configuration.

 required

Examples:

>>> from mobidic import load_config, configure_logger_from_config
>>> config = load_config("config.yaml")
>>> configure_logger_from_config(config)
Source code in mobidic/utils/logging.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def configure_logger_from_config(config: "MOBIDICConfig") -> None:
    """Configure the logger from a MOBIDICConfig object.

    This function reads the logging settings from the config's advanced section
    and configures the logger accordingly.

    Args:
        config: MOBIDICConfig object containing the configuration.

    Examples:
        >>> from mobidic import load_config, configure_logger_from_config
        >>> config = load_config("config.yaml")
        >>> configure_logger_from_config(config)
    """
    log_level = config.advanced.log_level if config.advanced else "INFO"
    log_file = config.advanced.log_file if config.advanced else None
    configure_logger(level=log_level, log_file=log_file)

Programmatic configuration

Configure logging directly in your Python code:

from mobidic import configure_logger

# Set logging level
configure_logger(level="DEBUG")  # or INFO, WARNING, ERROR, CRITICAL

# Configure with log file
configure_logger(level="INFO", log_file="mobidic.log")

# Disable colorization (e.g., for CI environments)
configure_logger(level="INFO", colorize=False)

YAML configuration

Configure logging via the advanced section in your configuration file:

advanced:
  log_level: DEBUG  # or INFO, WARNING, ERROR, CRITICAL
  log_file: logs/mobidic.log  # Optional

Then load configuration and apply logging settings:

from mobidic import load_config, configure_logger_from_config

# Load configuration
config = load_config("config.yaml")

# Apply logging settings from config
configure_logger_from_config(config)

# Proceed with preprocessing or simulation
gisdata = run_preprocessing(config)

Log levels

  • DEBUG: Detailed information for troubleshooting (includes function names and line numbers)
  • INFO: Progress updates and general information (default)
  • WARNING: Non-critical issues (e.g., missing optional files, automatic fallbacks)
  • ERROR: Critical failures that stop execution
  • CRITICAL: Severe errors that may lead to program termination

Log output

Console (stdout): - Always enabled with colorized output (unless colorize=False) - Automatically adjusts format based on log level (DEBUG shows more detail)

File (optional): - Specify via log_file parameter or config.advanced.log_file - Automatic log rotation: rotates when file reaches 10 MB - Log retention: keeps logs for 30 days - Compression: rotated logs are compressed to .zip format - Format adapts based on log level (DEBUG includes module/function/line)

Usage example

from mobidic import configure_logger, run_preprocessing, load_config

# Configure detailed logging for debugging
configure_logger(level="DEBUG", log_file="debug.log")

# Load configuration
config = load_config("basin.yaml")

# Run preprocessing (all operations will be logged)
gisdata = run_preprocessing(config)
# Output:
# 2026-01-21 10:30:45 | INFO     | Starting preprocessing workflow
# 2026-01-21 10:30:45 | DEBUG    | mobidic.config.parser:load_config:123 | Loading configuration from basin.yaml
# 2026-01-21 10:30:46 | INFO     | Stage 1/5: Loading configuration
# 2026-01-21 10:30:46 | INFO     | Stage 2/5: Reading raster data
# 2026-01-21 10:30:47 | DEBUG    | mobidic.preprocessing.gis_reader:grid_to_matrix:45 | Reading raster: dtm.tif
# ...