+import math
+import warnings
+from pathlib import Path
+import webbrowser
+
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+import scipy.optimize as spo
+from matplotlib.backends.backend_pdf import PdfPages
+
+from .arg_parser import heavy_rain_parser
+from .definitions import *
+from .idf_backend import IdfParameters
+from .in_out import import_series
+from .little_helpers import minutes_readable, height2rate, delta2min, rate2height, frame_looper, event_caption, \
+ event_caption_ger, duration_steps_readable, get_progress_bar
+from .plot_helpers import idf_bar_axes
+from .sww_utils import guess_freq, rain_events, agg_events, event_duration, resample_rain_series, rain_bar_plot, IdfError
+from .synthetic_rainseries import _BlockRain, _EulerRain
+
+
+########################################################################################################################
+
+
[docs]
+
class IntensityDurationFrequencyAnalyse:
+
"""
+
heavy rain as a function of the duration and the return period acc. to DWA-A 531 (2012)
+
+
This program reads the measurement data of the rainfall
+
and calculates the distribution of the rainfall as a function of the return period and the duration
+
+
for duration steps up to 12 hours (and more) and return period in a range of '0.5a <= T_n <= 100a'
+
+
Attributes:
+
_series (pandas.Series): rain time-series
+
_freq (pandas.DateOffset): frequency of the rain series
+
_return_periods_frame (pandas.DataFrame): with return periods of all given durations
+
_rain_events (pandas.DataFrame):
+
_rainfall_sum_frame (pandas.DataFrame): with rain sums of all given durations
+
+
"""
+
def __init__(self, series_kind=SERIES.PARTIAL, worksheet=METHOD.KOSTRA, extended_durations=False):
+
"""
+
Heavy rainfall intensity as a function of duration and return period acc. to DWA-A 531 (2012).
+
+
This program reads the measurement data of the rainfall
+
and calculates the distribution of the rainfall as a function of the return period and the duration
+
+
for duration steps up to 12 hours (and more) and return period in a range of 0.5a and 100a.
+
+
Args:
+
series_kind (str): ['partial', 'annual']
+
worksheet (str): ['DWA-A_531', 'ATV-A_121', 'DWA-A_531_advektiv']
+
extended_durations (bool): add [720, 1080, 1440, 2880, 4320, 5760, 7200, 8640] minutes to the calculation
+
"""
+
self._series = None
+
self._freq = None
+
+
# how to calculate the idf curves
+
self._parameters = IdfParameters(series_kind=series_kind, worksheet=worksheet,
+
extended_durations=extended_durations)
+
+
self._return_periods_frame = None
+
self._rain_events = None
+
self._rainfall_sum_frame = None
+
+
self._duration_steps_for_output = None
+
+
# __________________________________________________________________________________________________________________
+
@property
+
def series(self) -> pd.Series:
+
if self._series is None:
+
raise IdfError('No Series defined for IDF-Analysis!')
+
return self._series
+
+
@series.setter
+
def series(self, series: pd.Series):
+
self._series = series
+
+
+
[docs]
+
def set_series(self, series):
+
"""
+
set the series for the analysis
+
+
Args:
+
series (pandas.Series): precipitation time-series
+
"""
+
if not isinstance(series, pd.Series):
+
raise IdfError('The series has to be a pandas Series.')
+
+
if not isinstance(series.index, pd.DatetimeIndex):
+
raise IdfError('The series has to have a DatetimeIndex.')
+
+
# if series.index.tz is not None:
+
# series = remove_timezone(series)
+
+
self._freq = guess_freq(series.index)
+
freq_minutes = delta2min(self._freq)
+
self._parameters.filter_durations(freq_minutes)
+
self.series = series.replace(0, np.nan).dropna()
+
self._return_periods_frame = None
+
self._rain_events = None
+
self._rainfall_sum_frame = None
+
+
+
# __________________________________________________________________________________________________________________
+
@property
+
def duration_steps(self):
+
"""
+
get duration steps (in minutes) for the parameter calculation and basic evaluations
+
+
Returns:
+
list | numpy.ndarray: duration steps in minutes
+
"""
+
return self.parameters.durations
+
+
@duration_steps.setter
+
def duration_steps(self, durations):
+
"""
+
Set duration steps (in minutes) for the parameter calculation and basic evaluations.
+
+
Args:
+
durations (list | numpy.ndarray): duration steps in minutes
+
"""
+
if not isinstance(durations, (list, np.ndarray)):
+
raise IdfError(f'Duration steps have to be {(list, np.ndarray)} got "{type(durations)}"')
+
self.parameters.durations = durations
+
+
@property
+
def duration_steps_for_output(self):
+
"""
+
get duration steps (in minutes) for the parameter calculation and basic evaluations
+
+
Returns:
+
list | numpy.ndarray: duration steps in minutes
+
"""
+
if self._duration_steps_for_output is None:
+
self._duration_steps_for_output = self.duration_steps.copy()
+
+
return self._duration_steps_for_output
+
+
@duration_steps_for_output.setter
+
def duration_steps_for_output(self, durations):
+
"""
+
Set duration steps (in minutes) for the parameter calculation and basic evaluations.
+
+
Args:
+
durations (list | numpy.ndarray): duration steps in minutes
+
"""
+
if not isinstance(durations, (list, np.ndarray)):
+
raise IdfError(f'Duration steps have to be {(list, np.ndarray)} got "{type(durations)}"')
+
self._duration_steps_for_output = durations
+
+
# __________________________________________________________________________________________________________________
+
@property
+
def parameters(self):
+
"""
+
get the calculation parameters
+
+
calculation method depending on the used worksheet and on the duration
+
also the parameters for each method
+
+
to save some time and save the parameters with
+
:func:`IntensityDurationFrequencyAnalyse.write_parameters`
+
and read them later with :func:`IntensityDurationFrequencyAnalyse.read_parameters`
+
+
Returns:
+
IdfParameters: calculation parameters
+
"""
+
if not self._parameters.parameters_series and self._series is not None:
+
self._parameters.calc_from_series(self.series)
+
return self._parameters
+
+
+
[docs]
+
def write_parameters(self, filename):
+
"""
+
save parameters as yaml-file to save computation time.
+
+
Args:
+
filename (str): filename for the parameters yaml-file
+
"""
+
self.parameters.to_yaml(filename)
+
+
+
+
[docs]
+
def read_parameters(self, filename, worksheet=None):
+
"""
+
Read parameters from a .yaml-file to save computation time.
+
+
Extract interim results from parameters.
+
+
Args:
+
filename (str, Path): filename of the parameters yaml-file
+
"""
+
self._parameters = IdfParameters.from_yaml(filename, worksheet)
+
+
+
+
[docs]
+
def auto_save_parameters(self, filename: str or Path):
+
"""Auto-save the parameters as a yaml-file to save computation time."""
+
if isinstance(filename, str):
+
filename = Path(filename)
+
if filename.is_file():
+
self.read_parameters(filename)
+
else:
+
if not filename.parent.is_dir():
+
filename.parent.mkdir(parents=True)
+
self.write_parameters(filename)
+
+
+
# __________________________________________________________________________________________________________________
+
+
[docs]
+
def depth_of_rainfall(self, duration, return_period):
+
"""
+
calculate the height of the rainfall h in L/m² = mm
+
+
Args:
+
duration (int | float | list | numpy.ndarray | pandas.Series): duration: in minutes
+
return_period (float): in years
+
+
Returns:
+
int | float | list | numpy.ndarray | pandas.Series: height of the rainfall h in L/m² = mm
+
"""
+
if self.parameters.series_kind == SERIES.ANNUAL:
+
if return_period < 5:
+
print('WARNING: Using an annual series and a return period < 5 a will result in faulty values!')
+
+
if return_period <= 10:
+
return_period = np.exp(1.0 / return_period) / (np.exp(1.0 / return_period) - 1.0)
+
+
log_tn = -np.log(np.log(return_period / (return_period - 1.0)))
+
+
else:
+
log_tn = np.log(return_period)
+
+
u, w = self.parameters.get_u_w(duration)
+
return u + w * log_tn
+
+
+
# __________________________________________________________________________________________________________________
+
+
[docs]
+
def rain_flow_rate(self, duration, return_period):
+
"""
+
Convert the height of rainfall to the specific rain flow rate in [l/(s*ha)].
+
+
if 2 array-like parameters are give, an element-wise calculation will be made.
+
So the length of the array must be the same.
+
+
Args:
+
duration (int | float | list | numpy.ndarray | pandas.Series): in minutes
+
return_period (float): in years
+
+
Returns:
+
int | float | list | numpy.ndarray | pandas.Series: specific rain flow rate in [l/(s*ha)]
+
"""
+
height_of_rainfall = self.depth_of_rainfall(duration=duration, return_period=return_period)
+
return height2rate(height_of_rainfall=height_of_rainfall, duration=duration)
+
+
+
# __________________________________________________________________________________________________________________
+
+
[docs]
+
def r_720_1(self):
+
"""
+
rain flow rate in [l/(s*ha)] for a duration of 12h and a return period of 1 year
+
+
Returns:
+
float: rain flow rate in [l/(s*ha)]
+
"""
+
return self.rain_flow_rate(duration=720, return_period=1)
+
+
+
# __________________________________________________________________________________________________________________
+
+
[docs]
+
def get_return_period(self, height_of_rainfall, duration):
+
"""
+
calculate the return period, when the height of rainfall and the duration are given
+
+
Args:
+
height_of_rainfall (float): in [mm]
+
duration (int | float | list | numpy.ndarray | pandas.Series): in minutes
+
+
Returns:
+
int | float | list | numpy.ndarray | pandas.Series: return period in years
+
"""
+
u, w = self.parameters.get_u_w(duration)
+
return np.exp((height_of_rainfall - u) / w)
+
+
+
# __________________________________________________________________________________________________________________
+
+
[docs]
+
def get_duration(self, height_of_rainfall, return_period):
+
"""
+
calculate the duration, when the height of rainfall and the return period are given
+
+
Args:
+
height_of_rainfall (float): in [mm]
+
return_period (float): in years
+
+
Returns:
+
float: duration in minutes
+
"""
+
return spo.newton(lambda d: self.depth_of_rainfall(d, return_period) - height_of_rainfall, x0=1)
+
+
+
# __________________________________________________________________________________________________________________
+
+
[docs]
+
def result_table(self, durations=None, return_periods=None, add_names=False, add_unit=True, as_intensity=False):
+
"""
+
Get an idf-table of rainfall depth with return periods as columns and durations as rows.
+
+
Args:
+
durations (list | numpy.ndarray): list of durations in minutes for the table
+
return_periods (list): list of return periods in years for the table
+
add_names (bool): weather to use expressive names as index- & column-label
+
add_unit (bool): weather to add units to index- & column-label
+
+
Returns:
+
pandas.DataFrame: idf table
+
"""
+
if durations is None:
+
durations = self.duration_steps_for_output
+
+
if return_periods is None:
+
return_periods = [1, 2, 3, 5, 10, 20, 25, 30, 50, 75, 100]
+
+
result_table = {}
+
for t in return_periods:
+
result_table[t] = self.depth_of_rainfall(durations, t)
+
if as_intensity:
+
result_table[t] /= durations / 60 # mm/h
+
+
result_table = pd.DataFrame(result_table, index=durations)
+
result_table.index = result_table.index.astype(int) # there should be no float in minutes
+
+
if add_names:
+
result_table.index.name = 'duration' + (' (min)' if add_unit else '')
+
result_table.columns = pd.MultiIndex.from_tuples([(rp, round(1 / rp, 3)) for rp in result_table.columns])
+
result_table.columns.names = ['return period' + (' (a)' if add_unit else ''),
+
'frequency' + (' (1/a)' if add_unit else '')]
+
return result_table
+
+
+
####################################################################################################################
+
+
+
+
####################################################################################################################
+
# alias
+
result_figure = curve_figure
+
+
####################################################################################################################
+
@classmethod
+
def command_line_tool(cls):
+
user = heavy_rain_parser()
+
+
# --------------------------------------------------
+
# use the same directory as the input file and make as subdir with the name of the input_file + "_idf_data"
+
out = '{label}_idf_data'.format(label='.'.join(user.input.split('.')[:-1]))
+
out = Path(out)
+
+
if not out.is_dir():
+
out.mkdir()
+
action = 'Creating'
+
else:
+
action = 'Using'
+
+
print(f'{action} the subfolder "{out}" for the interim- and final-results.')
+
+
prefix = 'idf_'
+
+
# --------------------------------------------------
+
idf = cls(series_kind=user.series_kind, worksheet=user.worksheet, extended_durations=True)
+
+
# --------------------------------------------------
+
parameters_fn = out / f'{prefix}parameters.yaml'
+
+
if parameters_fn.is_file():
+
print(f'Found existing interim-results in "{parameters_fn}" and using them for calculations.')
+
else:
+
print(f'Start reading the time-series {user.input} for the analysis.')
+
ts = import_series(user.input).replace(0, np.nan).dropna()
+
# --------------------------------------------------
+
idf.set_series(ts)
+
print('Finished reading.')
+
+
# --------------------------------------------------
+
idf.auto_save_parameters(parameters_fn)
+
+
# --------------------------------------------------
+
h = user.height_of_rainfall
+
r = user.flow_rate_of_rainfall
+
d = user.duration
+
t = user.return_period
+
+
if r is not None:
+
if h is None and d is not None:
+
h = rate2height(rain_flow_rate=r, duration=d)
+
+
elif d is None and h is not None:
+
d = h / r * 1000 / 6
+
+
if user.r_720_1:
+
d = 720
+
t = 1
+
+
if any((h, d, t)):
+
if all((d, t)):
+
pass
+
+
elif all((d, h)):
+
t = idf.get_return_period(h, d)
+
print(f'The return period is {t:0.1f} years.')
+
+
elif all((h, t)):
+
d = idf.get_duration(h, t)
+
print(f'The duration is {d:0.1f} minutes.')
+
+
print(f'Resultierende Regenhöhe h_N(T_n={t:0.1f}a, D={d:0.1f}min) = {idf.depth_of_rainfall(d, t):0.2f} mm')
+
print(f'Resultierende Regenspende r_N(T_n={t:0.1f}a, D={d:0.1f}min) = {idf.rain_flow_rate(d, t):0.2f} L/(s*ha)')
+
+
# --------------------------------------------------
+
if user.plot:
+
fig, ax = idf.curve_figure()
+
plot_fn = out / f'{prefix}curves_plot.png'
+
fig.savefig(plot_fn, dpi=260)
+
plt.close(fig)
+
webbrowser.open(str(plot_fn))
+
print(f'Created the IDF-curves-plot and saved the file as "{plot_fn}".')
+
+
# --------------------------------------------------
+
if user.export_table:
+
table = idf.result_table(add_names=True)
+
print(table.round(2).to_string())
+
table_fn = out / f'{prefix}table.csv'
+
table.to_csv(table_fn, sep=';', decimal=',', float_format='%0.2f')
+
print(f'Created the IDF-curves-plot and saved the file as "{table_fn}".')
+
+
####################################################################################################################
+
+
[docs]
+
def get_rainfall_sum_frame(self, series=None, durations=None):
+
"""
+
Get a rainfall sum frame for any series with the duration steps as columns.
+
+
Default: The time-series and the duration-steps of the analysis.
+
+
Args:
+
series (pandas.Series, Optional): rainfall time-series
+
durations (list, Optional): list of durations in minutes which are of interest (default: pre-defined durations)
+
+
Returns:
+
pandas.DataFrame: Rain sum depending on the duration per datetime-index.
+
"""
+
if durations is None:
+
durations = self.duration_steps_for_output
+
+
if series is None:
+
if self._rainfall_sum_frame is not None:
+
return self._rainfall_sum_frame
+
ts = self.series.copy()
+
freq = self._freq
+
else:
+
freq = guess_freq(series.index)
+
ts = series.copy()
+
ts = ts.asfreq(freq).fillna(0)
+
# ts = series.replace(0, np.nan).dropna()
+
+
df = pd.DataFrame(index=ts.index)
+
# df = {}
+
+
freq_num = delta2min(freq)
+
+
for d in frame_looper(ts.index.size, columns=durations, label='rainfall_sum'):
+
if d % freq_num != 0:
+
warnings.warn(f'Using durations (= {d} minutes), '
+
f'which are not a multiple of the base frequency (= {freq_num} minutes) of the series, '
+
f'will lead to misinterpretations.')
+
df[d] = ts.rolling(pd.Timedelta(minutes=d)).sum()
+
+
# printable_names (bool): if durations should be as readable in dataframe, else in minutes
+
# df = df.rename(minutes_readable, axis=0)
+
+
return df#.round(2)
+
+
+
@property
+
def rainfall_sum_frame(self):
+
"""
+
Get the rainfall sum over the whole time-series for the default duration steps.
+
+
Returns:
+
pandas.DataFrame: Rain sum depending on the duration per datetime-index.
+
"""
+
if self._rainfall_sum_frame is None:
+
self._rainfall_sum_frame = self.get_rainfall_sum_frame()
+
return self._rainfall_sum_frame
+
+
####################################################################################################################
+
+
[docs]
+
def get_return_periods_frame(self, series=None, durations=None):
+
"""
+
Get the return periods for any time-series with the duration steps as columns.
+
+
Default: The time-series and the duration-steps of the analysis.
+
+
Is NaN if rainfall sum is smaller than 0.1 mm.
+
+
Args:
+
series (pandas.Series, Optional): rainfall time-series
+
durations (list, Optional): Durations in minutes which are of interest (default: pre-defined durations)
+
+
Returns:
+
pandas.DataFrame: Return periods depending on the duration per datetime-index.
+
"""
+
sums = self.get_rainfall_sum_frame(series=series, durations=durations)
+
df = pd.DataFrame(index=sums.index)
+
# df = {}
+
for d in frame_looper(sums.index.size, columns=sums.columns, label='return_periods'):
+
df[d] = self.get_return_period(height_of_rainfall=sums[d][sums[d] >= 0.1], duration=d)
+
return df#.fillna(0)#.round(2)
+
+
+
@property
+
def return_periods_frame(self):
+
"""
+
Get the return periods over the whole time-series for the default duration steps.
+
+
Is NaN if rainfall sum is smaller than 0.1 mm.
+
+
Returns:
+
pandas.DataFrame: data-frame of return periods where the columns are the duration steps
+
"""
+
if self._return_periods_frame is None:
+
self._return_periods_frame = self.get_return_periods_frame()
+
return self._return_periods_frame
+
+
+
[docs]
+
def write_return_periods_frame(self, filename, **kwargs):
+
"""save the return-periods dataframe as a parquet-file to save computation time."""
+
df = self.return_periods_frame.copy()
+
df.columns = df.columns.to_series().astype(str)
+
df.round(2).to_parquet(filename, **kwargs)
+
+
+
+
[docs]
+
def read_return_periods_frame(self, filename, **kwargs):
+
"""read the return-periods dataframe as a parquet-file to save computation time."""
+
df = pd.read_parquet(filename, **kwargs)
+
df.columns = df.columns.to_series().astype(int)
+
self._return_periods_frame = df
+
+
+
+
[docs]
+
def auto_save_return_periods_frame(self, filename: Path or str):
+
"""auto-save the return-periods dataframe as a parquet-file to save computation time."""
+
if isinstance(filename, str):
+
filename = Path(filename)
+
if filename.is_file():
+
self.read_return_periods_frame(filename)
+
else:
+
self.write_return_periods_frame(filename)
+
+
+
####################################################################################################################
+
@property
+
def rain_events(self):
+
"""
+
get the all the rain events of the time-series
+
+
default minimal gap between events is 4 hours
+
+
Returns:
+
pandas.DataFrame: data-frame of events with start-, end-time and duration
+
"""
+
if self._rain_events is None:
+
events = rain_events(self.series, min_gap=max(pd.Timedelta(hours=4), self._freq))
+
events[COL.DUR] = event_duration(events) + pd.Timedelta(self._freq)
+
events[COL.LP] = agg_events(events, self.series, 'sum').round(1)
+
events[COL.LAST] = events[COL.START] - events[COL.END].shift()
+
# events = events.sort_values(by=COL.LP, ascending=False)
+
self._rain_events = events
+
+
return self._rain_events
+
+
+
[docs]
+
def write_rain_events(self, filename, sep=';', decimal='.'):
+
"""save the rain-events dataframe as a csv-file for external use or to save computation time."""
+
self.rain_events.to_csv(filename, index=False, sep=sep, decimal=decimal)
+
+
+
+
[docs]
+
def read_rain_events(self, filename, sep=';', decimal='.'):
+
"""read the rain-events dataframe as a csv-file to save computation time."""
+
events = pd.read_csv(filename, skipinitialspace=True, sep=sep, decimal=decimal)
+
events[COL.START] = pd.to_datetime(events[COL.START])
+
events[COL.END] = pd.to_datetime(events[COL.END])
+
events[COL.DUR] = pd.to_timedelta(events[COL.DUR])
+
events[COL.LAST] = pd.to_timedelta(events[COL.LAST])
+
self._rain_events = events
+
+
+
+
[docs]
+
def auto_save_rain_events(self, filename: Path or str, sep=';', decimal='.'):
+
"""auto-save the rain-events dataframe as a csv-file to save computation time."""
+
if isinstance(filename, str):
+
filename = Path(filename)
+
if filename.is_file():
+
self.read_rain_events(filename, sep=sep, decimal=decimal)
+
else:
+
self.write_rain_events(filename, sep=sep, decimal=decimal)
+
+
+
def add_max_return_periods_to_events(self, events):
+
if COL.MAX_PERIOD not in events:
+
return_periods_frame = self.return_periods_frame
+
+
# maximum return period for every timestep
+
max_periods = return_periods_frame.max(axis=1).fillna(0) # fill NaN -> weil < 0.1 gefiltert wurde
+
+
datetime_max = agg_events(events, max_periods, 'idxmax')
+
datetime_max = np.where(np.isnan(datetime_max), events[COL.START].values, datetime_max)
+
# alternative:
+
# [xv if c else yv for c, xv, yv in zip(np.isnan(datetime_max), events[COL.START].values, datetime_max)]
+
if return_periods_frame.index.tz is not None:
+
datetime_max = pd.DatetimeIndex(datetime_max).tz_localize('utc').tz_convert(return_periods_frame.index.tz)
+
events[COL.MAX_PERIOD] = max_periods[datetime_max].values
+
+
events[COL.MAX_PERIOD_DURATION] = return_periods_frame.loc[datetime_max].idxmax(axis=1, skipna=True).values
+
+
def get_max_event_intensities_frame(self, events):
+
sum_frame = self.rainfall_sum_frame
+
di = {}
+
for duration in self.duration_steps:
+
di[duration] = agg_events(events, sum_frame[duration], 'max').round(2)
+
return pd.DataFrame(di, index=events.index)
+
+
+
[docs]
+
def add_max_intensities_to_events(self, events, column_format='max_sum_{:0.0f}'):
+
"""
+
Add the maximum intensities for all duration steps to the events table.
+
+
Args:
+
events (pandas.DataFrame): events table
+
column_format (str): format of the column names.
+
+
Returns:
+
pandas.DataFrame: events table including the columns with the maximum intensities
+
"""
+
return pd.concat([events, self.get_max_event_intensities_frame(events).rename(columns=column_format.format)], axis=1)
+
+
+
def get_max_return_periods_per_durations_frame(self, events):
+
return_periods_frame = self.return_periods_frame
+
di = {}
+
for duration in self.duration_steps:
+
di[duration] = agg_events(events, return_periods_frame[duration], 'max').round(2)
+
return pd.DataFrame(di, index=events.index)
+
+
+
[docs]
+
def add_max_return_periods_per_duration_to_events(self, events, column_format='max_return_period_{:0.0f}'):
+
"""
+
Add the maximum return periods for all duration steps to the events table.
+
+
Args:
+
events (pandas.DataFrame): events table
+
column_format (str): format of the column names.
+
+
Returns:
+
pandas.DataFrame: events table including the columns with the maximum return periods
+
"""
+
return pd.concat([events, self.get_max_return_periods_per_durations_frame(events).rename(columns=column_format.format)], axis=1)
+
+
+
####################################################################################################################
+
+
[docs]
+
def event_report(self, filename, min_event_rain_sum=25, min_return_period=0.5, durations=None):
+
"""
+
create pdf file with the biggest rain events
+
for each event is represented by a plot of the rain series
+
and a IDF analysis where the return periods are calculated
+
+
Args:
+
filename (str): path (directory + filename) for the created pdf-report
+
min_event_rain_sum (float): only events with a bigger rain sum will be created
+
min_return_period (float): only events with a bigger return period will be analysed
+
(the plot will be created anyway)
+
durations (list[int]): analysed durations
+
(default: [5, 10, 15, 20, 30, 45, 60, 90, 120, 180, 240, 360, 540, 720, 1080, 1440, 2880, 4320])
+
"""
+
events = self.rain_events
+
self.add_max_return_periods_to_events(events)
+
+
main_events = events[(events[COL.LP] > min_event_rain_sum) & (events[COL.MAX_PERIOD] > min_return_period)].sort_values(by=COL.MAX_PERIOD, ascending=False).to_dict(
+
orient='index')
+
+
unit = 'mm'
+
column_name = 'Precipitation'
+
+
pdf = PdfPages(filename)
+
+
for _, event in get_progress_bar(main_events.items()):
+
fig, caption = self.event_plot(event, min_return_period=min_return_period,
+
unit=unit, column_name=column_name)
+
+
# -------------------------------------
+
fig.get_axes()[0].set_title(caption + '\n\n\n')
+
+
# DIN A4
+
fig.set_size_inches(w=8.27, h=11.69)
+
# fig.tight_layout()
+
pdf.savefig(fig)
+
plt.close(fig)
+
+
pdf.close()
+
+
+
def event_plot(self, event, durations=None, unit='mm', column_name='Precipitation', min_return_period=1., german_caption=False, max_duration=None):
+
if isinstance(event, pd.Series):
+
event = event.to_dict()
+
+
plot_range = slice(event[COL.START] - pd.Timedelta(self._freq), event[COL.END] + pd.Timedelta(self._freq))
+
+
return_periods_frame = self.return_periods_frame[plot_range]
+
if COL.MAX_PERIOD not in event:
+
event[COL.MAX_PERIOD] = return_periods_frame.max().max()
+
event[COL.MAX_PERIOD_DURATION] = return_periods_frame.max().idxmax()
+
+
sum_frame_event = self.rainfall_sum_frame[plot_range]
+
+
ts = self.series[plot_range].resample(self._freq).sum().fillna(0).copy()
+
+
# -------------------------------------
+
fig = plt.figure()
+
+
if event[COL.MAX_PERIOD] < min_return_period:
+
rain_ax = fig.add_subplot(111)
+
+
else:
+
if max_duration is None:
+
if durations is not None:
+
max_dur = max(durations)
+
else:
+
max_dur = max(self.duration_steps_for_output)
+
else:
+
max_dur = max_duration
+
+
return_periods_frame_extended = self.get_return_periods_frame(
+
self.series[event[COL.START] - pd.Timedelta(minutes=max_dur):
+
event[COL.END] + pd.Timedelta(self._freq)].resample(self._freq).sum(),
+
durations=durations
+
).round(1)
+
+
idf_bar_ax = fig.add_subplot(211)
+
idf_bar_ax = idf_bar_axes(idf_bar_ax, return_periods_frame_extended)
+
rain_ax = fig.add_subplot(212, sharex=idf_bar_ax)
+
+
# -------------------------------------
+
ts_sum, minutes = resample_rain_series(ts)
+
rain_ax = rain_bar_plot(ts_sum, rain_ax)
+
rain_ax.set_ylabel(f'{column_name} in {unit}/{minutes if minutes != 1 else ""}min')
+
if ts.index.size > 1:
+
rain_ax.set_xlim(ts.index[0], ts.index[-1])
+
+
return fig, (event_caption_ger(event) if german_caption else event_caption(event, unit))
+
+
####################################################################################################################
+
def event_return_period_report(self, filename, min_return_period=1):
+
events = self.rain_events
+
self.add_max_return_periods_to_events(events)
+
+
main_events = events[events[COL.MAX_PERIOD] > min_return_period].sort_values(by=COL.MAX_PERIOD, ascending=False)
+
+
pdf = PdfPages(filename)
+
+
for _, event in get_progress_bar(main_events.to_dict(orient='index').items()):
+
fig, ax = self.return_period_event_figure(event)
+
+
# -------------------------------------
+
# DIN A4
+
fig.set_size_inches(h=11.69 / 2, w=8.27)
+
# fig.tight_layout()
+
pdf.savefig(fig)
+
plt.close(fig)
+
+
pdf.close()
+
+
def return_period_event_figure(self, event):
+
period_line = self.return_periods_frame[event[COL.START]:event[COL.END]].max()
+
+
# period_line[period_line < 0.75] = np.nan
+
period_line = period_line.dropna()
+
+
ax = period_line.plot() # type: plt.Axes
+
+
ax.set_title('rain event\n'
+
f'between {event[COL.START]:%Y-%m-%d %H:%M} and {event[COL.END]:%Y-%m-%d %H:%M}\n'
+
f'with a total sum of {event[COL.LP]:0.1f} mm\n'
+
f'and a duration of {event[COL.END] - event[COL.START]}\n'
+
f'The maximum return period was {period_line.max():0.2f}a\n'
+
f'at a duration of {period_line.idxmax()}.')
+
# ax.set_xscale('log')
+
# ax.set_yscale('log')
+
# print(ax.get_ylim())
+
# ax.set_ylim(0.01, 300)
+
# exit()
+
ax.set_xticks(period_line.index)
+
ax.set_xticklabels([minutes_readable(m) for m in period_line.index])
+
ax.set_xlabel('duration steps')
+
ax.set_ylabel('return period in years')
+
return ax.get_figure(), ax
+
+
+
[docs]
+
@classmethod
+
def from_idf_table(cls, idf_table, worksheet=METHOD.KOSTRA, linear_interpolation=True):
+
"""
+
Create an IDF-analysis-object based on an idf-tabel (i.e. from a given KOSTRA table)
+
+
Args:
+
idf_table (pandas.DataFrame): idf-table with index=Durations and columns=return Period and values=Rainheight
+
worksheet (str | optional): name of the worksheet to use. default: 'KOSTRA'
+
+
Returns:
+
IntensityDurationFrequencyAnalyse: idf-object
+
"""
+
idf = cls(worksheet=worksheet)
+
idf._parameters.reverse_engineering(idf_table, linear_interpolation=linear_interpolation)
+
idf.parameters.series_kind = 'from IDF table'
+
return idf
+
+
+
@property
+
def model_rain_block(self):
+
"""
+
Create a model block rain class.
+
+
Returns:
+
_BlockRain: Synthetic model block rain.
+
"""
+
return _BlockRain(self)
+
+
@property
+
def model_rain_euler(self):
+
"""
+
Create a model Euler rain class.
+
+
Returns:
+
_EulerRain: Synthetic model Euler rain.
+
"""
+
return _EulerRain(self)
+
+
def return_period_scatter(self, min_event_sum=25, durations=None):
+
if durations is None:
+
durations = [5, 10, 15, 20, 30, 45, 60, 90, 120, 180, 240, 360, 540, 720, 1080, 1440, 2880, 4320]
+
+
dur_short = durations[:durations.index(90)]
+
dur_long = durations[durations.index(90):]
+
+
events = self.rain_events
+
events[COL.LP] = agg_events(events, self.series, 'sum')
+
events = events[events[COL.LP] > min_event_sum].copy()
+
+
tn_long_list = {}
+
tn_short_list = {}
+
+
for _, event in events.iterrows():
+
start = event[COL.START]
+
end = event[COL.END]
+
# save true
+
idf_table = self.return_periods_frame[start:end]
+
# idf_table = idf_table.rename(columns=minutes_readable)
+
# idf_table[idf_table < min_return_period] = np.nan
+
+
tn = idf_table.loc[start:end]
+
tn_short = tn[dur_short].max().max()
+
tn_long = tn[dur_long].max().max()
+
+
if tn_long > tn_short:
+
tn_long_list[start] = tn_long
+
else:
+
tn_short_list[start] = tn_short
+
+
# check()
+
fig, ax = plt.subplots()
+
+
ax.scatter(x=list(tn_short_list.keys()), y=list(tn_short_list.values()), color='red')
+
ax.scatter(x=list(tn_long_list.keys()), y=list(tn_long_list.values()), color='blue')
+
fig = ax.get_figure()
+
+
ax.set_ylabel('Return Period in a')
+
+
def line_in_legend(color=None, marker=None, lw=None, ls=None, **kwargs):
+
from matplotlib.lines import Line2D
+
return Line2D([0], [0], color=color, marker=marker, linewidth=lw, linestyle=ls, **kwargs)
+
+
custom_lines = []
+
custom_lines.append(line_in_legend(color='red', marker='o', lw=0))
+
custom_lines.append(line_in_legend(color='blue', marker='o', lw=0))
+
# -----------------
+
l1 = ax.legend(custom_lines, ['< 60 min', '> 60 min'], loc='best', title='max Duration')
+
ax.add_artist(l1)
+
return fig, ax
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+