Add 'map_linear' and 'clamp' sensor filters (#5040)

esphome/components/sensor/__init__.py

@@ -31,6 +31,9 @@ from esphome.const import (
     CONF_MQTT_ID,
     CONF_FORCE_UPDATE,
     CONF_VALUE,
+    CONF_MIN_VALUE,
+    CONF_MAX_VALUE,
+    CONF_METHOD,
     DEVICE_CLASS_APPARENT_POWER,
     DEVICE_CLASS_AQI,
     DEVICE_CLASS_ATMOSPHERIC_PRESSURE,
@@ -227,6 +230,7 @@ OrFilter = sensor_ns.class_("OrFilter", Filter)
 CalibrateLinearFilter = sensor_ns.class_("CalibrateLinearFilter", Filter)
 CalibratePolynomialFilter = sensor_ns.class_("CalibratePolynomialFilter", Filter)
 SensorInRangeCondition = sensor_ns.class_("SensorInRangeCondition", Filter)
+ClampFilter = sensor_ns.class_("ClampFilter", Filter)
 
 validate_unit_of_measurement = cv.string_strict
 validate_accuracy_decimals = cv.int_
@@ -557,30 +561,60 @@ async def debounce_filter_to_code(config, filter_id):
     return var
 
 
-def validate_not_all_from_same(config):
-    if all(conf[CONF_FROM] == config[0][CONF_FROM] for conf in config):
-        raise cv.Invalid(
-            "The 'from' values of the calibrate_linear filter cannot all point "
-            "to the same value! Please add more values to the filter."
-        )
+CONF_DATAPOINTS = "datapoints"
+
+
+def validate_calibrate_linear(config):
+    datapoints = config[CONF_DATAPOINTS]
+    if config[CONF_METHOD] == "exact":
+        for i in range(len(datapoints) - 1):
+            if datapoints[i][CONF_FROM] > datapoints[i + 1][CONF_FROM]:
+                raise cv.Invalid(
+                    "The 'from' values of the calibrate_linear filter must be sorted in ascending order."
+                )
+        for i in range(len(datapoints) - 1):
+            if datapoints[i][CONF_FROM] == datapoints[i + 1][CONF_FROM]:
+                raise cv.Invalid(
+                    "The 'from' values of the calibrate_linear filter must not contain duplicates."
+                )
+    elif config[CONF_METHOD] == "least_squares":
+        if all(conf[CONF_FROM] == datapoints[0][CONF_FROM] for conf in datapoints):
+            raise cv.Invalid(
+                "The 'from' values of the calibrate_linear filter cannot all point "
+                "to the same value! Please add more values to the filter."
+            )
     return config
 
 
 @FILTER_REGISTRY.register(
     "calibrate_linear",
     CalibrateLinearFilter,
-    cv.All(
-        cv.ensure_list(validate_datapoint), cv.Length(min=2), validate_not_all_from_same
+    cv.maybe_simple_value(
+        {
+            cv.Required(CONF_DATAPOINTS): cv.All(
+                cv.ensure_list(validate_datapoint), cv.Length(min=2)
+            ),
+            cv.Optional(CONF_METHOD, default="least_squares"): cv.one_of(
+                "least_squares", "exact", lower=True
+            ),
+        },
+        validate_calibrate_linear,
+        key=CONF_DATAPOINTS,
     ),
 )
 async def calibrate_linear_filter_to_code(config, filter_id):
-    x = [conf[CONF_FROM] for conf in config]
-    y = [conf[CONF_TO] for conf in config]
-    k, b = fit_linear(x, y)
-    return cg.new_Pvariable(filter_id, k, b)
+    x = [conf[CONF_FROM] for conf in config[CONF_DATAPOINTS]]
+    y = [conf[CONF_TO] for conf in config[CONF_DATAPOINTS]]
+
+    linear_functions = []
+    if config[CONF_METHOD] == "least_squares":
+        k, b = fit_linear(x, y)
+        linear_functions = [[k, b, float("NaN")]]
+    elif config[CONF_METHOD] == "exact":
+        linear_functions = map_linear(x, y)
+    return cg.new_Pvariable(filter_id, linear_functions)
 
 
-CONF_DATAPOINTS = "datapoints"
 CONF_DEGREE = "degree"
 
 
@@ -619,6 +653,36 @@ async def calibrate_polynomial_filter_to_code(config, filter_id):
     return cg.new_Pvariable(filter_id, res)
 
 
+def validate_clamp(config):
+    if not math.isfinite(config[CONF_MIN_VALUE]) and not math.isfinite(
+        config[CONF_MAX_VALUE]
+    ):
+        raise cv.Invalid("Either 'min_value' or 'max_value' must be set to a number.")
+    if config[CONF_MIN_VALUE] > config[CONF_MAX_VALUE]:
+        raise cv.Invalid("The 'min_value' must not be larger than the 'max_value'.")
+    return config
+
+
+CLAMP_SCHEMA = cv.All(
+    cv.Schema(
+        {
+            cv.Optional(CONF_MIN_VALUE, default="NaN"): cv.float_,
+            cv.Optional(CONF_MAX_VALUE, default="NaN"): cv.float_,
+        }
+    ),
+    validate_clamp,
+)
+
+
+@FILTER_REGISTRY.register("clamp", ClampFilter, CLAMP_SCHEMA)
+async def clamp_filter_to_code(config, filter_id):
+    return cg.new_Pvariable(
+        filter_id,
+        config[CONF_MIN_VALUE],
+        config[CONF_MAX_VALUE],
+    )
+
+
 async def build_filters(config):
     return await cg.build_registry_list(FILTER_REGISTRY, config)
 
@@ -730,6 +794,22 @@ def fit_linear(x, y):
     return k, b
 
 
+def map_linear(x, y):
+    assert len(x) == len(y)
+    f = []
+    for i in range(len(x) - 1):
+        slope = (y[i + 1] - y[i]) / (x[i + 1] - x[i])
+        bias = y[i] - (slope * x[i])
+        next_x = x[i + 1]
+        if i == len(x) - 2:
+            next_x = float("NaN")
+        if f and f[-1][0] == slope and f[-1][1] == bias:
+            f[-1][2] = next_x
+        else:
+            f.append([slope, bias, next_x])
+    return f
+
+
 def _mat_copy(m):
     return [list(row) for row in m]