Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"weather_source": {},
"energy_unit_str": {},
},
output_mapping: {
"temp_sensitivity_params": {},
"n_days": {},
"average_daily_usages": {},
"estimated_average_daily_usages": {},
},
}
"""
meter = load(meter_yaml)
cd, params = generated_consumption_data_1
data_collection = DataCollection(
consumption_data=cd,
weather_source=gsod_722880_2012_2014_weather_source,
energy_unit_str="kWh")
result = meter.evaluate(data_collection)
assert_allclose(result.get_data('temp_sensitivity_params').value.to_list(), params.to_list(),
rtol=RTOL, atol=ATOL)
assert result.get_data('n_days') is not None
assert result.get_data('average_daily_usages') is not None
assert result.get_data('estimated_average_daily_usages') is not None
def test_data_collection_json():
dc1 = DataCollection(item1="item1_value", item2="item2_value", tags=["tag1"])
dc1.add_data_collection(DataCollection(item1="item1_value", item2="item2_value", tags=["tag2"]))
json_data = dc1.json()
json.dumps(json_data)
assert len(json_data["item1"]) == 2
assert len(json_data["item2"]) == 2
assert json_data["item1"][0]["value"] == "item1_value"
assert json_data["item1"][0]["tags"] in [["tag1"], ["tag2"]]
},
output_mapping: { result: { name: result1 } },
},
!obj:eemeter.meter.DummyMeter {
input_mapping: {
value: {}
},
output_mapping: { result: { name: result2 } },
}
]
}
}
"""
meter = load(meter_yaml)
data_collection = DataCollection(active_fuel_type="electricity")
data_collection_include = DataCollection(value="value_include")
data_collection_exclude = DataCollection(value="value_exclude")
data_collection.add_data_collection(data_collection_include, tagspace=["electricity"])
data_collection.add_data_collection(data_collection_exclude, tagspace=["natural_gas"])
output_data_collection = meter.evaluate(data_collection)
assert output_data_collection.get_data("result1").value == "electricity"
assert output_data_collection.get_data("result2").value == "value_include"
name: iterable,
},
meter: !obj:eemeter.meter.DummyMeter {
input_mapping: { value: {} },
output_mapping: { result: {} },
}
}
"""
meter = load(meter_yaml)
iterable = [
{
"value": 1, "tags": ["one"]
}, {
"value": 2, "tags": ["two"]
}]
data_collection = DataCollection(iterable=iterable)
result = meter.evaluate(data_collection)
assert 1 == result.get_data("result", tags=["one"]).value
assert 2 == result.get_data("result", tags=["two"]).value
def test_data_collection_creation_shortcut():
dc = DataCollection(item1="item1_value", item2="item2_value", tags=["tag"])
assert dc.get_data("item1",tags=["tag"]).name == "item1"
assert dc.get_data("item1",tags=["tag"]).value == "item1_value"
assert dc.get_data("item2",tags=["tag"]).name == "item2"
assert dc.get_data("item2",tags=["tag"]).value == "item2_value"
output_mapping: { result: { name: result1 } },
},
!obj:eemeter.meter.DummyMeter {
input_mapping: {
value: {}
},
output_mapping: { result: { name: result2 } },
}
]
}
}
"""
meter = load(meter_yaml)
data_collection = DataCollection(active_fuel_type="electricity")
data_collection_include = DataCollection(value="value_include")
data_collection_exclude = DataCollection(value="value_exclude")
data_collection.add_data_collection(data_collection_include, tagspace=["electricity"])
data_collection.add_data_collection(data_collection_exclude, tagspace=["natural_gas"])
output_data_collection = meter.evaluate(data_collection)
assert output_data_collection.get_data("result1").value == "electricity"
assert output_data_collection.get_data("result2").value == "value_include"
output_mapping: { result: {} },
},
},
default: !obj:eemeter.meter.DummyMeter {
input_mapping: { value: {name: value_default} },
output_mapping: { result: {} },
}
}
"""
meter = load(meter_yaml)
data_collection = DataCollection(target=1, value_one=1, value_two=2,
value_three=3, value_default=4)
result1 = meter.evaluate(data_collection)
data_collection = DataCollection(target=2, value_one=1, value_two=2,
value_three=3, value_default=4)
result2 = meter.evaluate(data_collection)
data_collection = DataCollection(target=3, value_one=1, value_two=2,
value_three=3, value_default=4)
result3 = meter.evaluate(data_collection)
data_collection = DataCollection(target=4, value_one=1, value_two=2,
value_three=3, value_default=4)
result4 = meter.evaluate(data_collection)
data_collection = DataCollection(value_one=1, value_two=2, value_three=3,
value_default=4)
result5 = meter.evaluate(data_collection)
assert 1 == result1.get_data("result").value
assert 2 == result2.get_data("result").value
assert 3 == result3.get_data("result").value
assert 4 == result4.get_data("result").value
def test_data_collection_add_data_collection(data_collection):
dc = DataCollection()
new_data = DataContainer("name", "new_data_value", ["tag1"])
dc.add_data(new_data)
data_collection.add_data_collection(dc,tagspace=["tagspace"])
assert data_collection.get_data("name", tags=["tag1","tagspace"]).name == "name"
with pytest.raises(TypeError):
data_collection.add_data_collection("name", "tagspace")
"""Overrides the evaluate method to evaluate either the `success` meter
or the `failure` meter on the boolean value of a particular meter
input stored in the element with the name `condition_parameter`.
Returns
-------
out : dict
Collected outputs from either the success or the failure meter.
"""
if data_collection.get_data(**self.condition).value:
meter = self.success
else:
meter = self.failure
if meter is None:
return DataCollection()
else:
return meter.evaluate(data_collection)
def evaluate(self, data_collection):
"""Overrides normal execution to evaluate meters in sequence,
collecting outputs.
Returns
-------
out : dict
Collected outputs from all meters in the sequence.
"""
input_data_collection = data_collection.copy()
output_data_collection = DataCollection()
for meter in self.sequence:
meter_result = meter.evaluate(input_data_collection)
input_data_collection.add_data_collection(meter_result)
output_data_collection.add_data_collection(meter_result)
return output_data_collection