feat: get price_euro_per_wh_accu from home-assistant/openhab

Author: igor-dueck-gcx

src/CONFIG_README.md

@@ -200,6 +200,18 @@ A default config file will be created with the first start, if there is no `conf
 - **`price_euro_per_wh_accu`**:
   Price for battery in €/Wh - can be used to shift the result over the day according to the available energy (more details follow).
 
+- **`battery.price_euro_per_wh_source`**:  
+  Determines where the battery energy cost should be read from.  
+  - `config` *(default)*: Use the static `price_euro_per_wh_accu` value.  
+  - `homeassistant`: Fetch the price from a Home Assistant entity defined via `battery.price_euro_per_wh_sensor`. Requires `battery.url` and `battery.access_token`.  
+  - `openhab`: Fetch the price from an OpenHAB item defined via `battery.price_euro_per_wh_sensor`. Requires `battery.url`.
+
+- **`battery.price_euro_per_wh_sensor`**:  
+  Sensor/item identifier that exposes the battery price in €/Wh.  
+  - For Home Assistant: Entity ID (e.g., `sensor.battery_price`).  
+  - For OpenHAB: Item name (e.g., `BatteryPrice`).  
+  Only used when `battery.price_euro_per_wh_source` is set to `homeassistant` or `openhab`.
+
 - **`battery.charging_curve_enabled`**:  
   Enables or disables the dynamic charging curve for the battery.  
   - `true`: The system will automatically reduce the maximum charging power as the battery SOC increases, helping to protect battery health and optimize efficiency.  
@@ -471,6 +483,8 @@ battery:
   min_soc_percentage: 5 # URL for battery soc in %
   max_soc_percentage: 100 # URL for battery soc in %
   price_euro_per_wh_accu: 0 # price for battery in €/Wh
+  price_euro_per_wh_source: config # use static config value or fetch from homeassistant
+  price_euro_per_wh_sensor: "" # Home Assistant entity (e.g. sensor.battery_price) providing €/Wh
   charging_curve_enabled: true # enable dynamic charging curve for battery
 # List of PV forecast source configuration
 pv_forecast_source:
@@ -548,6 +562,7 @@ battery:
   min_soc_percentage: 5 # URL for battery soc in %
   max_soc_percentage: 100 # URL for battery soc in %
   price_euro_per_wh_accu: 0 # price for battery in €/Wh
+  price_euro_per_wh_source: config # use static config value or fetch from homeassistant
   charging_curve_enabled: true # enable dynamic charging curve for battery
 # List of PV forecast source configuration
 pv_forecast_source:

src/config.py

@@ -89,6 +89,8 @@ def create_default_config(self):
                         "min_soc_percentage": 5,
                         "max_soc_percentage": 100,
                         "price_euro_per_wh_accu": 0.0,  # price for battery in euro/Wh
+                        "price_euro_per_wh_source": "config",  # config (static), homeassistant, or openhab
+                        "price_euro_per_wh_sensor": "",  # sensor/item providing battery energy cost in €/Wh
                         "charging_curve_enabled": True,  # enable charging curve
                     }
                 ),
@@ -281,6 +283,14 @@ def create_default_config(self):
         config["battery"].yaml_add_eol_comment(
             "price for battery in euro/Wh - default: 0.0", "price_euro_per_wh_accu"
         )
+        config["battery"].yaml_add_eol_comment(
+            "source for battery price: config (static), homeassistant, or openhab",
+            "price_euro_per_wh_source",
+        )
+        config["battery"].yaml_add_eol_comment(
+            "sensor/item providing the battery price (€/Wh) - HA entity or OpenHAB item",
+            "price_euro_per_wh_sensor",
+        )
         config["battery"].yaml_add_eol_comment(
             "enabling charging curve for controlled charging power"
             + " according to the SOC (default: true)",

src/eos_connect.py

@@ -493,9 +493,7 @@ def get_ems_data(dst_change_detected):
             "pv_prognose_wh": pv_prognose_wh,
             "strompreis_euro_pro_wh": strompreis_euro_pro_wh,
             "einspeiseverguetung_euro_pro_wh": einspeiseverguetung_euro_pro_wh,
-            "preis_euro_pro_wh_akku": config_manager.config["battery"][
-                "price_euro_per_wh_accu"
-            ],
+            "preis_euro_pro_wh_akku": battery_interface.get_price_euro_per_wh(),
             "gesamtlast": gesamtlast,
         }
 

src/interfaces/battery_interface.py

@@ -77,6 +77,9 @@ def __init__(self, config, on_bat_max_changed=None):
         self.on_bat_max_changed = on_bat_max_changed
         self.min_soc_set = config.get("min_soc_percentage", 0)
         self.max_soc_set = config.get("max_soc_percentage", 100)
+        self.price_euro_per_wh = float(config.get("price_euro_per_wh_accu", 0.0))
+        self.price_source = config.get("price_euro_per_wh_source", "config")
+        self.price_sensor = config.get("price_euro_per_wh_sensor", "")
 
         self.soc_fail_count = 0
 
@@ -192,6 +195,109 @@ def __battery_request_current_soc(self):
             )
         return self.current_soc
 
+    def __fetch_price_data_from_openhab(self):
+        """
+        Fetch the current battery energy price from an OpenHAB item.
+
+        Returns:
+            float: Battery energy cost in €/Wh provided by the configured item.
+        """
+        if not self.price_sensor:
+            raise ValueError("price_euro_per_wh_sensor must be configured for OpenHAB.")
+
+        logger.debug("[BATTERY-IF] getting price from openhab ...")
+        openhab_url = self.url + "/rest/items/" + self.price_sensor
+        try:
+            response = requests.get(openhab_url, timeout=6)
+            response.raise_for_status()
+            data = response.json()
+            raw_state = str(data["state"]).strip()
+            # Take only the first part before any space (handles "0.0001", "0.0001 €/Wh", etc.)
+            cleaned_value = raw_state.split()[0]
+            price = float(cleaned_value)
+            logger.debug("[BATTERY-IF] Fetched price from OpenHAB: %s €/Wh", price)
+            return price
+        except requests.exceptions.Timeout:
+            raise requests.exceptions.Timeout(
+                "Request timed out while fetching price from OpenHAB"
+            )
+        except requests.exceptions.RequestException as e:
+            raise requests.exceptions.RequestException(
+                f"Error fetching price from OpenHAB: {e}"
+            )
+
+    def __fetch_price_data_from_homeassistant(self):
+        """
+        Fetch the current battery energy price from a Home Assistant sensor.
+
+        Returns:
+            float: Battery energy cost in €/Wh provided by the configured sensor.
+        """
+        if not self.price_sensor:
+            raise ValueError(
+                "price_euro_per_wh_sensor must be configured for Home Assistant."
+            )
+
+        logger.debug("[BATTERY-IF] getting price from homeassistant ...")
+        homeassistant_url = f"{self.url}/api/states/{self.price_sensor}"
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+            "Content-Type": "application/json",
+        }
+        response = requests.get(homeassistant_url, headers=headers, timeout=6)
+        response.raise_for_status()
+        entity_data = response.json()
+        price = float(entity_data["state"])
+        logger.debug("[BATTERY-IF] Fetched price from Home Assistant: %s €/Wh", price)
+        return price
+
+    def __update_price_euro_per_wh(self):
+        """
+        Update the battery price from the configured source if needed.
+        """
+        if self.price_source == "config":
+            return self.price_euro_per_wh
+
+        source_name = self.price_source.upper()
+        try:
+            if self.price_source == "homeassistant":
+                latest_price = self.__fetch_price_data_from_homeassistant()
+            elif self.price_source == "openhab":
+                latest_price = self.__fetch_price_data_from_openhab()
+            else:
+                logger.warning(
+                    "[BATTERY-IF] Unknown price source '%s'. Keeping last value %s.",
+                    self.price_source,
+                    self.price_euro_per_wh,
+                )
+                return self.price_euro_per_wh
+        except requests.exceptions.Timeout:
+            logger.warning(
+                "[BATTERY-IF] %s - Request timed out while fetching "
+                + "price_euro_per_wh_accu. Keeping last value %s.",
+                source_name,
+                self.price_euro_per_wh,
+            )
+            return self.price_euro_per_wh
+        except (requests.exceptions.RequestException, ValueError, KeyError) as exc:
+            logger.warning(
+                "[BATTERY-IF] %s - Error fetching price sensor data: %s. "
+                + "Keeping last value %s.",
+                source_name,
+                exc,
+                self.price_euro_per_wh,
+            )
+            return self.price_euro_per_wh
+
+        self.price_euro_per_wh = latest_price
+        logger.debug(
+            "[BATTERY-IF] Updated price_euro_per_wh_accu from %s sensor %s: %s",
+            self.price_source,
+            self.price_sensor,
+            self.price_euro_per_wh,
+        )
+        return self.price_euro_per_wh
+
     def _handle_soc_error(self, source, error, last_soc):
         self.soc_fail_count += 1
         if self.soc_fail_count < 5:
@@ -236,6 +342,12 @@ def get_min_soc(self):
         """
         return self.min_soc_set
 
+    def get_price_euro_per_wh(self):
+        """
+        Returns the current battery price in €/Wh.
+        """
+        return self.price_euro_per_wh
+
     def set_min_soc(self, min_soc):
         """
         Sets the minimum state of charge (SOC) percentage of the battery.
@@ -402,6 +514,7 @@ def _update_state_loop(self):
                     ),
                 )
                 self.__get_max_charge_power_dyn()
+                self.__update_price_euro_per_wh()
 
             except (requests.exceptions.RequestException, ValueError, KeyError) as e:
                 logger.error("[BATTERY-IF] Error while updating state: %s", e)

tests/interfaces/test_battery_interface.py

@@ -7,6 +7,7 @@
 
 from unittest.mock import patch, MagicMock
 import pytest
+import requests
 from src.interfaces.battery_interface import BatteryInterface
 
 # Accessing protected members is fine in white-box tests.
@@ -28,6 +29,9 @@ def default_config():
         "max_soc_percentage": 90,
         "charging_curve_enabled": True,
         "discharge_efficiency": 1.0,
+        "price_euro_per_wh_accu": 0.0,
+        "price_euro_per_wh_source": "config",
+        "price_euro_per_wh_sensor": "",
     }
 
 
@@ -106,6 +110,125 @@ def test_homeassistant_fetch_success(default_config):
         assert soc == 55.0
 
 
+def test_homeassistant_price_sensor_success(default_config):
+    """
+    Ensure the Home Assistant price sensor value is fetched and stored.
+    """
+    test_config = default_config.copy()
+    test_config.update(
+        {
+            "url": "http://fake",
+            "access_token": "token",
+            "price_euro_per_wh_source": "homeassistant",
+            "price_euro_per_wh_sensor": "sensor.accu_price",
+        }
+    )
+    with patch("src.interfaces.battery_interface.requests.get") as mock_get:
+        mock_resp = MagicMock()
+        mock_resp.json.return_value = {"state": "0.002"}
+        mock_resp.raise_for_status.return_value = None
+        mock_get.return_value = mock_resp
+        bi = BatteryInterface(test_config)
+        # Ensure manual update works and the getter reflects the sensor value
+        bi._BatteryInterface__update_price_euro_per_wh()
+        assert bi.get_price_euro_per_wh() == pytest.approx(0.002)
+        bi.shutdown()
+
+
+def test_homeassistant_price_sensor_failure_keeps_last_value(default_config):
+    """
+    Ensure failing sensor updates keep the last configured price.
+    """
+    test_config = default_config.copy()
+    test_config.update(
+        {
+            "url": "http://fake",
+            "access_token": "token",
+            "price_euro_per_wh_source": "homeassistant",
+            "price_euro_per_wh_sensor": "sensor.accu_price",
+            "price_euro_per_wh_accu": 0.001,
+        }
+    )
+    with patch(
+        "src.interfaces.battery_interface.requests.get",
+        side_effect=requests.exceptions.RequestException("boom"),
+    ):
+        bi = BatteryInterface(test_config)
+        bi._BatteryInterface__update_price_euro_per_wh()
+        assert bi.get_price_euro_per_wh() == pytest.approx(0.001)
+        bi.shutdown()
+
+
+def test_openhab_price_sensor_success(default_config):
+    """
+    Ensure the OpenHAB price item value is fetched and stored.
+    """
+    test_config = default_config.copy()
+    test_config.update(
+        {
+            "url": "http://fake",
+            "price_euro_per_wh_source": "openhab",
+            "price_euro_per_wh_sensor": "BatteryPrice",
+        }
+    )
+    with patch("src.interfaces.battery_interface.requests.get") as mock_get:
+        mock_resp = MagicMock()
+        mock_resp.json.return_value = {"state": "0.00015"}
+        mock_resp.raise_for_status.return_value = None
+        mock_get.return_value = mock_resp
+        bi = BatteryInterface(test_config)
+        # Ensure manual update works and the getter reflects the item value
+        bi._BatteryInterface__update_price_euro_per_wh()
+        assert bi.get_price_euro_per_wh() == pytest.approx(0.00015)
+        bi.shutdown()
+
+
+def test_openhab_price_sensor_with_unit_success(default_config):
+    """
+    Ensure OpenHAB price item with unit (e.g., "0.00015 €/Wh") is parsed correctly.
+    """
+    test_config = default_config.copy()
+    test_config.update(
+        {
+            "url": "http://fake",
+            "price_euro_per_wh_source": "openhab",
+            "price_euro_per_wh_sensor": "BatteryPrice",
+        }
+    )
+    with patch("src.interfaces.battery_interface.requests.get") as mock_get:
+        mock_resp = MagicMock()
+        mock_resp.json.return_value = {"state": "0.00015 €/Wh"}
+        mock_resp.raise_for_status.return_value = None
+        mock_get.return_value = mock_resp
+        bi = BatteryInterface(test_config)
+        bi._BatteryInterface__update_price_euro_per_wh()
+        assert bi.get_price_euro_per_wh() == pytest.approx(0.00015)
+        bi.shutdown()
+
+
+def test_openhab_price_sensor_failure_keeps_last_value(default_config):
+    """
+    Ensure failing OpenHAB item updates keep the last configured price.
+    """
+    test_config = default_config.copy()
+    test_config.update(
+        {
+            "url": "http://fake",
+            "price_euro_per_wh_source": "openhab",
+            "price_euro_per_wh_sensor": "BatteryPrice",
+            "price_euro_per_wh_accu": 0.0001,
+        }
+    )
+    with patch(
+        "src.interfaces.battery_interface.requests.get",
+        side_effect=requests.exceptions.RequestException("boom"),
+    ):
+        bi = BatteryInterface(test_config)
+        bi._BatteryInterface__update_price_euro_per_wh()
+        assert bi.get_price_euro_per_wh() == pytest.approx(0.0001)
+        bi.shutdown()
+
+
 def test_soc_error_handling(default_config):
     """
     Test SOC error handling and fail count reset.