The Rainfall Runoff meteo layer¶

The meteo layer currently contains only support for the BUI file. The "bui" file contains the precipitation input data for Rainfall Runoff. It is represented by the classes below.

Model¶

`BuiModel (ParsableFileModel)` `pydantic-model` ¶

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

`get_station_events(self, station: str) -> Dict[datetime.datetime, List[float]]` ¶

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

Exceptions:

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

Returns:

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

Source code in hydrolib/core/io/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 (BaseModel)` `pydantic-model` ¶

`get_station_precipitations(self, station_idx: int) -> Tuple[datetime.datetime, List[float]]` ¶

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

Exceptions:

Type	Description
`ValueError`	If the station index does not exist.

Returns:

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

Source code in hydrolib/core/io/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
        ],
    )

Parser¶

`BuiEventListParser` ¶

A parser for .bui events which are like this: StartTime (YYYY mm dd HH MM SS) TimeSeriesLength (dd HH MM SS) PrecipitationPerTimestep StartTime (YYYY mm dd HH MM SS) TimeSeriesLength (dd HH MM SS) PrecipitationPerTimestep Example given: 2021 12 20 0 0 0 1 0 4 20 4.2 4.2 4.2 2021 12 21 0 0 0 1 0 4 20 2.4 2.4 2.4

`parse(raw_text: str, n_events: int, timestep: int) -> List[Dict]` `staticmethod` ¶

Parses a given raw text containing 0 to many text blocks representing a precipitation event.

Parameters:

Name	Type	Description	Default
`raw_text`	`str`	Text blocks representing precipitation events.	required
`n_events`	`int`	Number of events contained in the text block.	required
`timestep`	`int`	Number of seconds conforming a timestep.	required

Returns:

Type	Description
`List[Dict]`	List containing all the events represented as dictionaries.

Source code in hydrolib/core/io/rr/meteo/parser.py

@staticmethod
def parse(raw_text: str, n_events: int, timestep: int) -> List[Dict]:
    """
    Parses a given raw text containing 0 to many text blocks representing a precipitation event.

    Args:
        raw_text (str): Text blocks representing precipitation events.
        n_events (int): Number of events contained in the text block.
        timestep (int): Number of seconds conforming a timestep.

    Returns:
        List[Dict]: List containing all the events represented as dictionaries.
    """

    def get_event_timestep_length(raw_line: str) -> int:
        timereference = BuiEventParser.parse_event_time_reference(raw_line)
        ts_length: timedelta = timereference["timeseries_length"]
        return ts_length.total_seconds()

    def get_multiple_events(raw_lines: List[str]) -> Iterator[BuiEventParser]:
        n_line = 0
        while n_line < len(raw_lines):
            ts_seconds = get_event_timestep_length(raw_lines[n_line])
            event_lines = int(ts_seconds / timestep) + 1
            yield BuiEventParser.parse("\n".join(raw_lines[n_line:][:event_lines]))
            n_line += event_lines

    event_list = []
    if n_events == 1:
        event_list.append(BuiEventParser.parse(raw_text))
    elif n_events > 1:
        raw_lines = raw_text.splitlines(keepends=False)
        event_list = list(get_multiple_events(raw_lines))

    return event_list

`BuiEventParser` ¶

A parser for the precipitation event section within a .bui file. It resembles something like this: StartTime (YYYY mm dd HH MM SS) TimeSeriesLength (dd HH MM SS) PrecipitationPerTimestep Example given: 2021 12 20 0 0 0 1 0 4 20 4.2 2.4 4.2 2.4 4.2 2.4 (it should match the timeseries length based on the seconds per timstep.) Each column of the last three lines represents a station.

`parse(raw_text: str) -> Dict` `staticmethod` ¶

Given text representing a single BuiPrecipitationEvent parses it into a dictionary.

Parameters:

Name	Type	Description	Default
`raw_text`	`str`	Text containing a single precipitation event.	required

Returns:

Type	Description
`Dict`	Mapped contents of the text.

Source code in hydrolib/core/io/rr/meteo/parser.py

@staticmethod
def parse(raw_text: str) -> Dict:
    """
    Given text representing a single BuiPrecipitationEvent parses it into a dictionary.

    Args:
        raw_text (str): Text containing a single precipitation event.

    Returns:
        Dict: Mapped contents of the text.
    """

    def get_precipitations_per_ts(line: str) -> List[str]:
        return [prec for prec in line.split()]

    event_lines = raw_text.splitlines(keepends=False)
    time_reference = BuiEventParser.parse_event_time_reference(event_lines[0])
    return dict(
        start_time=time_reference["start_time"],
        timeseries_length=time_reference["timeseries_length"],
        precipitation_per_timestep=list(
            map(get_precipitations_per_ts, event_lines[1:])
        ),
    )

`parse_event_time_reference(raw_text: str) -> Dict` `staticmethod` ¶

Parses a single event time reference line containing both the start time and the timeseries length into a dictionary.

Parameters:

Name	Type	Description	Default
`raw_text`	`str`	Line representing both start time and timeseries length.	required

Returns:

Type	Description
`Dict`	Resulting dictionary with keys start_time and timeseries_length.

Source code in hydrolib/core/io/rr/meteo/parser.py

@staticmethod
def parse_event_time_reference(raw_text: str) -> Dict:
    """
    Parses a single event time reference line containing both the start time
    and the timeseries length into a dictionary.

    Args:
        raw_text (str): Line representing both start time and timeseries length.

    Returns:
        Dict: Resulting dictionary with keys start_time and timeseries_length.
    """

    def get_start_time(line: str) -> datetime:
        return datetime.strptime(line, "%Y %m %d %H %M %S")

    def get_timeseries_length(line: str) -> timedelta:
        time_fields = line.split()
        return timedelta(
            days=int(time_fields[0]),
            hours=int(time_fields[1]),
            minutes=int(time_fields[2]),
            seconds=int(time_fields[3]),
        )

    timeref = raw_text.split()
    return dict(
        start_time=get_start_time(" ".join(timeref[:6])),
        timeseries_length=get_timeseries_length(" ".join(timeref[6:])),
    )

`BuiParser` ¶

A parser for .bui files which are like this: * comments Dataset type to use (always 1). * comments Number of stations. * comments Name of stations * comments Number of events Number of seconds per timestep. * comments First datetime reference. Precipitation per timestep per station.

`parse(filepath: Path) -> Dict` `staticmethod` ¶

Parses a given file, in case valid, into a dictionary which can later be mapped to the BuiModel.

Parameters:

Name	Type	Description	Default
`filepath`	`Path`	Path to file containing the data to parse.	required

Returns:

Type	Description
`Dict`	Parsed values.

Source code in hydrolib/core/io/rr/meteo/parser.py

@staticmethod
def parse(filepath: Path) -> Dict:
    """
    Parses a given file, in case valid, into a dictionary which can later be mapped
    to the BuiModel.

    Args:
        filepath (Path): Path to file containing the data to parse.

    Returns:
        Dict: Parsed values.
    """

    def get_station_ids(line: str) -> List[str]:
        return [s_id for s_id in line.split(",")]

    def parse_events_and_timestep(line: str) -> Tuple[int, int]:
        n_events_timestep = line.split()
        return (int(n_events_timestep[0]), int(n_events_timestep[1]))

    bui_lines = [
        line
        for line in filepath.read_text(encoding="utf8").splitlines()
        if not line.startswith("*")
    ]

    n_events, timestep = parse_events_and_timestep(bui_lines[3])

    return dict(
        default_dataset=bui_lines[0],
        number_of_stations=bui_lines[1],
        name_of_stations=get_station_ids(bui_lines[2]),
        number_of_events=n_events,
        seconds_per_timestep=timestep,
        precipitation_events=BuiEventListParser.parse(
            "\n".join(bui_lines[4:]), n_events, timestep
        ),
    )

Serializer¶

`BuiEventSerializer` ¶

Serializer class to transform a bui event into a text block.

`get_timedelta_fields(duration: timedelta) -> Dict` `staticmethod` ¶

Gets a dictionary containing the time delta in days, hours, minutes and seconds. This means that the seconds field does not contain the accumulative value of days hours and minutes.

Parameters:

Name	Type	Description	Default
`duration`	`timedelta`	Timedelta to convert.	required

Returns:

Type	Description
`Dict`	Dictionary containing all fields.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def get_timedelta_fields(duration: timedelta) -> Dict:
    """
    Gets a dictionary containing the time delta in days, hours, minutes and seconds.
    This means that the seconds field does not contain the accumulative value of days
    hours and minutes.

    Args:
        duration (timedelta): Timedelta to convert.

    Returns:
        Dict: Dictionary containing all fields.
    """
    total_hours = int(duration.seconds / (60 * 60))
    total_minutes = int((duration.seconds / 60) - (total_hours * 60))
    total_seconds = int(
        duration.seconds - ((total_hours * 60 + total_minutes) * 60)
    )
    return dict(
        d_seconds=total_seconds,
        d_minutes=total_minutes,
        d_hours=total_hours,
        d_days=duration.days,
    )

`serialize(event_data: Dict) -> str` `staticmethod` ¶

Serializes a dictionary representing an event into a text block.

Parameters:

Name	Type	Description	Default
`event_data`	`Dict`	Dictionary representing precipitation event.	required

Returns:

Type	Description
`str`	Formatted string.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize(event_data: Dict) -> str:
    """
    Serializes a dictionary representing an event into a text block.

    Args:
        event_data (Dict): Dictionary representing precipitation event.

    Returns:
        str: Formatted string.
    """
    event_data["start_time"] = BuiEventSerializer.serialize_start_time(
        event_data["start_time"]
    )
    ts_duration = event_data["timeseries_length"]
    event_data = {
        **event_data,
        **BuiEventSerializer.get_timedelta_fields(ts_duration),
    }
    event_data[
        "timeseries_length"
    ] = BuiEventSerializer.serialize_timeseries_length(
        event_data["timeseries_length"]
    )
    event_data[
        "precipitation_per_timestep"
    ] = BuiEventSerializer.serialize_precipitation_per_timestep(
        event_data["precipitation_per_timestep"]
    )
    if "event_idx" not in event_data.keys():
        event_data["event_idx"] = 1
    return BuiEventSerializer.bui_event_template.format(**event_data)

`serialize_precipitation_per_timestep(data_to_serialize: List[List[str]]) -> str` `staticmethod` ¶

Serialized the data containing all the precipitations per timestep (and station) into a single string ready to be mapped.

Parameters:

Name	Type	Description	Default
`data_to_serialize`	`List[List[str]]`	Data to be mapped.	required

Returns:

Type	Description
`str`	Serialized string in .bui format.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize_precipitation_per_timestep(data_to_serialize: List[List[str]]) -> str:
    """
    Serialized the data containing all the precipitations per timestep (and station)
    into a single string ready to be mapped.

    Args:
        data_to_serialize (List[List[str]]): Data to be mapped.

    Returns:
        str: Serialized string in .bui format.
    """
    serialized_data = str.join(
        "\n",
        [str.join(" ", map(str, listed_data)) for listed_data in data_to_serialize],
    )
    return serialized_data

`serialize_start_time(data_to_serialize: datetime) -> str` `staticmethod` ¶

Serializes a datetime into the expected .bui format.

Parameters:

Name	Type	Description	Default
`data_to_serialize`	`datetime`	Datetime representing reference time.	required

Returns:

Type	Description
`str`	Converted datetime into string.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize_start_time(data_to_serialize: datetime) -> str:
    """
    Serializes a datetime into the expected .bui format.

    Args:
        data_to_serialize (datetime): Datetime representing reference time.

    Returns:
        str: Converted datetime into string.
    """
    # Not using the following format because we only want one digit instead of
    # double (day 1 -> 1, instead of 01).
    # data_to_serialize.strftime("%Y %m %d %H %M %S")
    dt = data_to_serialize
    return f"{dt.year} {dt.month} {dt.day} {dt.hour} {dt.minute} {dt.second}"

`serialize_timeseries_length(data_to_serialize: timedelta) -> str` `staticmethod` ¶

Serializes a given timedelta into the .bui format.

Parameters:

Name	Type	Description	Default
`data_to_serialize`	`timedelta`	Reference timespan to serialize.	required

Returns:

Type	Description
`str`	Converted timedelta in string.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize_timeseries_length(data_to_serialize: timedelta) -> str:
    """
    Serializes a given timedelta into the .bui format.

    Args:
        data_to_serialize (timedelta): Reference timespan to serialize.

    Returns:
        str: Converted timedelta in string.
    """
    fields_dict = BuiEventSerializer.get_timedelta_fields(data_to_serialize)
    total_hours = fields_dict["d_hours"]
    total_minutes = fields_dict["d_minutes"]
    total_seconds = fields_dict["d_seconds"]
    return f"{data_to_serialize.days} {total_hours} {total_minutes} {total_seconds}"

`BuiSerializer` ¶

Serializer class to transform an object into a .bui file text format.

`serialize(bui_data: Dict) -> str` `staticmethod` ¶

Formats the bui_template with the content of the given data. NOTE: It requires that caller injects file_path into bui_data prior to this call. Otherwise it will crash.

Parameters:

Name	Type	Description	Default
`bui_data`	`Dict`	Data to serialize.	required

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize(bui_data: Dict) -> str:
    """
    Formats the bui_template with the content of the given data.
    NOTE: It requires that caller injects file_path into bui_data prior to this call.
    Otherwise it will crash.

    Args:
        bui_data (Dict): Data to serialize.
    """
    bui_data["datetime_now"] = datetime.now().strftime("%d-%m-%y %H:%M:%S")
    bui_data["name_of_stations"] = BuiSerializer.serialize_stations_ids(
        bui_data["name_of_stations"]
    )
    bui_data["precipitation_events"] = BuiSerializer.serialize_event_list(
        bui_data["precipitation_events"]
    )
    return BuiSerializer.bui_template.format(**bui_data)

`serialize_event_list(data_to_serialize: List[Dict]) -> str` `staticmethod` ¶

Serializes a event list dictionary into a single text block.

Parameters:

Name	Type	Description	Default
`data_to_serialize`	`Dict`	Dictionary containing list of events.	required

Returns:

Type	Description
`str`	Text block representing all precipitation events.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize_event_list(data_to_serialize: List[Dict]) -> str:
    """
    Serializes a event list dictionary into a single text block.

    Args:
        data_to_serialize (Dict): Dictionary containing list of events.

    Returns:
        str: Text block representing all precipitation events.
    """
    serialized_list = []
    for n_event, event in enumerate(data_to_serialize):
        event["event_idx"] = n_event + 1
        serialized_list.append(BuiEventSerializer.serialize(event))
    return "\n".join(serialized_list)

`serialize_stations_ids(data_to_serialize: List[str]) -> str` `staticmethod` ¶

Serializes the stations ids into a single string as expected in a .bui file.

Parameters:

Name	Type	Description	Default
`data_to_serialize`	`List[str]`	List of station ids.	required

Returns:

Type	Description
`str`	Serialized string.

Source code in hydrolib/core/io/rr/meteo/serializer.py

@staticmethod
def serialize_stations_ids(data_to_serialize: List[str]) -> str:
    """
    Serializes the stations ids into a single string as expected in a .bui file.

    Args:
        data_to_serialize (List[str]): List of station ids.

    Returns:
        str: Serialized string.
    """
    return str.join(" ", data_to_serialize)

`write_bui_file(path: Path, data: Dict) -> None` ¶

Writes a .bui file in the given path based on the data given in a dictionary.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path where to output the text.	required
`data`	`Dict`	Data to serialize into the file.	required

Source code in hydrolib/core/io/rr/meteo/serializer.py

def write_bui_file(path: Path, data: Dict) -> None:
    """
    Writes a .bui file in the given path based on the data given in a dictionary.

    Args:
        path (Path): Path where to output the text.
        data (Dict): Data to serialize into the file.
    """
    data["filepath"] = path  # This is redundant as already exists in the data.
    serialized_bui_data = BuiSerializer.serialize(data)

    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(serialized_bui_data, encoding="utf8")

The Rainfall Runoff meteo layer¶

Model¶

BuiModel (ParsableFileModel) pydantic-model ¶

get_station_events(self, station: str) -> Dict[datetime.datetime, List[float]] ¶

BuiPrecipitationEvent (BaseModel) pydantic-model ¶

get_station_precipitations(self, station_idx: int) -> Tuple[datetime.datetime, List[float]] ¶

Parser¶

BuiEventListParser ¶

parse(raw_text: str, n_events: int, timestep: int) -> List[Dict] staticmethod ¶

BuiEventParser ¶

parse(raw_text: str) -> Dict staticmethod ¶

parse_event_time_reference(raw_text: str) -> Dict staticmethod ¶

BuiParser ¶

parse(filepath: Path) -> Dict staticmethod ¶

Serializer¶

BuiEventSerializer ¶

get_timedelta_fields(duration: timedelta) -> Dict staticmethod ¶

serialize(event_data: Dict) -> str staticmethod ¶

serialize_precipitation_per_timestep(data_to_serialize: List[List[str]]) -> str staticmethod ¶

serialize_start_time(data_to_serialize: datetime) -> str staticmethod ¶

serialize_timeseries_length(data_to_serialize: timedelta) -> str staticmethod ¶

BuiSerializer ¶

serialize(bui_data: Dict) -> str staticmethod ¶

serialize_event_list(data_to_serialize: List[Dict]) -> str staticmethod ¶

serialize_stations_ids(data_to_serialize: List[str]) -> str staticmethod ¶

write_bui_file(path: Path, data: Dict) -> None ¶

`BuiModel (ParsableFileModel)` `pydantic-model` ¶

`get_station_events(self, station: str) -> Dict[datetime.datetime, List[float]]` ¶

`BuiPrecipitationEvent (BaseModel)` `pydantic-model` ¶

`get_station_precipitations(self, station_idx: int) -> Tuple[datetime.datetime, List[float]]` ¶

`BuiEventListParser` ¶

`parse(raw_text: str, n_events: int, timestep: int) -> List[Dict]` `staticmethod` ¶

`BuiEventParser` ¶

`parse(raw_text: str) -> Dict` `staticmethod` ¶

`parse_event_time_reference(raw_text: str) -> Dict` `staticmethod` ¶

`BuiParser` ¶

`parse(filepath: Path) -> Dict` `staticmethod` ¶

`BuiEventSerializer` ¶

`get_timedelta_fields(duration: timedelta) -> Dict` `staticmethod` ¶

`serialize(event_data: Dict) -> str` `staticmethod` ¶

`serialize_precipitation_per_timestep(data_to_serialize: List[List[str]]) -> str` `staticmethod` ¶

`serialize_start_time(data_to_serialize: datetime) -> str` `staticmethod` ¶

`serialize_timeseries_length(data_to_serialize: timedelta) -> str` `staticmethod` ¶

`BuiSerializer` ¶

`serialize(bui_data: Dict) -> str` `staticmethod` ¶

`serialize_event_list(data_to_serialize: List[Dict]) -> str` `staticmethod` ¶

`serialize_stations_ids(data_to_serialize: List[str]) -> str` `staticmethod` ¶

`write_bui_file(path: Path, data: Dict) -> None` ¶