Add new power utility (#78867)

* Add power utility

* Fix tests

homeassistant/components/recorder/statistics.py

@@ -28,7 +28,6 @@ from homeassistant.const import (
     ENERGY_KILO_WATT_HOUR,
     ENERGY_MEGA_WATT_HOUR,
     ENERGY_WATT_HOUR,
-    POWER_KILO_WATT,
     POWER_WATT,
     PRESSURE_PA,
     TEMP_CELSIUS,
@@ -40,10 +39,13 @@ from homeassistant.helpers import entity_registry
 from homeassistant.helpers.json import JSONEncoder
 from homeassistant.helpers.storage import STORAGE_DIR
 from homeassistant.helpers.typing import UNDEFINED, UndefinedType
-import homeassistant.util.dt as dt_util
-import homeassistant.util.pressure as pressure_util
-import homeassistant.util.temperature as temperature_util
-import homeassistant.util.volume as volume_util
+from homeassistant.util import (
+    dt as dt_util,
+    power as power_util,
+    pressure as pressure_util,
+    temperature as temperature_util,
+    volume as volume_util,
+)
 
 from .const import DOMAIN, MAX_ROWS_TO_PURGE, SupportedDialect
 from .db_schema import Statistics, StatisticsMeta, StatisticsRuns, StatisticsShortTerm
@@ -156,9 +158,7 @@ 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
-    if to_unit == POWER_KILO_WATT:
-        return value / 1000
-    return value
+    return power_util.convert(value, POWER_WATT, to_unit)
 
 
 def _convert_pressure_from_pa(to_unit: str, value: float | None) -> float | None:

homeassistant/components/recorder/websocket_api.py

@@ -13,17 +13,18 @@ from homeassistant.const import (
     ENERGY_KILO_WATT_HOUR,
     ENERGY_MEGA_WATT_HOUR,
     ENERGY_WATT_HOUR,
-    POWER_KILO_WATT,
-    POWER_WATT,
     VOLUME_CUBIC_FEET,
     VOLUME_CUBIC_METERS,
 )
 from homeassistant.core import HomeAssistant, callback, valid_entity_id
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.json import JSON_DUMP
-from homeassistant.util import dt as dt_util
-import homeassistant.util.pressure as pressure_util
-import homeassistant.util.temperature as temperature_util
+from homeassistant.util import (
+    dt as dt_util,
+    power as power_util,
+    pressure as pressure_util,
+    temperature as temperature_util,
+)
 
 from .const import MAX_QUEUE_BACKLOG
 from .statistics import (
@@ -122,7 +123,7 @@ async def ws_handle_get_statistics_during_period(
                 vol.Optional("energy"): vol.Any(
                     ENERGY_WATT_HOUR, ENERGY_KILO_WATT_HOUR, ENERGY_MEGA_WATT_HOUR
                 ),
-                vol.Optional("power"): vol.Any(POWER_WATT, POWER_KILO_WATT),
+                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),
                 vol.Optional("volume"): vol.Any(VOLUME_CUBIC_FEET, VOLUME_CUBIC_METERS),

homeassistant/components/sensor/recorder.py

@@ -47,10 +47,13 @@ from homeassistant.const import (
 from homeassistant.core import HomeAssistant, State
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity import entity_sources
-import homeassistant.util.dt as dt_util
-import homeassistant.util.pressure as pressure_util
-import homeassistant.util.temperature as temperature_util
-import homeassistant.util.volume as volume_util
+from homeassistant.util import (
+    dt as dt_util,
+    power as power_util,
+    pressure as pressure_util,
+    temperature as temperature_util,
+    volume as volume_util,
+)
 
 from . import (
     ATTR_LAST_RESET,
@@ -89,8 +92,8 @@ UNIT_CONVERSIONS: dict[str, dict[str, Callable]] = {
     },
     # Convert power to W
     SensorDeviceClass.POWER: {
-        POWER_WATT: lambda x: x,
-        POWER_KILO_WATT: lambda x: x * 1000,
+        POWER_WATT: lambda x: x / power_util.UNIT_CONVERSION[POWER_WATT],
+        POWER_KILO_WATT: lambda x: x / power_util.UNIT_CONVERSION[POWER_KILO_WATT],
     },
     # Convert pressure to Pa
     # Note: pressure_util.convert is bypassed to avoid redundant error checking

homeassistant/util/power.py

@@ -0,0 +1,37 @@
+"""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,
+}
+
+
+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, "power"))
+    if unit_2 not in VALID_UNITS:
+        raise ValueError(UNIT_NOT_RECOGNIZED_TEMPLATE.format(unit_2, "power"))
+
+    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_power.py

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