From 8530ebc73f4ea0a0f7bd363ef5931e5869fa6039 Mon Sep 17 00:00:00 2001 From: MartinRinas <martrin@microsoft.com> Date: Tue, 21 Jan 2025 21:30:33 +0000 Subject: [PATCH] fixed hour tariffs --- .../fixedhours/__init__.py | 0 .../electricity_tariffs/fixedhours/config.py | 38 ++++++++++++ .../electricity_tariffs/fixedhours/tariff.py | 60 +++++++++++++++++++ 3 files changed, 98 insertions(+) create mode 100644 packages/modules/electricity_tariffs/fixedhours/__init__.py create mode 100644 packages/modules/electricity_tariffs/fixedhours/config.py create mode 100644 packages/modules/electricity_tariffs/fixedhours/tariff.py diff --git a/packages/modules/electricity_tariffs/fixedhours/__init__.py b/packages/modules/electricity_tariffs/fixedhours/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/packages/modules/electricity_tariffs/fixedhours/config.py b/packages/modules/electricity_tariffs/fixedhours/config.py new file mode 100644 index 0000000000..836d7e59e5 --- /dev/null +++ b/packages/modules/electricity_tariffs/fixedhours/config.py @@ -0,0 +1,38 @@ +from typing import Optional, List, Dict + + +class FixedHoursTariffConfiguration: + def __init__(self, default_price: Optional[float] = None, tariffs: Optional[List[Dict[str, any]]] = None): + self.default_price = default_price + self.tariffs = tariffs + ''' + Example configuration: + "tariffs": [ + { + "name": "high_tariff", + "price": 0.20, + "active_times": { + "quarters": [1, 2, 3, 4], # applicable quarters + "times": [("08:00", "12:00"), ("18:00", "22:00")] # active times during the day + } + }, + { + "name": "low_tariff", + "price": 0.05, + "active_times": { + "quarters": [1, 2, 3, 4], # applicable quarters + "times": [("00:00", "06:00"), ("22:00", "23:59")] # active times during the day + } + } + ] + ''' + + +class FixedHoursTariff: + def __init__(self, + name: str = "Feste Tarifstunden (z.b. ยง14a EnWG Modul3)", + type: str = "fixedhours", + configuration: FixedHoursTariffConfiguration = None) -> None: + self.name = name + self.type = type + self.configuration = configuration or FixedHoursTariffConfiguration() diff --git a/packages/modules/electricity_tariffs/fixedhours/tariff.py b/packages/modules/electricity_tariffs/fixedhours/tariff.py new file mode 100644 index 0000000000..d68de36af5 --- /dev/null +++ b/packages/modules/electricity_tariffs/fixedhours/tariff.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +import logging +import datetime +import time + +from modules.electricity_tariffs.fixedhours.config import FixedHoursTariff, FixedHoursTariffConfiguration +from modules.common.abstract_device import DeviceDescriptor +from modules.common.component_state import TariffState +# from modules.common.configurable_tariff import ConfigurableTariff + +log = logging.getLogger(__name__) + + +def to_time(time_str): + return datetime.datetime.strptime(time_str, "%H:%M").time() + + +def validate_tariff_times(config): + time_slots = [] + for tariff in config.tariffs: + for start, end in tariff["active_times"]["times"]: + start_time = to_time(start) + end_time = to_time(end) + for existing_start, existing_end in time_slots: + if (start_time < existing_end and end_time > existing_start): + raise ValueError(f"Overlapping time window detected: {start} - {end} in tariff '{tariff['name']}'") + time_slots.append((start_time, end_time)) + + +def fetch(config: FixedHoursTariffConfiguration) -> None: + validate_tariff_times(config) + + current_time = datetime.datetime.now().replace(minute=0, second=0, microsecond=0) + prices = {} + + for i in range(24): # get prices for the next 24 hours + time_slot = current_time + datetime.timedelta(hours=i) + epoch_time = int(time.mktime(time_slot.timetuple())) + quarter = (current_time.month - 1) // 3 + 1 + price = config.default_price/1000 + + for tariff in config.tariffs: + active_times = [(to_time(start), to_time(end)) for start, end in tariff["active_times"]["times"]] + if (any(start <= time_slot.time() < end for start, end in active_times) and + quarter in tariff["active_times"]["quarters"]): + price = tariff["price"]/1000 + break # Break since we found a matching tariff + + prices[str(epoch_time)] = price + + return TariffState(prices=prices) + + +def create_electricity_tariff(config: FixedHoursTariff): + def updater(): + return fetch(config.configuration) + return updater + + +device_descriptor = DeviceDescriptor(configuration_factory=FixedHoursTariff)