Merge pull request #180 from moritzmair/main

Adding Energyforecast.de as a data source

Author: suaveolent

README.md

@@ -26,6 +26,9 @@ You can choose between multiple sources:
 4. smartENERGY.at
    [smartENERGY.at](https://www.smartenergy.at/api-schnittstellen) provides a free of charge service for their customers. Market price data is available for Austria. So far no user identifiation is required.
 
+5. Energyforecast.de
+   [Energyforecast.de](https://www.energyforecast.de/api-docs/index.html) provides services to get market price data forecasts for Germany up to 96 hours into the future. An API token is required.
+
 If you like this component, please give it a star on [github](https://github.com/mampfes/hacs_epex_spot).
 
 ## Installation

custom_components/epex_spot/EPEXSpot/Energyforecast/__init__.py

@@ -0,0 +1,80 @@
+"""Energyforecast.de"""
+
+from datetime import datetime
+import logging
+
+import aiohttp
+
+
+from ...const import UOM_EUR_PER_KWH
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Marketprice:
+    """Marketprice class for Energyforecast."""
+
+    def __init__(self, data):
+        self._start_time = datetime.fromisoformat(data["start"])
+        self._end_time = datetime.fromisoformat(data["end"])
+        self._price_per_kwh = round(float(data["price"]), 6)
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(start: {self._start_time.isoformat()}, end: {self._end_time.isoformat()}, marketprice: {self._price_per_kwh} {UOM_EUR_PER_KWH})"  # noqa: E501
+
+    @property
+    def start_time(self):
+        return self._start_time
+
+    @property
+    def end_time(self):
+        return self._end_time
+
+    @property
+    def price_per_kwh(self):
+        return self._price_per_kwh
+
+class Energyforecast:
+    URL = "https://www.energyforecast.de/api/v1/predictions/prices_for_ha"
+
+    MARKET_AREAS = ("de",)
+
+    def __init__(self, market_area, token: str, session: aiohttp.ClientSession):
+        self._token = token
+        self._session = session
+        self._market_area = market_area
+        self._marketdata = []
+
+    @property
+    def name(self):
+        return "Energyforecast API V1"
+
+    @property
+    def market_area(self):
+        return self._market_area
+
+    @property
+    def duration(self):
+        return 60
+
+    @property
+    def currency(self):
+        return "EUR"
+
+    @property
+    def marketdata(self):
+        return self._marketdata
+
+    async def fetch(self):
+        data = await self._fetch_data(self.URL)
+        self._marketdata = self._extract_marketdata(data["forecast"]["data"])
+
+    async def _fetch_data(self, url):
+        async with self._session.get(
+            url, params={"token": self._token, "fixed_cost_cent": 0, "vat": 0}
+        ) as resp:
+            resp.raise_for_status()
+            return await resp.json()
+
+    def _extract_marketdata(self, data):
+        return [Marketprice(entry) for entry in data]

custom_components/epex_spot/SourceShell.py

@@ -22,6 +22,7 @@
     CONF_SOURCE_SMARD_DE,
     CONF_SOURCE_SMARTENERGY,
     CONF_SOURCE_TIBBER,
+    CONF_SOURCE_ENERGYFORECAST,
     CONF_SURCHARGE_ABS,
     CONF_SURCHARGE_PERC,
     CONF_TAX,
@@ -31,7 +32,7 @@
     DEFAULT_TAX,
     EMPTY_EXTREME_PRICE_INTERVAL_RESP,
 )
-from .EPEXSpot import SMARD, Awattar, EPEXSpotWeb, Tibber, smartENERGY
+from .EPEXSpot import SMARD, Awattar, EPEXSpotWeb, Tibber, smartENERGY, Energyforecast
 from .extreme_price_interval import find_extreme_price_interval, get_start_times
 
 _LOGGER = logging.getLogger(__name__)
@@ -68,6 +69,12 @@ def __init__(self, config_entry: ConfigEntry, session: aiohttp.ClientSession):
                 token=self._config_entry.data[CONF_TOKEN],
                 session=session,
             )
+        elif config_entry.data[CONF_SOURCE] == CONF_SOURCE_ENERGYFORECAST:
+            self._source = Energyforecast.Energyforecast(
+                market_area=config_entry.data[CONF_MARKET_AREA],
+                token=self._config_entry.data[CONF_TOKEN],
+                session=session
+            )
 
     @property
     def unique_id(self):

custom_components/epex_spot/config_flow.py

@@ -16,6 +16,7 @@
     CONF_SOURCE_SMARD_DE,
     CONF_SOURCE_SMARTENERGY,
     CONF_SOURCE_TIBBER,
+    CONF_SOURCE_ENERGYFORECAST,
     CONF_SURCHARGE_ABS,
     CONF_SURCHARGE_PERC,
     CONF_TAX,
@@ -26,14 +27,15 @@
     DEFAULT_TAX,
     DOMAIN,
 )
-from .EPEXSpot import SMARD, Awattar, EPEXSpotWeb, Tibber, smartENERGY
+from .EPEXSpot import SMARD, Awattar, EPEXSpotWeb, Tibber, smartENERGY, Energyforecast
 
 CONF_SOURCE_LIST = (
     CONF_SOURCE_AWATTAR,
     CONF_SOURCE_EPEX_SPOT_WEB,
     CONF_SOURCE_SMARD_DE,
     CONF_SOURCE_SMARTENERGY,
     CONF_SOURCE_TIBBER,
+    CONF_SOURCE_ENERGYFORECAST,
 )
 
 
@@ -79,6 +81,15 @@ async def async_step_source(self, user_input=None):
                     vol.Required(CONF_TOKEN): vol.Coerce(str),
                 }
             )
+        # Energyforecast API requires a token
+        elif self._source_name == CONF_SOURCE_ENERGYFORECAST:
+            areas = Energyforecast.Energyforecast.MARKET_AREAS
+            data_schema = vol.Schema(
+                {
+                    vol.Required(CONF_MARKET_AREA): vol.In(sorted(areas)),
+                    vol.Required(CONF_TOKEN): vol.Coerce(str)
+                }
+            )
         else:
             if self._source_name == CONF_SOURCE_AWATTAR:
                 areas = Awattar.Awattar.MARKET_AREAS

custom_components/epex_spot/const.py

@@ -24,6 +24,7 @@
 CONF_SOURCE_SMARD_DE = "SMARD.de"
 CONF_SOURCE_SMARTENERGY = "smartENERGY.at"
 CONF_SOURCE_TIBBER = "Tibber"
+CONF_SOURCE_ENERGYFORECAST = "Energyforecast.de"
 
 # configuration options for net price calculation
 CONF_SURCHARGE_PERC = "percentage_surcharge"

custom_components/epex_spot/test_energyforecast.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+
+import aiohttp
+import asyncio
+
+from .EPEXSpot import Energyforecast
+from .const import UOM_EUR_PER_KWH
+
+DEMO_TOKEN = "demo_token" # The "demo_token" token only provides up to 24 hours of forecast data into the future.
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        service = Energyforecast.Energyforecast(market_area="DE-LU", token=DEMO_TOKEN, session=session)
+
+        await service.fetch()
+        print(f"count = {len(service.marketdata)}")
+        for e in service.marketdata:
+            print(f"{e.start_time}: {e.price_per_kwh} {UOM_EUR_PER_KWH}")
+
+
+asyncio.run(main())