Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
period = Period(None, datetime(2015,1,3, tzinfo=pytz.UTC))
filtered_data = consumption_data_kWh_interval.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,1]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,1, tzinfo=pytz.UTC)
assert filtered_data.index[2] == datetime(2015,1,3, tzinfo=pytz.UTC)
period = Period(datetime(2015,1,9, tzinfo=pytz.UTC),None)
filtered_data = consumption_data_kWh_interval.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,9, tzinfo=pytz.UTC)
assert filtered_data.index[1] == datetime(2015,1,10, tzinfo=pytz.UTC)
period = Period(None,None)
filtered_data = consumption_data_kWh_interval.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,1,1,1,1,1,1,1,1]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,1, tzinfo=pytz.UTC)
assert filtered_data.index[9] == datetime(2015,1,10, tzinfo=pytz.UTC)
def test_consumption_data_filter_by_period_arbitrary(
consumption_data_kWh_arbitrary):
period = Period(datetime(2015,1,1), datetime(2015,1,3))
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,np.nan]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,1)
assert filtered_data.index[2] == datetime(2015,1,3)
period = Period(datetime(2014,1,1), datetime(2014,1,3,1))
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, [],
rtol=RTOL, atol=ATOL)
period = Period(None, datetime(2015,1,3))
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,np.nan]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,1)
assert filtered_data.index[2] == datetime(2015,1,3)
period = Period(datetime(2015,1,9),None)
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,np.nan]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,9)
assert filtered_data.index[2] == datetime(2015,1,11)
period = Period(None,None)
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,1,1,1,1,1,1,1,1,np.nan]),
def test_project_attributes(generated_consumption_data_1):
cd,params = generated_consumption_data_1
baseline_period = Period(datetime(2014,1,1),datetime(2014,1,1))
location = Location(zipcode="91104")
project = Project(location,[cd],baseline_period,None)
meter = ProjectAttributes(project)
result = meter.evaluate_raw(project=project)
assert result["location"].zipcode == location.zipcode
assert result["consumption"][0] is not None
assert result["baseline_period"] is not None
assert result["reporting_period"] is None
assert result["other_periods"] == []
assert result["weather_source"].station_id == "722880"
assert result["weather_normal_source"].station_id == "722880"
@pytest.mark.slow
def test_annualized_gross_savings_metric(
generated_consumption_data_pre_post_with_annualized_gross_savings_1,
gsod_722880_2012_2014_weather_source, tmy3_722880_weather_source):
model = AverageDailyTemperatureSensitivityModel(heating=True, cooling=True)
meter = AnnualizedGrossSavingsMeter(temperature_unit_str="degF", model=model)
cd, params_pre, params_post, retrofit_date, savings = \
generated_consumption_data_pre_post_with_annualized_gross_savings_1
reporting_period = Period(retrofit_date, datetime(2015,1,1))
result = meter.evaluate_raw(
model_params_baseline=params_pre,
model_params_reporting=params_post,
consumption_data_reporting=cd.filter_by_period(reporting_period),
weather_normal_source=tmy3_722880_weather_source,
energy_unit_str="kWh")
assert_allclose(result["annualized_gross_savings"], savings,
rtol=RTOL, atol=ATOL)
def test_generate_monthly_billing_datetimes():
period = Period(datetime(2012,1,1, tzinfo=pytz.UTC),datetime(2013,1,1, tzinfo=pytz.UTC))
datetimes_30d = generate_monthly_billing_datetimes(period,
randint(30,31))
assert datetimes_30d[0] == datetime(2012,1,1, tzinfo=pytz.UTC)
assert datetimes_30d[1] == datetime(2012,1,31, tzinfo=pytz.UTC)
assert datetimes_30d[11] == datetime(2012,11,26, tzinfo=pytz.UTC)
assert datetimes_30d[12] == datetime(2012,12,26, tzinfo=pytz.UTC)
datetimes_1d = generate_monthly_billing_datetimes(period, randint(1,2))
assert datetimes_1d[0] == datetime(2012,1,1, tzinfo=pytz.UTC)
assert datetimes_1d[1] == datetime(2012,1,2, tzinfo=pytz.UTC)
assert datetimes_1d[330] == datetime(2012,11,26, tzinfo=pytz.UTC)
assert datetimes_1d[331] == datetime(2012,11,27, tzinfo=pytz.UTC)
def test_period_name():
p = Period(name="baseline")
assert p.name == "baseline"
assert filtered_data.index[0] == datetime(2015,1,1, tzinfo=pytz.UTC)
assert filtered_data.index[2] == datetime(2015,1,3, tzinfo=pytz.UTC)
period = Period(datetime(2014,1,1, tzinfo=pytz.UTC), datetime(2014,1,3,1, tzinfo=pytz.UTC))
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, [],
rtol=RTOL, atol=ATOL)
period = Period(None, datetime(2015,1,3, tzinfo=pytz.UTC))
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,np.nan]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,1, tzinfo=pytz.UTC)
assert filtered_data.index[2] == datetime(2015,1,3, tzinfo=pytz.UTC)
period = Period(datetime(2015,1,9, tzinfo=pytz.UTC),None)
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,np.nan]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,9, tzinfo=pytz.UTC)
assert filtered_data.index[2] == datetime(2015,1,11, tzinfo=pytz.UTC)
period = Period(None,None)
filtered_data = consumption_data_kWh_arbitrary.filter_by_period(period).data
assert_allclose(filtered_data, np.array([1,1,1,1,1,1,1,1,1,1,np.nan]),
rtol=RTOL, atol=ATOL)
assert filtered_data.index[0] == datetime(2015,1,1, tzinfo=pytz.UTC)
assert filtered_data.index[10] == datetime(2015,1,11, tzinfo=pytz.UTC)
def test_project_attributes(generated_consumption_data_1):
cd,params = generated_consumption_data_1
baseline_period = Period(datetime(2014,1,1),datetime(2014,1,1))
location = Location(zipcode="91104")
project = Project(location,[cd],baseline_period,None)
meter = ProjectAttributes(project)
result = meter.evaluate_raw(project=project)
assert result["location"].zipcode == location.zipcode
assert result["consumption"][0] is not None
assert result["baseline_period"] is not None
assert result["reporting_period"] is None
assert result["other_periods"] == []
assert result["weather_source"].station_id == "722880"
assert result["weather_normal_source"].station_id == "722880"
def _normalize_period(self, period):
start = self._normalize_datetime(period.start)
year_offset = period.end.year - period.start.year
end = self._normalize_datetime(period.end, year_offset)
return Period(start, end)
"""Sums the total observed HDD in a normal year
Parameters
----------
weather_normal_source : eemeter.weather.WeatherSourceBase and eemeter.weather.WeatherNormalMixin
A weather data source from a location as geographically and
climatically close as possible to the target project.
Returns
-------
out : dict
- "normal_annual_hdd": the total heating degree days observed
during a typical meteorological year
"""
# year of this annual period will be ignored
annual_period = Period(datetime(2013,1,1), datetime(2014,1,1))
hdd = weather_normal_source.hdd(annual_period,
self.temperature_unit_str, self.base)
return { "normal_annual_hdd": hdd }