Skip to content

Commit

Permalink
test for ValueError
Browse files Browse the repository at this point in the history
  • Loading branch information
martin-springer committed Apr 30, 2024
1 parent f1378d9 commit b072b57
Show file tree
Hide file tree
Showing 2 changed files with 115 additions and 74 deletions.
87 changes: 51 additions & 36 deletions rdtools/test/energy_from_power_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

@pytest.fixture
def times():
return pd.date_range(start='20200101 12:00', end='20200101 13:00', freq='15T')
return pd.date_range(start="20200101 12:00", end="20200101 13:00", freq="15T")


@pytest.fixture
Expand All @@ -15,100 +15,115 @@ def power(times):


def test_energy_from_power_single_arg(power):
expected = power.iloc[1:]*0.25
expected.name = 'energy_Wh'
expected = power.iloc[1:] * 0.25
expected.name = "energy_Wh"
result = energy_from_power(power)
pd.testing.assert_series_equal(result, expected)


def test_energy_from_power_instantaneous(power):
expected = (0.25*(power + power.shift())/2).dropna()
expected.name = 'energy_Wh'
result = energy_from_power(power, power_type='instantaneous')
expected = (0.25 * (power + power.shift()) / 2).dropna()
expected.name = "energy_Wh"
result = energy_from_power(power, power_type="instantaneous")
pd.testing.assert_series_equal(result, expected)


def test_energy_from_power_max_timedelta_inference(power):
expected = power.iloc[1:]*0.25
expected.name = 'energy_Wh'
expected = power.iloc[1:] * 0.25
expected.name = "energy_Wh"
expected.iloc[:2] = np.nan
match = 'Fraction of excluded data (.*) exceeded threshold'
match = "Fraction of excluded data (.*) exceeded threshold"
with pytest.warns(UserWarning, match=match):
result = energy_from_power(power.drop(power.index[1]))
pd.testing.assert_series_equal(result, expected)


def test_energy_from_power_max_timedelta(power):
expected = power.iloc[1:]*0.25
expected.name = 'energy_Wh'
result = energy_from_power(power.drop(power.index[1]),
max_timedelta=pd.to_timedelta('30 minutes'))
expected = power.iloc[1:] * 0.25
expected.name = "energy_Wh"
result = energy_from_power(
power.drop(power.index[1]), max_timedelta=pd.to_timedelta("30 minutes")
)
pd.testing.assert_series_equal(result, expected)


def test_energy_from_power_upsample(power):
expected = power.resample('10T').asfreq().interpolate()/6
expected = power.resample("10T").asfreq().interpolate() / 6
expected = expected.iloc[1:]
expected.name = 'energy_Wh'
result = energy_from_power(power, target_frequency='10T')
expected.name = "energy_Wh"
result = energy_from_power(power, target_frequency="10T")
pd.testing.assert_series_equal(result, expected)


def test_energy_from_power_downsample(power):
expected = power.resample('20T').asfreq()
expected = power.resample("20T").asfreq()
expected = expected.iloc[1:]
expected = pd.Series([0.75, 0.833333333, 0.416666667], index=expected.index)
expected.name = 'energy_Wh'
result = energy_from_power(power, target_frequency='20T')
expected.name = "energy_Wh"
result = energy_from_power(power, target_frequency="20T")
pd.testing.assert_series_equal(result, expected)


def test_energy_from_power_max_timedelta_edge_case():
times = pd.date_range('2020-01-01 12:00', periods=4, freq='15T')
times = pd.date_range("2020-01-01 12:00", periods=4, freq="15T")
power = pd.Series(1, index=times)
power = power.drop(power.index[2])
result = energy_from_power(power, '30T', max_timedelta=pd.to_timedelta('20 minutes'))
result = energy_from_power(
power, "30T", max_timedelta=pd.to_timedelta("20 minutes")
)
assert result.isnull().all()


def test_energy_from_power_single_value_input():
times = pd.date_range('2019-01-01', freq='15T', periods=1)
power = pd.Series([100.], index=times)
expected_result = pd.Series([25.], index=times, name='energy_Wh')
times = pd.date_range("2019-01-01", freq="15T", periods=1)
power = pd.Series([100.0], index=times)
expected_result = pd.Series([25.0], index=times, name="energy_Wh")
result = energy_from_power(power)
pd.testing.assert_series_equal(result, expected_result)


def test_energy_from_power_single_value_input_no_freq():
power = pd.Series([1], pd.date_range('2019-01-01', periods=1, freq='15T'))
power = pd.Series([1], pd.date_range("2019-01-01", periods=1, freq="15T"))
power.index.freq = None
match = "Could not determine period of input power"
with pytest.raises(ValueError, match=match):
energy_from_power(power)


def test_energy_from_power_single_value_instantaneous():
power = pd.Series([1], pd.date_range('2019-01-01', periods=1, freq='15T'))
power = pd.Series([1], pd.date_range("2019-01-01", periods=1, freq="15T"))
power.index.freq = None
match = ("power_type='instantaneous' is incompatible with single element power. "
"Use power_type='right-labeled'")
match = (
"power_type='instantaneous' is incompatible with single element power. "
"Use power_type='right-labeled'"
)
with pytest.raises(ValueError, match=match):
energy_from_power(power, power_type='instantaneous')
energy_from_power(power, power_type="instantaneous")


def test_energy_from_power_single_value_with_target():
times = pd.date_range('2019-01-01', freq='15T', periods=1)
power = pd.Series([100.], index=times)
expected_result = pd.Series([100.], index=times, name='energy_Wh')
result = energy_from_power(power, target_frequency='H')
times = pd.date_range("2019-01-01", freq="15T", periods=1)
power = pd.Series([100.0], index=times)
expected_result = pd.Series([100.0], index=times, name="energy_Wh")
result = energy_from_power(power, target_frequency="H")
pd.testing.assert_series_equal(result, expected_result)


def test_energy_from_power_leading_nans():
# GH 244
power = pd.Series(1, pd.date_range('2019-01-01', freq='15min', periods=5))
power = pd.Series(1, pd.date_range("2019-01-01", freq="15min", periods=5))
power.iloc[:2] = np.nan
expected_result = pd.Series([np.nan, np.nan, 0.25, 0.25],
index=power.index[1:], name='energy_Wh')
expected_result = pd.Series(
[np.nan, np.nan, 0.25, 0.25], index=power.index[1:], name="energy_Wh"
)
result = energy_from_power(power)
pd.testing.assert_series_equal(result, expected_result)


def test_energy_from_power_series_index():
power = pd.Series([1, 2, 3, 4, 5])
pytest.raises(
ValueError,
energy_from_power,
power,
)
102 changes: 64 additions & 38 deletions rdtools/test/interpolate_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,29 +7,31 @@

@pytest.fixture
def time_series():
times = pd.date_range('2018-04-01 12:00', '2018-04-01 13:15', freq='15T')
time_series = pd.Series(data=[9, 6, 3, 3, 6, 9], index=times, name='foo')
times = pd.date_range("2018-04-01 12:00", "2018-04-01 13:15", freq="15T")
time_series = pd.Series(data=[9, 6, 3, 3, 6, 9], index=times, name="foo")
time_series = time_series.drop(times[4])
return time_series


@pytest.fixture
def target_index(time_series):
return pd.date_range(time_series.index.min(), time_series.index.max(), freq='20T')
return pd.date_range(time_series.index.min(), time_series.index.max(), freq="20T")


@pytest.fixture
def expected_series(target_index, time_series):
return pd.Series(data=[9.0, 5.0, 3.0, np.nan], index=target_index, name=time_series.name)
return pd.Series(
data=[9.0, 5.0, 3.0, np.nan], index=target_index, name=time_series.name
)


@pytest.fixture
def test_df(time_series):
time_series1 = time_series.copy()
time_series2 = time_series.copy()

time_series2.index = time_series2.index + pd.to_timedelta('30 minutes')
time_series2.name = 'bar'
time_series2.index = time_series2.index + pd.to_timedelta("30 minutes")
time_series2.name = "bar"

test_df = pd.concat([time_series1, time_series2], axis=1)

Expand All @@ -38,38 +40,47 @@ def test_df(time_series):

@pytest.fixture
def df_target_index(target_index):
return target_index + pd.to_timedelta('15 minutes')
return target_index + pd.to_timedelta("15 minutes")


@pytest.fixture
def df_expected_result(df_target_index, test_df):
col0 = test_df.columns[0]
col1 = test_df.columns[1]
expected_df_result = pd.DataFrame({
col0: [6.0, 3.0, np.nan, 9.0],
col1: [np.nan, 8.0, 4.0, 3.0]
}, index=df_target_index)
expected_df_result = pd.DataFrame(
{col0: [6.0, 3.0, np.nan, 9.0], col1: [np.nan, 8.0, 4.0, 3.0]},
index=df_target_index,
)

expected_df_result = expected_df_result[test_df.columns]
return expected_df_result


def test_interpolate_freq_specification(time_series, target_index, expected_series):
# test the string specification
interpolated = interpolate(time_series, target_index.freq.freqstr,
pd.to_timedelta('15 minutes'), warning_threshold=0.21)
interpolated = interpolate(
time_series,
target_index.freq.freqstr,
pd.to_timedelta("15 minutes"),
warning_threshold=0.21,
)
pd.testing.assert_series_equal(interpolated, expected_series)

# test the DateOffset specification
interpolated = interpolate(time_series, target_index.freq, pd.to_timedelta('15 minutes'),
warning_threshold=0.21)
interpolated = interpolate(
time_series,
target_index.freq,
pd.to_timedelta("15 minutes"),
warning_threshold=0.21,
)
pd.testing.assert_series_equal(interpolated, expected_series)


def test_interpolate_calculation(time_series, target_index, expected_series):

interpolated = interpolate(time_series, target_index, pd.to_timedelta('15 minutes'),
warning_threshold=0.21)
interpolated = interpolate(
time_series, target_index, pd.to_timedelta("15 minutes"), warning_threshold=0.21
)
pd.testing.assert_series_equal(interpolated, expected_series)


Expand All @@ -82,60 +93,75 @@ def test_interpolate_two_argument(time_series, target_index, expected_series):

def test_interpolate_tz_validation(time_series, target_index, expected_series):
with pytest.raises(ValueError):
interpolate(time_series, target_index.tz_localize('UTC'), pd.to_timedelta('15 minutes'))
interpolate(
time_series, target_index.tz_localize("UTC"), pd.to_timedelta("15 minutes")
)

time_series = time_series.copy()
time_series.index = time_series.index.tz_localize('UTC')
time_series.index = time_series.index.tz_localize("UTC")

with pytest.raises(ValueError):
interpolate(time_series, target_index, pd.to_timedelta('15 minutes'))
interpolate(time_series, target_index, pd.to_timedelta("15 minutes"))


def test_interpolate_same_tz(time_series, target_index, expected_series):
time_series = time_series.copy()
expected_series = expected_series.copy()

time_series.index = time_series.index.tz_localize('America/Denver')
target_index = target_index.tz_localize('America/Denver')
expected_series.index = expected_series.index.tz_localize('America/Denver')
time_series.index = time_series.index.tz_localize("America/Denver")
target_index = target_index.tz_localize("America/Denver")
expected_series.index = expected_series.index.tz_localize("America/Denver")

interpolated = interpolate(time_series, target_index, pd.to_timedelta('15 minutes'),
warning_threshold=0.21)
interpolated = interpolate(
time_series, target_index, pd.to_timedelta("15 minutes"), warning_threshold=0.21
)
pd.testing.assert_series_equal(interpolated, expected_series)


def test_interpolate_different_tz(time_series, target_index, expected_series):
time_series = time_series.copy()
expected_series = expected_series.copy()

time_series.index = time_series.index.tz_localize('America/Denver').tz_convert('UTC')
target_index = target_index.tz_localize('America/Denver')
expected_series.index = expected_series.index.tz_localize('America/Denver')
time_series.index = time_series.index.tz_localize("America/Denver").tz_convert(
"UTC"
)
target_index = target_index.tz_localize("America/Denver")
expected_series.index = expected_series.index.tz_localize("America/Denver")

interpolated = interpolate(time_series, target_index, pd.to_timedelta('15 minutes'),
warning_threshold=0.21)
interpolated = interpolate(
time_series, target_index, pd.to_timedelta("15 minutes"), warning_threshold=0.21
)
pd.testing.assert_series_equal(interpolated, expected_series)


def test_interpolate_dataframe(test_df, df_target_index, df_expected_result):
interpolated = interpolate(test_df, df_target_index, pd.to_timedelta('15 minutes'),
warning_threshold=0.21)
interpolated = interpolate(
test_df, df_target_index, pd.to_timedelta("15 minutes"), warning_threshold=0.21
)
pd.testing.assert_frame_equal(interpolated, df_expected_result)


def test_interpolate_warning(test_df, df_target_index, df_expected_result):
N = len(test_df)
all_idx = list(range(N))
# drop every other value in the first third of the dataset
index_with_gaps = all_idx[:N//3][::2] + all_idx[N//3:]
index_with_gaps = all_idx[: N // 3][::2] + all_idx[N // 3 :]
test_df = test_df.iloc[index_with_gaps, :]
with pytest.warns(UserWarning):
interpolate(test_df, df_target_index, pd.to_timedelta('15 minutes'),
warning_threshold=0.1)
interpolate(
test_df,
df_target_index,
pd.to_timedelta("15 minutes"),
warning_threshold=0.1,
)

with warnings.catch_warnings():
warnings.simplefilter("error")
interpolate(test_df, df_target_index, pd.to_timedelta('15 minutes'),
warning_threshold=0.5)
warnings.filterwarnings("error", message='Fraction of excluded data')
interpolate(
test_df,
df_target_index,
pd.to_timedelta("15 minutes"),
warning_threshold=0.5,
)
warnings.filterwarnings("error", message="Fraction of excluded data")
# if this test fails, it means a warning was raised that was not expected

0 comments on commit b072b57

Please sign in to comment.