Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def reporting_model_billing(il_electricity_cdd_hdd_billing_monthly):
meter_data = il_electricity_cdd_hdd_billing_monthly["meter_data"]
meter_data.value = meter_data.value - 50
temperature_data = il_electricity_cdd_hdd_billing_monthly["temperature_data"]
blackout_start_date = il_electricity_cdd_hdd_billing_monthly["blackout_start_date"]
baseline_meter_data, warnings = get_baseline_data(
meter_data, end=blackout_start_date
)
baseline_data = create_caltrack_billing_design_matrix(
baseline_meter_data, temperature_data
)
model_results = fit_caltrack_usage_per_day_model(
baseline_data, use_billing_presets=True, weights_col="n_days_kept"
)
return model_results
def baseline_model_billing_single_record_baseline_data(
il_electricity_cdd_hdd_billing_monthly
):
meter_data = il_electricity_cdd_hdd_billing_monthly["meter_data"]
temperature_data = il_electricity_cdd_hdd_billing_monthly["temperature_data"]
blackout_start_date = il_electricity_cdd_hdd_billing_monthly["blackout_start_date"]
baseline_meter_data, warnings = get_baseline_data(
meter_data, end=blackout_start_date
)
baseline_data = create_caltrack_billing_design_matrix(
baseline_meter_data, temperature_data
)
baseline_data = baseline_data[:2]
model_results = fit_caltrack_usage_per_day_model(
baseline_data, use_billing_presets=True, weights_col="n_days_kept"
)
return model_results
def test_get_baseline_data_with_overshoot(il_electricity_cdd_hdd_billing_monthly):
meter_data = il_electricity_cdd_hdd_billing_monthly["meter_data"]
baseline_data, warnings = get_baseline_data(
meter_data,
end=datetime(2016, 11, 9, tzinfo=pytz.UTC),
max_days=32,
allow_billing_period_overshoot=True,
)
assert baseline_data.shape == (2, 1)
assert round(baseline_data.value.sum(), 2) == 632.31
assert len(warnings) == 0
baseline_data, warnings = get_baseline_data(
meter_data,
end=datetime(2016, 11, 9, tzinfo=pytz.UTC),
max_days=32,
allow_billing_period_overshoot=False,
)
assert baseline_data.shape == (1, 1)
assert round(baseline_data.value.sum(), 2) == 0
assert len(warnings) == 0
baseline_data, warnings = get_baseline_data(
meter_data,
end=datetime(2016, 11, 9, tzinfo=pytz.UTC),
max_days=25,
allow_billing_period_overshoot=True,
)
assert baseline_data.shape == (1, 1)
def cdd_hdd_c65(il_electricity_cdd_hdd_daily):
meter_data = il_electricity_cdd_hdd_daily["meter_data"]
temperature_data = il_electricity_cdd_hdd_daily["temperature_data"]
blackout_start_date = il_electricity_cdd_hdd_daily["blackout_start_date"]
temperature_features = compute_temperature_features(
meter_data.index,
temperature_data,
heating_balance_points=[],
cooling_balance_points=[65],
use_mean_daily_values=True,
)
meter_data_feature = compute_usage_per_day_feature(meter_data, "meter_value")
data = merge_features([meter_data_feature, temperature_features])
baseline_data, warnings = get_baseline_data(data, end=blackout_start_date)
return baseline_data
def cdd_hdd_h60_c65(il_electricity_cdd_hdd_daily):
meter_data = il_electricity_cdd_hdd_daily["meter_data"]
temperature_data = il_electricity_cdd_hdd_daily["temperature_data"]
blackout_start_date = il_electricity_cdd_hdd_daily["blackout_start_date"]
temperature_features = compute_temperature_features(
meter_data.index,
temperature_data,
heating_balance_points=[60],
cooling_balance_points=[65],
use_mean_daily_values=True,
)
meter_data_feature = compute_usage_per_day_feature(meter_data, "meter_value")
data = merge_features([meter_data_feature, temperature_features])
baseline_data, warnings = get_baseline_data(data, end=blackout_start_date)
return baseline_data
def test_get_baseline_data_with_end_no_max_days(il_electricity_cdd_hdd_hourly):
meter_data = il_electricity_cdd_hdd_hourly["meter_data"]
blackout_start_date = il_electricity_cdd_hdd_hourly["blackout_start_date"]
baseline_data, warnings = get_baseline_data(
meter_data, end=blackout_start_date, max_days=None
)
assert meter_data.shape != baseline_data.shape == (9595, 1)
assert len(warnings) == 0
def test_get_baseline_data(il_electricity_cdd_hdd_hourly):
meter_data = il_electricity_cdd_hdd_hourly["meter_data"]
baseline_data, warnings = get_baseline_data(meter_data)
assert meter_data.shape == baseline_data.shape == (19417, 1)
assert len(warnings) == 0
def test_get_baseline_data_empty(il_electricity_cdd_hdd_hourly):
meter_data = il_electricity_cdd_hdd_hourly["meter_data"]
blackout_start_date = il_electricity_cdd_hdd_hourly["blackout_start_date"]
with pytest.raises(NoBaselineDataError):
get_baseline_data(meter_data, end=pd.Timestamp("2000").tz_localize("UTC"))
def test_get_baseline_data_with_end(il_electricity_cdd_hdd_hourly):
meter_data = il_electricity_cdd_hdd_hourly["meter_data"]
blackout_start_date = il_electricity_cdd_hdd_hourly["blackout_start_date"]
baseline_data, warnings = get_baseline_data(meter_data, end=blackout_start_date)
assert meter_data.shape != baseline_data.shape == (8761, 1)
assert len(warnings) == 0
)
assert baseline_data.shape == (2, 1)
assert round(baseline_data.value.sum(), 2) == 632.31
assert len(warnings) == 0
baseline_data, warnings = get_baseline_data(
meter_data,
end=datetime(2016, 11, 9, tzinfo=pytz.UTC),
max_days=45,
ignore_billing_period_gap_for_day_count=False,
)
assert baseline_data.shape == (1, 1)
assert round(baseline_data.value.sum(), 2) == 0
assert len(warnings) == 0
baseline_data, warnings = get_baseline_data(
meter_data,
end=datetime(2016, 11, 9, tzinfo=pytz.UTC),
max_days=25,
ignore_billing_period_gap_for_day_count=True,
)
assert baseline_data.shape == (1, 1)
assert round(baseline_data.value.sum(), 2) == 0
assert len(warnings) == 0