Skip to content

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

Bases: ParsableFileModel

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

Source code in hydrolib/core/rr/meteo/models.py 
 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
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 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
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
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 
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
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

Source code in hydrolib/core/rr/meteo/parser.py 
 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
class 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
    """

    @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

parse(raw_text, n_events, timestep) 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[Dict]: List containing all the events represented as dictionaries.
Source code in hydrolib/core/rr/meteo/parser.py 
 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
@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.

Source code in hydrolib/core/rr/meteo/parser.py 
 6
 7
 8
 9
10
11
12
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
class 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.
    """

    @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:])
            ),
        )

    @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:])),
        )

parse(raw_text) 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:

Name Type Description
Dict Dict

Mapped contents of the text.
Source code in hydrolib/core/rr/meteo/parser.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@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) 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:

Name Type Description
Dict Dict

Resulting dictionary with keys start_time and timeseries_length.
Source code in hydrolib/core/rr/meteo/parser.py 
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
@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.

Source code in hydrolib/core/rr/meteo/parser.py 
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
159
160
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
class 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.
    """

    @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(lines: List[str]) -> List[str]:
            return [s_id.strip("'\"") for s_id in lines]

        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("*")
        ]
        number_of_stations = int(bui_lines[1])
        last_station_line = 1 + number_of_stations

        n_events, timestep = parse_events_and_timestep(bui_lines[last_station_line + 1])

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

parse(filepath) 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:

Name Type Description
Dict Dict

Parsed values.
Source code in hydrolib/core/rr/meteo/parser.py 
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
178
179
180
181
182
183
184
185
186
187
188
@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(lines: List[str]) -> List[str]:
        return [s_id.strip("'\"") for s_id in lines]

    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("*")
    ]
    number_of_stations = int(bui_lines[1])
    last_station_line = 1 + number_of_stations

    n_events, timestep = parse_events_and_timestep(bui_lines[last_station_line + 1])

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

Serializer

BuiEventSerializer

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

Source code in hydrolib/core/rr/meteo/serializer.py 
  9
 10
 11
 12
 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
class BuiEventSerializer:
    """
    Serializer class to transform a bui event into a text block.
    """

    bui_event_template = inspect.cleandoc(
        """
        * Event {event_idx} duration days:{d_days} hours:{d_hours} minutes:{d_minutes} seconds:{d_seconds}
        * Start date and time of the event: yyyy mm dd hh mm ss
        * Duration of the event           : dd hh mm ss
        * Rainfall value per time step [mm/time step]
        {start_time} {timeseries_length}
        {precipitation_per_timestep}
    """
    )

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

        Args:
            event_data (Dict): Dictionary representing precipitation event.
            config (SerializerConfig): The serialization configuration.

        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"], config
            )
        )
        if "event_idx" not in event_data.keys():
            event_data["event_idx"] = 1
        return BuiEventSerializer.bui_event_template.format(**event_data)

    @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,
        )

    @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}"

    @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}"

    @staticmethod
    def serialize_precipitation_per_timestep(
        data_to_serialize: List[List[float]], config: SerializerConfig
    ) -> 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.
            config (SerializerConfig): The serialization configuration.

        Returns:
            str: Serialized string in .bui format.
        """
        float_format = lambda v: f"{v:{config.float_format}}"
        serialized_data = str.join(
            "\n",
            [
                str.join(" ", map(float_format, listed_data))
                for listed_data in data_to_serialize
            ],
        )
        return serialized_data

get_timedelta_fields(duration) 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:

Name Type Description
Dict Dict

Dictionary containing all fields.
Source code in hydrolib/core/rr/meteo/serializer.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@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, config) staticmethod

Serializes a dictionary representing an event into a text block.

Parameters:

Name Type Description Default
event_data Dict

Dictionary representing precipitation event.

 required
config SerializerConfig

The serialization configuration.

 required

Returns:

Name Type Description
str str

Formatted string.
Source code in hydrolib/core/rr/meteo/serializer.py 
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
@staticmethod
def serialize(event_data: Dict, config: SerializerConfig) -> str:
    """
    Serializes a dictionary representing an event into a text block.

    Args:
        event_data (Dict): Dictionary representing precipitation event.
        config (SerializerConfig): The serialization configuration.

    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"], config
        )
    )
    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, config) 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
config SerializerConfig

The serialization configuration.

 required

Returns:

Name Type Description
str str

Serialized string in .bui format.
Source code in hydrolib/core/rr/meteo/serializer.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@staticmethod
def serialize_precipitation_per_timestep(
    data_to_serialize: List[List[float]], config: SerializerConfig
) -> 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.
        config (SerializerConfig): The serialization configuration.

    Returns:
        str: Serialized string in .bui format.
    """
    float_format = lambda v: f"{v:{config.float_format}}"
    serialized_data = str.join(
        "\n",
        [
            str.join(" ", map(float_format, listed_data))
            for listed_data in data_to_serialize
        ],
    )
    return serialized_data

serialize_start_time(data_to_serialize) staticmethod

Serializes a datetime into the expected .bui format.

Parameters:

Name Type Description Default
data_to_serialize datetime

Datetime representing reference time.

 required

Returns:

Name Type Description
str str

Converted datetime into string.
Source code in hydrolib/core/rr/meteo/serializer.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@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) staticmethod

Serializes a given timedelta into the .bui format.

Parameters:

Name Type Description Default
data_to_serialize timedelta

Reference timespan to serialize.

 required

Returns:

Name Type Description
str str

Converted timedelta in string.
Source code in hydrolib/core/rr/meteo/serializer.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@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.

Source code in hydrolib/core/rr/meteo/serializer.py 
144
145
146
147
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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class BuiSerializer:
    """
    Serializer class to transform an object into a .bui file text format.
    """

    bui_template = inspect.cleandoc(
        """
        *Name of this file: {filepath}
        *Date and time of construction: {datetime_now}
        *Comments are following an * (asterisk) and written above variables
        {default_dataset}
        *Number of stations
        {number_of_stations}
        *Station Name
        {name_of_stations}
        *Number_of_events seconds_per_timestamp
        {number_of_events} {seconds_per_timestep}
        {precipitation_events}
        """
    )

    @staticmethod
    def serialize(bui_data: Dict, config: SerializerConfig) -> 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.
            config (SerializerConfig): The serialization configuration.

        Returns:
            str: The serialized data.
        """
        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"], config
        )
        return BuiSerializer.bui_template.format(**bui_data)

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

        Args:
            data_to_serialize (Dict): Dictionary containing list of events.
            config (SerializerConfig): The serialization configuration.

        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, config))
        return "\n".join(serialized_list)

    @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 "\n".join(f"'{station_id}'" for station_id in data_to_serialize)

serialize(bui_data, config) 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
config SerializerConfig

The serialization configuration.

 required

Returns:

Name Type Description
str str

The serialized data.
Source code in hydrolib/core/rr/meteo/serializer.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@staticmethod
def serialize(bui_data: Dict, config: SerializerConfig) -> 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.
        config (SerializerConfig): The serialization configuration.

    Returns:
        str: The serialized data.
    """
    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"], config
    )
    return BuiSerializer.bui_template.format(**bui_data)

serialize_event_list(data_to_serialize, config) 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
config SerializerConfig

The serialization configuration.

 required

Returns:

Name Type Description
str str

Text block representing all precipitation events.
Source code in hydrolib/core/rr/meteo/serializer.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@staticmethod
def serialize_event_list(
    data_to_serialize: List[Dict], config: SerializerConfig
) -> str:
    """
    Serializes a event list dictionary into a single text block.

    Args:
        data_to_serialize (Dict): Dictionary containing list of events.
        config (SerializerConfig): The serialization configuration.

    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, config))
    return "\n".join(serialized_list)

serialize_stations_ids(data_to_serialize) 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:

Name Type Description
str str

Serialized string.
Source code in hydrolib/core/rr/meteo/serializer.py 
208
209
210
211
212
213
214
215
216
217
218
219
@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 "\n".join(f"'{station_id}'" for station_id in data_to_serialize)

write_bui_file(path, data, config, save_settings)

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
config SerializerConfig

The serialization configuration.

 required
save_settings ModelSaveSettings

The model save settings.

 required
Source code in hydrolib/core/rr/meteo/serializer.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def write_bui_file(
    path: Path, data: Dict, config: SerializerConfig, save_settings: ModelSaveSettings
) -> 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.
        config (SerializerConfig): The serialization configuration.
        save_settings (ModelSaveSettings): The model save settings.
    """
    data["filepath"] = path  # This is redundant as already exists in the data.
    serialized_bui_data = BuiSerializer.serialize(data, config)

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