RainfallRunoff file models

The rr module provides the various file models for files that are input to the Rainfall Runoff kernel. Support is still basic, but growing. The input 'layers' listed below correspond with the input description in the SOBEK UM.

Main input (sobek_3b.fnm)

models.py defines the RainfallRunoffModel and supporting structures.

ImmutableDiskOnlyFileModel

Bases: DiskOnlyFileModel

ImmutableDiskOnlyFileModel modifies the DiskOnlyFileModel to provide faux immutablitity.

This behaviour is required for the mappix properties, which should always have the same name and path and should not be modified by users.

Source code in hydrolib/core/rr/models.py

class ImmutableDiskOnlyFileModel(DiskOnlyFileModel, allow_mutation=False):
    """
    ImmutableDiskOnlyFileModel modifies the DiskOnlyFileModel to provide faux
    immutablitity.

    This behaviour is required for the mappix properties, which should always
    have the same name and path and should not be modified by users.
    """

    pass

RainfallRunoffModel

Bases: ParsableFileModel

The RainfallRunoffModel contains all paths and sub-models related to the Rainfall Runoff model.

Source code in hydrolib/core/rr/models.py

class RainfallRunoffModel(ParsableFileModel):
    """The RainfallRunoffModel contains all paths and sub-models related to the
    Rainfall Runoff model.
    """

    _disk_only_file_model_cannot_be_none = (
        validator_set_default_disk_only_file_model_when_none()
    )

    # Note that order is defined by the .fnm file type and is used for parsing the data.
    control_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("delft_3b.ini"))
    )
    node_data: Optional[NodeFile] = Field(None)
    link_data: Optional[LinkFile] = Field(None)
    open_water_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3brunoff.tp"))
    )
    paved_area_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("paved.3b"))
    )
    paved_area_storage: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("paved.sto"))
    )
    paved_area_dwa: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("paved.dwa"))
    )
    paved_area_sewer_pump_capacity: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("paved.tbl"))
    )
    boundaries: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("pluvius.dwa"))
    )
    pluvius: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("pluvius.3b"))
    )
    pluvius_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("pluvius.alg"))
    )
    kasklasse: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("kasklass"))
    )
    bui_file: Optional[BuiModel] = Field(None)
    verdampings_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("default.evp"))
    )
    unpaved_area_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.3b"))
    )
    unpaved_area_storage: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.sto"))
    )
    green_house_area_initialisation: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("kasinit"))
    )
    green_house_area_usage_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("kasgebr"))
    )
    crop_factors: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("cropfact"))
    )
    table_bergingscoef: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("bergcoef"))
    )
    unpaved_alpha_factor_definitions: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.alf"))
    )
    run_log_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sobek_3b.log"))
    )
    schema_overview: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3b_gener.out"))
    )
    output_results_paved_area: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("paved.out"))
    )
    output_results_unpaved_area: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.out"))
    )
    output_results_green_houses: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("grnhous.out"))
    )
    output_results_open_water: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("openwate.out"))
    )
    output_results_structures: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("struct3b.out"))
    )
    output_results_boundaries: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("bound3b.out"))
    )
    output_results_pluvius: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("pluvius.out"))
    )
    infiltration_definitions_unpaved: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.inf"))
    )
    run_debug_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sobek_3b.dbg"))
    )
    unpaved_seepage: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.sep"))
    )
    unpaved_initial_values_tables: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unpaved.tbl"))
    )
    green_house_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("greenhse.3b"))
    )
    green_house_roof_storage: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("greenhse.rf"))
    )
    pluvius_sewage_entry: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("runoff.out"))
    )
    input_variable_gauges_on_edge_nodes: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sbk_rtc.his"))
    )
    input_salt_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("salt.3b"))
    )
    input_crop_factors_open_water: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("crop_ow.prn"))
    )
    restart_file_input: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("RSRR_IN"))
    )
    restart_file_output: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("RSRR_OUT"))
    )
    binary_file_input: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3b_input.bin"))
    )
    sacramento_input: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sacrmnto.3b"))
    )
    output_flow_rates_edge_nodes: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("aanvoer.abr"))
    )
    output_salt_concentration_edge: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("saltbnd.out"))
    )
    output_salt_exportation: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("salt.out"))
    )
    greenhouse_silo_definitions: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("greenhse.sil"))
    )
    open_water_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("openwate.3b"))
    )
    open_water_seepage_definitions: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("openwate.sep"))
    )
    open_water_tables_target_levels: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("openwate.tbl"))
    )
    structure_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("struct3b.dat"))
    )
    structure_definitions: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("struct3b.def"))
    )
    controller_definitions: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("contr3b.def"))
    )
    structure_tables: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("struct3b.tbl"))
    )
    boundary_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("bound3b.3b"))
    )
    boundary_tables: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("bound3b.tbl"))
    )
    sobek_location_rtc: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sbk_loc.rtc"))
    )
    wwtp_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("wwtp.3b"))
    )
    wwtp_tables: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("wwtp.tbl"))
    )
    industry_general: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("industry.3b"))
    )

    # Mappix paths should not change and always hold the defined values.
    _validate_mappix_paved_area_sewage_storage = _validator_mappix_value(
        "mappix_paved_area_sewage_storage", _mappix_paved_area_sewage_storage_name
    )
    mappix_paved_area_sewage_storage: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_paved_area_sewage_storage_name)),
        allow_mutation=False,
    )

    _validate_mappix_paved_area_flow_rates = _validator_mappix_value(
        "mappix_paved_area_flow_rates", _mappix_paved_area_flow_rates_name
    )
    mappix_paved_area_flow_rates: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_paved_area_flow_rates_name)),
        allow_mutation=False,
    )
    _validate_mappix_unpaved_area_flow_rates = _validator_mappix_value(
        "mappix_unpaved_area_flow_rates", _mappix_unpaved_area_flow_rates_name
    )
    mappix_unpaved_area_flow_rates: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_unpaved_area_flow_rates_name)),
        allow_mutation=False,
    )
    _validate_mappix_ground_water_levels = _validator_mappix_value(
        "mappix_ground_water_levels", _mappix_ground_water_levels_name
    )
    mappix_ground_water_levels: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_ground_water_levels_name)),
        allow_mutation=False,
    )
    _validate_mappix_green_house_bassins_storage = _validator_mappix_value(
        "mappix_green_house_bassins_storage", _mappix_green_house_bassins_storage_name
    )
    mappix_green_house_bassins_storage: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_green_house_bassins_storage_name)),
        allow_mutation=False,
    )
    _validate_mappix_green_house_bassins_results = _validator_mappix_value(
        "mappix_green_house_bassins_results", _mappix_green_house_bassins_results_name
    )
    mappix_green_house_bassins_results: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_green_house_bassins_results_name)),
        allow_mutation=False,
    )
    _validate_mappix_open_water_details = _validator_mappix_value(
        "mappix_open_water_details", _mappix_open_water_details_name
    )
    mappix_open_water_details: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_open_water_details_name)),
        allow_mutation=False,
    )
    _validate_mappix_exceedance_time_reference_levels = _validator_mappix_value(
        "mappix_exceedance_time_reference_levels",
        _mappix_exceedance_time_reference_levels_name,
    )
    mappix_exceedance_time_reference_levels: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_exceedance_time_reference_levels_name)),
        allow_mutation=False,
    )
    _validate_mappix_flow_rates_over_structures = _validator_mappix_value(
        "mappix_flow_rates_over_structures", _mappix_flow_rates_over_structures_name
    )
    mappix_flow_rates_over_structures: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_flow_rates_over_structures_name)),
        allow_mutation=False,
    )
    _validate_mappix_flow_rates_to_edge = _validator_mappix_value(
        "mappix_flow_rates_to_edge", _mappix_flow_rates_to_edge_name
    )
    mappix_flow_rates_to_edge: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_flow_rates_to_edge_name)),
        allow_mutation=False,
    )
    _validate_mappix_pluvius_max_sewage_storage = _validator_mappix_value(
        "mappix_pluvius_max_sewage_storage", _mappix_pluvius_max_sewage_storage_name
    )
    mappix_pluvius_max_sewage_storage: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_pluvius_max_sewage_storage_name)),
        allow_mutation=False,
    )
    _validate_mappix_pluvius_max_flow_rates = _validator_mappix_value(
        "mappix_pluvius_max_flow_rates", _mappix_pluvius_max_flow_rates_name
    )
    mappix_pluvius_max_flow_rates: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_pluvius_max_flow_rates_name)),
        allow_mutation=False,
    )
    _validate_mappix_balance = _validator_mappix_value(
        "mappix_balance", _mappix_balance_name
    )
    mappix_balance: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_balance_name)),
        allow_mutation=False,
    )
    _validate_mappix_cumulative_balance = _validator_mappix_value(
        "mappix_cumulative_balance", _mappix_cumulative_balance_name
    )
    mappix_cumulative_balance: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_cumulative_balance_name)),
        allow_mutation=False,
    )
    _validate_mappix_salt_concentrations = _validator_mappix_value(
        "mappix_salt_concentrations", _mappix_salt_concentrations_name
    )
    mappix_salt_concentrations: ImmutableDiskOnlyFileModel = Field(
        ImmutableDiskOnlyFileModel(Path(_mappix_salt_concentrations_name)),
        allow_mutation=False,
    )

    industry_tables: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("industry.tbl"))
    )
    maalstop: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("rtc_3b.his"))
    )
    time_series_temperature: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("default.tmp"))
    )
    time_series_runoff: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("rnff.#"))
    )
    discharges_totals_at_edge_nodes: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("bndfltot.his"))
    )
    language_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sobek_3b.lng"))
    )
    ow_volume: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("ow_vol.his"))
    )
    ow_levels: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("ow_level.his"))
    )
    balance_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3b_bal.out"))
    )
    his_3B_area_length: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3bareas.his"))
    )
    his_3B_structure_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3bstrdim.his"))
    )
    his_rr_runoff: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("rrrunoff.his"))
    )
    his_sacramento: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sacrmnto.his"))
    )
    his_rwzi: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("wwtpdt.his"))
    )
    his_industry: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("industdt.his"))
    )
    ctrl_ini: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("ctrl.ini"))
    )
    root_capsim_input_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("root_sim.inp"))
    )
    unsa_capsim_input_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("unsa_sim.inp"))
    )
    capsim_message_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("capsim.msg"))
    )
    capsim_debug_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("capsim.dbg"))
    )
    restart_1_hour: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("restart1.out"))
    )
    restart_12_hours: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("restart12.out"))
    )
    rr_ready: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("RR-ready"))
    )
    nwrw_areas: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("NwrwArea.His"))
    )
    link_flows: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3blinks.his"))
    )
    modflow_rr: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("modflow_rr.His"))
    )
    rr_modflow: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("rr_modflow.His"))
    )
    rr_wlm_balance: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("rr_wlmbal.His"))
    )
    sacramento_ascii_output: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sacrmnto.out"))
    )
    nwrw_input_dwa_table: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("pluvius.tbl"))
    )
    rr_balance: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("rrbalans.his"))
    )
    green_house_classes: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("KasKlasData.dat"))
    )
    green_house_init: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("KasInitData.dat"))
    )
    green_house_usage: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("KasGebrData.dat"))
    )
    crop_factor: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("CropData.dat"))
    )
    crop_ow: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("CropOWData.dat"))
    )
    soil_data: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("SoilData.dat"))
    )
    dio_config_ini_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("dioconfig.ini"))
    )
    bui_file_for_continuous_calculation_series: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("NWRWCONT.#"))
    )
    nwrw_output: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("NwrwSys.His"))
    )
    rr_routing_link_definitions: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3b_rout.3b"))
    )
    cell_input_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3b_cel.3b"))
    )
    cell_output_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("3b_cel.his"))
    )
    rr_simulate_log_file: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("sobek3b_progress.txt"))
    )
    rtc_coupling_wq_salt: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("wqrtc.his"))
    )
    rr_boundary_conditions_sobek3: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(Path("BoundaryConditions.bc"))
    )

    rr_ascii_restart_openda: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(filepath=None)
    )
    lgsi_cachefile: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(filepath=None)
    )
    meteo_input_file_rainfall: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(filepath=None)
    )
    meteo_input_file_evaporation: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(filepath=None)
    )
    meteo_input_file_temperature: DiskOnlyFileModel = Field(
        default_factory=lambda: DiskOnlyFileModel(filepath=None)
    )

    @classmethod
    def property_keys(cls) -> Iterable[str]:
        # Skip first two elements corresponding with file_path and serializer_config introduced by the FileModel.
        return list(cls.schema()["properties"].keys())[2:]

    @classmethod
    def _ext(cls) -> str:
        return ".fnm"

    @classmethod
    def _filename(cls) -> str:
        return "rr"

    @classmethod
    def _get_parser(cls) -> Callable[[FilePath], Dict]:
        return lambda path: read(cls.property_keys(), path)

    @classmethod
    def _get_serializer(
        cls,
    ) -> Callable[[Path, Dict, SerializerConfig, ModelSaveSettings], None]:
        return write

General layer

rainfall file

BuiModel

Bases: ParsableFileModel

Model that represents the file structure of a .bui file.

Source code in hydrolib/core/rr/meteo/models.py

class BuiModel(ParsableFileModel):
    """
    Model that represents the file structure of a .bui file.
    """

    default_dataset: int = 1  # Default value (always)
    number_of_stations: int
    name_of_stations: List[str]
    number_of_events: int
    seconds_per_timestep: int
    precipitation_events: List[BuiPrecipitationEvent]

    @classmethod
    def _filename(cls):
        return "bui_file"

    @classmethod
    def _ext(cls) -> str:
        return ".bui"

    @classmethod
    def _get_serializer(
        cls,
    ) -> Callable[[Path, Dict, SerializerConfig, ModelSaveSettings], None]:
        return write_bui_file

    @classmethod
    def _get_parser(cls) -> Callable:
        return BuiParser.parse

    def get_station_events(self, station: str) -> Dict[datetime, List[float]]:
        """
        Returns all the events (start time and precipitations) related to a given station.

        Args:
            station (str): Name of the station to retrieve.

        Raises:
            ValueError: If the station name does not exist in the BuiModel.

        Returns:
            Dict[datetime, List[float]]: Dictionary with the start time and its precipitations.
        """
        if station not in self.name_of_stations:
            raise ValueError("Station {} not found BuiModel.".format(station))
        station_idx = self.name_of_stations.index(station)
        station_events = {}
        for event in self.precipitation_events:
            start_time, precipitations = event.get_station_precipitations(station_idx)
            station_events[start_time] = precipitations
        return station_events

get_station_events(station)

Returns all the events (start time and precipitations) related to a given station.

Parameters:

Name	Type	Description	Default
station	str	Name of the station to retrieve.	required

Raises:

Type	Description
ValueError	If the station name does not exist in the BuiModel.

Returns:

Type	Description
Dict[datetime, List[float]]	Dict[datetime, List[float]]: Dictionary with the start time and its precipitations.

Source code in hydrolib/core/rr/meteo/models.py

def get_station_events(self, station: str) -> Dict[datetime, List[float]]:
    """
    Returns all the events (start time and precipitations) related to a given station.

    Args:
        station (str): Name of the station to retrieve.

    Raises:
        ValueError: If the station name does not exist in the BuiModel.

    Returns:
        Dict[datetime, List[float]]: Dictionary with the start time and its precipitations.
    """
    if station not in self.name_of_stations:
        raise ValueError("Station {} not found BuiModel.".format(station))
    station_idx = self.name_of_stations.index(station)
    station_events = {}
    for event in self.precipitation_events:
        start_time, precipitations = event.get_station_precipitations(station_idx)
        station_events[start_time] = precipitations
    return station_events

BuiPrecipitationEvent

Bases: BaseModel

Source code in hydrolib/core/rr/meteo/models.py

class BuiPrecipitationEvent(BaseModel):
    start_time: datetime
    timeseries_length: timedelta
    precipitation_per_timestep: List[List[float]]

    def get_station_precipitations(
        self, station_idx: int
    ) -> Tuple[datetime, List[float]]:
        """
        Returns all the precipitations related to the given station index (column).

        Args:
            station_idx (int): Index of the column which values need to be retrieved.

        Raises:
            ValueError: If the station index does not exist.

        Returns:
            Tuple[datetime, List[float]]: Tuple with the start time and its precipitations.
        """
        number_of_stations = len(self.precipitation_per_timestep[0])
        if station_idx >= number_of_stations:
            raise ValueError(
                "Station index not found, number of stations: {}".format(
                    number_of_stations
                )
            )
        return (
            self.start_time,
            [
                ts_precipitations[station_idx]
                for ts_precipitations in self.precipitation_per_timestep
            ],
        )

get_station_precipitations(station_idx)

Returns all the precipitations related to the given station index (column).

Parameters:

Name	Type	Description	Default
station_idx	int	Index of the column which values need to be retrieved.	required

Raises:

Type	Description
ValueError	If the station index does not exist.

Returns:

Type	Description
Tuple[datetime, List[float]]	Tuple[datetime, List[float]]: Tuple with the start time and its precipitations.

Source code in hydrolib/core/rr/meteo/models.py

def get_station_precipitations(
    self, station_idx: int
) -> Tuple[datetime, List[float]]:
    """
    Returns all the precipitations related to the given station index (column).

    Args:
        station_idx (int): Index of the column which values need to be retrieved.

    Raises:
        ValueError: If the station index does not exist.

    Returns:
        Tuple[datetime, List[float]]: Tuple with the start time and its precipitations.
    """
    number_of_stations = len(self.precipitation_per_timestep[0])
    if station_idx >= number_of_stations:
        raise ValueError(
            "Station index not found, number of stations: {}".format(
                number_of_stations
            )
        )
    return (
        self.start_time,
        [
            ts_precipitations[station_idx]
            for ts_precipitations in self.precipitation_per_timestep
        ],
    )

Topology layer ("Topography" in SOBEK UM)

nodetypes_netter_to_rr = {43: 1, 44: 2, 45: 3, 46: 4, -5: 5, 34: 6, 35: 6, 47: 6, 48: 8, 49: 9, 50: 10, 51: 11, 52: 12, 56: 14, 55: 15, 54: 16, -21: 21, 69: 23} module-attribute

Dictionary with nt mapped against the expected mt.

Some model types mt do not have a related netter type; in that case the dict key is a dummy value of -.

Link

Bases: BaseModel

Represents a link from the topology link file.

Source code in hydrolib/core/rr/topology/models.py

class Link(BaseModel):
    """Represents a link from the topology link file."""

    id: str = Field(alias="id")
    name: Optional[str] = Field(alias="nm")
    branchid: int = Field(alias="ri")
    modellinktype: int = Field(alias="mt")
    branchtype: int = Field(alias="bt")
    objectid: str = Field(alias="ObID")
    beginnode: str = Field(alias="bn")
    endnode: str = Field(alias="en")

    def _get_identifier(self, data: dict) -> Optional[str]:
        return data.get("id") or data.get("nm")

    def dict(self, *args, **kwargs):
        kwargs["by_alias"] = True
        return super().dict(*args, **kwargs)

LinkFile

Bases: ParsableFileModel

Represents the file with the RR link topology data.

Source code in hydrolib/core/rr/topology/models.py

class LinkFile(ParsableFileModel):
    """Represents the file with the RR link topology data."""

    _parser = NetworkTopologyFileParser(enclosing_tag="brch")
    link: List[Link] = Field([], alias="brch")

    @classmethod
    def _ext(cls) -> str:
        return ".tp"

    @classmethod
    def _filename(cls) -> str:
        return "3b_link"

    @classmethod
    def _get_serializer(
        cls,
    ) -> Callable[[Path, Dict, SerializerConfig, ModelSaveSettings], None]:
        return LinkFileSerializer.serialize

    @classmethod
    def _get_parser(cls) -> Callable:
        return cls._parser.parse

Node

Bases: BaseModel

Represents a node from the topology node file.

Source code in hydrolib/core/rr/topology/models.py

class Node(BaseModel):
    """Represents a node from the topology node file."""

    id: str = Field(alias="id")
    name: Optional[str] = Field(alias="nm")
    branchid: int = Field(alias="ri")
    modelnodetype: int = Field(alias="mt")
    netternodetype: int = Field(alias="nt")
    objectid: str = Field(alias="ObID")
    xposition: float = Field(alias="px")
    yposition: float = Field(alias="py")

    def _get_identifier(self, data: dict) -> Optional[str]:
        return data.get("id") or data.get("nm")

    def dict(self, *args, **kwargs):
        kwargs["by_alias"] = True
        return super().dict(*args, **kwargs)

    @root_validator()
    @classmethod
    def _validate_node_type(cls, values):

        cls._raise_if_invalid_type(
            values,
            "modelnodetype",
            set(nodetypes_netter_to_rr.values()),
            "model node type (mt)",
        )

        modelnodetype = values.get("modelnodetype")

        # modelnodetype=6 ("boundary node") is a special case that allows various netter nodetypes,
        # so therefore it always validates well.
        if modelnodetype == 6:
            return values

        cls._raise_if_invalid_type(
            values,
            "netternodetype",
            set(nodetypes_netter_to_rr.keys()),
            "netter node type (nt)",
        )

        netternodetype = values.get("netternodetype")
        modelnodetype_expected = nodetypes_netter_to_rr[netternodetype]

        if modelnodetype != modelnodetype_expected:
            raise ValueError(
                f"{modelnodetype} is not a supported model node type (mt) when netter node type (nt) is {netternodetype}. Supported value: {modelnodetype_expected}."
            )

        return values

    @classmethod
    def _raise_if_invalid_type(
        cls, values, field_name: str, supported_values: set, description: str
    ):
        """Validates the node type for the provided `field_name`.
        The specified node type should contain a supported value,
        otherwise a `ValueError` is raised.


        Args:
            values ([type]): Dictionary with values that are used to create this `Node`.
            field_name (str): Field name of the node type to validate.
            supported_values (set): Set of all the supported values for this node type.
            description (str): Description of this node type that will be readable for the user.

        Raises:
            ValueError: Thrown when `supported_values` does node contain the node type.
        """
        field_value = values.get(field_name)

        if field_value not in supported_values:
            str_supported_values = ", ".join([str(t) for t in supported_values])
            raise ValueError(
                f"{field_value} is not a supported {description}. Supported values: {str_supported_values}."
            )

NodeFile

Bases: ParsableFileModel

Represents the file with the RR node topology data.

Source code in hydrolib/core/rr/topology/models.py

class NodeFile(ParsableFileModel):
    """Represents the file with the RR node topology data."""

    _parser = NetworkTopologyFileParser(enclosing_tag="node")
    node: List[Node] = Field([], alias="node")

    @classmethod
    def _ext(cls) -> str:
        return ".tp"

    @classmethod
    def _filename(cls) -> str:
        return "3b_nod"

    @classmethod
    def _get_serializer(
        cls,
    ) -> Callable[[Path, Dict, SerializerConfig, ModelSaveSettings], None]:
        return NodeFileSerializer.serialize

    @classmethod
    def _get_parser(cls) -> Callable:
        return cls._parser.parse

(Other input layers foreseen in future releases)