Added new calc sensor method, added token for webif, choose cubic spl…

…ines when scipy available

custom_components/weishaupt_modbus/__init__.py

@@ -2,7 +2,7 @@
 
 import json
 import logging
-from pathlib import Path
+# from pathlib import Path
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -51,7 +51,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: MyConfigEntry) -> bool:
     mbapi = ModbusAPI(config_entry=entry)
 
     if entry.data[CONF.CB_WEBIF]:
-        print
+        # print
         webapi = WebifConnection(config_entry=entry)
         await webapi.login()
     else:
@@ -158,6 +158,7 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: MyConfigEntry):
         new_data[CONF.CB_WEBIF] = False
         new_data[CONF.USERNAME] = ""
         new_data[CONF.PASSWORD] = ""
+        new_data[CONF.WEBIF_TOKEN] = ""
         hass.config_entries.async_update_entry(
             config_entry, data=new_data, minor_version=1, version=5
         )
@@ -242,7 +243,8 @@ def create_string_json() -> None:
     #    ...
 
     # load strings.json into string
-    with Path.open(
+    # replaced Path.open by open
+    with open(
         file="config/custom_components/weishaupt_modbus/strings.json",
         encoding="utf-8",
     ) as file:
@@ -252,7 +254,8 @@ def create_string_json() -> None:
     # overwrite entiy dict
     data_dict["entity"] = myEntity
     # write whole json to file again
-    with Path.open(
+    # replaced Path.open by open
+    with open(
         file="config/custom_components/weishaupt_modbus/strings.json",
         mode="w",
         encoding="utf-8",

custom_components/weishaupt_modbus/config_flow.py

@@ -64,7 +64,7 @@ async def validate_input(data: dict) -> dict[str, Any]:
     return {"title": data["host"]}
 
 
-class ConfigFlow(config_entries.ConfigFlow, domain=CONST.DOMAIN):
+class ConfigFlow(config_entries.ConfigFlow, domain=CONST.DOMAIN):  # pylint: disable=W0223
     """Class config flow."""
 
     VERSION = 5
@@ -105,6 +105,7 @@ async def async_step_user(self, user_input=None) -> config_entries.ConfigFlowRes
                 vol.Optional(schema=CONF.CB_WEBIF, default=False): bool,
                 vol.Optional(schema=CONF.USERNAME, default=""): str,
                 vol.Optional(schema=CONF.PASSWORD, default=""): str,
+                vol.Optional(schema=CONF.WEBIF_TOKEN, default=""): str,
             }
         )
 
@@ -119,7 +120,8 @@ async def async_step_user(self, user_input=None) -> config_entries.ConfigFlowRes
             except Exception:  # noqa: BLE001
                 errors["base"] = "unknown error"
 
-        # If there is no user input or there were errors, show the form again, including any errors that were found with the input.
+        # If there is no user input or there were errors, show the form again,
+        # #including any errors that were found with the input.
         return self.async_show_form(
             step_id="user", data_schema=data_schema, errors=errors
         )
@@ -187,6 +189,10 @@ async def async_step_reconfigure(
                 vol.Optional(
                     schema=CONF.PASSWORD, default=reconfigure_entry.data[CONF.PASSWORD]
                 ): str,
+                vol.Optional(
+                    schema=CONF.WEBIF_TOKEN,
+                    default=reconfigure_entry.data[CONF.WEBIF_TOKEN],
+                ): str,
             }
         )
 

custom_components/weishaupt_modbus/const.py

@@ -30,6 +30,7 @@ class ConfConstants:
     CB_WEBIF = "enable-webif"
     PASSWORD = CONF_PASSWORD
     USERNAME = CONF_USERNAME
+    WEBIF_TOKEN = "Web-IF-Token"
 
 
 CONF = ConfConstants()
@@ -120,15 +121,3 @@ class DeviceNameConstants:
 
 
 DEVICENAMES = DeviceNameConstants()
-
-
-@dataclass(frozen=True)
-class CalcConstants:
-    """Main constants."""
-
-    POWER = "power"
-    QUOTIENT = "quotient"
-    DIFFERENCE = "difference"
-
-
-CALCTYPES = CalcConstants()

custom_components/weishaupt_modbus/entities.py

@@ -11,7 +11,7 @@
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .configentry import MyConfigEntry
-from .const import CONF, CONST, FORMATS, CALCTYPES
+from .const import CONF, CONST, FORMATS
 from .coordinator import MyCoordinator, MyWebIfCoordinator
 from .hpconst import reverse_device_list
 from .items import ModbusItem, WebItem
@@ -85,7 +85,8 @@ def __init__(
         else:
             if self._api_item.params is not None:
                 self._attr_state_class = self._api_item.params.get(
-                    "stateclass",SensorStateClass.MEASUREMENT)
+                    "stateclass", SensorStateClass.MEASUREMENT
+                )
                 self._attr_native_unit_of_measurement = self._api_item.params.get(
                     "unit", ""
                 )
@@ -212,8 +213,8 @@ class MyCalcSensorEntity(MySensorEntity):
     """
 
     # calculates output from map
-    my_map = None
-    _calculation_type = None
+    _calculation_source = None
+    _calculation = None
 
     def __init__(
         self,
@@ -224,11 +225,17 @@ def __init__(
     ) -> None:
         """Initialize MyCalcSensorEntity."""
         MySensorEntity.__init__(self, config_entry, modbus_item, coordinator, idx)
+
         if self._api_item.params is not None:
-            self._calculation_type = self._api_item.params.get("calculation_type", None)
+            self._calculation_source = self._api_item.params.get("calculation", None)
 
-        if self._calculation_type == CALCTYPES.POWER:
-            self.my_map = config_entry.runtime_data.powermap
+        if self._calculation_source is not None:
+            try:
+                self._calculation = compile(
+                    self._calculation_source, "calulation", "eval"
+                )
+            except SyntaxError:
+                log.warning("Syntax error %s", self._calculation_source)
 
     @callback
     def _handle_coordinator_update(self) -> None:
@@ -237,64 +244,61 @@ def _handle_coordinator_update(self) -> None:
         self.async_write_ha_state()
 
     def translate_val(self, val):
-        if val is None:
-            return None
-        match self._calculation_type:
-            case CALCTYPES.POWER:
-                return self.power_val(val)
-            case CALCTYPES.QUOTIENT:
-                return self.quotient_val(val)
-            case CALCTYPES.DIFFERENCE:
-                return self.diff_val(val)
-
-    def power_val(self, val):
         """Translate a value from the modbus."""
-        val_0 = val / self._divider
-        val_x = self._config_entry.runtime_data.coordinator.get_value_from_item(
-            self._api_item.params.get("x", 1)
-        )
-        if val_x is None:
+        if self._calculation_source is None:
             return None
-        val_y = self._config_entry.runtime_data.coordinator.get_value_from_item(
-            self._api_item.params.get("y", 1)
-        )
-        if val_y is None:
+        if self._api_item.params is None:
             return None
-        return round(
-            (val_0 / 100) * self.my_map.map(val_x, val_y),
-            self._attr_suggested_display_precision,
-        )
+        if "val_1" in self._calculation_source:
+            val_1 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_1", 1)
+            )
+        if "val_2" in self._calculation_source:
+            val_2 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_2", 1)
+            )
+        if "val_3" in self._calculation_source:
+            val_3 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_3", 1)
+            )
+        if "val_4" in self._calculation_source:
+            val_4 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_4", 1)
+            )
+        if "val_5" in self._calculation_source:
+            val_5 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_5", 1)
+            )
+        if "val_6" in self._calculation_source:
+            val_6 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_6", 1)
+            )
+        if "val_7" in self._calculation_source:
+            val_7 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_7", 1)
+            )
+        if "val_8" in self._calculation_source:
+            val_8 = self._config_entry.runtime_data.coordinator.get_value_from_item(  # noqa F841 pylint: disable=W0612
+                self._api_item.params.get("val_8", 1)
+            )
+        if "power" in self._calculation_source:
+            power = self._config_entry.runtime_data.powermap  # noqa F841 pylint: disable=W0612
 
-    def quotient_val(self, val):
-        """Translate a value from the modbus."""
-        val_0 = val / self._divider
-        val_x = self._config_entry.runtime_data.coordinator.get_value_from_item(
-            self._api_item.params["denominator"]
-        )
-        if val_x is None:
+        try:
+            val_0 = val / self._divider  # noqa F841 pylint: disable=W0612
+            y = eval(self._calculation)  # pylint: disable=W0123
+        except ZeroDivisionError:
             return None
-        if val_0 is None:
+        except NameError:
+            log.warning("Variable not defined %s", self._calculation_source)
             return None
-        if val_x == 0:
+        except TypeError:
+            log.warning("No valid calulation string")
             return None
-        return round(val_0 / val_x, self._attr_suggested_display_precision)
-
-    def diff_val(self, val):
-        """Translate a value from the modbus."""
-        val_0 = val / self._divider
-        val_x = self._config_entry.runtime_data.coordinator.get_value_from_item(
-            self._api_item.params["diff_val"]
-        )
-        if val_x is None:
-            return None
-        if val_0 is None:
-            return None
-        return round(
-            val_0 - val_x / self._divider, self._attr_suggested_display_precision
-        )
+        return round(y, self._attr_suggested_display_precision)
 
 
-class MyNumberEntity(CoordinatorEntity, NumberEntity, MyEntity):
+class MyNumberEntity(CoordinatorEntity, NumberEntity, MyEntity):  # pylint: disable=W0223
     """Represent a Number Entity.
 
     Class that represents a sensor entity derived from Sensorentity
@@ -331,7 +335,7 @@ def device_info(self) -> DeviceInfo:
         return MyEntity.my_device_info(self)
 
 
-class MySelectEntity(CoordinatorEntity, SelectEntity, MyEntity):
+class MySelectEntity(CoordinatorEntity, SelectEntity, MyEntity):  # pylint: disable=W0223
     """Class that represents a sensor entity.
 
     Class that represents a sensor entity derived from Sensorentity
@@ -433,7 +437,7 @@ def _handle_coordinator_update(self) -> None:
                 "Update of %s failed. None response from server", self._api_item.name
             )
 
-    async def async_turn_on(self, **kwargs):
+    async def async_turn_on(self, **kwargs):  # pylint: disable=W0613
         """Turn the light on.
 
         Example method how to request data updates.