Add new energy utility (#78883)

* Add new energy utility

* Adjust STATISTIC_UNIT_TO_VALID_UNITS

homeassistant/components/recorder/statistics.py

@@ -26,8 +26,6 @@ import voluptuous as vol
 
 from homeassistant.const import (
     ENERGY_KILO_WATT_HOUR,
-    ENERGY_MEGA_WATT_HOUR,
-    ENERGY_WATT_HOUR,
     POWER_WATT,
     PRESSURE_PA,
     TEMP_CELSIUS,
@@ -41,6 +39,7 @@ 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,
@@ -138,20 +137,12 @@ 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
-    if to_unit == ENERGY_MEGA_WATT_HOUR:
-        return value / 1000
-    if to_unit == ENERGY_WATT_HOUR:
-        return value * 1000
-    return value
+    return energy_util.convert(value, ENERGY_KILO_WATT_HOUR, to_unit)
 
 
 def _convert_energy_to_kwh(from_unit: str, value: float) -> float:
     """Convert energy in from_unit to kWh."""
-    if from_unit == ENERGY_MEGA_WATT_HOUR:
-        return value * 1000
-    if from_unit == ENERGY_WATT_HOUR:
-        return value / 1000
-    return value
+    return energy_util.convert(value, from_unit, ENERGY_KILO_WATT_HOUR)
 
 
 def _convert_power_from_w(to_unit: str, value: float | None) -> float | None:
@@ -196,11 +187,7 @@ STATISTIC_UNIT_TO_UNIT_CLASS: dict[str | None, str] = {
 }
 
 STATISTIC_UNIT_TO_VALID_UNITS: dict[str | None, Iterable[str | None]] = {
-    ENERGY_KILO_WATT_HOUR: [
-        ENERGY_KILO_WATT_HOUR,
-        ENERGY_MEGA_WATT_HOUR,
-        ENERGY_WATT_HOUR,
-    ],
+    ENERGY_KILO_WATT_HOUR: energy_util.VALID_UNITS,
     POWER_WATT: power_util.VALID_UNITS,
     PRESSURE_PA: pressure_util.VALID_UNITS,
     TEMP_CELSIUS: temperature_util.VALID_UNITS,

homeassistant/components/recorder/websocket_api.py

@@ -21,6 +21,7 @@ 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,
@@ -120,9 +121,7 @@ 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.Any(
-                    ENERGY_WATT_HOUR, ENERGY_KILO_WATT_HOUR, ENERGY_MEGA_WATT_HOUR
-                ),
+                vol.Optional("energy"): vol.In(energy_util.VALID_UNITS),
                 vol.Optional("power"): vol.In(power_util.VALID_UNITS),
                 vol.Optional("pressure"): vol.In(pressure_util.VALID_UNITS),
                 vol.Optional("temperature"): vol.In(temperature_util.VALID_UNITS),

homeassistant/components/sensor/recorder.py

@@ -49,6 +49,7 @@ 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,
@@ -86,9 +87,11 @@ DEVICE_CLASS_UNITS: dict[str, str] = {
 UNIT_CONVERSIONS: dict[str, dict[str, Callable]] = {
     # Convert energy to kWh
     SensorDeviceClass.ENERGY: {
-        ENERGY_KILO_WATT_HOUR: lambda x: x,
-        ENERGY_MEGA_WATT_HOUR: lambda x: x * 1000,
-        ENERGY_WATT_HOUR: lambda x: x / 1000,
+        ENERGY_KILO_WATT_HOUR: lambda x: x
+        / energy_util.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],
     },
     # Convert power to W
     SensorDeviceClass.POWER: {

homeassistant/util/energy.py

@@ -0,0 +1,40 @@
+"""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,
+}
+
+
+def convert(value: float, unit_1: str, unit_2: str) -> float:
+    """Convert one unit of measurement to another."""
+    if unit_1 not in VALID_UNITS:
+        raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(unit_1, "energy"))
+    if unit_2 not in VALID_UNITS:
+        raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(unit_2, "energy"))
+
+    if not isinstance(value, Number):
+        raise TypeError(f"{value} is not of numeric type")
+
+    if unit_1 == unit_2:
+        return value
+
+    watts = value / UNIT_CONVERSION[unit_1]
+    return watts * UNIT_CONVERSION[unit_2]

tests/util/test_energy.py

@@ -0,0 +1,72 @@
+"""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
+    )