Merge pull request #706 from davidusb-geek/davidusb-geek/fix/typical_lists_warning_logs

Fix typical load when maximum_power_from_grid is a list and uneeded w…

Author: davidusb-geek

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.16.2 - 2026-01-29
+### Improvement
+- Added support for a thermal inertia parameter for the basic thermal model
+- Handle time-dependent battery weights (@rmounce)
+### Fix
+- Fix Docker signal handling with tini and pin uvicorn version (@rmounce)
+- Fix "typical" load forecast method fails with maximum_power_from_grid as vector (list)
+- Fix warning logging messages on sensor_replace_zero
+- Fix for correct handling of open-meteo temp_air variable
+
 ## 0.16.1 - 2026-01-23
 ### Fix
 - Hot fix for InfluxDB not passed credentials

conda/meta.yaml

@@ -1,5 +1,5 @@
 {% set name = "emhass" %}
-{% set version = "0.16.1" %}
+{% set version = "0.16.2" %}
 
 package:
   name: {{ name|lower }}
@@ -45,7 +45,7 @@ requirements:
     - aiohttp
     - orjson
     - websockets
-    - uvicorn >=0.30.0
+    - uvicorn==0.30.6
     - influxdb >=5.3.1
 
 test:

docs/conf.py

@@ -23,7 +23,7 @@
 author = "David HERNANDEZ TORRES"
 
 # The full version, including alpha/beta/rc tags
-release = "0.16.1"
+release = "0.16.2"
 
 # -- General configuration ---------------------------------------------------
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "emhass"
-version = "0.16.1"
+version = "0.16.2"
 description = "An Energy Management System for Home Assistant"
 readme = "README.md"
 requires-python = ">=3.10, <3.13"

src/emhass/forecast.py

@@ -1380,14 +1380,30 @@ async def _get_load_forecast_typical(self) -> pd.DataFrame:
             data.columns = ["load"]
             forecast_tmp, used_days = Forecast.get_typical_load_forecast(data, forecast_date)
             self.logger.debug(f"Using {len(used_days)} days of data to generate the forecast.")
-            forecast_tmp = forecast_tmp * self.plant_conf["maximum_power_from_grid"] / 9000
             if len(forecast) == 0:
                 forecast = forecast_tmp
             else:
                 forecast = pd.concat([forecast, forecast_tmp], axis=0)
         forecast_out = forecast.loc[forecast.index.intersection(self.forecast_dates)]
         forecast_out.index = self.forecast_dates
         forecast_out.index.name = "ts"
+        max_power = self.plant_conf["maximum_power_from_grid"]
+        # Normalize list/array inputs to a Series aligned with forecast index
+        if isinstance(max_power, list | tuple | np.ndarray):
+            # Validate length to prevent confusing errors later
+            if len(max_power) != len(forecast_out):
+                raise ValueError(
+                    f"The length of 'maximum_power_from_grid' ({len(max_power)}) "
+                    f"does not match the forecast horizon length ({len(forecast_out)})."
+                )
+            # Create Series for explicit temporal alignment
+            scaling_factor = pd.Series(max_power, index=forecast_out.index)
+        else:
+            # Assume scalar (int/float)
+            scaling_factor = max_power
+        # Apply scaling
+        # The '9000' divisor appears to be a normalization constant specific to the 'typical' model data
+        forecast_out["load"] = forecast_out["load"] * scaling_factor / 9000
         return forecast_out.rename(columns={"load": "yhat"})
 
     def _get_load_forecast_naive(self, df: pd.DataFrame) -> pd.DataFrame:

src/emhass/retrieve_hass.py

@@ -846,15 +846,23 @@ def _map_variable_names(
     ) -> list | None:
         """Helper to map old variable names to new ones (if renaming occurred)."""
         if not target_list:
-            self.logger.warning(f"Unable to find all the sensors in {param_name} parameter")
+            # Dynamic Warning Message
+            # Don't hardcode "sensor_power_photovoltaics". Use the actual parameter name.
+            self.logger.warning(f"The list of sensors for parameter '{param_name}' is empty.")
             self.logger.warning(
-                f"Confirm sure all sensors in {param_name} are sensor_power_photovoltaics and/or sensor_power_load_no_var_loads"
+                f"Please verify that the sensors defined in '{param_name}' match "
+                "the sensors connected to EMHASS."
             )
             return None
         new_list = []
         for string in target_list:
             if not skip_renaming:
-                new_list.append(string.replace(var_load, var_load + "_positive"))
+                # Exact Match Logic
+                # Prevent dangerous substring replacements (e.g. 'sensor.power' inside 'sensor.power_meter')
+                if string == var_load:
+                    new_list.append(var_load + "_positive")
+                else:
+                    new_list.append(string)
             else:
                 new_list.append(string)
         return new_list
@@ -896,8 +904,13 @@ def prepare_data(
         # Instead of calling self._validate_sensor_list (which warns),
         # we silently drop sensors that are not in the current fetched data.
         if var_replace_zero:
+            # Optional: Log if we are dropping items to help debugging
+            missing_sensors = [x for x in var_replace_zero if x not in self.var_list]
+            if missing_sensors:
+                self.logger.debug(
+                    f"Sensors in 'sensor_replace_zero' not found in data: {missing_sensors}"
+                )
             var_replace_zero = [x for x in var_replace_zero if x in self.var_list]
-
         if var_interp:
             var_interp = [x for x in var_interp if x in self.var_list]
         # Rename Load Columns (Handle sign change)
@@ -908,17 +921,19 @@ def prepare_data(
             self.df_final.clip(lower=0.0, inplace=True, axis=1)
             self.df_final.replace(to_replace=0.0, value=np.nan, inplace=True)
         # Map Variable Names (Update lists to match new column names)
-        # Note: We still use _map_variable_names to handle the renaming logic
-        # (e.g. adding '_positive' suffix), but the input lists are now clean.
-        new_var_replace_zero = self._map_variable_names(
-            var_replace_zero, var_load, skip_renaming, "sensor_replace_zero"
-        )
-        new_var_interp = self._map_variable_names(
-            var_interp, var_load, skip_renaming, "sensor_linear_interp"
-        )
+        # Only call mapping if the list is not empty to avoid spurious warnings
+        new_var_replace_zero = None
+        if var_replace_zero:
+            new_var_replace_zero = self._map_variable_names(
+                var_replace_zero, var_load, skip_renaming, "sensor_replace_zero"
+            )
+        new_var_interp = None
+        if var_interp:
+            new_var_interp = self._map_variable_names(
+                var_interp, var_load, skip_renaming, "sensor_linear_interp"
+            )
         # Apply Data Cleaning (FillNA / Interpolate)
         if new_var_replace_zero:
-            # Intersection check to ensure columns exist before assigning
             cols_to_fix = [c for c in new_var_replace_zero if c in self.df_final.columns]
             if cols_to_fix:
                 self.df_final[cols_to_fix] = self.df_final[cols_to_fix].fillna(0.0)