Add Compensation Integration

CODEOWNERS

@@ -88,6 +88,7 @@ homeassistant/components/cloud/* @home-assistant/cloud
 homeassistant/components/cloudflare/* @ludeeus @ctalkington
 homeassistant/components/color_extractor/* @GenericStudent
 homeassistant/components/comfoconnect/* @michaelarnauts
+homeassistant/components/compensation/* @Petro31
 homeassistant/components/config/* @home-assistant/core
 homeassistant/components/configurator/* @home-assistant/core
 homeassistant/components/control4/* @lawtancool

homeassistant/components/compensation/__init__.py

@@ -0,0 +1,121 @@
+"""The Compensation integration."""
+import logging
+import warnings
+
+import numpy as np
+import voluptuous as vol
+
+from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.const import (
+    CONF_ATTRIBUTE,
+    CONF_ENTITY_ID,
+    CONF_NAME,
+    CONF_UNIT_OF_MEASUREMENT,
+)
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.discovery import async_load_platform
+
+from .const import (
+    CONF_COMPENSATION,
+    CONF_DATAPOINTS,
+    CONF_DEGREE,
+    CONF_POLYNOMIAL,
+    CONF_PRECISION,
+    DATA_COMPENSATION,
+    DEFAULT_DEGREE,
+    DEFAULT_NAME,
+    DEFAULT_PRECISION,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def datapoints_greater_than_degree(value: dict) -> dict:
+    """Validate data point list is greater than polynomial degrees."""
+    if len(value[CONF_DATAPOINTS]) <= value[CONF_DEGREE]:
+        raise vol.Invalid(
+            f"{CONF_DATAPOINTS} must have at least {value[CONF_DEGREE]+1} {CONF_DATAPOINTS}"
+        )
+
+    return value
+
+
+COMPENSATION_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_ENTITY_ID): cv.entity_id,
+        vol.Required(CONF_DATAPOINTS): [
+            vol.ExactSequence([vol.Coerce(float), vol.Coerce(float)])
+        ],
+        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
+        vol.Optional(CONF_ATTRIBUTE): cv.string,
+        vol.Optional(CONF_PRECISION, default=DEFAULT_PRECISION): cv.positive_int,
+        vol.Optional(CONF_DEGREE, default=DEFAULT_DEGREE): vol.All(
+            vol.Coerce(int),
+            vol.Range(min=1, max=7),
+        ),
+        vol.Optional(CONF_UNIT_OF_MEASUREMENT): cv.string,
+    }
+)
+
+CONFIG_SCHEMA = vol.Schema(
+    {
+        DOMAIN: vol.Schema(
+            {cv.slug: vol.All(COMPENSATION_SCHEMA, datapoints_greater_than_degree)}
+        )
+    },
+    extra=vol.ALLOW_EXTRA,
+)
+
+
+async def async_setup(hass, config):
+    """Set up the Compensation sensor."""
+    hass.data[DATA_COMPENSATION] = {}
+
+    for compensation, conf in config.get(DOMAIN).items():
+        _LOGGER.debug("Setup %s.%s", DOMAIN, compensation)
+
+        degree = conf[CONF_DEGREE]
+
+        # get x values and y values from the x,y point pairs
+        x_values, y_values = zip(*conf[CONF_DATAPOINTS])
+
+        # try to get valid coefficients for a polynomial
+        coefficients = None
+        with np.errstate(all="raise"):
+            with warnings.catch_warnings(record=True) as all_warnings:
+                warnings.simplefilter("always")
+                try:
+                    coefficients = np.polyfit(x_values, y_values, degree)
+                except FloatingPointError as error:
+                    _LOGGER.error(
+                        "Setup of %s encountered an error, %s",
+                        compensation,
+                        error,
+                    )
+                for warning in all_warnings:
+                    _LOGGER.warning(
+                        "Setup of %s encountered a warning, %s",
+                        compensation,
+                        str(warning.message).lower(),
+                    )
+
+        if coefficients is not None:
+            data = {
+                k: v for k, v in conf.items() if k not in [CONF_DEGREE, CONF_DATAPOINTS]
+            }
+            data[CONF_POLYNOMIAL] = np.poly1d(coefficients)
+
+            hass.data[DATA_COMPENSATION][compensation] = data
+
+            hass.async_create_task(
+                async_load_platform(
+                    hass,
+                    SENSOR_DOMAIN,
+                    DOMAIN,
+                    {CONF_COMPENSATION: compensation},
+                    config,
+                )
+            )
+
+    return True

homeassistant/components/compensation/const.py

@@ -0,0 +1,18 @@
+"""Compensation constants."""
+DOMAIN = "compensation"
+
+SENSOR = "compensation"
+
+CONF_COMPENSATION = "compensation"
+CONF_DATAPOINTS = "data_points"
+CONF_DEGREE = "degree"
+CONF_PRECISION = "precision"
+CONF_POLYNOMIAL = "polynomial"
+
+DATA_COMPENSATION = "compensation_data"
+
+DEFAULT_NAME = "Compensation"
+DEFAULT_DEGREE = 1
+DEFAULT_PRECISION = 2
+
+MATCH_DATAPOINT = r"([-+]?[0-9]+\.?[0-9]*){1} -> ([-+]?[0-9]+\.?[0-9]*){1}"

homeassistant/components/compensation/manifest.json

@@ -0,0 +1,7 @@
+{
+  "domain": "compensation",
+  "name": "Compensation",
+  "documentation": "https://www.home-assistant.io/integrations/compensation",
+  "requirements": ["numpy==1.19.2"],
+  "codeowners": ["@Petro31"]
+}

homeassistant/components/compensation/sensor.py

@@ -0,0 +1,144 @@
+"""Support for compensation sensor."""
+import logging
+
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    ATTR_UNIT_OF_MEASUREMENT,
+    CONF_ATTRIBUTE,
+    CONF_ENTITY_ID,
+    CONF_NAME,
+    CONF_UNIT_OF_MEASUREMENT,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import callback
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.event import async_track_state_change_event
+
+from .const import CONF_COMPENSATION, CONF_POLYNOMIAL, CONF_PRECISION, DATA_COMPENSATION
+
+_LOGGER = logging.getLogger(__name__)
+
+ATTR_ATTRIBUTE = "attribute"
+ATTR_COEFFICIENTS = "coefficients"
+
+
+async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
+    """Set up the Compensation sensor."""
+    if discovery_info is None:
+        return
+
+    compensation = discovery_info.get(CONF_COMPENSATION)
+    conf = hass.data[DATA_COMPENSATION][compensation]
+
+    async_add_entities(
+        [
+            CompensationSensor(
+                hass,
+                conf[CONF_ENTITY_ID],
+                conf.get(CONF_NAME),
+                conf.get(CONF_ATTRIBUTE),
+                conf[CONF_PRECISION],
+                conf[CONF_POLYNOMIAL],
+                conf.get(CONF_UNIT_OF_MEASUREMENT),
+            )
+        ]
+    )
+
+
+class CompensationSensor(Entity):
+    """Representation of a Compensation sensor."""
+
+    def __init__(
+        self,
+        hass,
+        entity_id,
+        name,
+        attribute,
+        precision,
+        polynomial,
+        unit_of_measurement,
+    ):
+        """Initialize the Compensation sensor."""
+        self._entity_id = entity_id
+        self._name = name
+        self._precision = precision
+        self._attribute = attribute
+        self._unit_of_measurement = unit_of_measurement
+        self._poly = polynomial
+        self._coefficients = polynomial.coefficients.tolist()
+        self._state = STATE_UNKNOWN
+
+    async def async_added_to_hass(self):
+        """Handle added to Hass."""
+        self.async_on_remove(
+            async_track_state_change_event(
+                self.hass,
+                [self._entity_id],
+                self._async_compensation_sensor_state_listener,
+            )
+        )
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def should_poll(self):
+        """No polling needed."""
+        return False
+
+    @property
+    def state(self):
+        """Return the state of the sensor."""
+        return self._state
+
+    @property
+    def device_state_attributes(self):
+        """Return the state attributes of the sensor."""
+        ret = {
+            ATTR_ENTITY_ID: self._entity_id,
+            ATTR_COEFFICIENTS: self._coefficients,
+        }
+        if self._attribute:
+            ret[ATTR_ATTRIBUTE] = self._attribute
+        return ret
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return self._unit_of_measurement
+
+    @callback
+    def _async_compensation_sensor_state_listener(self, event):
+        """Handle sensor state changes."""
+        new_state = event.data.get("new_state")
+        if new_state is None:
+            return
+
+        if self._unit_of_measurement is None and self._attribute is None:
+            self._unit_of_measurement = new_state.attributes.get(
+                ATTR_UNIT_OF_MEASUREMENT
+            )
+
+        try:
+            if self._attribute:
+                value = float(new_state.attributes.get(self._attribute))
+            else:
+                value = (
+                    None if new_state.state == STATE_UNKNOWN else float(new_state.state)
+                )
+            self._state = round(self._poly(value), self._precision)
+
+        except (ValueError, TypeError):
+            self._state = STATE_UNKNOWN
+            if self._attribute:
+                _LOGGER.warning(
+                    "%s attribute %s is not numerical",
+                    self._entity_id,
+                    self._attribute,
+                )
+            else:
+                _LOGGER.warning("%s state is not numerical", self._entity_id)
+
+        self.async_write_ha_state()

requirements_all.txt

@@ -1009,6 +1009,7 @@ nuheat==0.3.0
 # homeassistant.components.numato
 numato-gpio==0.10.0
 
+# homeassistant.components.compensation
 # homeassistant.components.iqvia
 # homeassistant.components.opencv
 # homeassistant.components.tensorflow

requirements_test_all.txt

@@ -518,6 +518,7 @@ nuheat==0.3.0
 # homeassistant.components.numato
 numato-gpio==0.10.0
 
+# homeassistant.components.compensation
 # homeassistant.components.iqvia
 # homeassistant.components.opencv
 # homeassistant.components.tensorflow

tests/components/compensation/__init__.py

@@ -0,0 +1 @@
+"""Tests for the compensation component."""