"""
Base Heat Generator Module
==========================
Abstract base classes for heat generation technologies.
:author: Dipl.-Ing. (FH) Jonas Pfeiffer
"""
import logging
from typing import Any
import numpy as np
from districtheatingsim.heat_generators.annuity import annuity
[docs]
class BaseHeatGenerator:
"""
Abstract base class for heat generators.
:param name: Unique identifier
:type name: str
.. note::
Derived classes must implement: calculate(), set_parameters(), add_optimization_parameters()
"""
[docs]
def __init__(self, name: str) -> None:
"""
Initialize the base heat generator.
:param name: Unique identifier for the heat generator instance
:type name: str
"""
self.name = name
[docs]
def annuity(self, *args, **kwargs) -> float:
"""
VDI 2067 compliant economic evaluation wrapper.
:param args: Positional arguments for annuity function
:param kwargs: Keyword arguments for annuity function
:return: Annual equivalent cost [€/year]
:rtype: float
.. note::
See annuity.py for complete parameter documentation.
"""
return annuity(*args, **kwargs)
[docs]
def generate(self, t: int, **kwargs) -> tuple:
"""
Generate heat for a single time step (used in storage-coupled dispatch).
:param t: Time step index
:type t: int
:param kwargs: Technology-specific context (remaining_load, VLT_L, RLT_L, etc.)
:return: (heat_output_kW, electricity_output_kW)
:rtype: tuple
:raises NotImplementedError: Must be implemented by derived classes that support STES coupling.
"""
raise NotImplementedError("generate() must be implemented for STES-coupled dispatch.")
[docs]
def calculate(self, economic_parameters: dict[str, Any], duration: float, load_profile, **kwargs) -> dict[str, Any]:
"""
Full-profile calculation including economic and environmental analysis (abstract).
:param economic_parameters: Economic parameters (electricity_price, gas_price, etc.)
:type economic_parameters: dict
:param duration: Time step duration [hours]
:type duration: float
:param load_profile: Remaining heat demand profile [kW]
:type load_profile: numpy.ndarray
:param kwargs: Technology-specific parameters
:return: Results dict with Wärmemenge, WGK, spec_co2_total, etc.
:rtype: dict
:raises NotImplementedError: Must be implemented by derived classes.
"""
raise NotImplementedError("The method 'calculate' must be implemented in the derived class.")
[docs]
def load_economic_parameters(self, economic_parameters: dict[str, Any]) -> None:
"""
Extract and store common economic parameters from the shared parameters dict.
:param economic_parameters: Dict with keys electricity_price, gas_price, wood_price,
capital_interest_rate, inflation_rate, time_period, subsidy_eligibility, hourly_rate.
:type economic_parameters: dict
.. note::
Call this at the start of calculate_heat_generation_cost() in each subclass
instead of repeating the same 7-8 assignment lines.
"""
self.Strompreis = economic_parameters["electricity_price"]
self.Gaspreis = economic_parameters["gas_price"]
self.Holzpreis = economic_parameters["wood_price"]
self.q = economic_parameters["capital_interest_rate"]
self.r = economic_parameters["inflation_rate"]
self.T = economic_parameters["time_period"]
self.BEW = economic_parameters["subsidy_eligibility"]
self.stundensatz = economic_parameters["hourly_rate"]
[docs]
def set_parameters(self, variables: list[float], variables_order: list[str], idx: int) -> None:
"""
Set optimization parameters for the heat generator (abstract).
:param variables: List of optimization variable values
:type variables: list of float
:param variables_order: Order and assignment of optimization variables
:type variables_order: list of str
:param idx: Technology index in the system list
:type idx: int
:raises NotImplementedError: Must be implemented by derived classes
"""
raise NotImplementedError("set_parameters must be implemented in the derived class.")
[docs]
def add_optimization_parameters(self, idx: int) -> dict[str, Any]:
"""
Define optimization variables and constraints for the technology (abstract).
:param idx: Technology index in the system list
:type idx: int
:return: Dict with 'variables', 'bounds', 'constraints' keys
:rtype: dict
:raises NotImplementedError: Must be implemented by derived classes
"""
raise NotImplementedError("add_optimization_parameters must be implemented in the derived class.")
[docs]
def update_parameters(self, optimized_values: list[float], variables_order: list[str]) -> None:
"""
Update technology parameters from optimization results.
:param optimized_values: Optimized values for all system variables
:type optimized_values: list of float
:param variables_order: Order of variables defining parameter assignment
:type variables_order: list of str
"""
# Extract technology index from name
idx = self.name.split("_")[-1]
# Filter variables belonging to this technology
relevant_vars = [var for var in variables_order if var.endswith(f"_{idx}")]
relevant_values = [
value for var, value in zip(variables_order, optimized_values, strict=False) if var in relevant_vars
]
if not relevant_vars:
logging.debug("No relevant variables found for %s.", self.name)
return
# Update parameters with optimized values
for var, value in zip(relevant_vars, relevant_values, strict=False):
# Extract parameter name without technology index
param_name = var.rsplit("_", 1)[0]
if param_name in self.__dict__:
setattr(self, param_name, value)
logging.debug("Set %s for %s to %s", param_name, self.name, value)
[docs]
def get_plot_data(self) -> dict[str, list | np.ndarray]:
"""
Extract time-series data for visualization.
:return: Dict mapping variable names to time-series arrays
:rtype: dict
"""
return {
var_name: getattr(self, var_name)
for var_name in self.__dict__
if isinstance(getattr(self, var_name), (list, np.ndarray))
}
[docs]
def to_dict(self) -> dict[str, Any]:
"""
Convert heat generator to dictionary for serialization.
:return: Dictionary representation excluding non-serializable attributes
:rtype: dict
.. note::
Numpy arrays are converted to lists for JSON compatibility.
"""
# Create copy of object dictionary
data = self.__dict__.copy()
# Store class name for reliable deserialization (avoids fragile prefix-matching)
data["tech_type"] = type(self).__name__
# Remove non-serializable attributes
data.pop("scene_item", None) # GUI elements
data.pop("buffer", None) # BufferStorage — contains ThermalStorage1D model;
# rebuilt from constructor params on deserialization
# Convert numpy arrays to lists for JSON compatibility
for key, value in data.items():
if isinstance(value, np.ndarray):
data[key] = value.tolist()
return data
[docs]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "BaseHeatGenerator":
"""
Create heat generator from dictionary representation.
:param data: Dictionary containing heat generator attributes
:type data: dict
:return: New heat generator object
:rtype: BaseHeatGenerator
"""
# Create new object without calling __init__
obj = cls.__new__(cls)
# Update object dictionary with provided data
obj.__dict__.update(data)
# Convert lists back to numpy arrays for array attributes
for key, value in obj.__dict__.items():
if isinstance(value, list) and key.endswith(("_array", "_data", "_profile")):
setattr(obj, key, np.array(value))
# Restore strategy: JSON round-trip leaves it as a plain dict.
# Reconstruct the original strategy subclass so the correct
# decide_operation() override is preserved.
if isinstance(getattr(obj, "strategy", None), dict):
obj.strategy = BaseStrategy.from_dict(obj.strategy)
# Subclasses register automatically via __init_subclass__ when their
# modules are imported — all strategy subclasses are available by
# the time from_dict() is called during deserialization.
return obj
def __deepcopy__(self, memo: dict[int, Any]) -> "BaseHeatGenerator":
"""
Create deep copy of heat generator.
:param memo: Memoization dict for deepcopy operation
:type memo: dict
:return: Deep copy with independent memory allocation
:rtype: BaseHeatGenerator
"""
return self.from_dict(self.to_dict())
[docs]
class BaseStrategy:
"""
Base control strategy with hysteresis logic.
:param charge_on: Temperature threshold for activation [°C]
:type charge_on: float
:param charge_off: Temperature threshold for deactivation [°C]
:type charge_off: float
"""
# Auto-populated registry of all subclasses — used for deserialization.
_registry: dict = {}
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
BaseStrategy._registry[cls.__name__] = cls
[docs]
def __init__(self, charge_on: float, charge_off: float) -> None:
"""
Initialize control strategy with temperature thresholds.
:param charge_on: Storage temperature threshold for activation [°C]
:type charge_on: float
:param charge_off: Storage temperature threshold for deactivation [°C]
:type charge_off: float
"""
self.charge_on = charge_on
self.charge_off = charge_off
[docs]
def decide_operation(
self, current_state: bool, upper_storage_temp: float, lower_storage_temp: float, remaining_demand: float
) -> bool:
"""
Decide heat generator operation based on storage conditions and demand.
:param current_state: Current operational state of the heat generator
:type current_state: bool
:param upper_storage_temp: Upper storage layer temperature [°C]
:type upper_storage_temp: float
:param lower_storage_temp: Lower storage layer temperature [°C]
:type lower_storage_temp: float
:param remaining_demand: Remaining heat demand [kW]
:type remaining_demand: float
:return: True to operate, False to stop
:rtype: bool
.. note::
ON: Continue if lower_temp < charge_off AND demand > 0.
OFF: Start if upper_temp ≤ charge_on AND demand > 0.
"""
# Check current operational state and apply hysteresis logic
if current_state:
# Generator is currently operating
charge_off = getattr(self, "charge_off", None)
if (charge_off is None or lower_storage_temp < charge_off) and remaining_demand > 0:
return True # Continue operation
else:
return False # Stop operation (overheating protection or no demand)
else:
# Generator is currently stopped
if upper_storage_temp <= self.charge_on and remaining_demand > 0:
return True # Start operation (low storage temp and demand present)
else:
return False # Remain stopped (sufficient storage temp or no demand)
[docs]
def to_dict(self) -> dict[str, Any]:
"""
Convert strategy to dictionary for serialization.
:return: Dictionary representation of the strategy
:rtype: dict
"""
data = self.__dict__.copy()
data["_strategy_class"] = type(self).__name__
return data
[docs]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "BaseStrategy":
"""
Create strategy from dictionary representation.
Looks up the concrete subclass via the ``_strategy_class`` key written by
``to_dict()`` so the correct ``decide_operation()`` override is preserved.
Falls back to *cls* (typically ``BaseStrategy``) for legacy JSON files that
do not contain the key.
:param data: Dictionary containing strategy attributes
:type data: dict
:return: New strategy object
:rtype: BaseStrategy
"""
data = dict(data) # copy so we don't mutate the caller's dict
strategy_class_name = data.pop("_strategy_class", None)
target_cls = BaseStrategy._registry.get(strategy_class_name, cls)
obj = target_cls.__new__(target_cls)
obj.__dict__.update(data)
return obj
def __deepcopy__(self, memo: dict[int, Any]) -> "BaseStrategy":
"""
Create deep copy of strategy.
:param memo: Memoization dict for deepcopy operation
:type memo: dict
:return: Independent copy of strategy
:rtype: BaseStrategy
"""
return self.from_dict(self.to_dict())