Add quantiles to Statistics integration (#52189)

* Add quantiles as another Statistics attribute

Quantiles divide states into intervals of equal probability. The
statistics.quantiles() function was added in Python 3.8 and can now
be included in the Statistics integration without new dependencies.

Quantiles can be used in conjunction with other distribution metrics to
create box plots (quartiles) and other graphical resources for
visualizing the distribution of states.

* Add quantiles reference to basic tests

homeassistant/components/statistics/sensor.py

@@ -39,6 +39,7 @@ ATTR_MEAN = "mean"
 ATTR_MEDIAN = "median"
 ATTR_MIN_AGE = "min_age"
 ATTR_MIN_VALUE = "min_value"
+ATTR_QUANTILES = "quantiles"
 ATTR_SAMPLING_SIZE = "sampling_size"
 ATTR_STANDARD_DEVIATION = "standard_deviation"
 ATTR_TOTAL = "total"
@@ -47,10 +48,14 @@ ATTR_VARIANCE = "variance"
 CONF_SAMPLING_SIZE = "sampling_size"
 CONF_MAX_AGE = "max_age"
 CONF_PRECISION = "precision"
+CONF_QUANTILE_INTERVALS = "quantile_intervals"
+CONF_QUANTILE_METHOD = "quantile_method"
 
 DEFAULT_NAME = "Stats"
 DEFAULT_SIZE = 20
 DEFAULT_PRECISION = 2
+DEFAULT_QUANTILE_INTERVALS = 4
+DEFAULT_QUANTILE_METHOD = "exclusive"
 ICON = "mdi:calculator"
 
 PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@@ -62,6 +67,12 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
         ),
         vol.Optional(CONF_MAX_AGE): cv.time_period,
         vol.Optional(CONF_PRECISION, default=DEFAULT_PRECISION): vol.Coerce(int),
+        vol.Optional(
+            CONF_QUANTILE_INTERVALS, default=DEFAULT_QUANTILE_INTERVALS
+        ): vol.All(vol.Coerce(int), vol.Range(min=2)),
+        vol.Optional(CONF_QUANTILE_METHOD, default=DEFAULT_QUANTILE_METHOD): vol.In(
+            ["exclusive", "inclusive"]
+        ),
     }
 )
 
@@ -76,9 +87,22 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
     sampling_size = config.get(CONF_SAMPLING_SIZE)
     max_age = config.get(CONF_MAX_AGE)
     precision = config.get(CONF_PRECISION)
+    quantile_intervals = config.get(CONF_QUANTILE_INTERVALS)
+    quantile_method = config.get(CONF_QUANTILE_METHOD)
 
     async_add_entities(
-        [StatisticsSensor(entity_id, name, sampling_size, max_age, precision)], True
+        [
+            StatisticsSensor(
+                entity_id,
+                name,
+                sampling_size,
+                max_age,
+                precision,
+                quantile_intervals,
+                quantile_method,
+            )
+        ],
+        True,
     )
 
     return True
@@ -87,7 +111,16 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
 class StatisticsSensor(SensorEntity):
     """Representation of a Statistics sensor."""
 
-    def __init__(self, entity_id, name, sampling_size, max_age, precision):
+    def __init__(
+        self,
+        entity_id,
+        name,
+        sampling_size,
+        max_age,
+        precision,
+        quantile_intervals,
+        quantile_method,
+    ):
         """Initialize the Statistics sensor."""
         self._entity_id = entity_id
         self.is_binary = self._entity_id.split(".")[0] == "binary_sensor"
@@ -95,12 +128,14 @@ class StatisticsSensor(SensorEntity):
         self._sampling_size = sampling_size
         self._max_age = max_age
         self._precision = precision
+        self._quantile_intervals = quantile_intervals
+        self._quantile_method = quantile_method
         self._unit_of_measurement = None
         self.states = deque(maxlen=self._sampling_size)
         self.ages = deque(maxlen=self._sampling_size)
 
         self.count = 0
-        self.mean = self.median = self.stdev = self.variance = None
+        self.mean = self.median = self.quantiles = self.stdev = self.variance = None
         self.total = self.min = self.max = None
         self.min_age = self.max_age = None
         self.change = self.average_change = self.change_rate = None
@@ -191,6 +226,7 @@ class StatisticsSensor(SensorEntity):
                 ATTR_COUNT: self.count,
                 ATTR_MEAN: self.mean,
                 ATTR_MEDIAN: self.median,
+                ATTR_QUANTILES: self.quantiles,
                 ATTR_STANDARD_DEVIATION: self.stdev,
                 ATTR_VARIANCE: self.variance,
                 ATTR_TOTAL: self.total,
@@ -257,9 +293,18 @@ class StatisticsSensor(SensorEntity):
             try:  # require at least two data points
                 self.stdev = round(statistics.stdev(self.states), self._precision)
                 self.variance = round(statistics.variance(self.states), self._precision)
+                if self._quantile_intervals < self.count:
+                    self.quantiles = [
+                        round(quantile, self._precision)
+                        for quantile in statistics.quantiles(
+                            self.states,
+                            n=self._quantile_intervals,
+                            method=self._quantile_method,
+                        )
+                    ]
             except statistics.StatisticsError as err:
                 _LOGGER.debug("%s: %s", self.entity_id, err)
-                self.stdev = self.variance = STATE_UNKNOWN
+                self.stdev = self.variance = self.quantiles = STATE_UNKNOWN
 
             if self.states:
                 self.total = round(sum(self.states), self._precision)

tests/components/statistics/test_sensor.py

@@ -48,6 +48,9 @@ class TestStatisticsSensor(unittest.TestCase):
         self.median = round(statistics.median(self.values), 2)
         self.deviation = round(statistics.stdev(self.values), 2)
         self.variance = round(statistics.variance(self.values), 2)
+        self.quantiles = [
+            round(quantile, 2) for quantile in statistics.quantiles(self.values)
+        ]
         self.change = round(self.values[-1] - self.values[0], 2)
         self.average_change = round(self.change / (len(self.values) - 1), 2)
         self.change_rate = round(self.change / (60 * (self.count - 1)), 2)
@@ -112,6 +115,7 @@ class TestStatisticsSensor(unittest.TestCase):
         assert self.variance == state.attributes.get("variance")
         assert self.median == state.attributes.get("median")
         assert self.deviation == state.attributes.get("standard_deviation")
+        assert self.quantiles == state.attributes.get("quantiles")
         assert self.mean == state.attributes.get("mean")
         assert self.count == state.attributes.get("count")
         assert self.total == state.attributes.get("total")
@@ -188,6 +192,7 @@ class TestStatisticsSensor(unittest.TestCase):
         # require at least two data points
         assert state.attributes.get("variance") == STATE_UNKNOWN
         assert state.attributes.get("standard_deviation") == STATE_UNKNOWN
+        assert state.attributes.get("quantiles") == STATE_UNKNOWN
 
     def test_max_age(self):
         """Test value deprecation."""