Electricity tariffs (EKZ, Groupe E)

packages/modules/electricity_tariffs/ekz/config.py

@@ -0,0 +1,16 @@
+class EkzTariffConfiguration:
+    def __init__(self):
+        self.country = "ch"
+        self.unit = "rp"
+
+
+class EkzTariff:
+    def __init__(self,
+                 name: str = "EKZ (CH)",
+                 type: str = "ekz",
+                 official: bool = False,
+                 configuration: EkzTariffConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.official = official
+        self.configuration = configuration or EkzTariffConfiguration()

packages/modules/electricity_tariffs/ekz/tariff.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+from datetime import datetime, timezone, timedelta
+from dateutil import tz
+from urllib.parse import quote
+from typing import Dict
+from modules.common import req
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_state import TariffState
+from modules.electricity_tariffs.ekz.config import EkzTariffConfiguration
+from modules.electricity_tariffs.ekz.config import EkzTariff
+
+
+# Extract timestamp from power price entry
+def timestamp(power):
+    return str(int(datetime.strptime(power['start_timestamp'], "%Y-%m-%dT%H:%M:%S%z")
+                   .astimezone(tz.tzutc()).timestamp()))
+
+
+# Read prices from EKZ API
+def readApi() -> list[tuple[str, float]]:
+    endpoint = "https://api.tariffs.ekz.ch/v1/tariffs"
+    tariff_power = "electricity_dynamic"
+    tariff_grid = "grid_400D_inclFees"
+    utcnow = datetime.now(timezone.utc)
+    startDate = utcnow.strftime("%Y-%m-%dT%H:00:00Z")
+    endDate = (utcnow + timedelta(days=2)).strftime("%Y-%m-%dT%H:00:00Z")
+    session = req.get_http_session()
+    power_raw = session.get(
+        url=endpoint +
+        f"?tariff_name={tariff_power}&start_timestamp={quote(startDate)}&end_timestamp={quote(endDate)}",
+        ).json()["prices"]
+    grid_raw = session.get(
+        url=endpoint +
+        f"?tariff_name={tariff_grid}&start_timestamp={quote(startDate)}&end_timestamp={quote(endDate)}",
+        ).json()["prices"]
+    return [(timestamp(power), (power['electricity'][1]['value']+grid['grid'][1]['value'])/1000)
+            for power, grid in zip(power_raw, grid_raw)]
+
+
+# Fetch electricity prices from EKZ API
+# API Reference: https://api.tariffs.ekz.ch/swagger
+def fetch_prices(config: EkzTariffConfiguration) -> Dict[str, float]:
+    pricelist = readApi()
+    prices: Dict[str, float] = dict(pricelist)
+    return prices
+
+
+def create_electricity_tariff(config: EkzTariff):
+    def updater():
+        return TariffState(prices=fetch_prices(config.configuration))
+    return updater
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=EkzTariff)

packages/modules/electricity_tariffs/groupe_e/config.py

@@ -0,0 +1,15 @@
+class GroupeETariffConfiguration:
+    def __init__(self):
+        self.country = "ch"
+
+
+class GroupeETariff:
+    def __init__(self,
+                 name: str = "Groupe E (CH)",
+                 type: str = "groupe_e",
+                 official: bool = False,
+                 configuration: GroupeETariffConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.official = official
+        self.configuration = configuration or GroupeETariffConfiguration()

packages/modules/electricity_tariffs/groupe_e/tariff.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+from datetime import datetime, timedelta
+from dateutil import tz
+from urllib.parse import quote
+from typing import Dict
+from modules.common import req
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_state import TariffState
+from modules.electricity_tariffs.groupe_e.config import GroupeETariffConfiguration
+from modules.electricity_tariffs.groupe_e.config import GroupeETariff
+
+
+# Extract timestamp from power price entry
+def timestamp(power):
+    return str(int(datetime.strptime(power['start_timestamp'], "%Y-%m-%dT%H:%M:%S%z")
+                   .astimezone(tz.tzutc()).timestamp()))
+
+
+# Read prices from Groupe E API
+def readApi() -> list[tuple[str, float]]:
+    endpoint = "https://api.tariffs.groupe-e.ch/v1/tariffs"
+    tariffName = "vario_plus"
+    startDate = datetime.now().strftime("%Y-%m-%dT%H:00:00+01:00")
+    endDate = (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%dT%H:00:00+01:00")
+    session = req.get_http_session()
+    prices_raw = session.get(
+        url=endpoint +
+        f"?start_timestamp={ quote(startDate) }&end_timestamp={ quote(endDate) }",
+        ).json()
+    return [(timestamp(power), (power[tariffName]/100000))
+            for power in prices_raw]
+
+
+# Fetch prices and return as a dictionary
+def fetch_prices(config: GroupeETariffConfiguration) -> Dict[str, float]:
+    pricelist = readApi()
+    prices: Dict[str, float] = dict(pricelist)
+    return prices
+
+
+def create_electricity_tariff(config: GroupeETariff):
+    def updater():
+        return TariffState(prices=fetch_prices(config.configuration))
+    return updater
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=GroupeETariff)