BUI file¶

The "bui" file contains the precipitation input data for Rainfall Runoff. It is represented by the classes below.

Model¶

`BuiModel (FileModel)` `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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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/bui/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")

BUI file¶

Model¶

BuiModel (FileModel) 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 (FileModel)` `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` ¶