Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2026-07-13 05:18:12 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat: add fixed electricity prediction with time window support (#930)
Some checks failed
Bump Version / Bump Version Workflow (push) Has been cancelled
Details
docker-build / platform-excludes (push) Has been cancelled
Details
docker-build / build (push) Has been cancelled
Details
docker-build / merge (push) Has been cancelled
Details
pre-commit / pre-commit (push) Has been cancelled
Details
Run Pytest on Pull Request / test (push) Has been cancelled
Details
Close stale pull requests/issues / Find Stale issues and PRs (push) Has been cancelled
Details

Browse Source 
Add a fixed electricity prediction that supports prices per time window.
The time windows may flexible be defined by day or date.

The prediction documentation is updated to also cover the ElecPriceFixed
provider.

The feature includes several changes that are not directly related to the
electricity price prediction implementation but are necessary to keep
EOS running properly and to test and document the changes.

* feat: add value time windows

    Add time windows with an associated float value.

* feat: harden eos measurements endpoints error detection and reporting

    Cover more errors that may be raised during endpoint access. Report the
    errors including trace information to ease debugging.

* feat: extend server configuration to cover all arguments

    Make the argument controlled options also available in server configuration.

* fix: eos config configuration by cli arguments

    Move the command line argument handling to config eos so that it is
    excuted whenever eos config is rebuild or reset.

* chore: extend measurement endpoint system test

* chore: refactor time windows

    Move time windows to configabc as they are only used in configurations.
    Also move all tests to test_configabc.

* chore: provide config update errors in eosdash with summarized error text

    If there is an update error provide the error text as a summary. On click
    provide the full error text.

* chore: force eosdash ip address and port in makefile dev run

    Ensure eosdash ip address and port are correctly set for development runs.

Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
This commit is contained in:
Bobby Noelte
2026-03-11 17:18:45 +01:00
committed by GitHub
parent 850d6b7c74
commit cf477d91a3
35 changed files with 3778 additions and 1491 deletions
Download Patch File Download Diff File Expand all files Collapse all files

635
src/akkudoktoreos/utils/datetimeutil.py
View File

@@ -10,7 +10,7 @@ Features:
- Convert durations from strings or numerics into `pendulum.Duration`.
- Infer timezone from UTC offset or geolocation.
- Support for custom output formats (ISO 8601, UTC normalized, or user-specified formats).
- Makes pendulum types usable in pydantic models using `pydantic_extra_types.pendulum_dt`
- Makes pendulum types usable in pydantic models using `pydantic_extra_types.pe ndulum_dt`
and the `Time` class.
Types:
@@ -19,7 +19,6 @@ Types:
- `DateTime`: Pendulum's timezone-aware datetime type.
- `Date`: Pendulum's date type.
- `Duration`: Pendulum's representation of a time delta.
- `TimeWindow`: Daily or specific date time window with optional localization support.
Functions:
----------
@@ -45,22 +44,15 @@ Usage Examples:
See each function's docstring for detailed argument options and examples.
"""
import calendar
import datetime
import re
from typing import Any, Iterator, List, Literal, Optional, Tuple, Union, overload
from typing import Any, List, Literal, Optional, Tuple, Union, overload
import pendulum
from babel.dates import get_day_names
from loguru import logger
from pendulum.tz.timezone import Timezone
from pydantic import (
BaseModel,
Field,
GetCoreSchemaHandler,
field_serializer,
field_validator,
model_validator,
)
from pydantic_core import core_schema
from pydantic_extra_types.pendulum_dt import ( # make pendulum types pydantic
@@ -830,627 +822,6 @@ def to_time(
raise ValueError(f"Invalid time value: {value!r} of type: {type(value)}") from e
class TimeWindow(BaseModel):
"""Model defining a daily or specific date time window with optional localization support.
Represents a time interval starting at `start_time` and lasting for `duration`.
Can restrict applicability to a specific day of the week or a specific calendar date.
Supports day names in multiple languages via locale-aware parsing.
"""
start_time: Time = Field(
..., json_schema_extra={"description": "Start time of the time window (time of day)."}
)
duration: Duration = Field(
...,
json_schema_extra={
"description": "Duration of the time window starting from `start_time`."
},
)
day_of_week: Optional[Union[int, str]] = Field(
default=None,
json_schema_extra={
"description": (
"Optional day of the week restriction. "
"Can be specified as integer (0=Monday to 6=Sunday) or localized weekday name. "
"If None, applies every day unless `date` is set."
)
},
)
date: Optional[Date] = Field(
default=None,
json_schema_extra={
"description": (
"Optional specific calendar date for the time window. Overrides `day_of_week` if set."
)
},
)
locale: Optional[str] = Field(
default=None,
json_schema_extra={
"description": (
"Locale used to parse weekday names in `day_of_week` when given as string. "
"If not set, Pendulum's default locale is used. "
"Examples: 'en', 'de', 'fr', etc."
)
},
)
@field_validator("duration", mode="before")
@classmethod
def transform_to_duration(cls, value: Any) -> Duration:
"""Converts various duration formats into Duration.
Args:
value: The value to convert to Duration.
Returns:
Duration: The converted Duration object.
"""
return to_duration(value)
@model_validator(mode="after")
def validate_day_of_week_with_locale(self) -> "TimeWindow":
"""Validates and normalizes the `day_of_week` field using the specified locale.
This method supports both integer (0–6) and string inputs for `day_of_week`.
String inputs are matched first against English weekday names (case-insensitive),
and then against localized weekday names using the provided `locale`.
If a valid match is found, `day_of_week` is converted to its corresponding
integer value (0 for Monday through 6 for Sunday).
Returns:
TimeWindow: The validated instance with `day_of_week` normalized to an integer.
Raises:
ValueError: If `day_of_week` is an invalid integer (not in 0–6),
or an unrecognized string (not matching English or localized names),
or of an unsupported type.
"""
if self.day_of_week is None:
return self
if isinstance(self.day_of_week, int):
if not 0 <= self.day_of_week <= 6:
raise ValueError("day_of_week must be in 0 (Monday) to 6 (Sunday)")
return self
if isinstance(self.day_of_week, str):
# Try matching against English names first (lowercase)
english_days = {name.lower(): i for i, name in enumerate(calendar.day_name)}
lowercase_value = self.day_of_week.lower()
if lowercase_value in english_days:
self.day_of_week = english_days[lowercase_value]
return self
# Try localized names
if self.locale:
localized_days = {
get_day_names("wide", locale=self.locale)[i].lower(): i for i in range(7)
}
if lowercase_value in localized_days:
self.day_of_week = localized_days[lowercase_value]
return self
raise ValueError(
f"Invalid weekday name '{self.day_of_week}' for locale '{self.locale}'. "
f"Expected English names (monday–sunday) or localized names."
)
raise ValueError(f"Invalid type for day_of_week: {type(self.day_of_week)}")
@field_serializer("duration")
def serialize_duration(self, value: Duration) -> str:
"""Serialize duration to string."""
return str(value)
def _window_start_end(self, reference_date: DateTime) -> tuple[DateTime, DateTime]:
"""Get the actual start and end datetimes for the time window on a given date.
This method computes the concrete start and end datetimes of the configured
time window for a specific date, taking into account timezone information.
Handles timezone-aware and naive `DateTime` and `Time` objects:
- If both `reference_date` and `start_time` have timezones but differ,
`start_time` is converted to the timezone of `reference_date`.
- If only one has a timezone, the other inherits it.
- If both are naive, UTC is assumed for both.
Args:
reference_date: The reference date on which to calculate the window.
Returns:
tuple[DateTime, DateTime]: A tuple containing the start and end datetimes
for the time window, both timezone-aware.
"""
ref_tz = reference_date.timezone
start_tz = self.start_time.tzinfo
# --- Timezone resolution logic ---
if ref_tz and start_tz:
# Both aware: align start_time to reference_date's tz
if ref_tz != start_tz:
start_time = self.start_time.in_timezone(ref_tz)
else:
start_time = self.start_time
elif ref_tz and not start_tz:
# Only reference_date aware → assume same tz for time
start_time = self.start_time.replace_timezone(ref_tz)
elif not ref_tz and start_tz:
# Only start_time aware → apply its tz to reference_date
reference_date = reference_date.replace(tzinfo=start_tz)
start_time = self.start_time
else:
# Both naive → default to UTC
reference_date = reference_date.replace(tzinfo="UTC")
start_time = self.start_time.replace_timezone("UTC")
# --- Build window start ---
start = reference_date.replace(
hour=start_time.hour,
minute=start_time.minute,
second=start_time.second,
microsecond=start_time.microsecond,
)
# --- Compute window end ---
end = start + self.duration
return start, end
def contains(self, date_time: DateTime, duration: Optional[Duration] = None) -> bool:
"""Check whether a datetime (and optional duration) fits within the time window.
This method checks if a given datetime `date_time` lies within the start time and duration
defined by the `TimeWindow`. If `duration` is provided, it also ensures that
the full duration starting at `date_time` ends before or at the end of the time window.
Handles timezone-aware and naive datetimes:
- If both `date_time` and `start_time` are timezone-aware but differ → align `start_time`
to `date_time`’s timezone.
- If only one has a timezone → assign it to the other.
- If both are naive → assume UTC for both.
If `day_of_week` or `date` are specified in the time window, the method will also
ensure that `date_time` falls on the correct day or matches the exact date.
Args:
date_time: The datetime to test.
duration: An optional duration that must fit entirely within the time window
starting from `date_time`.
Returns:
bool: True if the datetime (and optional duration) is fully contained in the
time window, False otherwise.
"""
start_time = self.start_time # work on a local copy to avoid mutating self
start_tz = getattr(start_time, "tzinfo", None)
ref_tz = date_time.timezone
# --- Handle timezone logic ---
if ref_tz and start_tz:
# Both aware but different → align start_time to date_time's timezone
if ref_tz != start_tz:
start_time = start_time.in_timezone(ref_tz)
elif ref_tz and not start_tz:
# Only date_time aware → assign its timezone to start_time
start_time = start_time.replace_timezone(ref_tz)
elif not ref_tz and start_tz:
# Only start_time aware → assign its timezone to date_time
date_time = date_time.replace(tzinfo=start_tz)
else:
# Both naive → assume UTC
date_time = date_time.replace(tzinfo="UTC")
start_time = start_time.replace_timezone("UTC")
# --- Date and weekday constraints ---
if self.date and date_time.date() != self.date:
return False
if self.day_of_week is not None and date_time.day_of_week != self.day_of_week:
return False
# --- Compute window start and end for this date ---
start, end = self._window_start_end(date_time)
# --- Check containment ---
if not (start <= date_time < end):
return False
if duration is not None:
date_time_end = date_time + duration
return date_time_end <= end
return True
def earliest_start_time(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> Optional[DateTime]:
"""Get the earliest datetime that allows a duration to fit within the time window.
Args:
duration: The duration that needs to fit within the window.
reference_date: The date to check for the time window. Defaults to today.
Returns:
The earliest start time for the duration, or None if it doesn't fit.
"""
if reference_date is None:
reference_date = pendulum.today()
# Check if the reference date matches our constraints
if self.date and reference_date.date() != self.date:
return None
if self.day_of_week is not None and reference_date.day_of_week != self.day_of_week:
return None
# Check if the duration can fit within the time window
if duration > self.duration:
return None
window_start, window_end = self._window_start_end(reference_date)
# The earliest start time is simply the window start time
return window_start
def latest_start_time(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> Optional[DateTime]:
"""Get the latest datetime that allows a duration to fit within the time window.
Args:
duration: The duration that needs to fit within the window.
reference_date: The date to check for the time window. Defaults to today.
Returns:
The latest start time for the duration, or None if it doesn't fit.
"""
if reference_date is None:
reference_date = pendulum.today()
# Check if the reference date matches our constraints
if self.date and reference_date.date() != self.date:
return None
if self.day_of_week is not None and reference_date.day_of_week != self.day_of_week:
return None
# Check if the duration can fit within the time window
if duration > self.duration:
return None
window_start, window_end = self._window_start_end(reference_date)
# The latest start time is the window end minus the duration
latest_start = window_end - duration
# Ensure the latest start time is not before the window start
if latest_start < window_start:
return None
return latest_start
def can_fit_duration(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> bool:
"""Check if a duration can fit within the time window on a given date.
Args:
duration: The duration to check.
reference_date: The date to check for the time window. Defaults to today.
Returns:
bool: True if the duration can fit, False otherwise.
"""
return self.earliest_start_time(duration, reference_date) is not None
def available_duration(self, reference_date: Optional[DateTime] = None) -> Optional[Duration]:
"""Get the total available duration for the time window on a given date.
Args:
reference_date: The date to check for the time window. Defaults to today.
Returns:
The available duration, or None if the date doesn't match constraints.
"""
if reference_date is None:
reference_date = pendulum.today()
if self.date and reference_date.date() != self.date:
return None
if self.day_of_week is not None and reference_date.day_of_week != self.day_of_week:
return None
return self.duration
class TimeWindowSequence(BaseModel):
"""Model representing a sequence of time windows with collective operations.
Manages multiple TimeWindow objects and provides methods to work with them
as a cohesive unit for scheduling and availability checking.
"""
windows: Optional[list[TimeWindow]] = Field(
default_factory=list,
json_schema_extra={"description": "List of TimeWindow objects that make up this sequence."},
)
@field_validator("windows")
@classmethod
def validate_windows(cls, v: Optional[list[TimeWindow]]) -> list[TimeWindow]:
"""Validate windows and convert None to empty list."""
if v is None:
return []
return v
def model_post_init(self, __context: Any) -> None:
"""Ensure windows is always a list after initialization."""
if self.windows is None:
self.windows = []
def __iter__(self) -> Iterator[TimeWindow]:
"""Allow iteration over the time windows."""
return iter(self.windows or [])
def __len__(self) -> int:
"""Return the number of time windows in the sequence."""
return len(self.windows or [])
def __getitem__(self, index: int) -> TimeWindow:
"""Allow indexing into the time windows."""
if not self.windows:
raise IndexError("list index out of range")
return self.windows[index]
def contains(self, date_time: DateTime, duration: Optional[Duration] = None) -> bool:
"""Check if any time window in the sequence contains the given datetime and duration.
Args:
date_time: The datetime to test.
duration: An optional duration that must fit entirely within one of the time windows.
Returns:
bool: True if any time window contains the datetime (and optional duration), False if no windows.
"""
if not self.windows:
return False
return any(window.contains(date_time, duration) for window in self.windows)
def earliest_start_time(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> Optional[DateTime]:
"""Get the earliest datetime across all windows that allows a duration to fit.
Args:
duration: The duration that needs to fit within a window.
reference_date: The date to check for the time windows. Defaults to today.
Returns:
The earliest start time across all windows, or None if no window can fit the duration.
"""
if not self.windows:
return None
if reference_date is None:
reference_date = pendulum.today()
earliest_times = []
for window in self.windows:
earliest = window.earliest_start_time(duration, reference_date)
if earliest is not None:
earliest_times.append(earliest)
return min(earliest_times) if earliest_times else None
def latest_start_time(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> Optional[DateTime]:
"""Get the latest datetime across all windows that allows a duration to fit.
Args:
duration: The duration that needs to fit within a window.
reference_date: The date to check for the time windows. Defaults to today.
Returns:
The latest start time across all windows, or None if no window can fit the duration.
"""
if not self.windows:
return None
if reference_date is None:
reference_date = pendulum.today()
latest_times = []
for window in self.windows:
latest = window.latest_start_time(duration, reference_date)
if latest is not None:
latest_times.append(latest)
return max(latest_times) if latest_times else None
def can_fit_duration(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> bool:
"""Check if the duration can fit within any time window in the sequence.
Args:
duration: The duration to check.
reference_date: The date to check for the time windows. Defaults to today.
Returns:
bool: True if any window can fit the duration, False if no windows.
"""
if not self.windows:
return False
return any(window.can_fit_duration(duration, reference_date) for window in self.windows)
def available_duration(self, reference_date: Optional[DateTime] = None) -> Optional[Duration]:
"""Get the total available duration across all applicable windows.
Args:
reference_date: The date to check for the time windows. Defaults to today.
Returns:
The sum of available durations from all applicable windows, or None if no windows apply.
"""
if not self.windows:
return None
if reference_date is None:
reference_date = pendulum.today()
total_duration = Duration()
has_applicable_windows = False
for window in self.windows:
window_duration = window.available_duration(reference_date)
if window_duration is not None:
total_duration += window_duration
has_applicable_windows = True
return total_duration if has_applicable_windows else None
def get_applicable_windows(self, reference_date: Optional[DateTime] = None) -> list[TimeWindow]:
"""Get all windows that apply to the given reference date.
Args:
reference_date: The date to check for the time windows. Defaults to today.
Returns:
List of TimeWindow objects that apply to the reference date.
"""
if not self.windows:
return []
if reference_date is None:
reference_date = pendulum.today()
applicable_windows = []
for window in self.windows:
if window.available_duration(reference_date) is not None:
applicable_windows.append(window)
return applicable_windows
def find_windows_for_duration(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> list[TimeWindow]:
"""Find all windows that can accommodate the given duration.
Args:
duration: The duration that needs to fit.
reference_date: The date to check for the time windows. Defaults to today.
Returns:
List of TimeWindow objects that can fit the duration.
"""
if not self.windows:
return []
if reference_date is None:
reference_date = pendulum.today()
fitting_windows = []
for window in self.windows:
if window.can_fit_duration(duration, reference_date):
fitting_windows.append(window)
return fitting_windows
def get_all_possible_start_times(
self, duration: Duration, reference_date: Optional[DateTime] = None
) -> list[tuple[DateTime, DateTime, TimeWindow]]:
"""Get all possible start time ranges for a duration across all windows.
Args:
duration: The duration that needs to fit.
reference_date: The date to check for the time windows. Defaults to today.
Returns:
List of tuples containing (earliest_start, latest_start, window) for each
window that can accommodate the duration.
"""
if not self.windows:
return []
if reference_date is None:
reference_date = pendulum.today()
possible_times = []
for window in self.windows:
earliest = window.earliest_start_time(duration, reference_date)
latest = window.latest_start_time(duration, reference_date)
if earliest is not None and latest is not None:
possible_times.append((earliest, latest, window))
return possible_times
def add_window(self, window: TimeWindow) -> None:
"""Add a new time window to the sequence.
Args:
window: The TimeWindow to add.
"""
if self.windows is None:
self.windows = []
self.windows.append(window)
def remove_window(self, index: int) -> TimeWindow:
"""Remove a time window from the sequence by index.
Args:
index: The index of the window to remove.
Returns:
The removed TimeWindow.
Raises:
IndexError: If the index is out of range.
"""
if not self.windows:
raise IndexError("pop from empty list")
return self.windows.pop(index)
def clear_windows(self) -> None:
"""Remove all windows from the sequence."""
if self.windows is not None:
self.windows.clear()
def sort_windows_by_start_time(self, reference_date: Optional[DateTime] = None) -> None:
"""Sort the windows by their start time on the given reference date.
Windows that don't apply to the reference date are placed at the end.
Args:
reference_date: The date to use for sorting. Defaults to today.
"""
if not self.windows:
return
if reference_date is None:
reference_date = pendulum.today()
def sort_key(window: TimeWindow) -> tuple[int, DateTime]:
"""Sort key: (priority, start_time) where priority 0 = applicable, 1 = not applicable."""
start_time = window.earliest_start_time(Duration(), reference_date)
if start_time is None:
# Non-applicable windows get a high priority (sorted last) and a dummy time
return (1, reference_date)
return (0, start_time)
self.windows.sort(key=sort_key)
@overload
def to_datetime(
date_input: Optional[Any] = None,
@@ -1782,7 +1153,7 @@ def to_duration(
elif isinstance(input_value, str):
# first try pendulum.parse
try:
parsed = pendulum.parse(input_value)
parsed = pendulum.parse(input_value, strict=False)
if isinstance(parsed, pendulum.Duration):
duration = parsed # Already a duration
else: