Move energy and power utilites to unit_conversion (#78950)

* Move energy and power utilites to unit_conversion

* Move tests
This commit is contained in:
epenet 2022-09-22 15:39:49 +02:00 committed by GitHub
parent c97817bb0e
commit 0767cdd935
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 176 additions and 230 deletions

View File

@ -32,8 +32,6 @@ from homeassistant.helpers.storage import STORAGE_DIR
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
from homeassistant.util import (
dt as dt_util,
energy as energy_util,
power as power_util,
pressure as pressure_util,
temperature as temperature_util,
volume as volume_util,
@ -138,19 +136,19 @@ def _convert_energy_from_kwh(to_unit: str, value: float | None) -> float | None:
"""Convert energy in kWh to to_unit."""
if value is None:
return None
return energy_util.convert(value, energy_util.NORMALIZED_UNIT, to_unit)
return EnergyConverter.convert(value, EnergyConverter.NORMALIZED_UNIT, to_unit)
def _convert_energy_to_kwh(from_unit: str, value: float) -> float:
"""Convert energy in from_unit to kWh."""
return energy_util.convert(value, from_unit, energy_util.NORMALIZED_UNIT)
return EnergyConverter.convert(value, from_unit, EnergyConverter.NORMALIZED_UNIT)
def _convert_power_from_w(to_unit: str, value: float | None) -> float | None:
"""Convert power in W to to_unit."""
if value is None:
return None
return power_util.convert(value, power_util.NORMALIZED_UNIT, to_unit)
return PowerConverter.convert(value, PowerConverter.NORMALIZED_UNIT, to_unit)
def _convert_pressure_from_pa(to_unit: str, value: float | None) -> float | None:
@ -180,16 +178,16 @@ def _convert_volume_to_m3(from_unit: str, value: float) -> float:
STATISTIC_UNIT_TO_UNIT_CLASS: dict[str | None, str] = {
energy_util.NORMALIZED_UNIT: "energy",
power_util.NORMALIZED_UNIT: "power",
EnergyConverter.NORMALIZED_UNIT: "energy",
PowerConverter.NORMALIZED_UNIT: "power",
pressure_util.NORMALIZED_UNIT: "pressure",
temperature_util.NORMALIZED_UNIT: "temperature",
volume_util.NORMALIZED_UNIT: "volume",
}
STATISTIC_UNIT_TO_UNIT_CONVERTER: dict[str | None, type[BaseUnitConverter]] = {
energy_util.NORMALIZED_UNIT: EnergyConverter,
power_util.NORMALIZED_UNIT: PowerConverter,
EnergyConverter.NORMALIZED_UNIT: EnergyConverter,
PowerConverter.NORMALIZED_UNIT: PowerConverter,
pressure_util.NORMALIZED_UNIT: PressureConverter,
temperature_util.NORMALIZED_UNIT: TemperatureConverter,
volume_util.NORMALIZED_UNIT: VolumeConverter,
@ -200,8 +198,8 @@ STATISTIC_UNIT_TO_UNIT_CONVERTER: dict[str | None, type[BaseUnitConverter]] = {
STATISTIC_UNIT_TO_DISPLAY_UNIT_FUNCTIONS: dict[
str, Callable[[str, float | None], float | None]
] = {
energy_util.NORMALIZED_UNIT: _convert_energy_from_kwh,
power_util.NORMALIZED_UNIT: _convert_power_from_w,
EnergyConverter.NORMALIZED_UNIT: _convert_energy_from_kwh,
PowerConverter.NORMALIZED_UNIT: _convert_power_from_w,
pressure_util.NORMALIZED_UNIT: _convert_pressure_from_pa,
temperature_util.NORMALIZED_UNIT: _convert_temperature_from_c,
volume_util.NORMALIZED_UNIT: _convert_volume_from_m3,
@ -211,7 +209,7 @@ STATISTIC_UNIT_TO_DISPLAY_UNIT_FUNCTIONS: dict[
# to the normalized unit used for statistics.
# This is used to support adjusting statistics in the display unit
DISPLAY_UNIT_TO_STATISTIC_UNIT_FUNCTIONS: dict[str, Callable[[str, float], float]] = {
energy_util.NORMALIZED_UNIT: _convert_energy_to_kwh,
EnergyConverter.NORMALIZED_UNIT: _convert_energy_to_kwh,
volume_util.NORMALIZED_UNIT: _convert_volume_to_m3,
}

View File

@ -21,11 +21,10 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.json import JSON_DUMP
from homeassistant.util import (
dt as dt_util,
energy as energy_util,
power as power_util,
pressure as pressure_util,
temperature as temperature_util,
)
from homeassistant.util.unit_conversion import EnergyConverter, PowerConverter
from .const import MAX_QUEUE_BACKLOG
from .statistics import (
@ -121,8 +120,8 @@ async def ws_handle_get_statistics_during_period(
vol.Required("period"): vol.Any("5minute", "hour", "day", "month"),
vol.Optional("units"): vol.Schema(
{
vol.Optional("energy"): vol.In(energy_util.VALID_UNITS),
vol.Optional("power"): vol.In(power_util.VALID_UNITS),
vol.Optional("energy"): vol.In(EnergyConverter.VALID_UNITS),
vol.Optional("power"): vol.In(PowerConverter.VALID_UNITS),
vol.Optional("pressure"): vol.In(pressure_util.VALID_UNITS),
vol.Optional("temperature"): vol.In(temperature_util.VALID_UNITS),
vol.Optional("volume"): vol.Any(VOLUME_CUBIC_FEET, VOLUME_CUBIC_METERS),

View File

@ -49,8 +49,6 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import entity_sources
from homeassistant.util import (
dt as dt_util,
energy as energy_util,
power as power_util,
pressure as pressure_util,
temperature as temperature_util,
volume as volume_util,
@ -95,15 +93,16 @@ UNIT_CONVERSIONS: dict[str, dict[str, Callable]] = {
# Convert energy to kWh
SensorDeviceClass.ENERGY: {
ENERGY_KILO_WATT_HOUR: lambda x: x
/ energy_util.UNIT_CONVERSION[ENERGY_KILO_WATT_HOUR],
/ EnergyConverter.UNIT_CONVERSION[ENERGY_KILO_WATT_HOUR],
ENERGY_MEGA_WATT_HOUR: lambda x: x
/ energy_util.UNIT_CONVERSION[ENERGY_MEGA_WATT_HOUR],
ENERGY_WATT_HOUR: lambda x: x / energy_util.UNIT_CONVERSION[ENERGY_WATT_HOUR],
/ EnergyConverter.UNIT_CONVERSION[ENERGY_MEGA_WATT_HOUR],
ENERGY_WATT_HOUR: lambda x: x
/ EnergyConverter.UNIT_CONVERSION[ENERGY_WATT_HOUR],
},
# Convert power to W
SensorDeviceClass.POWER: {
POWER_WATT: lambda x: x / power_util.UNIT_CONVERSION[POWER_WATT],
POWER_KILO_WATT: lambda x: x / power_util.UNIT_CONVERSION[POWER_KILO_WATT],
POWER_WATT: lambda x: x / PowerConverter.UNIT_CONVERSION[POWER_WATT],
POWER_KILO_WATT: lambda x: x / PowerConverter.UNIT_CONVERSION[POWER_KILO_WATT],
},
# Convert pressure to Pa
# Note: pressure_util.convert is bypassed to avoid redundant error checking

View File

@ -1,42 +0,0 @@
"""Energy util functions."""
from __future__ import annotations
from numbers import Number
from homeassistant.const import (
ENERGY_KILO_WATT_HOUR,
ENERGY_MEGA_WATT_HOUR,
ENERGY_WATT_HOUR,
UNIT_NOT_RECOGNIZED_TEMPLATE,
)
VALID_UNITS: tuple[str, ...] = (
ENERGY_WATT_HOUR,
ENERGY_KILO_WATT_HOUR,
ENERGY_MEGA_WATT_HOUR,
)
UNIT_CONVERSION: dict[str, float] = {
ENERGY_WATT_HOUR: 1 * 1000,
ENERGY_KILO_WATT_HOUR: 1,
ENERGY_MEGA_WATT_HOUR: 1 / 1000,
}
NORMALIZED_UNIT = ENERGY_KILO_WATT_HOUR
def convert(value: float, from_unit: str, to_unit: str) -> float:
"""Convert one unit of measurement to another."""
if from_unit not in VALID_UNITS:
raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(from_unit, "energy"))
if to_unit not in VALID_UNITS:
raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(to_unit, "energy"))
if not isinstance(value, Number):
raise TypeError(f"{value} is not of numeric type")
if from_unit == to_unit:
return value
watthours = value / UNIT_CONVERSION[from_unit]
return watthours * UNIT_CONVERSION[to_unit]

View File

@ -1,39 +0,0 @@
"""Power util functions."""
from __future__ import annotations
from numbers import Number
from homeassistant.const import (
POWER_KILO_WATT,
POWER_WATT,
UNIT_NOT_RECOGNIZED_TEMPLATE,
)
VALID_UNITS: tuple[str, ...] = (
POWER_WATT,
POWER_KILO_WATT,
)
UNIT_CONVERSION: dict[str, float] = {
POWER_WATT: 1,
POWER_KILO_WATT: 1 / 1000,
}
NORMALIZED_UNIT = POWER_WATT
def convert(value: float, from_unit: str, to_unit: str) -> float:
"""Convert one unit of measurement to another."""
if from_unit not in VALID_UNITS:
raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(from_unit, "power"))
if to_unit not in VALID_UNITS:
raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(to_unit, "power"))
if not isinstance(value, Number):
raise TypeError(f"{value} is not of numeric type")
if from_unit == to_unit:
return value
watts = value / UNIT_CONVERSION[from_unit]
return watts * UNIT_CONVERSION[to_unit]

View File

@ -2,10 +2,21 @@
from __future__ import annotations
from collections.abc import Callable
from numbers import Number
from homeassistant.const import (
ENERGY_KILO_WATT_HOUR,
ENERGY_MEGA_WATT_HOUR,
ENERGY_WATT_HOUR,
POWER_KILO_WATT,
POWER_WATT,
PRESSURE_PA,
TEMP_CELSIUS,
UNIT_NOT_RECOGNIZED_TEMPLATE,
VOLUME_CUBIC_METERS,
)
from . import (
energy as energy_util,
power as power_util,
pressure as pressure_util,
temperature as temperature_util,
volume as volume_util,
@ -20,26 +31,70 @@ class BaseUnitConverter:
convert: Callable[[float, str, str], float]
class EnergyConverter(BaseUnitConverter):
class BaseUnitConverterWithUnitConversion(BaseUnitConverter):
"""Define the format of a conversion utility."""
DEVICE_CLASS: str
UNIT_CONVERSION: dict[str, float]
@classmethod
def convert(cls, value: float, from_unit: str, to_unit: str) -> float:
"""Convert one unit of measurement to another."""
if from_unit not in cls.VALID_UNITS:
raise ValueError(
UNIT_NOT_RECOGNIZED_TEMPLATE.format(from_unit, cls.DEVICE_CLASS)
)
if to_unit not in cls.VALID_UNITS:
raise ValueError(
UNIT_NOT_RECOGNIZED_TEMPLATE.format(to_unit, cls.DEVICE_CLASS)
)
if not isinstance(value, Number):
raise TypeError(f"{value} is not of numeric type")
if from_unit == to_unit:
return value
new_value = value / cls.UNIT_CONVERSION[from_unit]
return new_value * cls.UNIT_CONVERSION[to_unit]
class EnergyConverter(BaseUnitConverterWithUnitConversion):
"""Utility to convert energy values."""
NORMALIZED_UNIT = energy_util.NORMALIZED_UNIT
VALID_UNITS = energy_util.VALID_UNITS
convert = energy_util.convert
DEVICE_CLASS = "energy"
NORMALIZED_UNIT = ENERGY_KILO_WATT_HOUR
UNIT_CONVERSION: dict[str, float] = {
ENERGY_WATT_HOUR: 1 * 1000,
ENERGY_KILO_WATT_HOUR: 1,
ENERGY_MEGA_WATT_HOUR: 1 / 1000,
}
VALID_UNITS: tuple[str, ...] = (
ENERGY_WATT_HOUR,
ENERGY_KILO_WATT_HOUR,
ENERGY_MEGA_WATT_HOUR,
)
class PowerConverter(BaseUnitConverter):
class PowerConverter(BaseUnitConverterWithUnitConversion):
"""Utility to convert power values."""
NORMALIZED_UNIT = power_util.NORMALIZED_UNIT
VALID_UNITS = power_util.VALID_UNITS
convert = power_util.convert
DEVICE_CLASS = "power"
NORMALIZED_UNIT = POWER_WATT
UNIT_CONVERSION: dict[str, float] = {
POWER_WATT: 1,
POWER_KILO_WATT: 1 / 1000,
}
VALID_UNITS: tuple[str, ...] = (
POWER_WATT,
POWER_KILO_WATT,
)
class PressureConverter(BaseUnitConverter):
"""Utility to convert pressure values."""
NORMALIZED_UNIT = pressure_util.NORMALIZED_UNIT
NORMALIZED_UNIT = PRESSURE_PA
VALID_UNITS = pressure_util.VALID_UNITS
convert = pressure_util.convert
@ -47,7 +102,7 @@ class PressureConverter(BaseUnitConverter):
class TemperatureConverter(BaseUnitConverter):
"""Utility to convert temperature values."""
NORMALIZED_UNIT = temperature_util.NORMALIZED_UNIT
NORMALIZED_UNIT = TEMP_CELSIUS
VALID_UNITS = temperature_util.VALID_UNITS
convert = temperature_util.convert
@ -55,6 +110,6 @@ class TemperatureConverter(BaseUnitConverter):
class VolumeConverter(BaseUnitConverter):
"""Utility to convert volume values."""
NORMALIZED_UNIT = volume_util.NORMALIZED_UNIT
NORMALIZED_UNIT = VOLUME_CUBIC_METERS
VALID_UNITS = volume_util.VALID_UNITS
convert = volume_util.convert

View File

@ -1,72 +0,0 @@
"""Test Home Assistant eneergy utility functions."""
import pytest
from homeassistant.const import (
ENERGY_KILO_WATT_HOUR,
ENERGY_MEGA_WATT_HOUR,
ENERGY_WATT_HOUR,
)
import homeassistant.util.energy as energy_util
INVALID_SYMBOL = "bob"
VALID_SYMBOL = ENERGY_KILO_WATT_HOUR
def test_convert_same_unit():
"""Test conversion from any unit to same unit."""
assert energy_util.convert(2, ENERGY_WATT_HOUR, ENERGY_WATT_HOUR) == 2
assert energy_util.convert(3, ENERGY_KILO_WATT_HOUR, ENERGY_KILO_WATT_HOUR) == 3
assert energy_util.convert(4, ENERGY_MEGA_WATT_HOUR, ENERGY_MEGA_WATT_HOUR) == 4
def test_convert_invalid_unit():
"""Test exception is thrown for invalid units."""
with pytest.raises(ValueError):
energy_util.convert(5, INVALID_SYMBOL, VALID_SYMBOL)
with pytest.raises(ValueError):
energy_util.convert(5, VALID_SYMBOL, INVALID_SYMBOL)
def test_convert_nonnumeric_value():
"""Test exception is thrown for nonnumeric type."""
with pytest.raises(TypeError):
energy_util.convert("a", ENERGY_WATT_HOUR, ENERGY_KILO_WATT_HOUR)
def test_convert_from_wh():
"""Test conversion from Wh to other units."""
watthours = 10
assert (
energy_util.convert(watthours, ENERGY_WATT_HOUR, ENERGY_KILO_WATT_HOUR) == 0.01
)
assert (
energy_util.convert(watthours, ENERGY_WATT_HOUR, ENERGY_MEGA_WATT_HOUR)
== 0.00001
)
def test_convert_from_kwh():
"""Test conversion from kWh to other units."""
kilowatthours = 10
assert (
energy_util.convert(kilowatthours, ENERGY_KILO_WATT_HOUR, ENERGY_WATT_HOUR)
== 10000
)
assert (
energy_util.convert(kilowatthours, ENERGY_KILO_WATT_HOUR, ENERGY_MEGA_WATT_HOUR)
== 0.01
)
def test_convert_from_mwh():
"""Test conversion from W to other units."""
megawatthours = 10
assert (
energy_util.convert(megawatthours, ENERGY_MEGA_WATT_HOUR, ENERGY_WATT_HOUR)
== 10000000
)
assert (
energy_util.convert(megawatthours, ENERGY_MEGA_WATT_HOUR, ENERGY_KILO_WATT_HOUR)
== 10000
)

View File

@ -1,41 +0,0 @@
"""Test Home Assistant power utility functions."""
import pytest
from homeassistant.const import POWER_KILO_WATT, POWER_WATT
import homeassistant.util.power as power_util
INVALID_SYMBOL = "bob"
VALID_SYMBOL = POWER_WATT
def test_convert_same_unit():
"""Test conversion from any unit to same unit."""
assert power_util.convert(2, POWER_WATT, POWER_WATT) == 2
assert power_util.convert(3, POWER_KILO_WATT, POWER_KILO_WATT) == 3
def test_convert_invalid_unit():
"""Test exception is thrown for invalid units."""
with pytest.raises(ValueError):
power_util.convert(5, INVALID_SYMBOL, VALID_SYMBOL)
with pytest.raises(ValueError):
power_util.convert(5, VALID_SYMBOL, INVALID_SYMBOL)
def test_convert_nonnumeric_value():
"""Test exception is thrown for nonnumeric type."""
with pytest.raises(TypeError):
power_util.convert("a", POWER_WATT, POWER_KILO_WATT)
def test_convert_from_kw():
"""Test conversion from kW to other units."""
kilowatts = 10
assert power_util.convert(kilowatts, POWER_KILO_WATT, POWER_WATT) == 10000
def test_convert_from_w():
"""Test conversion from W to other units."""
watts = 10
assert power_util.convert(watts, POWER_WATT, POWER_KILO_WATT) == 0.01

View File

@ -0,0 +1,89 @@
"""Test Home Assistant eneergy utility functions."""
import pytest
from homeassistant.const import (
ENERGY_KILO_WATT_HOUR,
ENERGY_MEGA_WATT_HOUR,
ENERGY_WATT_HOUR,
POWER_KILO_WATT,
POWER_WATT,
)
from homeassistant.util.unit_conversion import (
BaseUnitConverter,
EnergyConverter,
PowerConverter,
)
INVALID_SYMBOL = "bob"
@pytest.mark.parametrize(
"converter,valid_unit",
[
(EnergyConverter, ENERGY_WATT_HOUR),
(EnergyConverter, ENERGY_KILO_WATT_HOUR),
(EnergyConverter, ENERGY_MEGA_WATT_HOUR),
(PowerConverter, POWER_WATT),
(PowerConverter, POWER_KILO_WATT),
],
)
def test_convert_same_unit(converter: type[BaseUnitConverter], valid_unit: str) -> None:
"""Test conversion from any valid unit to same unit."""
assert converter.convert(2, valid_unit, valid_unit) == 2
@pytest.mark.parametrize(
"converter,valid_unit",
[
(EnergyConverter, ENERGY_KILO_WATT_HOUR),
(PowerConverter, POWER_WATT),
],
)
def test_convert_invalid_unit(
converter: type[BaseUnitConverter], valid_unit: str
) -> None:
"""Test exception is thrown for invalid units."""
with pytest.raises(ValueError):
converter.convert(5, INVALID_SYMBOL, valid_unit)
with pytest.raises(ValueError):
EnergyConverter.convert(5, valid_unit, INVALID_SYMBOL)
@pytest.mark.parametrize(
"converter,from_unit,to_unit",
[
(EnergyConverter, ENERGY_WATT_HOUR, ENERGY_KILO_WATT_HOUR),
(PowerConverter, POWER_WATT, POWER_KILO_WATT),
],
)
def test_convert_nonnumeric_value(
converter: type[BaseUnitConverter], from_unit: str, to_unit: str
) -> None:
"""Test exception is thrown for nonnumeric type."""
with pytest.raises(TypeError):
converter.convert("a", from_unit, to_unit)
@pytest.mark.parametrize(
"converter,value,from_unit,expected,to_unit",
[
(EnergyConverter, 10, ENERGY_WATT_HOUR, 0.01, ENERGY_KILO_WATT_HOUR),
(EnergyConverter, 10, ENERGY_WATT_HOUR, 0.00001, ENERGY_MEGA_WATT_HOUR),
(EnergyConverter, 10, ENERGY_KILO_WATT_HOUR, 10000, ENERGY_WATT_HOUR),
(EnergyConverter, 10, ENERGY_KILO_WATT_HOUR, 0.01, ENERGY_MEGA_WATT_HOUR),
(EnergyConverter, 10, ENERGY_MEGA_WATT_HOUR, 10000000, ENERGY_WATT_HOUR),
(EnergyConverter, 10, ENERGY_MEGA_WATT_HOUR, 10000, ENERGY_KILO_WATT_HOUR),
(PowerConverter, 10, POWER_KILO_WATT, 10000, POWER_WATT),
(PowerConverter, 10, POWER_WATT, 0.01, POWER_KILO_WATT),
],
)
def test_convert(
converter: type[BaseUnitConverter],
value: float,
from_unit: str,
expected: float,
to_unit: str,
) -> None:
"""Test conversion to other units."""
assert converter.convert(value, from_unit, to_unit) == expected