Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions packages/modules/electricity_tariffs/fixed_hours/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,15 @@ def __init__(self, default_price: Optional[float] = None, tariffs: List[Dict[str
"name": "high_tariff",
"price": 0.20,
"active_times": {
"quarters": [1, 2, 3, 4], # applicable quarters
"dates": [("01-01", "31-03"), ("01-07", "30-09")], # applicable date ranges (day-month)
"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
"dates": [("01-04", "30-06"), ("01-10", "31-12")], # applicable date ranges (day-month)
"times": [("00:00", "06:00"), ("22:00", "23:59")] # active times during the day
}
}
Expand Down
40 changes: 27 additions & 13 deletions packages/modules/electricity_tariffs/fixed_hours/tariff.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import logging
import datetime
import time
from typing import List, Tuple, Dict

from modules.electricity_tariffs.fixed_hours.config import FixedHoursTariff, FixedHoursTariffConfiguration
from modules.common.abstract_device import DeviceDescriptor
Expand All @@ -11,39 +12,52 @@
log = logging.getLogger(__name__)


def to_time(time_str):
def to_time(time_str: str) -> datetime.time:
if time_str == "24:00":
return datetime.time(23, 59, 59)
return datetime.datetime.strptime(time_str, "%H:%M").time()


def validate_tariff_times(config):
time_slots = []
def to_date(date_str: str, time_slot: datetime.datetime) -> datetime.date:
date = datetime.datetime.strptime(date_str, "%d-%m").date().replace(year=datetime.datetime.now().year)
if date.year < time_slot.year: # Beim Jahreswechsel das korrekte Jahr setzen
date = date.replace(year=time_slot.year)
return date


def validate_tariff_times(config: FixedHoursTariffConfiguration) -> None:
time_slots: List[Tuple[datetime.time, datetime.time, List[Tuple[str, str]]]] = []
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):
for existing_start, existing_end, existing_dates in time_slots:
if (start_time < existing_end and end_time > existing_start and
any(start <= existing_end and end >= existing_start for start, end in existing_dates)):
raise ValueError(f"Overlapping time window detected: {start} - {end} in tariff '{tariff['name']}'")
time_slots.append((start_time, end_time))
time_slots.append((start_time, end_time, tariff["active_times"]["dates"]))


def fetch(config: FixedHoursTariffConfiguration) -> None:
def fetch(config: FixedHoursTariffConfiguration) -> TariffState:
validate_tariff_times(config)

current_time = datetime.datetime.now().replace(minute=0, second=0, microsecond=0)
prices = {}
prices: Dict[str, float] = {}

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
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"]]
active_dates = [
(to_date(start, time_slot), to_date(end, time_slot))
for start, end in tariff["active_times"]["dates"]
]
if (any(start <= time_slot.time() < end for start, end in active_times) and
quarter in tariff["active_times"]["quarters"]):
price = tariff["price"]/1000
any(start <= time_slot.date() <= end for start, end in active_dates)):
price = tariff["price"] / 1000
break # Break since we found a matching tariff

prices[str(epoch_time)] = price
Expand All @@ -52,7 +66,7 @@ def fetch(config: FixedHoursTariffConfiguration) -> None:


def create_electricity_tariff(config: FixedHoursTariff):
def updater():
def updater() -> TariffState:
return fetch(config.configuration)
return updater

Expand Down