Add Derivative component (#26456)

* create derivation component based on integration component remove left and right * Update test (was'n saved) * add some functionnal point test * Change derivation to derivative * Continue migration from derivation to derivative * Add codeowners info * fix tests typo and values * Improve code from reviewer add test case fix test values still a prefix issue that should not * create derivation component based on integration component remove left and right * Update test (was'n saved) * add some functionnal point test * Change derivation to derivative * Continue migration from derivation to derivative * Add codeowners info * fix tests typo and values * Improve code from reviewer add test case fix test values still a prefix issue that should not * Update homeassistant/components/derivative/sensor.py Fix test issue with unit of measurement Co-Authored-By: Santobert <tobhaase@gmail.com> * Fix review Move ValueError to SyntaxError * precise state test * un comment original tests and remove error tests * Fix isort issue * Fix review - update doc link - migrate to general const import * Rollback import conf_unit, just defined localy Co-authored-by: Santobert <tobhaase@gmail.com>
2025-04-24 01:08:12 +00:00 · 2020-01-20 07:17:06 +01:00 · 2020-01-20 07:17:06 +01:00 · b2212ad445
commit b2212ad445
parent 5f31b48f1d
6 changed files with 481 additions and 0 deletions
--- a/1
+++ b/1
@ -75,6 +75,7 @@ homeassistant/components/darksky/* @fabaff
 homeassistant/components/deconz/* @kane610
 homeassistant/components/delijn/* @bollewolle
 homeassistant/components/demo/* @home-assistant/core
+homeassistant/components/derivative/* @afaucogney
 homeassistant/components/device_automation/* @home-assistant/core
 homeassistant/components/digital_ocean/* @fabaff
 homeassistant/components/discogs/* @thibmaek
--- a/homeassistant/components/derivative/init.py
+++ b/homeassistant/components/derivative/init.py
@ -0,0 +1 @@
+"""The derivative component."""
--- a/homeassistant/components/derivative/manifest.json
+++ b/homeassistant/components/derivative/manifest.json
@ -0,0 +1,10 @@
+{
+  "domain": "derivative",
+  "name": "Derivative",
+  "documentation": "https://www.home-assistant.io/integrations/derivative",
+  "requirements": [],
+  "dependencies": [],
+  "codeowners": [
+    "@afaucogney"
+  ]
+}
--- a/homeassistant/components/derivative/sensor.py
+++ b/homeassistant/components/derivative/sensor.py
@ -0,0 +1,177 @@
+"""Numeric derivative of data coming from a source sensor over time."""
+from decimal import Decimal, DecimalException
+import logging
+
+import voluptuous as vol
+
+from homeassistant.components.sensor import PLATFORM_SCHEMA
+from homeassistant.const import (
+    ATTR_UNIT_OF_MEASUREMENT,
+    CONF_NAME,
+    CONF_SOURCE,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import callback
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.event import async_track_state_change
+from homeassistant.helpers.restore_state import RestoreEntity
+
+# mypy: allow-untyped-defs, no-check-untyped-defs
+
+_LOGGER = logging.getLogger(__name__)
+
+ATTR_SOURCE_ID = "source"
+
+CONF_ROUND_DIGITS = "round"
+CONF_UNIT_PREFIX = "unit_prefix"
+CONF_UNIT_TIME = "unit_time"
+CONF_UNIT = "unit"
+
+# SI Metric prefixes
+UNIT_PREFIXES = {None: 1, "k": 10 ** 3, "G": 10 ** 6, "T": 10 ** 9}
+
+# SI Time prefixes
+UNIT_TIME = {"s": 1, "min": 60, "h": 60 * 60, "d": 24 * 60 * 60}
+
+ICON = "mdi:chart-line"
+
+DEFAULT_ROUND = 3
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
+    {
+        vol.Optional(CONF_NAME): cv.string,
+        vol.Required(CONF_SOURCE): cv.entity_id,
+        vol.Optional(CONF_ROUND_DIGITS, default=DEFAULT_ROUND): vol.Coerce(int),
+        vol.Optional(CONF_UNIT_PREFIX, default=None): vol.In(UNIT_PREFIXES),
+        vol.Optional(CONF_UNIT_TIME, default="h"): vol.In(UNIT_TIME),
+        vol.Optional(CONF_UNIT): cv.string,
+    }
+)
+
+
+async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
+    """Set up the derivative sensor."""
+    derivative = DerivativeSensor(
+        config[CONF_SOURCE],
+        config.get(CONF_NAME),
+        config[CONF_ROUND_DIGITS],
+        config[CONF_UNIT_PREFIX],
+        config[CONF_UNIT_TIME],
+        config.get(CONF_UNIT),
+    )
+
+    async_add_entities([derivative])
+
+
+class DerivativeSensor(RestoreEntity):
+    """Representation of an derivative sensor."""
+
+    def __init__(
+        self,
+        source_entity,
+        name,
+        round_digits,
+        unit_prefix,
+        unit_time,
+        unit_of_measurement,
+    ):
+        """Initialize the derivative sensor."""
+        self._sensor_source_id = source_entity
+        self._round_digits = round_digits
+        self._state = 0
+
+        self._name = name if name is not None else f"{source_entity} derivative"
+
+        if unit_of_measurement is None:
+            final_unit_prefix = "" if unit_prefix is None else unit_prefix
+            self._unit_template = f"{final_unit_prefix}{{}}/{unit_time}"
+            # we postpone the definition of unit_of_measurement to later
+            self._unit_of_measurement = None
+        else:
+            self._unit_of_measurement = unit_of_measurement
+
+        self._unit_prefix = UNIT_PREFIXES[unit_prefix]
+        self._unit_time = UNIT_TIME[unit_time]
+
+    async def async_added_to_hass(self):
+        """Handle entity which will be added."""
+        await super().async_added_to_hass()
+        state = await self.async_get_last_state()
+        if state is not None:
+            try:
+                self._state = Decimal(state.state)
+            except SyntaxError as err:
+                _LOGGER.warning("Could not restore last state: %s", err)
+
+        @callback
+        def calc_derivative(entity, old_state, new_state):
+            """Handle the sensor state changes."""
+            if (
+                old_state is None
+                or old_state.state in [STATE_UNKNOWN, STATE_UNAVAILABLE]
+                or new_state.state in [STATE_UNKNOWN, STATE_UNAVAILABLE]
+            ):
+                return
+
+            if self._unit_of_measurement is None:
+                unit = new_state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
+                self._unit_of_measurement = self._unit_template.format(
+                    "" if unit is None else unit
+                )
+
+            try:
+                # derivative of previous measures.
+                gradient = 0
+                elapsed_time = (
+                    new_state.last_updated - old_state.last_updated
+                ).total_seconds()
+                gradient = Decimal(new_state.state) - Decimal(old_state.state)
+                derivative = gradient / (
+                    Decimal(elapsed_time) * (self._unit_prefix * self._unit_time)
+                )
+                assert isinstance(derivative, Decimal)
+            except ValueError as err:
+                _LOGGER.warning("While calculating derivative: %s", err)
+            except DecimalException as err:
+                _LOGGER.warning(
+                    "Invalid state (%s > %s): %s", old_state.state, new_state.state, err
+                )
+            except AssertionError as err:
+                _LOGGER.error("Could not calculate derivative: %s", err)
+            else:
+                self._state = derivative
+                self.async_schedule_update_ha_state()
+
+        async_track_state_change(self.hass, self._sensor_source_id, calc_derivative)
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def state(self):
+        """Return the state of the sensor."""
+        return round(self._state, self._round_digits)
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return self._unit_of_measurement
+
+    @property
+    def should_poll(self):
+        """No polling needed."""
+        return False
+
+    @property
+    def device_state_attributes(self):
+        """Return the state attributes of the sensor."""
+        state_attr = {ATTR_SOURCE_ID: self._sensor_source_id}
+        return state_attr
+
+    @property
+    def icon(self):
+        """Return the icon to use in the frontend."""
+        return ICON
--- a/tests/components/derivative/init.py
+++ b/tests/components/derivative/init.py
@ -0,0 +1 @@
+"""Tests for the derivative component."""
--- a/tests/components/derivative/test_sensor.py
+++ b/tests/components/derivative/test_sensor.py
@ -0,0 +1,291 @@
+"""The tests for the derivative sensor platform."""
+from datetime import timedelta
+from unittest.mock import patch
+
+from homeassistant.setup import async_setup_component
+import homeassistant.util.dt as dt_util
+
+
+async def test_state(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "derivative",
+            "source": "sensor.energy",
+            "unit": "kW",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 1, {})
+    await hass.async_block_till_done()
+
+    now = dt_util.utcnow() + timedelta(seconds=3600)
+    with patch("homeassistant.util.dt.utcnow", return_value=now):
+        hass.states.async_set(entity_id, 1, {}, force_update=True)
+        await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.derivative")
+    assert state is not None
+
+    # Testing a energy sensor at 1 kWh for 1hour = 0kW
+    assert round(float(state.state), config["sensor"]["round"]) == 0.0
+
+    assert state.attributes.get("unit_of_measurement") == "kW"
+
+
+async def test_dataSet1(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "power",
+            "source": "sensor.energy",
+            "unit_time": "s",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 0, {})
+    await hass.async_block_till_done()
+
+    # Testing a energy sensor with non-monotonic intervals and values
+    for time, value in [(20, 10), (30, 30), (40, 5), (50, 0)]:
+        now = dt_util.utcnow() + timedelta(seconds=time)
+        with patch("homeassistant.util.dt.utcnow", return_value=now):
+            hass.states.async_set(entity_id, value, {}, force_update=True)
+            await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.power")
+    assert state is not None
+
+    assert round(float(state.state), config["sensor"]["round"]) == -0.5
+
+
+async def test_dataSet2(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "power",
+            "source": "sensor.energy",
+            "unit_time": "s",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 0, {})
+    await hass.async_block_till_done()
+
+    # Testing a energy sensor with non-monotonic intervals and values
+    for time, value in [(20, 5), (30, 0)]:
+        now = dt_util.utcnow() + timedelta(seconds=time)
+        with patch("homeassistant.util.dt.utcnow", return_value=now):
+            hass.states.async_set(entity_id, value, {}, force_update=True)
+            await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.power")
+    assert state is not None
+
+    assert round(float(state.state), config["sensor"]["round"]) == -0.5
+
+
+async def test_dataSet3(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "power",
+            "source": "sensor.energy",
+            "unit_time": "s",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 0, {})
+    await hass.async_block_till_done()
+
+    # Testing a energy sensor with non-monotonic intervals and values
+    for time, value in [(20, 5), (30, 10)]:
+        now = dt_util.utcnow() + timedelta(seconds=time)
+        with patch("homeassistant.util.dt.utcnow", return_value=now):
+            hass.states.async_set(entity_id, value, {}, force_update=True)
+            await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.power")
+    assert state is not None
+
+    assert round(float(state.state), config["sensor"]["round"]) == 0.5
+
+    assert state.attributes.get("unit_of_measurement") == "/s"
+
+
+async def test_dataSet4(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "power",
+            "source": "sensor.energy",
+            "unit_time": "s",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 0, {})
+    await hass.async_block_till_done()
+
+    # Testing a energy sensor with non-monotonic intervals and values
+    for time, value in [(20, 5), (30, 5)]:
+        now = dt_util.utcnow() + timedelta(seconds=time)
+        with patch("homeassistant.util.dt.utcnow", return_value=now):
+            hass.states.async_set(entity_id, value, {}, force_update=True)
+            await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.power")
+    assert state is not None
+
+    assert round(float(state.state), config["sensor"]["round"]) == 0
+
+
+async def test_dataSet5(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "power",
+            "source": "sensor.energy",
+            "unit_time": "s",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 0, {})
+    await hass.async_block_till_done()
+
+    # Testing a energy sensor with non-monotonic intervals and values
+    for time, value in [(20, 10), (30, -10)]:
+        now = dt_util.utcnow() + timedelta(seconds=time)
+        with patch("homeassistant.util.dt.utcnow", return_value=now):
+            hass.states.async_set(entity_id, value, {}, force_update=True)
+            await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.power")
+    assert state is not None
+
+    assert round(float(state.state), config["sensor"]["round"]) == -2
+
+
+async def test_dataSet6(hass):
+    """Test derivative sensor state."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "power",
+            "source": "sensor.energy",
+            "round": 2,
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 0, {})
+    await hass.async_block_till_done()
+
+    # Testing a energy sensor with non-monotonic intervals and values
+    for time, value in [(20, 0), (30, 36000)]:
+        now = dt_util.utcnow() + timedelta(seconds=time)
+        with patch("homeassistant.util.dt.utcnow", return_value=now):
+            hass.states.async_set(entity_id, value, {}, force_update=True)
+            await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.power")
+    assert state is not None
+
+    assert round(float(state.state), config["sensor"]["round"]) == 1
+
+
+async def test_prefix(hass):
+    """Test derivative sensor state using a power source."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "derivative",
+            "source": "sensor.power",
+            "round": 2,
+            "unit_prefix": "k",
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(
+        entity_id, 1000, {"unit_of_measurement": "W"}, force_update=True
+    )
+    await hass.async_block_till_done()
+
+    now = dt_util.utcnow() + timedelta(seconds=3600)
+    with patch("homeassistant.util.dt.utcnow", return_value=now):
+        hass.states.async_set(
+            entity_id, 1000, {"unit_of_measurement": "W"}, force_update=True
+        )
+        await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.derivative")
+    assert state is not None
+
+    # Testing a power sensor at 1000 Watts for 1hour = 0kW/h
+    assert round(float(state.state), config["sensor"]["round"]) == 0.0
+    assert state.attributes.get("unit_of_measurement") == "kW/h"
+
+
+async def test_suffix(hass):
+    """Test derivative sensor state using a network counter source."""
+    config = {
+        "sensor": {
+            "platform": "derivative",
+            "name": "derivative",
+            "source": "sensor.bytes_per_second",
+            "round": 2,
+            "unit_prefix": "k",
+            "unit_time": "s",
+        }
+    }
+
+    assert await async_setup_component(hass, "sensor", config)
+
+    entity_id = config["sensor"]["source"]
+    hass.states.async_set(entity_id, 1000, {})
+    await hass.async_block_till_done()
+
+    now = dt_util.utcnow() + timedelta(seconds=10)
+    with patch("homeassistant.util.dt.utcnow", return_value=now):
+        hass.states.async_set(entity_id, 1000, {}, force_update=True)
+        await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.derivative")
+    assert state is not None
+
+    # Testing a network speed sensor at 1000 bytes/s over 10s  = 10kbytes/s2
+    assert round(float(state.state), config["sensor"]["round"]) == 0.0