Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2026-10-10 08:26:38 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

merge: integrate device configuration PR #1256 onto pinned main

Browse Source
This commit is contained in:
Andreas
2026-09-16 12:06:11 +02:00
parent afb7bcb8bc 6ebd343047
commit d3f96c4902
47 changed files with 4836 additions and 2128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/akkudoktoreos/config/configabc.py
+218
View File
@@ -59,6 +59,15 @@ def runtime_environment() -> str:
return f"Standalone Python (Python {python_version})"
class ConfigScope(StrEnum):
"""Configuration scope for x-scope json_schema_extra."""
GENERAL = "GENERAL"
GENETIC = "GENETIC"
GENETIC0 = "GENETIC0"
UNUSED = "UNUSED"
class SettingsBaseModel(PydanticBaseModel):
"""Base model class for all settings configurations."""
@@ -1037,3 +1046,212 @@ class ValueTimeWindowSequence(TimeWindowSequence[ValueTimeWindow]):
index=pd.DatetimeIndex(timestamps),
dtype=np.float64,
)
class CycleTimeWindowSequence(ValueTimeWindowSequence):
"""Sequence of time windows associated to cycles.
This model specializes ``ValueTimeWindowSequence`` so that the ``value``
field of each ``ValueTimeWindow`` encodes the **cycle index** (0-based
integer) the window belongs to.
Typical use: an appliance that must run ``n`` times per day, each run
constrained to a distinct time window. Assign ``value=0`` to windows
for the first cycle, ``value=1`` for the second, and so on. Multiple
windows may share the same cycle index (their allowed regions are unioned).
Windows with ``value=None`` are silently ignored by all cycle-aware methods.
"""
def num_cycles(self) -> int:
"""Return the number of distinct cycles defined in the sequence.
Cycles are derived from the integer part of the window values.
Windows without a value are ignored.
Returns:
int: Number of unique cycles.
"""
cycles = {int(window.value) for window in self.windows if window.value is not None}
return len(cycles)
def cycle_to_array(
self,
cycle: int,
start_datetime: DateTime,
end_datetime: DateTime,
interval: Duration,
dropna: bool = True,
boundary: str = "context",
align_to_interval: bool = True,
) -> np.ndarray:
"""Return a binary 1-D array indicating when *cycle* is active.
The time grid and alignment semantics are identical to
``TimeWindowSequence.to_array``: ``start_datetime`` is floored to the
nearest interval boundary in wall-clock time when
``align_to_interval=True``, and the timezone (or naivety) of
``start_datetime`` is preserved without any UTC conversion.
The step count uses ``math.ceil`` so that a partially-covered final
interval is included, consistent with ``to_array``'s while-loop
termination condition.
Args:
cycle: Integer cycle index to query.
start_datetime: First step of the time grid (inclusive).
end_datetime: Upper bound of the time grid (exclusive).
interval: Fixed step size.
dropna: Accepted for signature compatibility; has no effect.
boundary: Accepted for signature compatibility; has no effect.
align_to_interval: When ``True`` (default), floor
``start_datetime`` to the nearest interval boundary in
wall-clock time before building the grid.
Returns:
``np.ndarray`` of shape ``(n_steps,)`` with ``dtype=float64``.
``1.0`` at step ``t`` means step ``t`` falls inside a window
belonging to ``cycle``; ``0.0`` otherwise.
"""
import math
interval_s = interval.total_seconds()
if align_to_interval and interval_s > 0:
# Floor purely in wall-clock seconds — identical to to_array's
# alignment so all three methods produce consistent grids.
wall_s = (
start_datetime.hour * 3600
+ start_datetime.minute * 60
+ start_datetime.second
+ start_datetime.microsecond / 1_000_000
)
remainder_s = wall_s % interval_s
if remainder_s:
start_datetime = start_datetime.subtract(seconds=remainder_s)
# Use ceil so a partially-covered final step is included, matching the
# while-loop semantics of to_array.
steps = (
math.ceil((end_datetime - start_datetime).total_seconds() / interval_s)
if interval_s > 0
else 0
)
result = np.zeros(steps, dtype=np.float64)
# Anchor window start times to the calendar day of the (aligned) grid start.
base_day = start_datetime.start_of("day")
for window in self.windows:
if window.value is None:
continue
if int(window.value) != cycle:
continue
t = window.start_time
win_start = base_day.replace(
hour=t.hour,
minute=t.minute,
second=t.second,
microsecond=t.microsecond,
)
win_end = win_start + window.duration
# Convert to step indices relative to the aligned grid start.
idx_start = int((win_start - start_datetime).total_seconds() / interval_s)
idx_end = math.ceil((win_end - start_datetime).total_seconds() / interval_s)
idx_start = max(idx_start, 0)
idx_end = min(idx_end, steps)
if idx_start < idx_end:
result[idx_start:idx_end] = 1.0
return result
def cycles_to_matrix(
self,
start_datetime: DateTime,
end_datetime: DateTime,
interval: Duration,
) -> tuple[list[int], np.ndarray]:
"""Return a ``(cycle_indices, matrix)`` pair over the simulation horizon.
The matrix encodes, for each cycle and each time step, whether that
step falls inside a window belonging to that cycle. It is the
vectorised equivalent of calling ``cycle_to_array`` for every cycle.
Alignment and step-count semantics are identical to ``to_array`` and
``cycle_to_array``: ``start_datetime`` is floored to the nearest
interval boundary in wall-clock time, and the step count uses
``math.ceil`` for consistency with ``to_array``'s while-loop.
Args:
start_datetime: First step of the time grid (inclusive).
end_datetime: Upper bound of the time grid (exclusive).
interval: Fixed step size.
Returns:
``(cycle_indices, matrix)`` where
* ``cycle_indices`` is a sorted ``list[int]`` of the distinct
cycle numbers found in the sequence (e.g. ``[0, 1, 2]``).
* ``matrix`` is a ``np.ndarray`` of shape
``(len(cycle_indices), n_steps)`` with ``dtype=float64``.
``matrix[k, t] == 1.0`` iff step ``t`` falls inside a window
belonging to ``cycle_indices[k]``; ``0.0`` otherwise.
"""
import math
interval_s = interval.total_seconds()
if interval_s > 0:
# Apply the same wall-clock floor as to_array and cycle_to_array.
wall_s = (
start_datetime.hour * 3600
+ start_datetime.minute * 60
+ start_datetime.second
+ start_datetime.microsecond / 1_000_000
)
remainder_s = wall_s % interval_s
if remainder_s:
start_datetime = start_datetime.subtract(seconds=remainder_s)
cycles = sorted({int(w.value) for w in self.windows if w.value is not None})
steps = (
math.ceil((end_datetime - start_datetime).total_seconds() / interval_s)
if interval_s > 0
else 0
)
matrix = np.zeros((len(cycles), steps), dtype=np.float64)
cycle_index = {c: i for i, c in enumerate(cycles)}
# Anchor window start times to the calendar day of the aligned grid start.
base_day = start_datetime.start_of("day")
for window in self.windows:
if window.value is None:
continue
c = int(window.value)
row = cycle_index[c]
t = window.start_time
win_start = base_day.replace(
hour=t.hour,
minute=t.minute,
second=t.second,
microsecond=t.microsecond,
)
win_end = win_start + window.duration
idx_start = int((win_start - start_datetime).total_seconds() / interval_s)
idx_end = math.ceil((win_end - start_datetime).total_seconds() / interval_s)
idx_start = max(idx_start, 0)
idx_end = min(idx_end, steps)
if idx_start < idx_end:
matrix[row, idx_start:idx_end] = 1.0
return cycles, matrix
src/akkudoktoreos/config/configmigrate.py
+53
View File
@@ -30,6 +30,38 @@ if TYPE_CHECKING:
_KEEP_DEFAULT = object()
# -----------------------------
# Migration helpers
# -----------------------------
def _list_to_device_dict(
prefix: str,
) -> Callable[[Any], Any]:
"""Return a transform that converts a list of device dicts to a keyed dict.
Each item must be a dict. The key is taken from the item's own
``device_id`` field when present; otherwise a key is synthesised
from *prefix* + the zero-based index (e.g. ``"bat0"``, ``"bat1"``).
"""
def _transform(value: Any) -> Any:
if not isinstance(value, list):
return value # already a dict or something unexpected – leave as-is
result: Dict[str, Any] = {}
for i, item in enumerate(value):
if not isinstance(item, dict):
continue
key = item.get("device_id") or f"{prefix}{i}"
# Ensure device_id is stored inside the dict so Pydantic can validate it
item.setdefault("device_id", key)
result[key] = item
return result
return _transform
# -----------------------------
# Global migration map constant
# -----------------------------
@@ -58,10 +90,31 @@ MIGRATION_MAP: Dict[
# - NodeRed
# devices
# =======
# List → dict migration (all device collections)
# These must come *before* any sub-path entries that reference the old list indices,
# so the whole collection is moved first; the sub-path None-drops clean up leftovers.
# - batteries
"devices/batteries": (
"devices/batteries",
_list_to_device_dict("bat"),
),
"devices/batteries/0/initial_soc_percentage": None,
# - electric_vehicles
"devices/electric_vehicles": (
"devices/electric_vehicles",
_list_to_device_dict("ev"),
),
"devices/electric_vehicles/0/initial_soc_percentage": None,
# - inverters
"devices/inverters": (
"devices/inverters",
_list_to_device_dict("inv"),
),
# - home_appliances
"devices/home_appliances": (
"devices/home_appliances",
_list_to_device_dict("appliance"),
),
# elecfee
# =======
# - ElecFeeFixed
src/akkudoktoreos/devices/devices.py
+71 -294
View File
@@ -1,354 +1,131 @@
"""General configuration settings for simulated devices for optimization."""
import json
import re
from typing import Any, Optional, TextIO, cast
import numpy as np
from loguru import logger
from numpydantic import NDArray, Shape
from pydantic import Field, computed_field, field_validator, model_validator
from pydantic import Field, computed_field, model_validator
from akkudoktoreos.config.configabc import SettingsBaseModel, TimeWindowSequence
from akkudoktoreos.config.configabc import ConfigScope, SettingsBaseModel
from akkudoktoreos.core.cache import CacheFileStore
from akkudoktoreos.core.coreabc import ConfigMixin, SingletonMixin
from akkudoktoreos.core.emplan import ResourceStatus
from akkudoktoreos.core.pydantic import ConfigDict, PydanticBaseModel
from akkudoktoreos.devices.devicesabc import DevicesBaseSettings
from akkudoktoreos.devices.settings.batterysettings import BatteriesCommonSettings
from akkudoktoreos.devices.settings.homeappliancesettings import (
HomeApplianceCommonSettings,
)
from akkudoktoreos.devices.settings.invertersettings import InverterCommonSettings
from akkudoktoreos.utils.datetimeutil import DateTime, to_datetime
# Default charge rates for battery
BATTERY_DEFAULT_CHARGE_RATES: list[float] = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
class BatteriesCommonSettings(DevicesBaseSettings):
"""Battery devices base settings."""
capacity_wh: int = Field(
default=8000, gt=0, json_schema_extra={"description": "Capacity [Wh].", "examples": [8000]}
)
charging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "Charging efficiency [0.01 ... 1.00].",
"examples": [0.88],
},
)
discharging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "Discharge efficiency [0.01 ... 1.00].",
"examples": [0.88],
},
)
levelized_cost_of_storage_kwh: float = Field(
default=0.0,
json_schema_extra={
"description": "Levelized cost of storage (LCOS), the average lifetime cost of delivering one kWh [amount/kWh].",
"examples": [0.12],
},
)
max_charge_power_w: Optional[float] = Field(
default=5000,
gt=0,
json_schema_extra={"description": "Maximum charging power [W].", "examples": [5000]},
)
min_charge_power_w: Optional[float] = Field(
default=50,
gt=0,
json_schema_extra={"description": "Minimum charging power [W].", "examples": [50]},
)
charge_rates: Optional[list[float]] = Field(
default=BATTERY_DEFAULT_CHARGE_RATES,
json_schema_extra={
"description": (
"Charge rates as factor of maximum charging power [0.00 ... 1.00]. "
"None triggers fallback to default charge-rates."
),
"examples": [[0.0, 0.25, 0.5, 0.75, 1.0], None],
},
)
min_soc_percentage: int = Field(
default=0,
ge=0,
le=100,
json_schema_extra={
"description": (
"Minimum state of charge (SOC) as percentage of capacity [%]. "
"This is the target SoC for charging"
),
"examples": [10],
},
)
max_soc_percentage: int = Field(
default=100,
ge=0,
le=100,
json_schema_extra={
"description": "Maximum state of charge (SOC) as percentage of capacity [%].",
"examples": [100],
},
)
@field_validator("charge_rates", mode="before")
def validate_and_sort_charge_rates(cls, v: Any) -> NDArray[Shape["*"], float]:
# None means fallback to default values
if v is None:
return np.asarray(BATTERY_DEFAULT_CHARGE_RATES, dtype=float)
# Convert to numpy array
if isinstance(v, str):
# Remove brackets and split by comma or whitespace
numbers = re.split(r"[,\s]+", v.strip("[]"))
# Filter out any empty strings and convert to floats
arr = np.array([float(x) for x in numbers if x])
else:
arr = np.array(v, dtype=float)
# Must not be empty
if arr.size == 0:
raise ValueError("charge_rates must contain at least one value.")
# Enforce bounds: 0.0 ≤ x ≤ 1.0
if (arr < 0.0).any() or (arr > 1.0).any():
raise ValueError("charge_rates must be within [0.0, 1.0].")
# Remove duplicates + sort
arr = np.unique(arr)
arr.sort()
return arr
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_soc_factor(self) -> str:
"""Measurement key for the battery state of charge (SoC) as factor of total capacity [0.0 ... 1.0]."""
return f"{self.device_id}-soc-factor"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_l1_w(self) -> str:
"""Measurement key for the L1 power the battery is charged or discharged with [W]."""
return f"{self.device_id}-power-l1-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_l2_w(self) -> str:
"""Measurement key for the L2 power the battery is charged or discharged with [W]."""
return f"{self.device_id}-power-l2-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_l3_w(self) -> str:
"""Measurement key for the L3 power the battery is charged or discharged with [W]."""
return f"{self.device_id}-power-l3-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_3_phase_sym_w(self) -> str:
"""Measurement key for the symmetric 3 phase power the battery is charged or discharged with [W]."""
return f"{self.device_id}-power-3-phase-sym-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> Optional[list[str]]:
"""Measurement keys for the battery stati that are measurements."""
keys: list[str] = [
self.measurement_key_soc_factor,
self.measurement_key_power_l1_w,
self.measurement_key_power_l2_w,
self.measurement_key_power_l3_w,
self.measurement_key_power_3_phase_sym_w,
]
return keys
class InverterCommonSettings(DevicesBaseSettings):
"""Inverter devices base settings."""
max_power_w: Optional[float] = Field(
default=None,
gt=0,
json_schema_extra={"description": "Maximum power [W].", "examples": [10000]},
)
battery_id: Optional[str] = Field(
default=None,
json_schema_extra={
"description": "ID of battery controlled by this inverter.",
"examples": [None, "battery1"],
},
)
ac_to_dc_efficiency: float = Field(
default=1.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of AC to DC conversion for grid-to-battery AC charging (0-1). "
"Set to 0 to disable AC charging. Default 1.0 (no additional inverter loss)."
),
"examples": [0.95, 1.0, 0.0],
},
)
dc_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of DC to AC conversion for battery discharging to AC load/grid (0-1). "
"Default 1.0 (no additional inverter loss)."
),
"examples": [0.95, 1.0],
},
)
max_ac_charge_power_w: Optional[float] = Field(
default=None,
ge=0,
json_schema_extra={
"description": (
"Maximum AC charging power in watts. "
"null means no additional limit. Set to 0 to disable AC charging."
),
"examples": [None, 0, 5000],
},
)
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> Optional[list[str]]:
"""Measurement keys for the inverter stati that are measurements."""
keys: list[str] = []
return keys
class HomeApplianceCommonSettings(DevicesBaseSettings):
"""Home Appliance devices base settings."""
consumption_wh: int = Field(
gt=0, json_schema_extra={"description": "Energy consumption [Wh].", "examples": [2000]}
)
duration_h: int = Field(
gt=0,
le=24,
json_schema_extra={"description": "Usage duration in hours [0 ... 24].", "examples": [1]},
)
time_windows: Optional[TimeWindowSequence] = Field(
default=None,
json_schema_extra={
"description": "Sequence of allowed time windows. Defaults to optimization general time window.",
"examples": [
{
"windows": [
{"start_time": "10:00", "duration": "2 hours"},
],
},
],
},
)
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> Optional[list[str]]:
"""Measurement keys for the home appliance stati that are measurements."""
keys: list[str] = []
return keys
class DevicesCommonSettings(SettingsBaseModel):
"""Base configuration for devices simulation settings."""
"""Configuration for all controllable devices in the simulation.
batteries: Optional[list[BatteriesCommonSettings]] = Field(
Every device collection is a ``dict[str, <Settings>]`` keyed by
``device_id``. This makes config paths stable regardless of
declaration order and lets each device settings class build its own
config path from ``self.device_id`` without needing an external index.
"""
# ---- Batteries ----
batteries: Optional[dict[str, BatteriesCommonSettings]] = Field(
default=None,
json_schema_extra={
"description": "List of battery devices",
"examples": [[{"device_id": "battery1", "capacity_wh": 8000}]],
"description": "Stationary battery storage devices, keyed by device_id.",
"examples": [{"bat0": {"device_id": "bat0", "capacity_wh": 8000, "ports": []}}],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
max_batteries: Optional[int] = Field(
default=None,
ge=0,
json_schema_extra={
"description": "Maximum number of batteries that can be set",
"examples": [1, 2],
"description": "Maximum number of batteries allowed.",
"examples": [1],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
electric_vehicles: Optional[list[BatteriesCommonSettings]] = Field(
# ---- Electric vehicles ----
electric_vehicles: Optional[dict[str, BatteriesCommonSettings]] = Field(
default=None,
json_schema_extra={
"description": "List of electric vehicle devices",
"examples": [[{"device_id": "battery1", "capacity_wh": 8000}]],
"description": "Electric vehicle battery packs, keyed by device_id.",
"examples": [{"ev0": {"device_id": "ev0", "capacity_wh": 60000, "ports": []}}],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
max_electric_vehicles: Optional[int] = Field(
default=None,
ge=0,
json_schema_extra={
"description": "Maximum number of electric vehicles that can be set",
"examples": [1, 2],
"description": "Maximum number of EVs allowed.",
"examples": [1],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
inverters: Optional[list[InverterCommonSettings]] = Field(
default=None, json_schema_extra={"description": "List of inverters", "examples": [[]]}
# ---- Inverters ----
inverters: Optional[dict[str, InverterCommonSettings]] = Field(
default=None,
json_schema_extra={
"description": "Inverter devices, keyed by device_id.",
"examples": [{}],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
max_inverters: Optional[int] = Field(
default=None,
ge=0,
json_schema_extra={
"description": "Maximum number of inverters that can be set",
"examples": [1, 2],
"description": "Maximum number of inverters allowed.",
"examples": [1],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
home_appliances: Optional[list[HomeApplianceCommonSettings]] = Field(
default=None, json_schema_extra={"description": "List of home appliances", "examples": [[]]}
# ---- Controllable home appliances ----
home_appliances: dict[str, HomeApplianceCommonSettings] = Field(
default_factory=dict,
json_schema_extra={
"description": "Shiftable home appliance devices, keyed by device_id.",
"examples": [
{
"dishwasher": {
"device_id": "dishwasher",
"consumption_wh": 1500,
"duration_h": 2.0, # required field
"ports": [{"bus_id": "bus_ac", "port_id": "p_ac", "direction": "sink"}],
},
},
],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
max_home_appliances: Optional[int] = Field(
default=None,
ge=0,
json_schema_extra={
"description": "Maximum number of home_appliances that can be set",
"examples": [1, 2],
"description": "Maximum number of home appliances allowed.",
"examples": [3],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> Optional[list[str]]:
"""Return the measurement keys for the resource/ device stati that are measurements."""
def measurement_keys(self) -> list[str]:
"""All measurement keys across all configured devices."""
keys: list[str] = []
if self.max_batteries and self.batteries:
for battery in self.batteries:
keys.extend(battery.measurement_keys or [])
if self.max_electric_vehicles and self.electric_vehicles:
for electric_vehicle in self.electric_vehicles:
keys.extend(electric_vehicle.measurement_keys or [])
for device_dict in [
self.batteries,
self.electric_vehicles,
self.inverters,
self.home_appliances,
]:
for device in (device_dict or {}).values():
keys.extend(device.measurement_keys)
return keys
src/akkudoktoreos/devices/devicesabc.py
-25
View File
@@ -1,32 +1,7 @@
"""Abstract and base classes for devices."""
import secrets
import string
from enum import StrEnum
from pydantic import Field
from akkudoktoreos.config.configabc import SettingsBaseModel
def device_default_id() -> str:
"""Provide random default device id."""
alphabet = string.ascii_letters + string.digits
device_id = "".join(secrets.choice(alphabet) for _ in range(10))
return device_id
class DevicesBaseSettings(SettingsBaseModel):
"""Base devices setting."""
device_id: str = Field(
default_factory=device_default_id,
json_schema_extra={
"description": "ID of device",
"examples": ["battery1", "ev1", "inverter1", "dishwasher"],
},
)
class BatteryOperationMode(StrEnum):
"""Battery Operation Mode.
src/akkudoktoreos/devices/genetic/battery.py
+96 -5
View File
@@ -1,12 +1,103 @@
from typing import Any, Iterator, Optional
import numpy as np
from pydantic import Field
from akkudoktoreos.devices.devices import BATTERY_DEFAULT_CHARGE_RATES
from akkudoktoreos.optimization.genetic.geneticdevices import (
BaseBatteryParameters,
SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.settings.batterysettings import BATTERY_DEFAULT_CHARGE_RATES
from akkudoktoreos.optimization.genetic.geneticdevices import DeviceParameters
def max_charging_power_field(description: Optional[str] = None) -> float:
if description is None:
description = "Maximum charging power in watts."
return Field(default=5000, gt=0, json_schema_extra={"description": description})
def initial_soc_percentage_field(description: str) -> int:
return Field(
default=0, ge=0, le=100, json_schema_extra={"description": description, "examples": [42]}
)
def discharging_efficiency_field(default_value: float) -> float:
return Field(
default=default_value,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the discharge efficiency of the battery."
},
)
class BaseBatteryParameters(DeviceParameters):
"""Battery Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of battery", "examples": ["battery1"]}
)
capacity_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the capacity of the battery in watt-hours.",
"examples": [8000],
},
)
charging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the charging efficiency of the battery."
},
)
discharging_efficiency: float = discharging_efficiency_field(0.88)
max_charge_power_w: Optional[float] = max_charging_power_field()
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the state of charge of the battery at the **start** of the current hour (not the current state)."
)
min_soc_percentage: int = Field(
default=0,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the minimum state of charge (SOC) of the battery in percentage.",
"examples": [10],
},
)
max_soc_percentage: int = Field(
default=100,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the maximum state of charge (SOC) of the battery in percentage."
},
)
charge_rates: Optional[list[float]] = Field(
default=None,
json_schema_extra={
"description": "Charge rates as factor of maximum charging power [0.00 ... 1.00]. None denotes all charge rates are available.",
"examples": [[0.0, 0.25, 0.5, 0.75, 1.0], None],
},
)
class SolarPanelBatteryParameters(BaseBatteryParameters):
"""PV battery device simulation configuration."""
max_charge_power_w: Optional[float] = max_charging_power_field()
class ElectricVehicleParameters(BaseBatteryParameters):
"""Battery Electric Vehicle Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of electric vehicle", "examples": ["ev1"]}
)
discharging_efficiency: float = discharging_efficiency_field(1.0)
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the current state of charge (SOC) of the battery in percentage."
)
class Battery:
src/akkudoktoreos/devices/genetic/homeappliance.py
+582 -57
View File
@@ -1,100 +1,625 @@
import numpy as np
"""Simulation of a home appliance that runs one or more fixed-duration cycles.
from akkudoktoreos.config.configabc import TimeWindow, TimeWindowSequence
from akkudoktoreos.optimization.genetic.geneticdevices import HomeApplianceParameters
from akkudoktoreos.utils.datetimeutil import to_datetime, to_duration, to_time
This module models household devices such as dishwashers or washing
machines that must run for a fixed duration, possibly multiple times
(cycles), within one or more allowed time windows. Given a set of
requested start times, `HomeAppliance` repairs them into a
feasible, chronologically ordered schedule and produces the resulting
hourly load curve.
Time windows are always expressed as a `CycleTimeWindowSequence`
(see ``akkudoktoreos.config.configabc``): each contained window's
``value`` encodes the 0-based cycle index it applies to, so different
cycles of the same appliance can be constrained to different windows.
When no windows are configured, every cycle defaults to a single
window spanning the full prediction horizon (i.e. unconstrained).
Cycle start times are repaired according to the following rules:
1. Round and clip the requested start to the simulation horizon.
2. Snap each cycle's start to the nearest start allowed by that
cycle's own time window.
3. Sort all cycle starts chronologically, keeping each start paired
with the cycle (and therefore the allowed-start mask) it belongs
to.
4. Walk the sorted starts and push any cycle that starts too soon
after its predecessor to the next start allowed by its own
window, enforcing the appliance duration plus the configured
minimum idle gap.
5. Generate the combined hourly load curve from the final starts.
"""
from typing import Optional
import numpy as np
from pydantic import Field
from akkudoktoreos.config.configabc import CycleTimeWindowSequence, ValueTimeWindow
from akkudoktoreos.optimization.genetic.geneticdevices import DeviceParameters
from akkudoktoreos.utils.datetimeutil import (
DateTime,
Duration,
to_datetime,
to_duration,
to_time,
)
class HomeApplianceParameters(DeviceParameters):
"""Configuration for a simulated home appliance device."""
device_id: str = Field(
json_schema_extra={
"description": "ID of home appliance",
"examples": ["dishwasher"],
}
)
consumption_wh: int = Field(
gt=0,
json_schema_extra={
"description": (
"An integer representing the energy consumption "
"of a household device in watt-hours."
),
"examples": [2000],
},
)
duration_h: int = Field(
gt=0,
json_schema_extra={
"description": (
"An integer representing the usage duration of a household device in hours."
),
"examples": [3],
},
)
num_cycles: int = Field(
default=1,
gt=0,
json_schema_extra={
"description": "Number of cycles the appliance must run.",
"examples": [2],
},
)
min_cycle_gap_h: int = Field(
default=0,
ge=0,
json_schema_extra={
"description": (
"Minimum idle time between the end of one cycle and the start of the next cycle."
),
"examples": [1],
},
)
time_windows: Optional[CycleTimeWindowSequence] = Field(
default=None,
json_schema_extra={
"description": (
"Allowed per-cycle time windows. Each window's `value` "
"encodes the 0-based cycle index it applies to; multiple "
"windows may share a cycle index. When omitted, every "
"cycle is unconstrained across the full prediction "
"horizon."
),
"examples": [
[
{
"start_time": "10:00",
"duration": "3 hours",
"value": 0,
},
],
],
},
)
class HomeAppliance:
"""Non-vectorized simulation of a multi-cycle home appliance.
A home appliance may execute multiple fixed-duration cycles during
the simulation horizon. See the module docstring for the start-time
repair algorithm.
"""
def __init__(
self,
parameters: HomeApplianceParameters,
optimization_hours: int,
prediction_hours: int,
):
self.parameters: HomeApplianceParameters = parameters
) -> None:
"""Initializes the appliance and builds its allowed-start masks.
Args:
parameters: The appliance's configuration.
optimization_hours: Number of hours under active
optimization.
prediction_hours: Length of the simulation horizon, in
hours.
"""
self.parameters = parameters
self.optimization_hours = optimization_hours
self.prediction_hours = prediction_hours
self.duration_h = parameters.duration_h
self.consumption_wh = parameters.consumption_wh
self.num_cycles = parameters.num_cycles
self.min_cycle_gap_h = parameters.min_cycle_gap_h
self.completed_cycles = 0
self.load_curve = np.zeros(prediction_hours)
# Start times for remaining cycles, in chronological order.
self.start_hours: list[int] = []
# Absolute cycle index corresponding to each remaining cycle.
#
# Example:
# num_cycles = 4
# completed_cycles = 2
#
# remaining_cycle_indices = [2, 3]
self.remaining_cycle_indices: list[int] = []
# start_allowed[k][hour]
#
# k is the index into remaining_cycle_indices (NOT into the
# chronologically-sorted self.start_hours).
self.start_allowed: list[np.ndarray] = []
self.start_earliest: list[int] = []
self.start_latest: list[int] = []
self._setup()
# ------------------------------------------------------------------
# Setup
# ------------------------------------------------------------------
def _setup(self) -> None:
"""Sets up the home appliance parameters based provided parameters."""
self.load_curve = np.zeros(self.prediction_hours) # Initialize the load curve with zeros
self.duration_h = self.parameters.duration_h
self.consumption_wh = self.parameters.consumption_wh
# setup possible start times
"""Sets up appliance parameters and default time windows.
When no ``time_windows`` are configured, builds a single
placeholder window spanning the full prediction horizon with
``value`` left unset. ``CycleTimeWindowSequence.cycles_to_matrix``
ignores windows whose ``value`` is ``None``, so this window never
matches any cycle; every remaining cycle then falls through to
the "no window for this cycle" branch in
``_build_cycle_start_allowed``, which treats it as unconstrained.
The net effect is the same as having no windows at all, without
assigning cycles an explicit (and misleadingly meaningful)
``value``.
"""
if self.parameters.time_windows is None:
self.parameters.time_windows = TimeWindowSequence(
self.parameters.time_windows = CycleTimeWindowSequence(
windows=[
TimeWindow(
ValueTimeWindow(
start_time=to_time("00:00"),
duration=to_duration(f"{self.prediction_hours} hours"),
),
]
)
start_datetime = to_datetime().set(hour=0, minute=0, second=0)
duration = to_duration(f"{self.duration_h} hours")
self.start_allowed: list[bool] = []
for hour in range(0, self.prediction_hours):
self.start_allowed.append(
self.parameters.time_windows.contains(
start_datetime.add(hours=hour), duration=duration
)
)
start_earliest = self.parameters.time_windows.earliest_start_time(duration, start_datetime)
if start_earliest:
self.start_earliest = start_earliest.hour
else:
self.start_earliest = 0
start_latest = self.parameters.time_windows.latest_start_time(duration, start_datetime)
if start_latest:
self.start_latest = start_latest.hour
else:
self.start_latest = 23
def set_starting_time(self, start_hour: int, global_start_hour: int = 0) -> int:
"""Sets the start time of the device and generates the corresponding load curve.
self._build_start_allowed()
:param start_hour: The hour at which the device should start.
@property
def num_remaining_cycles(self) -> int:
"""int: Number of cycles which still have to be scheduled."""
return max(0, self.num_cycles - self.completed_cycles)
def set_completed_cycles(self, completed_cycles: int) -> None:
"""Sets the number of cycles already completed.
Clears any previously scheduled start times and load curve,
and rebuilds the allowed-start masks for the cycles that
remain.
Args:
completed_cycles: Number of cycles completed so far.
Clamped to ``[0, num_cycles]``.
"""
if not self.start_allowed[start_hour]:
# It is not allowed (by the time windows) to start the application at this time
if global_start_hour <= self.start_latest:
# There is a time window left to start the appliance. Use it
start_hour = self.start_latest
else:
# There is no time window left to run the application
# Set the start into tomorrow
start_hour = self.start_earliest + 24
self.completed_cycles = max(
0,
min(completed_cycles, self.num_cycles),
)
self.start_hours = []
self.reset_load_curve()
self._build_start_allowed()
# ------------------------------------------------------------------
# Time-window handling
# ------------------------------------------------------------------
def _build_start_allowed(self) -> None:
"""Builds allowed start positions for all remaining cycles."""
if self.parameters.time_windows is None:
raise ValueError("Expected time windows in parameters, got {self.parameters}.")
self.start_allowed = []
self.start_earliest = []
self.start_latest = []
self.remaining_cycle_indices = list(
range(
self.completed_cycles,
self.num_cycles,
)
)
if not self.remaining_cycle_indices:
return
start_datetime = to_datetime().set(
hour=0,
minute=0,
second=0,
microsecond=0,
)
end_datetime = start_datetime.add(
hours=self.prediction_hours,
)
interval = to_duration("1 hour")
self._build_cycle_start_allowed(
self.parameters.time_windows,
start_datetime,
end_datetime,
interval,
)
def _build_cycle_start_allowed(
self,
time_windows: CycleTimeWindowSequence,
start_datetime: DateTime,
end_datetime: DateTime,
interval: Duration,
) -> None:
"""Builds allowed starts from per-cycle windows.
Args:
time_windows: Windows associated with individual cycles.
start_datetime: Start of the simulation horizon.
end_datetime: End of the simulation horizon.
interval: Step size used to sample the cycle windows.
"""
cycle_indices, matrix = time_windows.cycles_to_matrix(
start_datetime=start_datetime,
end_datetime=end_datetime,
interval=interval,
)
cycle_to_row = {cycle: row for row, cycle in enumerate(cycle_indices)}
max_start = max(
0,
self.prediction_hours - self.duration_h,
)
# The matrix tells us which individual *steps* are inside
# the cycle window. We still have to check that the complete
# appliance duration fits inside the window.
for cycle in self.remaining_cycle_indices:
row_index = cycle_to_row.get(cycle)
if row_index is None:
# No window for this cycle -> unconstrained.
allowed = np.zeros(
self.prediction_hours,
dtype=bool,
)
allowed[: max_start + 1] = True
else:
allowed = self._build_duration_feasibility(
matrix[row_index],
)
self.start_allowed.append(allowed)
allowed_indices = np.flatnonzero(allowed)
if len(allowed_indices):
self.start_earliest.append(int(allowed_indices[0]))
self.start_latest.append(int(allowed_indices[-1]))
else:
self.start_earliest.append(0)
self.start_latest.append(max_start)
def _build_duration_feasibility(
self,
window_steps: np.ndarray,
) -> np.ndarray:
"""Returns starts where the complete appliance fits in a window.
A start ``s`` is feasible when every step in
``window_steps[s : s + duration_h]`` lies inside the cycle's
window (i.e. sums to ``duration_h``).
Args:
window_steps: Per-hour membership of the cycle's window
(1.0 inside the window, 0.0 outside), one value per
hour of the prediction horizon.
Returns:
Boolean mask, one entry per hour of the prediction
horizon, ``True`` where a cycle of length ``duration_h``
can start without leaving the window.
"""
horizon = len(window_steps)
max_start = max(
0,
horizon - self.duration_h,
)
allowed = np.zeros(
horizon,
dtype=bool,
)
if self.duration_h > horizon:
return allowed
# Rolling sum of `duration_h` consecutive steps, aligned so
# that window_sums[s] == sum(window_steps[s : s + duration_h]).
cumulative = np.concatenate(([0.0], np.cumsum(window_steps)))
window_sums = cumulative[self.duration_h :] - cumulative[: -self.duration_h]
allowed[: max_start + 1] = window_sums[: max_start + 1] == float(self.duration_h)
return allowed
# ------------------------------------------------------------------
# Scheduling / repair
# ------------------------------------------------------------------
def set_starting_times(
self,
start_hours: list[int],
) -> list[int]:
"""Sets and repairs the start times of all remaining cycles.
See the module docstring for the repair algorithm.
Args:
start_hours: Requested start hour for each remaining
cycle, in the same order as ``remaining_cycle_indices``
(i.e. matching ``self.start_allowed``).
Returns:
The repaired, chronologically ordered start hours.
Raises:
ValueError: If ``start_hours`` does not have exactly
``num_remaining_cycles`` entries.
"""
if len(start_hours) != self.num_remaining_cycles:
raise ValueError(
f"Expected {self.num_remaining_cycles} start times, got {len(start_hours)}."
)
if self.num_remaining_cycles == 0:
self.start_hours = []
self.reset_load_curve()
return []
max_start = max(
0,
self.prediction_hours - self.duration_h,
)
# 1. Round and clip.
starts = [
max(
0,
min(
int(round(start)),
max_start,
),
)
for start in start_hours
]
# 2. Snap each cycle to its nearest allowed start, keeping the
# start paired with the cycle (start_allowed index) it
# belongs to.
repaired = [
(
self._repair_start(
start,
cycle_index,
max_start,
),
cycle_index,
)
for cycle_index, start in enumerate(starts)
]
# 3. Sort by start time, keeping each cycle's own index
# attached so its allowed-start mask is still used
# correctly in step 4.
repaired.sort(key=lambda pair: pair[0])
# 4. Enforce duration + minimum idle gap. Each cycle is
# pushed forward, if needed, to the next start allowed by
# its *own* window.
min_next_start = self.duration_h + self.min_cycle_gap_h
final_starts = [repaired[0][0]]
for index in range(1, len(repaired)):
_, cycle_index = repaired[index]
earliest = final_starts[index - 1] + min_next_start
candidate = self._first_allowed_start_at_or_after(
cycle_index=cycle_index,
earliest=earliest,
)
if candidate is None:
# No valid start remains for this cycle.
final_starts.append(max_start)
else:
final_starts.append(candidate)
# 5. Reconstruct load curve from the final schedule.
self.start_hours = final_starts
self._build_load_curve()
return list(self.start_hours)
def _repair_start(
self,
start: int,
cycle_index: int,
max_start: int,
) -> int:
"""Snaps a start to the nearest allowed start.
Args:
start: Requested (already rounded and clipped) start
hour.
cycle_index: Index into ``self.start_allowed`` for the
cycle being repaired.
max_start: Latest hour at which any cycle may start
without exceeding the prediction horizon, used as a
fallback when the cycle has no allowed start at all.
Returns:
The nearest hour allowed for this cycle, or ``max_start``
if the cycle has no allowed start.
"""
allowed = self.start_allowed[cycle_index]
if not np.any(allowed):
# Same fallback as the vectorized implementation.
return max_start
if allowed[start]:
return start
allowed_indices = np.flatnonzero(allowed)
distances = np.abs(allowed_indices - start)
return int(allowed_indices[np.argmin(distances)])
def _first_allowed_start_at_or_after(
self,
cycle_index: int,
earliest: int,
) -> int | None:
"""Returns the first allowed start at or after ``earliest``.
Args:
cycle_index: Index into ``self.start_allowed`` for the
cycle being scheduled.
earliest: Earliest acceptable start hour.
Returns:
The first allowed hour ``>= earliest``, or ``None`` if no
such hour exists.
"""
allowed = self.start_allowed[cycle_index]
allowed_indices = np.flatnonzero(allowed)
if len(allowed_indices) == 0:
return None
position = np.searchsorted(
allowed_indices,
max(0, earliest),
side="left",
)
if position >= len(allowed_indices):
return None
return int(allowed_indices[position])
# ------------------------------------------------------------------
# Backwards-compatible single-cycle interface
# ------------------------------------------------------------------
def set_starting_time(
self,
start_hour: int,
global_start_hour: int = 0,
) -> int:
"""Sets the start time of the first remaining cycle.
Args:
start_hour: Requested start hour for the first remaining
cycle.
global_start_hour: Retained for API compatibility with
the old, single-cycle implementation. Unused.
Returns:
The repaired start hour of the first remaining cycle, or
``start_hour`` unchanged if there are no remaining
cycles.
"""
if self.num_remaining_cycles == 0:
self.reset_load_curve()
return start_hour
if self.start_hours:
starts = list(self.start_hours)
else:
starts = [self.start_earliest[index] for index in range(self.num_remaining_cycles)]
starts[0] = start_hour
repaired = self.set_starting_times(starts)
return repaired[0]
# ------------------------------------------------------------------
# Load curve
# ------------------------------------------------------------------
def _build_load_curve(self) -> None:
"""Builds the load curve from all scheduled cycles."""
self.reset_load_curve()
# Calculate power per hour based on total consumption and duration
power_per_hour = self.consumption_wh / self.duration_h # Convert to watt-hours
power_per_hour = self.consumption_wh / self.duration_h
# Set the power for the duration of use in the load curve array
if start_hour < len(self.load_curve):
end_hour = min(start_hour + self.duration_h, self.prediction_hours)
self.load_curve[start_hour:end_hour] = power_per_hour
for start_hour in self.start_hours:
if start_hour >= self.prediction_hours:
continue
return start_hour
end_hour = min(
start_hour + self.duration_h,
self.prediction_hours,
)
self.load_curve[start_hour:end_hour] += power_per_hour
def reset_load_curve(self) -> None:
"""Resets the load curve."""
"""Resets the load curve to all zeros."""
self.load_curve = np.zeros(self.prediction_hours)
def get_load_curve(self) -> np.ndarray:
"""Returns the current load curve."""
"""Returns the current hourly load curve, in watts."""
return self.load_curve
def get_load_for_hour(self, hour: int) -> float:
"""Returns the load for a specific hour.
:param hour: The hour for which the load is queried.
:return: The load in watts for the specified hour.
Args:
hour: Hour of the prediction horizon to look up.
Returns:
The load, in watts, at ``hour``.
Raises:
ValueError: If ``hour`` is outside
``[0, prediction_hours)``.
"""
if hour < 0 or hour >= self.prediction_hours:
raise ValueError(
f"The specified hour {hour} is outside the available time frame {self.prediction_hours}."
f"The specified hour {hour} is outside the available "
f"time frame {self.prediction_hours}."
)
return self.load_curve[hour]
return float(self.load_curve[hour])
src/akkudoktoreos/devices/genetic/inverter.py
+52 -1
View File
@@ -1,12 +1,63 @@
from typing import Optional
from loguru import logger
from pydantic import Field
from akkudoktoreos.devices.genetic.battery import Battery
from akkudoktoreos.optimization.genetic.geneticdevices import InverterParameters
from akkudoktoreos.optimization.genetic.geneticdevices import DeviceParameters
from akkudoktoreos.prediction.interpolator import get_eos_load_interpolator
class InverterParameters(DeviceParameters):
"""Inverter Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of inverter", "examples": ["inverter1"]}
)
max_power_wh: float = Field(gt=0, json_schema_extra={"examples": [10000]})
battery_id: Optional[str] = Field(
default=None,
json_schema_extra={"description": "ID of battery", "examples": [None, "battery1"]},
)
ac_to_dc_efficiency: float = Field(
default=1.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of AC to DC conversion (for AC/grid charging of battery). "
"Set to 0 to disable AC charging via inverter. "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0, 0.0],
},
)
dc_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of DC to AC conversion (for battery discharging to AC load/grid). "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0],
},
)
max_ac_charge_power_w: Optional[float] = Field(
default=None,
ge=0,
json_schema_extra={
"description": (
"Maximum AC charging power in watts. "
"None means no additional limit (battery's own max_charge_power_w applies). "
"Set to 0 to disable AC charging."
),
"examples": [None, 0, 5000],
},
)
class Inverter:
def __init__(
self,
src/akkudoktoreos/devices/genetic0/genetic0battery.py
+96 -5
View File
@@ -1,12 +1,103 @@
from typing import Any, Iterator, Optional
import numpy as np
from pydantic import Field
from akkudoktoreos.devices.devices import BATTERY_DEFAULT_CHARGE_RATES
from akkudoktoreos.optimization.genetic0.genetic0devices import (
Genetic0BaseBatteryParameters,
Genetic0SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.settings.batterysettings import BATTERY_DEFAULT_CHARGE_RATES
from akkudoktoreos.optimization.genetic0.genetic0devices import Genetic0DeviceParameters
def max_charging_power_field(description: Optional[str] = None) -> float:
if description is None:
description = "Maximum charging power in watts."
return Field(default=5000, gt=0, json_schema_extra={"description": description})
def initial_soc_percentage_field(description: str) -> int:
return Field(
default=0, ge=0, le=100, json_schema_extra={"description": description, "examples": [42]}
)
def discharging_efficiency_field(default_value: float) -> float:
return Field(
default=default_value,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the discharge efficiency of the battery."
},
)
class Genetic0BaseBatteryParameters(Genetic0DeviceParameters):
"""Battery Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of battery", "examples": ["battery1"]}
)
capacity_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the capacity of the battery in watt-hours.",
"examples": [8000],
},
)
charging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the charging efficiency of the battery."
},
)
discharging_efficiency: float = discharging_efficiency_field(0.88)
max_charge_power_w: Optional[float] = max_charging_power_field()
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the state of charge of the battery at the **start** of the current hour (not the current state)."
)
min_soc_percentage: int = Field(
default=0,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the minimum state of charge (SOC) of the battery in percentage.",
"examples": [10],
},
)
max_soc_percentage: int = Field(
default=100,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the maximum state of charge (SOC) of the battery in percentage."
},
)
charge_rates: Optional[list[float]] = Field(
default=None,
json_schema_extra={
"description": "Charge rates as factor of maximum charging power [0.00 ... 1.00]. None denotes all charge rates are available.",
"examples": [[0.0, 0.25, 0.5, 0.75, 1.0], None],
},
)
class Genetic0SolarPanelBatteryParameters(Genetic0BaseBatteryParameters):
"""PV battery device simulation configuration."""
max_charge_power_w: Optional[float] = max_charging_power_field()
class Genetic0ElectricVehicleParameters(Genetic0BaseBatteryParameters):
"""Battery Electric Vehicle Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of electric vehicle", "examples": ["ev1"]}
)
discharging_efficiency: float = discharging_efficiency_field(1.0)
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the current state of charge (SOC) of the battery in percentage."
)
class Genetic0Battery:
src/akkudoktoreos/devices/genetic0/genetic0homeappliance.py
+37 -3
View File
@@ -1,12 +1,46 @@
from typing import Optional
import numpy as np
from pydantic import Field
from akkudoktoreos.config.configabc import TimeWindow, TimeWindowSequence
from akkudoktoreos.optimization.genetic0.genetic0devices import (
Genetic0HomeApplianceParameters,
)
from akkudoktoreos.optimization.genetic0.genetic0devices import Genetic0DeviceParameters
from akkudoktoreos.utils.datetimeutil import to_datetime, to_duration, to_time
class Genetic0HomeApplianceParameters(Genetic0DeviceParameters):
"""Home Appliance Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of home appliance", "examples": ["dishwasher"]}
)
consumption_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the energy consumption of a household device in watt-hours.",
"examples": [2000],
},
)
duration_h: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the usage duration of a household device in hours.",
"examples": [3],
},
)
time_windows: Optional[TimeWindowSequence] = Field(
default=None,
json_schema_extra={
"description": "List of allowed time windows. Defaults to optimization general time window.",
"examples": [
[
{"start_time": "10:00", "duration": "3 hours"},
],
],
},
)
class Genetic0HomeAppliance:
def __init__(
self,
src/akkudoktoreos/devices/genetic0/genetic0inverter.py
+52 -3
View File
@@ -1,16 +1,65 @@
from typing import Optional
from loguru import logger
from pydantic import Field
from akkudoktoreos.devices.genetic0.genetic0battery import Genetic0Battery
from akkudoktoreos.optimization.genetic0.genetic0devices import (
Genetic0InverterParameters,
)
from akkudoktoreos.optimization.genetic0.genetic0devices import Genetic0DeviceParameters
from akkudoktoreos.optimization.genetic0.genetic0loadinterpolator import (
get_genetic0_load_interpolator,
)
class Genetic0InverterParameters(Genetic0DeviceParameters):
"""Inverter Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of inverter", "examples": ["inverter1"]}
)
max_power_wh: float = Field(gt=0, json_schema_extra={"examples": [10000]})
battery_id: Optional[str] = Field(
default=None,
json_schema_extra={"description": "ID of battery", "examples": [None, "battery1"]},
)
ac_to_dc_efficiency: float = Field(
default=1.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of AC to DC conversion (for AC/grid charging of battery). "
"Set to 0 to disable AC charging via inverter. "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0, 0.0],
},
)
dc_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of DC to AC conversion (for battery discharging to AC load/grid). "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0],
},
)
max_ac_charge_power_w: Optional[float] = Field(
default=None,
ge=0,
json_schema_extra={
"description": (
"Maximum AC charging power in watts. "
"None means no additional limit (battery's own max_charge_power_w applies). "
"Set to 0 to disable AC charging."
),
"examples": [None, 0, 5000],
},
)
class Genetic0Inverter:
def __init__(
self,
src/akkudoktoreos/devices/settings/__init__.py
View File
src/akkudoktoreos/devices/settings/batterysettings.py
+271
View File
@@ -0,0 +1,271 @@
"""Battery and electric vehicle device settings.
Note: Used for the GENETIC and GENETIC0 algorithm.
"""
import re
from typing import TYPE_CHECKING, Any, Optional
import numpy as np
from numpydantic import NDArray, Shape
from pydantic import Field, computed_field, field_validator, model_validator
from akkudoktoreos.devices.settings.devicebasesettings import DevicesBaseSettings
if TYPE_CHECKING:
from akkudoktoreos.devices.genetic0.genetic0battery import (
Genetic0ElectricVehicleParameters,
Genetic0SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.genetic.battery import (
ElectricVehicleParameters,
SolarPanelBatteryParameters,
)
# Default charge rates for battery
BATTERY_DEFAULT_CHARGE_RATES: list[float] = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
class BatteriesCommonSettings(DevicesBaseSettings):
"""Battery and electric vehicle device settings.
Used for both stationary batteries and EV battery packs.
Note: Used for the GENETIC and GENETIC0 algorithm.
"""
capacity_wh: int = Field(
default=8000,
gt=0,
json_schema_extra={"description": "Capacity [Wh].", "examples": [8000]},
)
charging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "Charging efficiency [0.01 ... 1.00].",
"examples": [0.88],
},
)
discharging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "Discharge efficiency [0.01 ... 1.00].",
"examples": [0.88],
},
)
levelized_cost_of_storage_amt_kwh: float = Field(
default=0.0,
json_schema_extra={
"description": (
"Levelized cost of storage (LCOS), the average lifetime cost "
"of delivering one kWh [€/kWh]."
),
"examples": [0.12],
},
)
max_charge_power_w: Optional[float] = Field(
default=5000,
gt=0,
json_schema_extra={
"description": "Maximum charging power [W].",
"examples": [5000],
},
)
min_charge_power_w: Optional[float] = Field(
default=50,
gt=0,
json_schema_extra={
"description": "Minimum charging power [W].",
"examples": [50],
},
)
charge_rates: Optional[list[float]] = Field(
default=BATTERY_DEFAULT_CHARGE_RATES,
json_schema_extra={
"description": (
"Charge rates as factor of maximum charging power [0.00 ... 1.00]. "
"None triggers fallback to default charge-rates."
),
"examples": [[0.0, 0.25, 0.5, 0.75, 1.0], None],
},
)
min_soc_percentage: int = Field(
default=0,
ge=0,
le=100,
json_schema_extra={
"description": (
"Minimum state of charge (SOC) as percentage of capacity [%]. "
"This is the target SoC for charging."
),
"examples": [10],
},
)
max_soc_percentage: int = Field(
default=100,
ge=0,
le=100,
json_schema_extra={
"description": "Maximum state of charge (SOC) as percentage of capacity [%].",
"examples": [100],
},
)
# ------------------------------------------------------------------
# GENETIC domain conversion
# ------------------------------------------------------------------
def to_genetic_pv_bat_param(self) -> "SolarPanelBatteryParameters":
"""Return SolarPanelBatteryParameters for the GENETIC optimizer."""
from akkudoktoreos.devices.genetic.battery import SolarPanelBatteryParameters
return SolarPanelBatteryParameters(
device_id=self.device_id,
capacity_wh=self.capacity_wh,
charging_efficiency=self.charging_efficiency,
discharging_efficiency=self.discharging_efficiency,
max_charge_power_w=self.max_charge_power_w,
min_soc_percentage=self.min_soc_percentage,
max_soc_percentage=self.max_soc_percentage,
)
def to_genetic_ev_bat_param(self) -> "ElectricVehicleParameters":
"""Return ElectricVehicleParameters for the GENETIC optimizer."""
from akkudoktoreos.devices.genetic.battery import ElectricVehicleParameters
return ElectricVehicleParameters(
device_id=self.device_id,
capacity_wh=self.capacity_wh,
charging_efficiency=self.charging_efficiency,
discharging_efficiency=self.discharging_efficiency,
charge_rates=self.charge_rates,
max_charge_power_w=self.max_charge_power_w,
min_soc_percentage=self.min_soc_percentage,
max_soc_percentage=self.max_soc_percentage,
)
# ------------------------------------------------------------------
# GENETIC0 domain conversion
# ------------------------------------------------------------------
def to_genetic0_pv_bat_param(self) -> "Genetic0SolarPanelBatteryParameters":
"""Return Genetic0SolarPanelBatteryParameters for the GENETIC0 optimizer."""
from akkudoktoreos.devices.genetic0.genetic0battery import (
Genetic0SolarPanelBatteryParameters,
)
return Genetic0SolarPanelBatteryParameters(
device_id=self.device_id,
capacity_wh=self.capacity_wh,
charging_efficiency=self.charging_efficiency,
discharging_efficiency=self.discharging_efficiency,
max_charge_power_w=self.max_charge_power_w,
min_soc_percentage=self.min_soc_percentage,
max_soc_percentage=self.max_soc_percentage,
)
def to_genetic0_ev_bat_param(self) -> "Genetic0ElectricVehicleParameters":
"""Return Genetic0ElectricVehicleParameters for the GENETI0C optimizer."""
from akkudoktoreos.devices.genetic0.genetic0battery import (
Genetic0ElectricVehicleParameters,
)
return Genetic0ElectricVehicleParameters(
device_id=self.device_id,
capacity_wh=self.capacity_wh,
charging_efficiency=self.charging_efficiency,
discharging_efficiency=self.discharging_efficiency,
charge_rates=self.charge_rates,
max_charge_power_w=self.max_charge_power_w,
min_soc_percentage=self.min_soc_percentage,
max_soc_percentage=self.max_soc_percentage,
)
@field_validator("charge_rates", mode="before")
def validate_and_sort_charge_rates(cls, v: Any) -> NDArray[Shape["*"], float]:
# None means fallback to default values
if v is None:
return BATTERY_DEFAULT_CHARGE_RATES.copy()
# Convert to numpy array
if isinstance(v, str):
# Remove brackets and split by comma or whitespace
numbers = re.split(r"[,\s]+", v.strip("[]"))
# Filter out any empty strings and convert to floats
arr = np.array([float(x) for x in numbers if x])
else:
arr = np.array(v, dtype=float)
# Must not be empty
if arr.size == 0:
raise ValueError("charge_rates must contain at least one value.")
# Enforce bounds: 0.0 ≤ x ≤ 1.0
if (arr < 0.0).any() or (arr > 1.0).any():
raise ValueError("charge_rates must be within [0.0, 1.0].")
# Remove duplicates + sort
arr = np.unique(arr)
arr.sort()
return arr
@model_validator(mode="after")
def _validate_soc_range(self) -> "BatteriesCommonSettings":
if self.min_soc_percentage >= self.max_soc_percentage:
raise ValueError("min_soc_percentage must be < max_soc_percentage")
if (
self.min_charge_power_w is not None
and self.max_charge_power_w is not None
and self.min_charge_power_w > self.max_charge_power_w
):
raise ValueError("min_charge_power_w must be <= max_charge_power_w")
return self
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_soc_factor(self) -> str:
"""Measurement key for SoC as factor of total capacity [0.0 ... 1.0]."""
return f"{self.device_id}-soc-factor"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_l1_w(self) -> str:
"""Measurement key for L1 power [W]."""
return f"{self.device_id}-power-l1-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_l2_w(self) -> str:
"""Measurement key for L2 power [W]."""
return f"{self.device_id}-power-l2-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_l3_w(self) -> str:
"""Measurement key for L3 power [W]."""
return f"{self.device_id}-power-l3-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_key_power_3_phase_sym_w(self) -> str:
"""Measurement key for symmetric 3-phase power [W]."""
return f"{self.device_id}-power-3-phase-sym-w"
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> list[str]:
"""All measurement keys for this battery."""
return [
self.measurement_key_soc_factor,
self.measurement_key_power_l1_w,
self.measurement_key_power_l2_w,
self.measurement_key_power_l3_w,
self.measurement_key_power_3_phase_sym_w,
]
src/akkudoktoreos/devices/settings/devicebasesettings.py
+39
View File
@@ -0,0 +1,39 @@
"""Base classe for all device settings.
This module contains the building blocks that every device settings class
depends on.:
- ``DevicesBaseSettings``: common ``device_id`` field for all devices.
Nothing in this module is optimizer-specific.
"""
import secrets
import string
from pydantic import Field
from akkudoktoreos.config.configabc import SettingsBaseModel
# ============================================================
# Base settings
# ============================================================
def device_default_id() -> str:
"""Provide random default device id."""
alphabet = string.ascii_letters + string.digits
device_id = "".join(secrets.choice(alphabet) for _ in range(10))
return device_id
class DevicesBaseSettings(SettingsBaseModel):
"""Base devices setting."""
device_id: str = Field(
default_factory=device_default_id,
json_schema_extra={
"description": "ID of device",
"examples": ["battery1", "ev1", "inverter1", "dishwasher"],
},
)
src/akkudoktoreos/devices/settings/homeappliancesettings.py
+258
View File
@@ -0,0 +1,258 @@
"""Controllable home appliance device settings.
Note: Used for the GENETIC and GENETIC0 algorithm.
"""
from typing import TYPE_CHECKING, Optional
from pydantic import Field, computed_field, model_validator
from akkudoktoreos.config.configabc import ConfigScope, CycleTimeWindowSequence
from akkudoktoreos.devices.settings.devicebasesettings import DevicesBaseSettings
if TYPE_CHECKING:
from akkudoktoreos.devices.genetic0.genetic0homeappliance import (
Genetic0HomeApplianceParameters,
)
from akkudoktoreos.devices.genetic.homeappliance import HomeApplianceParameters
class HomeApplianceCommonSettings(DevicesBaseSettings):
"""Controllable home appliance device settings.
Represents a shiftable load whose start time — and optionally the
start times for multiple sequential runs — can be deferred by the
optimiser within per-cycle allowed time windows.
The number of remaining cycles to plan is determined at runtime by
reading ``cycles_completed_measurement_key`` from the measurement
store inside ``HomeApplianceDevice.setup_run``.
``num_cycles`` is required when ``cycle_time_windows`` is ``None``
(unconstrained); when windows are provided it is derived from
``cycle_time_windows.num_cycles()``.
Single-cycle, unconstrained (start any time)
---------------------------------------------
::
device_id: dishwasher
consumption_wh: 1500
duration_h: 2
num_cycles: 1
ports:
- port_id: p_ac
bus_id: bus_ac
direction: sink
Single-cycle, constrained to one window
----------------------------------------
Each window's ``value`` field carries the **cycle index** (0-based).
Windows without a ``value`` are ignored by the optimizer::
device_id: dishwasher
consumption_wh: 1500
duration_h: 2
ports:
- port_id: p_ac
bus_id: bus_ac
direction: sink
cycle_time_windows:
windows:
- start_time: "10:00"
duration: "12 hours"
value: 0
Multi-cycle, per-cycle windows
--------------------------------
Two cycles, each with its own window. Cycle 0 runs in the morning,
cycle 1 in the evening::
device_id: washing_machine
consumption_wh: 2000
duration_h: 2
min_cycle_gap_h: 1
ports:
- port_id: p_ac
bus_id: bus_ac
direction: sink
cycle_time_windows:
windows:
- start_time: "07:00"
duration: "5 hours"
value: 0
- start_time: "17:00"
duration: "5 hours"
value: 1
Multi-cycle, shared window (both cycles may run any time 10:00-20:00)
-----------------------------------------------------------------------
Assign the same-shaped windows to distinct cycle indices so the
optimizer can place them independently::
cycle_time_windows:
windows:
- start_time: "10:00"
duration: "10 hours"
value: 0
- start_time: "10:00"
duration: "10 hours"
value: 1
"""
consumption_wh: int = Field(
default=3000,
gt=0,
json_schema_extra={
"description": "Energy consumption per run cycle [Wh].",
"examples": [2000],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
duration_h: int = Field(
default=3,
gt=0,
le=24,
json_schema_extra={
"description": "Run duration per cycle [h] (1-24).",
"examples": [2],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
num_cycles: int = Field(
default=1,
ge=1,
json_schema_extra={
"description": (
"Number of times the appliance must run within the horizon. "
"Required when cycle_time_windows is null (unconstrained). "
"Ignored when cycle_time_windows is provided -- the number "
"of distinct cycle indices in the windows defines num_cycles. "
"Defaults to 1."
),
"examples": [1, 2],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
cycle_time_windows: Optional[CycleTimeWindowSequence] = Field(
default=None,
json_schema_extra={
"description": (
"Per-cycle allowed scheduling time windows. "
"Each window's value field specifies the cycle index (0-based). "
"When null, the appliance may start at any step and num_cycles "
"must be set explicitly."
),
"examples": [
None,
{
"windows": [
{"start_time": "07:00", "duration": "5 hours", "value": 0},
{"start_time": "17:00", "duration": "5 hours", "value": 1},
]
},
],
"x-scope": [
str(ConfigScope.GENETIC),
],
},
)
min_cycle_gap_h: int = Field(
default=0,
ge=0,
json_schema_extra={
"description": (
"Minimum idle time between the end of one cycle and the start "
"of the next [h]. Applies uniformly between all consecutive cycles. "
"0 means back-to-back runs are permitted."
),
"examples": [0, 1, 4],
"x-scope": [
str(ConfigScope.GENETIC),
],
},
)
cycles_completed_measurement_key: Optional[str] = Field(
default=None,
json_schema_extra={
"description": (
"Measurement store key holding the number of cycles already "
"completed in the current planning day. Read by "
"HomeApplianceDevice.setup_run via context.resolve_measurement. "
"Defaults to '{device_id}.cycles_completed' when null."
),
"examples": ["dishwasher.cycles_completed", None],
"x-scope": [
str(ConfigScope.GENETIC),
],
},
)
@model_validator(mode="after")
def _validate_num_cycles_specified(self) -> "HomeApplianceCommonSettings":
"""Require num_cycles when windows are not provided."""
if self.cycle_time_windows is None and self.num_cycles is None:
raise ValueError(
"num_cycles must be set when cycle_time_windows is null. "
"Provide either cycle_time_windows (windows define cycle count) "
"or set num_cycles explicitly."
)
return self
@computed_field # type: ignore[prop-decorator]
@property
def effective_num_cycles(self) -> int:
"""Number of cycles as seen by the optimizer.
Derived from cycle_time_windows.num_cycles() when windows are
provided; falls back to the explicit num_cycles field otherwise.
"""
if self.cycle_time_windows is not None:
return self.cycle_time_windows.num_cycles()
return self.num_cycles
# ------------------------------------------------------------------
# GENETIC domain conversion
# ------------------------------------------------------------------
def to_genetic_param(self) -> "HomeApplianceParameters":
"""Return HomeApplianceParameters for the GENETIC optimizer."""
from akkudoktoreos.devices.genetic.homeappliance import HomeApplianceParameters
return HomeApplianceParameters(
device_id=self.device_id,
consumption_wh=self.consumption_wh,
duration_h=self.duration_h,
num_cycles=self.effective_num_cycles,
min_cycle_gap_h=self.min_cycle_gap_h,
time_windows=self.cycle_time_windows,
)
# ------------------------------------------------------------------
# GENETIC0 domain conversion
# ------------------------------------------------------------------
def to_genetic0_param(self) -> "Genetic0HomeApplianceParameters":
"""Return Genetic0HomeApplianceParameters for the GENETIC0 optimizer."""
from akkudoktoreos.devices.genetic0.genetic0homeappliance import (
Genetic0HomeApplianceParameters,
)
return Genetic0HomeApplianceParameters(
device_id=self.device_id,
consumption_wh=float(self.consumption_wh),
duration_h=self.duration_h,
time_windows=self.cycle_time_windows,
)
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> Optional[list[str]]:
"""Measurement keys for the home appliance stati that are measurements."""
keys: list[str] = []
return keys
src/akkudoktoreos/devices/settings/invertersettings.py
+393
View File
@@ -0,0 +1,393 @@
"""Inverter device settings."""
from typing import TYPE_CHECKING, Optional
from pydantic import Field, computed_field, model_validator
from akkudoktoreos.config.configabc import ConfigScope
from akkudoktoreos.devices.settings.devicebasesettings import (
DevicesBaseSettings,
)
if TYPE_CHECKING:
from akkudoktoreos.devices.genetic0.genetic0inverter import (
Genetic0InverterParameters,
)
from akkudoktoreos.devices.genetic.inverter import InverterParameters
class InverterCommonSettings(DevicesBaseSettings):
"""Inverter device settings.
An inverter bridges a DC bus (PV / battery) and an AC bus (grid /
household). It must therefore have at least one DC port and one AC
port.
"""
# ------------------------------------------------------------------
# Shared fields (GENETIC + GENETIC0)
# ------------------------------------------------------------------
max_power_w: Optional[float] = Field(
default=None,
gt=0,
json_schema_extra={
"description": "Maximum AC output power [W].",
"examples": [10000],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
ac_to_dc_efficiency: float = Field(
default=1.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of AC→DC conversion for grid-to-battery charging (0–1). "
"Set to 0 to disable AC charging. Default 1.0."
),
"examples": [0.95, 1.0, 0.0],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
dc_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of DC→AC conversion for battery discharging (0–1). Default 1.0."
),
"examples": [0.95, 1.0],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
max_ac_charge_power_w: Optional[float] = Field(
default=None,
ge=0,
json_schema_extra={
"description": (
"Maximum AC charging power [W]. "
"null means no additional limit. 0 disables AC charging."
),
"examples": [None, 0, 5000],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
battery_id: Optional[str] = Field(
default=None,
json_schema_extra={
"description": ("Device ID of the battery."),
"examples": [None],
"x-scope": [str(ConfigScope.GENETIC), str(ConfigScope.GENETIC0)],
},
)
# ------------------------------------------------------------------
# UNUSED-only fields
# ------------------------------------------------------------------
# Auxiliary power consumption
off_state_power_consumption_w: float = Field(
default=0.0,
ge=0,
json_schema_extra={
"description": (
"Standby power consumed when the inverter is fully idle "
"(battery=0 and PV=0) [W]. Default 0.0."
),
"examples": [5.0, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
on_state_power_consumption_w: float = Field(
default=0.0,
ge=0,
json_schema_extra={
"description": (
"Auxiliary power consumed whenever the inverter is active "
"(non-zero AC power) [W]. Default 0.0."
),
"examples": [10.0, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
# PV parameters (used for SOLAR and HYBRID inverter types)
pv_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of PV DC→AC conversion (0–1). "
"Required when pv_power_w_key is set (SOLAR or HYBRID). Default 1.0."
),
"examples": [0.97, 1.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
pv_to_battery_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of PV DC→battery charging path (0–1). "
"Used for HYBRID inverters only. Default 1.0."
),
"examples": [0.98, 1.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
pv_max_power_w: Optional[float] = Field(
default=None,
gt=0,
json_schema_extra={
"description": (
"Maximum DC PV power fed into the inverter [W]. "
"Required when pv_power_w_key is set (SOLAR or HYBRID). "
"Values from pv_power_w_key are clipped to this limit."
),
"examples": [8000.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
pv_min_power_w: float = Field(
default=0.0,
ge=0,
json_schema_extra={
"description": (
"Minimum DC PV power threshold [W]. Steps with available PV "
"below this value are treated as zero. Default 0.0."
),
"examples": [50.0, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
pv_power_w_key: Optional[str] = Field(
default=None,
json_schema_extra={
"description": (
"SimulationContext prediction key resolving to a per-step PV "
"power forecast array [W] of shape (horizon,). "
"Set for SOLAR and HYBRID inverter types; leave None for BATTERY."
),
"examples": ["pv_forecast_w", None],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
# Battery parameters (used for BATTERY and HYBRID inverter types)
battery_capacity_wh: Optional[float] = Field(
default=None,
gt=0,
json_schema_extra={
"description": (
"Usable battery capacity [Wh]. Required for BATTERY and HYBRID inverter types."
),
"examples": [10000.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_charge_rates: Optional[list[float]] = Field(
default=None,
json_schema_extra={
"description": (
"Optional list of discrete charge rate fractions (each in (0, 1]). "
"When set, the battery is constrained to these specific fractions "
"of battery_max_charge_rate. null means continuous charging. "
"All values must be in (0, 1]."
),
"examples": [None, [0.25, 0.5, 1.0]],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_min_charge_rate: float = Field(
default=0.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Minimum non-zero charge rate as a fraction of the 1C rate "
"(1C = battery_capacity_wh W). "
"Charge commands below this threshold are rounded to zero. Default 0.0."
),
"examples": [0.1, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_max_charge_rate: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Maximum charge rate as a fraction of the 1C rate "
"(1C = battery_capacity_wh W). "
"bat_factor=+1 maps to this rate. Default 1.0."
),
"examples": [0.5, 1.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_min_discharge_rate: float = Field(
default=0.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Minimum discharge rate as a fraction of the 1C rate "
"(1C = battery_capacity_wh W). "
"Discharge commands below this threshold are rounded to zero. Default 0.0."
),
"examples": [0.1, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_max_discharge_rate: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Maximum discharge rate as a fraction of the 1C rate. "
"(1C = battery_capacity_wh W). "
"bat_factor=−1 maps to this rate. Default 1.0."
),
"examples": [0.5, 1.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_min_soc_factor: float = Field(
default=0.0,
ge=0,
lt=1,
json_schema_extra={
"description": (
"Minimum allowed state of charge as a fraction of battery_capacity_wh. "
"Must be < battery_max_soc_factor. Default 0.0."
),
"examples": [0.1, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_max_soc_factor: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Maximum allowed state of charge as a fraction of battery_capacity_wh. "
"Must be > battery_min_soc_factor. Default 1.0."
),
"examples": [0.9, 1.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_initial_soc_factor_key: str = Field(
default="",
json_schema_extra={
"description": (
"SimulationContext measurement key resolving to the initial battery "
"SoC as a fraction of battery_capacity_wh, in [min_soc_factor, max_soc_factor]. "
"An empty string means the device uses battery_min_soc_factor as the "
"initial SoC (fully depleted to the minimum)."
),
"examples": ["battery1_soc_factor", ""],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_lcos_amt_kwh: float = Field(
default=0.0,
ge=0,
json_schema_extra={
"description": (
"Levelized cost of battery storage [Amt./kWh cycled]. "
"Penalises unnecessary charging/discharging so the GA avoids "
"grid-charge→discharge cycles with no price-spread benefit. "
"Typical residential Li-ion value: 0.05 Amt./kWh. "
"Set to 0.0 to encourage the optimizer to use the battery. "
"Defaults to 0.0."
),
"examples": [0.05, 0.0],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
battery_discharge_reward_amt_kwh: float = Field(
default=0.02,
ge=0,
json_schema_extra={
"description": (
"Shadow price rewarding battery discharge [Amt./kWh discharged AC]. "
"Adds a direct fitness benefit per kWh the battery delivers, on top of "
"the grid import cost reduction already captured by GridConnectionDevice. "
"Helps the GA discover discharge when the load-matching rate is small "
"relative to mutation noise. "
"Suggested value: import_price - export_price - lcos "
"(e.g. 0.30 - 0.08 - 0.05 = 0.17). Set to 0.0 to disable."
),
"examples": [0.0, 0.17],
"x-scope": [str(ConfigScope.UNUSED)],
},
)
# ------------------------------------------------------------------
# Validators
# ------------------------------------------------------------------
@model_validator(mode="after")
def _validate_soc_factors(self) -> "InverterCommonSettings":
if self.battery_min_soc_factor >= self.battery_max_soc_factor:
raise ValueError(
"battery_min_soc_factor must be strictly less than battery_max_soc_factor"
)
return self
# ------------------------------------------------------------------
# GENETIC domain conversion
# ------------------------------------------------------------------
def to_genetic_param(self) -> "InverterParameters":
"""Return InverterParameters for the GENETIC optimizer."""
from akkudoktoreos.devices.genetic.inverter import InverterParameters
return InverterParameters(
device_id=self.device_id,
max_power_wh=self.max_power_w,
battery_id=self.battery_id,
ac_to_dc_efficiency=self.ac_to_dc_efficiency,
dc_to_ac_efficiency=self.dc_to_ac_efficiency,
max_ac_charge_power_w=self.max_ac_charge_power_w,
)
# ------------------------------------------------------------------
# GENETIC0 domain conversion
# ------------------------------------------------------------------
def to_genetic0_param(self) -> "Genetic0InverterParameters":
"""Return Genetic0InverterParameters for the GENETIC0 optimizer."""
from akkudoktoreos.devices.genetic0.genetic0inverter import (
Genetic0InverterParameters,
)
return Genetic0InverterParameters(
device_id=self.device_id,
max_power_wh=self.max_power_w,
battery_id=self.battery_id,
ac_to_dc_efficiency=self.ac_to_dc_efficiency,
dc_to_ac_efficiency=self.dc_to_ac_efficiency,
max_ac_charge_power_w=self.max_ac_charge_power_w,
)
@computed_field # type: ignore[prop-decorator]
@property
def measurement_keys(self) -> list[str]:
"""Measurement keys for this inverter.
Returns the ``battery_initial_soc_factor_key`` if non-empty, so
the EMS measurement store knows to watch for this key.
"""
if self.battery_initial_soc_factor_key:
return [self.battery_initial_soc_factor_key]
return []
src/akkudoktoreos/optimization/genetic/genetic.py
+8 -8
View File
@@ -1102,12 +1102,12 @@ class GeneticOptimization(OptimizationBase):
self.ev_possible_charge_values = parameters.ev.charge_rates
elif (
self.config.devices.electric_vehicles
and self.config.devices.electric_vehicles[0]
and self.config.devices.electric_vehicles[0].charge_rates is not None
and len(self.config.devices.electric_vehicles) > 0
and list(self.config.devices.electric_vehicles.values())[0].charge_rates is not None
):
self.ev_possible_charge_values = self.config.devices.electric_vehicles[
0
].charge_rates
self.ev_possible_charge_values = list(
self.config.devices.electric_vehicles.values()
)[0].charge_rates
else:
warning_msg = "No charge rates provided for electric vehicle - using default."
logger.warning(warning_msg)
@@ -1136,11 +1136,11 @@ class GeneticOptimization(OptimizationBase):
] or [1.0]
elif (
self.config.devices.batteries
and self.config.devices.batteries[0]
and self.config.devices.batteries[0].charge_rates
and len(self.config.devices.batteries) > 0
and list(self.config.devices.batteries.values())[0].charge_rates
):
self.bat_possible_charge_values = [
r for r in self.config.devices.batteries[0].charge_rates if r > 0.0
r for r in list(self.config.devices.batteries.values())[0].charge_rates if r > 0.0
] or [1.0]
else:
self.bat_possible_charge_values = [1.0]
src/akkudoktoreos/optimization/genetic/geneticdevices.py
-177
View File
@@ -4,7 +4,6 @@ from typing import Optional
from pydantic import Field
from akkudoktoreos.config.configabc import TimeWindowSequence
from akkudoktoreos.optimization.genetic.geneticabc import GeneticParametersBaseModel
@@ -18,179 +17,3 @@ class DeviceParameters(GeneticParametersBaseModel):
"examples": [None],
},
)
def max_charging_power_field(description: Optional[str] = None) -> float:
if description is None:
description = "Maximum charging power in watts."
return Field(default=5000, gt=0, json_schema_extra={"description": description})
def initial_soc_percentage_field(description: str) -> int:
return Field(
default=0, ge=0, le=100, json_schema_extra={"description": description, "examples": [42]}
)
def discharging_efficiency_field(default_value: float) -> float:
return Field(
default=default_value,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the discharge efficiency of the battery."
},
)
class BaseBatteryParameters(DeviceParameters):
"""Battery Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of battery", "examples": ["battery1"]}
)
capacity_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the capacity of the battery in watt-hours.",
"examples": [8000],
},
)
charging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the charging efficiency of the battery."
},
)
discharging_efficiency: float = discharging_efficiency_field(0.88)
max_charge_power_w: Optional[float] = max_charging_power_field()
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the state of charge of the battery at the **start** of the current hour (not the current state)."
)
min_soc_percentage: int = Field(
default=0,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the minimum state of charge (SOC) of the battery in percentage.",
"examples": [10],
},
)
max_soc_percentage: int = Field(
default=100,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the maximum state of charge (SOC) of the battery in percentage."
},
)
charge_rates: Optional[list[float]] = Field(
default=None,
json_schema_extra={
"description": "Charge rates as factor of maximum charging power [0.00 ... 1.00]. None denotes all charge rates are available.",
"examples": [[0.0, 0.25, 0.5, 0.75, 1.0], None],
},
)
class SolarPanelBatteryParameters(BaseBatteryParameters):
"""PV battery device simulation configuration."""
max_charge_power_w: Optional[float] = max_charging_power_field()
class ElectricVehicleParameters(BaseBatteryParameters):
"""Battery Electric Vehicle Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of electric vehicle", "examples": ["ev1"]}
)
discharging_efficiency: float = discharging_efficiency_field(1.0)
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the current state of charge (SOC) of the battery in percentage."
)
class HomeApplianceParameters(DeviceParameters):
"""Home Appliance Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of home appliance", "examples": ["dishwasher"]}
)
consumption_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the energy consumption of a household device in watt-hours.",
"examples": [2000],
},
)
duration_h: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the usage duration of a household device in hours.",
"examples": [3],
},
)
time_windows: Optional[TimeWindowSequence] = Field(
default=None,
json_schema_extra={
"description": "List of allowed time windows. Defaults to optimization general time window.",
"examples": [
[
{"start_time": "10:00", "duration": "3 hours"},
],
],
},
)
class InverterParameters(DeviceParameters):
"""Inverter Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of inverter", "examples": ["inverter1"]}
)
max_power_wh: float = Field(gt=0, json_schema_extra={"examples": [10000]})
battery_id: Optional[str] = Field(
default=None,
json_schema_extra={"description": "ID of battery", "examples": [None, "battery1"]},
)
ac_to_dc_efficiency: float = Field(
default=1.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of AC to DC conversion (for AC/grid charging of battery). "
"Set to 0 to disable AC charging via inverter. "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0, 0.0],
},
)
dc_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of DC to AC conversion (for battery discharging to AC load/grid). "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0],
},
)
max_ac_charge_power_w: Optional[float] = Field(
default=None,
ge=0,
json_schema_extra={
"description": (
"Maximum AC charging power in watts. "
"None means no additional limit (battery's own max_charge_power_w applies). "
"Set to 0 to disable AC charging."
),
"examples": [None, 0, 5000],
},
)
src/akkudoktoreos/optimization/genetic/geneticparams.py
+64 -75
View File
@@ -27,13 +27,13 @@ from akkudoktoreos.core.coreabc import (
PredictionMixin,
get_ems,
)
from akkudoktoreos.optimization.genetic.geneticabc import GeneticParametersBaseModel
from akkudoktoreos.optimization.genetic.geneticdevices import (
from akkudoktoreos.devices.genetic.battery import (
ElectricVehicleParameters,
HomeApplianceParameters,
InverterParameters,
SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.genetic.homeappliance import HomeApplianceParameters
from akkudoktoreos.devices.genetic.inverter import InverterParameters
from akkudoktoreos.optimization.genetic.geneticabc import GeneticParametersBaseModel
from akkudoktoreos.utils.datetimeutil import to_duration
# Do not import directly from akkudoktoreos.core.coreabc
@@ -269,6 +269,11 @@ class GeneticOptimizationParameters(
if "ev_soc_miss" not in cls.config.optimization.genetic.penalties:
logger.info("Genetic penalties unknown - defaulting to ev_soc_miss = 10.")
cls.config.optimization.genetic.penalties["ev_soc_miss"] = 10
# Setup some basic providers if not set
if not cls.config.weather.provider:
cls.config.weather.provider = "OpenMeteo"
if not cls.config.load.provider:
cls.config.load.provider = "LoadAkkudoktor"
# Get start solution from last run
start_solution = None
@@ -557,36 +562,38 @@ class GeneticOptimizationParameters(
else:
if cls.config.devices.batteries is None:
logger.info("No battery device data available - defaulting to demo data.")
cls.config.devices.batteries = [{"device_id": "battery1", "capacity_wh": 8000}]
cls.config.devices.batteries = {
"battery1": {
"device_id": "battery1",
"capacity_wh": 8000,
},
}
try:
battery_config = cls.config.devices.batteries[0]
battery_params = SolarPanelBatteryParameters(
device_id=battery_config.device_id,
capacity_wh=battery_config.capacity_wh,
charging_efficiency=battery_config.charging_efficiency,
discharging_efficiency=battery_config.discharging_efficiency,
max_charge_power_w=battery_config.max_charge_power_w,
min_soc_percentage=battery_config.min_soc_percentage,
max_soc_percentage=battery_config.max_soc_percentage,
charge_rates=battery_config.charge_rates,
)
# Take first battery
battery_config = list(cls.config.devices.batteries.values())[0]
battery_params = battery_config.to_genetic_pv_bat_param()
except Exception as e:
logger.info(
"No battery device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
attempt,
e,
)
cls.config.devices.batteries = [{"device_id": "battery1", "capacity_wh": 8000}]
cls.config.devices.batteries = {
"battery1": {
"device_id": "battery1",
"capacity_wh": 8000,
},
}
# Retry
continue
# Levelized cost of ownership
if battery_config.levelized_cost_of_storage_kwh is None:
if battery_config.levelized_cost_of_storage_amt_kwh is None:
logger.info(
"No battery device LCOS data available - defaulting to 0 [amount/kWh]. Parameter preparation attempt {}.",
attempt,
)
battery_config.levelized_cost_of_storage_kwh = 0
battery_lcos_kwh = battery_config.levelized_cost_of_storage_kwh
battery_config.levelized_cost_of_storage_amt_kwh = 0
battery_lcos_kwh = battery_config.levelized_cost_of_storage_amt_kwh
# Initial SOC
try:
initial_soc_factor = await cls.measurement.key_to_value(
@@ -623,26 +630,18 @@ class GeneticOptimizationParameters(
"No electric vehicle device data available - defaulting to demo data."
)
cls.config.devices.max_electric_vehicles = 1
cls.config.devices.electric_vehicles = [
{
"device_id": "ev11",
cls.config.devices.electric_vehicles = {
"ev1": {
"device_id": "ev1",
"capacity_wh": 50000,
"charge_rates": [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
"min_soc_percentage": 70,
}
]
},
}
try:
electric_vehicle_config = cls.config.devices.electric_vehicles[0]
electric_vehicle_params = ElectricVehicleParameters(
device_id=electric_vehicle_config.device_id,
capacity_wh=electric_vehicle_config.capacity_wh,
charging_efficiency=electric_vehicle_config.charging_efficiency,
discharging_efficiency=electric_vehicle_config.discharging_efficiency,
charge_rates=electric_vehicle_config.charge_rates,
max_charge_power_w=electric_vehicle_config.max_charge_power_w,
min_soc_percentage=electric_vehicle_config.min_soc_percentage,
max_soc_percentage=electric_vehicle_config.max_soc_percentage,
)
# Take first electric_vehicle
electric_vehicle_config = list(cls.config.devices.electric_vehicles.values())[0]
electric_vehicle_params = electric_vehicle_config.to_genetic_ev_bat_param()
except Exception as e:
logger.info(
"No electric_vehicle device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
@@ -650,14 +649,14 @@ class GeneticOptimizationParameters(
e,
)
cls.config.devices.max_electric_vehicles = 1
cls.config.devices.electric_vehicles = [
{
"device_id": "ev12",
cls.config.devices.electric_vehicles = {
"ev1": {
"device_id": "ev1",
"capacity_wh": 50000,
"charge_rates": [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
"min_soc_percentage": 70,
}
]
},
}
# Retry
continue
# Initial SOC
@@ -693,36 +692,30 @@ class GeneticOptimizationParameters(
else:
if cls.config.devices.inverters is None:
logger.info("No inverter device data available - defaulting to demo data.")
cls.config.devices.inverters = [
{
cls.config.devices.inverters = {
"inverter1": {
"device_id": "inverter1",
"max_power_w": 10000,
"battery_id": battery_config.device_id,
}
]
},
}
try:
inverter_config = cls.config.devices.inverters[0]
inverter_params = InverterParameters(
device_id=inverter_config.device_id,
max_power_wh=inverter_config.max_power_w,
battery_id=inverter_config.battery_id,
ac_to_dc_efficiency=inverter_config.ac_to_dc_efficiency,
dc_to_ac_efficiency=inverter_config.dc_to_ac_efficiency,
max_ac_charge_power_w=inverter_config.max_ac_charge_power_w,
)
# Take first inverter
inverter_config = list(cls.config.devices.inverters.values())[0]
inverter_params = inverter_config.to_genetic_param()
except Exception as e:
logger.info(
"No inverter device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
attempt,
e,
)
cls.config.devices.inverters = [
{
cls.config.devices.inverters = {
"inverter1": {
"device_id": "inverter1",
"max_power_w": 10000,
"battery_id": battery_config.device_id,
}
]
},
}
# Retry
continue
@@ -739,12 +732,12 @@ class GeneticOptimizationParameters(
logger.info(
"No home appliance device data available - defaulting to demo data."
)
cls.config.devices.home_appliances = [
{
cls.config.devices.home_appliances = {
"dishwasher1": {
"device_id": "dishwasher1",
"consumption_wh": 2000,
"duration_h": 3.0,
"time_windows": {
"cycle_time_windows": {
"windows": [
{
"start_time": "08:00",
@@ -756,30 +749,26 @@ class GeneticOptimizationParameters(
},
],
},
}
]
},
}
try:
home_appliance_config = cls.config.devices.home_appliances[0]
home_appliance_params = HomeApplianceParameters(
device_id=home_appliance_config.device_id,
consumption_wh=home_appliance_config.consumption_wh,
duration_h=home_appliance_config.duration_h,
time_windows=home_appliance_config.time_windows,
)
# Take first appliance
home_appliance_config = list(cls.config.devices.home_appliances.values())[0]
home_appliance_params = home_appliance_config.to_genetic_param()
except Exception as e:
logger.info(
"No home appliance device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
attempt,
e,
)
cls.config.devices.home_appliances = [
{
cls.config.devices.home_appliances = {
"dishwasher1": {
"device_id": "dishwasher1",
"consumption_wh": 2000,
"duration_h": 3.0,
"time_windows": None,
}
]
"cycle_time_windows": None,
},
}
# Retry
continue
src/akkudoktoreos/optimization/genetic/geneticsolution.py
+6 -6
View File
@@ -350,21 +350,21 @@ class GeneticSolution(ConfigMixin, GeneticParametersBaseModel):
def _battery_device_id(self) -> str:
"""Get battery device id."""
try:
return self.config.devices.batteries[0].device_id
return list(self.config.devices.batteries.values())[0].device_id
except Exception:
return "battery1"
def _ev_device_id(self) -> str:
"""Get electric vehicle device id."""
try:
return self.config.devices.electric_vehicles[0].device_id
return self.config.devices.electric_vehicles.values()[0].device_id
except Exception:
return "ev1"
def _homeappliance_device_id(self) -> str:
"""Get home appliance device id."""
try:
return self.config.devices.home_appliances[0].device_id
return self.config.devices.home_appliances.values()[0].device_id
except Exception:
return "homeappliance1"
@@ -453,11 +453,11 @@ class GeneticSolution(ConfigMixin, GeneticParametersBaseModel):
(the inverter curtails automatically, but this makes intent clear).
- Discharge: blocked when SOC is at or below min SOC.
"""
bat_list = self.config.devices.batteries
if not bat_list:
bat_dict = self.config.devices.batteries
if bat_dict is None or len(bat_dict) <= 0:
return ac_charge, dc_charge, discharge_allowed
bat = bat_list[0]
bat = list(bat_dict.values())[0]
min_soc = float(bat.min_soc_percentage)
max_soc = float(bat.max_soc_percentage)
capacity_wh = float(bat.capacity_wh)
src/akkudoktoreos/optimization/genetic0/genetic0.py
+8 -8
View File
@@ -1106,12 +1106,12 @@ class Genetic0Optimization(OptimizationBase):
self.ev_possible_charge_values = parameters.ev.charge_rates
elif (
self.config.devices.electric_vehicles
and self.config.devices.electric_vehicles[0]
and self.config.devices.electric_vehicles[0].charge_rates is not None
and len(self.config.devices.electric_vehicles) > 0
and list(self.config.devices.electric_vehicles.values())[0].charge_rates is not None
):
self.ev_possible_charge_values = self.config.devices.electric_vehicles[
0
].charge_rates
self.ev_possible_charge_values = list(
self.config.devices.electric_vehicles.values()
)[0].charge_rates
else:
warning_msg = "No charge rates provided for electric vehicle - using default."
logger.warning(warning_msg)
@@ -1140,11 +1140,11 @@ class Genetic0Optimization(OptimizationBase):
] or [1.0]
elif (
self.config.devices.batteries
and self.config.devices.batteries[0]
and self.config.devices.batteries[0].charge_rates
and len(self.config.devices.batteries) > 0
and list(self.config.devices.batteries.values())[0].charge_rates
):
self.bat_possible_charge_values = [
r for r in self.config.devices.batteries[0].charge_rates if r > 0.0
r for r in list(self.config.devices.batteries.values())[0].charge_rates if r > 0.0
] or [1.0]
else:
self.bat_possible_charge_values = [1.0]
src/akkudoktoreos/optimization/genetic0/genetic0devices.py
-177
View File
@@ -4,7 +4,6 @@ from typing import Optional
from pydantic import Field
from akkudoktoreos.config.configabc import TimeWindowSequence
from akkudoktoreos.optimization.genetic0.genetic0abc import Genetic0ParametersBaseModel
@@ -18,179 +17,3 @@ class Genetic0DeviceParameters(Genetic0ParametersBaseModel):
"examples": [None],
},
)
def max_charging_power_field(description: Optional[str] = None) -> float:
if description is None:
description = "Maximum charging power in watts."
return Field(default=5000, gt=0, json_schema_extra={"description": description})
def initial_soc_percentage_field(description: str) -> int:
return Field(
default=0, ge=0, le=100, json_schema_extra={"description": description, "examples": [42]}
)
def discharging_efficiency_field(default_value: float) -> float:
return Field(
default=default_value,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the discharge efficiency of the battery."
},
)
class Genetic0BaseBatteryParameters(Genetic0DeviceParameters):
"""Battery Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of battery", "examples": ["battery1"]}
)
capacity_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the capacity of the battery in watt-hours.",
"examples": [8000],
},
)
charging_efficiency: float = Field(
default=0.88,
gt=0,
le=1,
json_schema_extra={
"description": "A float representing the charging efficiency of the battery."
},
)
discharging_efficiency: float = discharging_efficiency_field(0.88)
max_charge_power_w: Optional[float] = max_charging_power_field()
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the state of charge of the battery at the **start** of the current hour (not the current state)."
)
min_soc_percentage: int = Field(
default=0,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the minimum state of charge (SOC) of the battery in percentage.",
"examples": [10],
},
)
max_soc_percentage: int = Field(
default=100,
ge=0,
le=100,
json_schema_extra={
"description": "An integer representing the maximum state of charge (SOC) of the battery in percentage."
},
)
charge_rates: Optional[list[float]] = Field(
default=None,
json_schema_extra={
"description": "Charge rates as factor of maximum charging power [0.00 ... 1.00]. None denotes all charge rates are available.",
"examples": [[0.0, 0.25, 0.5, 0.75, 1.0], None],
},
)
class Genetic0SolarPanelBatteryParameters(Genetic0BaseBatteryParameters):
"""PV battery device simulation configuration."""
max_charge_power_w: Optional[float] = max_charging_power_field()
class Genetic0ElectricVehicleParameters(Genetic0BaseBatteryParameters):
"""Battery Electric Vehicle Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of electric vehicle", "examples": ["ev1"]}
)
discharging_efficiency: float = discharging_efficiency_field(1.0)
initial_soc_percentage: int = initial_soc_percentage_field(
"An integer representing the current state of charge (SOC) of the battery in percentage."
)
class Genetic0HomeApplianceParameters(Genetic0DeviceParameters):
"""Home Appliance Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of home appliance", "examples": ["dishwasher"]}
)
consumption_wh: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the energy consumption of a household device in watt-hours.",
"examples": [2000],
},
)
duration_h: int = Field(
gt=0,
json_schema_extra={
"description": "An integer representing the usage duration of a household device in hours.",
"examples": [3],
},
)
time_windows: Optional[TimeWindowSequence] = Field(
default=None,
json_schema_extra={
"description": "List of allowed time windows. Defaults to optimization general time window.",
"examples": [
[
{"start_time": "10:00", "duration": "3 hours"},
],
],
},
)
class Genetic0InverterParameters(Genetic0DeviceParameters):
"""Inverter Device Simulation Configuration."""
device_id: str = Field(
json_schema_extra={"description": "ID of inverter", "examples": ["inverter1"]}
)
max_power_wh: float = Field(gt=0, json_schema_extra={"examples": [10000]})
battery_id: Optional[str] = Field(
default=None,
json_schema_extra={"description": "ID of battery", "examples": [None, "battery1"]},
)
ac_to_dc_efficiency: float = Field(
default=1.0,
ge=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of AC to DC conversion (for AC/grid charging of battery). "
"Set to 0 to disable AC charging via inverter. "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0, 0.0],
},
)
dc_to_ac_efficiency: float = Field(
default=1.0,
gt=0,
le=1,
json_schema_extra={
"description": (
"Efficiency of DC to AC conversion (for battery discharging to AC load/grid). "
"Default 1.0 for backward compatibility (no additional inverter loss)."
),
"examples": [0.95, 1.0],
},
)
max_ac_charge_power_w: Optional[float] = Field(
default=None,
ge=0,
json_schema_extra={
"description": (
"Maximum AC charging power in watts. "
"None means no additional limit (battery's own max_charge_power_w applies). "
"Set to 0 to disable AC charging."
),
"examples": [None, 0, 5000],
},
)
src/akkudoktoreos/optimization/genetic0/genetic0params.py
+66 -75
View File
@@ -27,13 +27,15 @@ from akkudoktoreos.core.coreabc import (
PredictionMixin,
get_ems,
)
from akkudoktoreos.optimization.genetic0.genetic0abc import Genetic0ParametersBaseModel
from akkudoktoreos.optimization.genetic0.genetic0devices import (
from akkudoktoreos.devices.genetic0.genetic0battery import (
Genetic0ElectricVehicleParameters,
Genetic0HomeApplianceParameters,
Genetic0InverterParameters,
Genetic0SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.genetic0.genetic0homeappliance import (
Genetic0HomeApplianceParameters,
)
from akkudoktoreos.devices.genetic0.genetic0inverter import Genetic0InverterParameters
from akkudoktoreos.optimization.genetic0.genetic0abc import Genetic0ParametersBaseModel
from akkudoktoreos.utils.datetimeutil import to_duration
# Do not import directly from akkudoktoreos.core.coreabc
@@ -260,6 +262,11 @@ class Genetic0OptimizationParameters(
if "ev_soc_miss" not in cls.config.optimization.genetic0.penalties:
logger.info("Genetic penalties unknown - defaulting to ev_soc_miss = 10.")
cls.config.optimization.genetic0.penalties["ev_soc_miss"] = 10
# Setup some basic providers if not set
if not cls.config.weather.provider:
cls.config.weather.provider = "OpenMeteo"
if not cls.config.load.provider:
cls.config.load.provider = "LoadAkkudoktor"
# Get start solution from last run
start_solution = None
@@ -548,36 +555,38 @@ class Genetic0OptimizationParameters(
else:
if cls.config.devices.batteries is None:
logger.info("No battery device data available - defaulting to demo data.")
cls.config.devices.batteries = [{"device_id": "battery1", "capacity_wh": 8000}]
cls.config.devices.batteries = {
"battery1": {
"device_id": "battery1",
"capacity_wh": 8000,
},
}
try:
battery_config = cls.config.devices.batteries[0]
battery_params = Genetic0SolarPanelBatteryParameters(
device_id=battery_config.device_id,
capacity_wh=battery_config.capacity_wh,
charging_efficiency=battery_config.charging_efficiency,
discharging_efficiency=battery_config.discharging_efficiency,
max_charge_power_w=battery_config.max_charge_power_w,
min_soc_percentage=battery_config.min_soc_percentage,
max_soc_percentage=battery_config.max_soc_percentage,
charge_rates=battery_config.charge_rates,
)
# Take first battery
battery_config = list(cls.config.devices.batteries.values())[0]
battery_params = battery_config.to_genetic0_pv_bat_param()
except Exception as e:
logger.info(
"No battery device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
attempt,
e,
)
cls.config.devices.batteries = [{"device_id": "battery1", "capacity_wh": 8000}]
cls.config.devices.batteries = {
"battery1": {
"device_id": "battery1",
"capacity_wh": 8000,
},
}
# Retry
continue
# Levelized cost of ownership
if battery_config.levelized_cost_of_storage_kwh is None:
if battery_config.levelized_cost_of_storage_amt_kwh is None:
logger.info(
"No battery device LCOS data available - defaulting to 0 [amount/kWh]. Parameter preparation attempt {}.",
attempt,
)
battery_config.levelized_cost_of_storage_kwh = 0
battery_lcos_kwh = battery_config.levelized_cost_of_storage_kwh
battery_config.levelized_cost_of_storage_amt_kwh = 0
battery_lcos_kwh = battery_config.levelized_cost_of_storage_amt_kwh
# Initial SOC
try:
initial_soc_factor = await cls.measurement.key_to_value(
@@ -614,26 +623,18 @@ class Genetic0OptimizationParameters(
"No electric vehicle device data available - defaulting to demo data."
)
cls.config.devices.max_electric_vehicles = 1
cls.config.devices.electric_vehicles = [
{
"device_id": "ev11",
cls.config.devices.electric_vehicles = {
"ev1": {
"device_id": "ev1",
"capacity_wh": 50000,
"charge_rates": [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
"min_soc_percentage": 70,
}
]
},
}
try:
electric_vehicle_config = cls.config.devices.electric_vehicles[0]
electric_vehicle_params = Genetic0ElectricVehicleParameters(
device_id=electric_vehicle_config.device_id,
capacity_wh=electric_vehicle_config.capacity_wh,
charging_efficiency=electric_vehicle_config.charging_efficiency,
discharging_efficiency=electric_vehicle_config.discharging_efficiency,
charge_rates=electric_vehicle_config.charge_rates,
max_charge_power_w=electric_vehicle_config.max_charge_power_w,
min_soc_percentage=electric_vehicle_config.min_soc_percentage,
max_soc_percentage=electric_vehicle_config.max_soc_percentage,
)
# Take first electric_vehicle
electric_vehicle_config = list(cls.config.devices.electric_vehicles.values())[0]
electric_vehicle_params = electric_vehicle_config.to_genetic0_ev_bat_param()
except Exception as e:
logger.info(
"No electric_vehicle device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
@@ -641,14 +642,14 @@ class Genetic0OptimizationParameters(
e,
)
cls.config.devices.max_electric_vehicles = 1
cls.config.devices.electric_vehicles = [
{
"device_id": "ev12",
cls.config.devices.electric_vehicles = {
"ev1": {
"device_id": "ev1",
"capacity_wh": 50000,
"charge_rates": [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
"min_soc_percentage": 70,
}
]
},
}
# Retry
continue
# Initial SOC
@@ -684,36 +685,30 @@ class Genetic0OptimizationParameters(
else:
if cls.config.devices.inverters is None:
logger.info("No inverter device data available - defaulting to demo data.")
cls.config.devices.inverters = [
{
cls.config.devices.inverters = {
"inverter1": {
"device_id": "inverter1",
"max_power_w": 10000,
"battery_id": battery_config.device_id,
}
]
},
}
try:
inverter_config = cls.config.devices.inverters[0]
inverter_params = Genetic0InverterParameters(
device_id=inverter_config.device_id,
max_power_wh=inverter_config.max_power_w,
battery_id=inverter_config.battery_id,
ac_to_dc_efficiency=inverter_config.ac_to_dc_efficiency,
dc_to_ac_efficiency=inverter_config.dc_to_ac_efficiency,
max_ac_charge_power_w=inverter_config.max_ac_charge_power_w,
)
# Take first inverter
inverter_config = list(cls.config.devices.inverters.values())[0]
inverter_params = inverter_config.to_genetic0_param()
except Exception as e:
logger.info(
"No inverter device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
attempt,
e,
)
cls.config.devices.inverters = [
{
cls.config.devices.inverters = {
"inverter1": {
"device_id": "inverter1",
"max_power_w": 10000,
"battery_id": battery_config.device_id,
}
]
},
}
# Retry
continue
@@ -730,12 +725,12 @@ class Genetic0OptimizationParameters(
logger.info(
"No home appliance device data available - defaulting to demo data."
)
cls.config.devices.home_appliances = [
{
cls.config.devices.home_appliances = {
"dishwasher1": {
"device_id": "dishwasher1",
"consumption_wh": 2000,
"duration_h": 3.0,
"time_windows": {
"cycle_time_windows": {
"windows": [
{
"start_time": "08:00",
@@ -747,30 +742,26 @@ class Genetic0OptimizationParameters(
},
],
},
}
]
},
}
try:
home_appliance_config = cls.config.devices.home_appliances[0]
home_appliance_params = Genetic0HomeApplianceParameters(
device_id=home_appliance_config.device_id,
consumption_wh=home_appliance_config.consumption_wh,
duration_h=home_appliance_config.duration_h,
time_windows=home_appliance_config.time_windows,
)
# Take first appliance
home_appliance_config = list(cls.config.devices.home_appliances.values())[0]
home_appliance_params = home_appliance_config.to_genetic0_param()
except Exception as e:
logger.info(
"No home appliance device data available - defaulting to demo data. Parameter preparation attempt {}: {}",
attempt,
e,
)
cls.config.devices.home_appliances = [
{
cls.config.devices.home_appliances = {
"dishwasher1": {
"device_id": "dishwasher1",
"consumption_wh": 2000,
"duration_h": 3.0,
"time_windows": None,
}
]
"cycle_time_windows": None,
},
}
# Retry
continue
src/akkudoktoreos/optimization/genetic0/genetic0solution.py
+6 -6
View File
@@ -427,21 +427,21 @@ class Genetic0Solution(ConfigMixin, Genetic0ParametersBaseModel):
def _battery_device_id(self) -> str:
"""Get battery device id."""
try:
return self.config.devices.batteries[0].device_id
return list(self.config.devices.batteries.values())[0].device_id
except Exception:
return "battery1"
def _ev_device_id(self) -> str:
"""Get electric vehicle device id."""
try:
return self.config.devices.electric_vehicles[0].device_id
return self.config.devices.electric_vehicles.values()[0].device_id
except Exception:
return "ev1"
def _homeappliance_device_id(self) -> str:
"""Get home appliance device id."""
try:
return self.config.devices.home_appliances[0].device_id
return self.config.devices.home_appliances.values()[0].device_id
except Exception:
return "homeappliance1"
@@ -530,11 +530,11 @@ class Genetic0Solution(ConfigMixin, Genetic0ParametersBaseModel):
(the inverter curtails automatically, but this makes intent clear).
- Discharge: blocked when SOC is at or below min SOC.
"""
bat_list = self.config.devices.batteries
if not bat_list:
bat_dict = self.config.devices.batteries
if bat_dict is None or len(bat_dict) <= 0:
return ac_charge, dc_charge, discharge_allowed
bat = bat_list[0]
bat = list(bat_dict.values())[0]
min_soc = float(bat.min_soc_percentage)
max_soc = float(bat.max_soc_percentage)
capacity_wh = float(bat.capacity_wh)
src/akkudoktoreos/server/dash/itemscard.py
+3 -1
View File
@@ -445,9 +445,11 @@ def ConfigItemsCard(
description = config["description"]
item_model = resolve_item_model(hint)
if item_model is None:
raise ValueError(f"Hint for {config_name} needs item_model to be listed. Got {hint}")
item_path = hint.item_path # e.g. "pvforecast.planes"
if item_path is None:
raise ValueError(f"Hint needs item_path to be listed. Got {hint}")
raise ValueError(f"Hint for {config_name} needs item_path to be listed. Got {hint}")
path_parts = item_path.split(".") # e.g. ["pvforecast", "planes"]
items_list = json.loads(value) or []
src/akkudoktoreos/server/dash/mapcard.py
+3 -1
View File
@@ -453,9 +453,11 @@ def ConfigMapCard(
config_id = config_name.lower().replace(".", "-")
item_model = resolve_item_model(hint)
if item_model is None:
raise ValueError(f"Hint for {config_name} needs item_model to be listed. Got {hint}")
item_path = hint.item_path # e.g. "devices.batteries"
if item_path is None:
raise ValueError(f"Hint needs item_path to be mapped. Got {hint}")
raise ValueError(f"Hint for {config_name} needs item_path to be listed. Got {hint}")
path_parts = item_path.split(".") # e.g. ["devices", "batteries"]
items_map = json.loads(value) or {}
src/akkudoktoreos/server/dash/plan.py
+4 -3
View File
@@ -535,7 +535,8 @@ def InstructionCard(
config.devices
and config.devices.batteries
and any(
battery_config.device_id == resource_id for battery_config in config.devices.batteries
battery_config.device_id == resource_id
for battery_config in config.devices.batteries.values()
)
):
# This is a battery
@@ -548,7 +549,7 @@ def InstructionCard(
and config.devices.electric_vehicles
and any(
electric_vehicle_config.device_id == resource_id
for electric_vehicle_config in config.devices.electric_vehicles
for electric_vehicle_config in config.devices.electric_vehicles.values()
)
):
# This is a car battery
@@ -558,7 +559,7 @@ def InstructionCard(
and config.devices.home_appliances
and any(
home_appliance.device_id == resource_id
for home_appliance in config.devices.home_appliances
for home_appliance in config.devices.home_appliances.values()
)
):
# This is a home appliance
src/akkudoktoreos/server/dash/uihints.py
+28 -6
View File
@@ -207,16 +207,18 @@ UI_HINTS: dict[str, UiHint] = {
# ------------------------------------------------------------------
# Devices
# ------------------------------------------------------------------
# - batteries
"devices.batteries": UiHint(
form="items",
form="map_items",
item_path="devices.batteries",
),
"devices.electric_vehicles": UiHint(
form="items",
form="map_items",
item_path="devices.electric_vehicles",
),
# - home_appliances
"devices.home_appliances": UiHint(
form="items",
form="map_items",
item_path="devices.home_appliances",
),
# Sub-field hint for the time_windows field inside each appliance entry
@@ -224,6 +226,19 @@ UI_HINTS: dict[str, UiHint] = {
form="time_windows",
value_description="cycle index (0-based)",
),
# - inverters
"devices.inverters": UiHint(
form="map_items",
item_path="devices.inverters",
),
# Sub-field hint for the load_power_w_key field inside each inverter entry
"devices.inverters.load_power_w_key": UiHint(
form="select", options=["loadforecast_power_w", "null"]
),
# Sub-field hint for the pv_power_w_key field inside each inverter entry
"devices.inverters.pv_power_w_key": UiHint(
form="select", options=["pvforecast_dc_power_w", "null"]
),
# ------------------------------------------------------------------
# Electricity fee
# ------------------------------------------------------------------
@@ -362,26 +377,33 @@ def _ensure_item_models() -> None:
UI_HINTS["pvforecast.planes"].item_model = PVForecastPlaneSetting
if UI_HINTS["devices.batteries"].item_model is None:
from akkudoktoreos.devices.devices import (
from akkudoktoreos.devices.settings.batterysettings import (
BatteriesCommonSettings,
)
UI_HINTS["devices.batteries"].item_model = BatteriesCommonSettings
if UI_HINTS["devices.electric_vehicles"].item_model is None:
from akkudoktoreos.devices.devices import (
from akkudoktoreos.devices.settings.batterysettings import (
BatteriesCommonSettings,
)
UI_HINTS["devices.electric_vehicles"].item_model = BatteriesCommonSettings
if UI_HINTS["devices.home_appliances"].item_model is None:
from akkudoktoreos.devices.devices import (
from akkudoktoreos.devices.settings.homeappliancesettings import (
HomeApplianceCommonSettings,
)
UI_HINTS["devices.home_appliances"].item_model = HomeApplianceCommonSettings
if UI_HINTS["devices.inverters"].item_model is None:
from akkudoktoreos.devices.settings.invertersettings import (
InverterCommonSettings,
)
UI_HINTS["devices.inverters"].item_model = InverterCommonSettings
def resolve_item_model(hint: UiHint) -> Optional[Any]:
"""Return the ``item_model`` for an ``"items"`` hint, resolving lazily.