"""
Pandapipes time series simulation for district heating networks including controller updates,
temperature control, and result processing.
:author: Dipl.-Ing. (FH) Jonas Pfeiffer
"""
from typing import Any
import numpy as np
import pandas as pd
from pandapipes.timeseries import run_time_series
from pandapower.control.controller.const_control import ConstControl
from pandapower.timeseries import DFData, OutputWriter
from districtheatingsim.constants import CP_WATER_KJ_KGK, KELVIN_OFFSET
from districtheatingsim.net_simulation_pandapipes.controllers import MinimumSupplyTemperatureController
from districtheatingsim.net_simulation_pandapipes.result_validation import (
validate_design_state,
validate_simulation_results,
)
from districtheatingsim.net_simulation_pandapipes.utilities import COP_WP
from districtheatingsim.utilities.test_reference_year import import_TRY
[docs]
def update_heat_consumer_qext_controller(
net, qext_w_profiles: list[np.ndarray], time_steps: range, start: int, end: int
) -> None:
"""
Update external heat demand controllers with time-dependent profiles.
:param net: Pandapipes network with controllers
:type net: pandapipes.pandapipesNet
:param qext_w_profiles: List of heat demand profiles for each consumer [W]
:type qext_w_profiles: List[np.ndarray]
:param time_steps: Time steps for simulation period
:type time_steps: range
:param start: Start index for slicing profiles
:type start: int
:param end: End index for slicing profiles
:type end: int
"""
for i, qext_w_profile in enumerate(qext_w_profiles):
df = pd.DataFrame(index=time_steps, data={f"qext_w_{i}": qext_w_profile[start:end]})
data_source = DFData(df)
for ctrl in net.controller.object.values:
if isinstance(ctrl, ConstControl) and ctrl.element_index == i and ctrl.variable == "qext_w":
ctrl.data_source = data_source
[docs]
def update_heat_consumer_temperature_controller(
net, min_supply_temperature_heat_consumer: np.ndarray | list, time_steps: range, start: int, end: int
) -> None:
"""
Update minimum supply temperature controllers with static or time-dependent profiles.
:param net: Pandapipes network object
:type net: pandapipes.pandapipesNet
:param min_supply_temperature_heat_consumer: Min supply temperature profiles [°C]
:type min_supply_temperature_heat_consumer: Union[np.ndarray, List]
:param time_steps: Time steps range
:type time_steps: range
:param start: Start index for slicing
:type start: int
:param end: End index for slicing
:type end: int
.. note::
Handles both scalar (static) and array (dynamic) temperature requirements.
"""
controller_count = 0
for ctrl in net.controller.object.values:
if isinstance(ctrl, MinimumSupplyTemperatureController):
profile = min_supply_temperature_heat_consumer[controller_count]
# Check if static or time-dependent
if np.isscalar(profile) or (isinstance(profile, np.ndarray) and profile.ndim == 0):
# Single value: repeat for all time steps
values = np.full(len(time_steps), profile)
elif isinstance(profile, np.ndarray) and profile.ndim == 1 and len(profile) == 1:
values = np.full(len(time_steps), profile[0])
else:
# Time series: take correct slice
values = profile[start:end]
df_return_temp = pd.DataFrame(index=time_steps, data={"min_supply_temperature": values})
data_source_return_temp = DFData(df_return_temp)
ctrl.data_source = data_source_return_temp
controller_count += 1
[docs]
def update_heat_consumer_return_temperature_controller(
net, return_temperature_heat_consumer: np.ndarray | list, time_steps: range, start: int, end: int
) -> None:
"""
Update return temperature controllers with static or dynamic profiles.
:param net: Pandapipes network object
:type net: pandapipes.pandapipesNet
:param return_temperature_heat_consumer: Return temperature profiles [°C]
:type return_temperature_heat_consumer: Union[np.ndarray, List]
:param time_steps: Time steps range
:type time_steps: range
:param start: Start index for slicing
:type start: int
:param end: End index for slicing
:type end: int
.. note::
Automatically converts temperatures from °C to K (adds KELVIN_OFFSET).
"""
for i, return_temp_profile in enumerate(return_temperature_heat_consumer):
# Check if static or time-dependent
if np.isscalar(return_temp_profile) or (
isinstance(return_temp_profile, np.ndarray) and return_temp_profile.ndim == 0
):
# Single value: repeat for all time steps
values = np.full(len(time_steps), return_temp_profile + KELVIN_OFFSET)
elif (
isinstance(return_temp_profile, np.ndarray)
and return_temp_profile.ndim == 1
and len(return_temp_profile) == 1
):
values = np.full(len(time_steps), return_temp_profile[0] + KELVIN_OFFSET)
else:
# Time series: take correct slice and convert
values = return_temp_profile[start:end] + KELVIN_OFFSET
# Build DataFrame for all time steps
df_return_temp = pd.DataFrame(index=time_steps, data={f"treturn_k_{i}": values})
data_source_return_temp = DFData(df_return_temp)
for ctrl in net.controller.object.values:
if (
isinstance(ctrl, ConstControl)
and ctrl.element == "heat_consumer"
and ctrl.element_index == i
and ctrl.variable == "treturn_k"
):
# Update the data source of the existing ConstControl
ctrl.data_source = data_source_return_temp
[docs]
def update_secondary_producer_controller(
net, secondary_producers: list[Any], time_steps: range, start: int, end: int
) -> None:
"""
Update secondary producer mass flow controllers for time series simulation.
:param net: Pandapipes network object
:type net: pandapipes.pandapipesNet
:param secondary_producers: List of producer configs with index and mass_flow
:type secondary_producers: List[Any]
:param time_steps: Time steps range
:type time_steps: range
:param start: Start index for slicing
:type start: int
:param end: End index for slicing
:type end: int
.. note::
Updates both circ_pump_mass and flow_control controllers for each producer.
"""
for producer in secondary_producers:
producer_index = producer.index if hasattr(producer, "index") else 0
mass_flow_data = producer.mass_flow if hasattr(producer, "mass_flow") else np.zeros(len(time_steps))
# Stelle sicher, dass mass_flow_data die richtige Form hat
if np.isscalar(mass_flow_data):
mass_flow_slice = np.full(len(time_steps), mass_flow_data)
elif isinstance(mass_flow_data, np.ndarray):
mass_flow_slice = mass_flow_data[start:end]
else:
mass_flow_slice = np.full(len(time_steps), mass_flow_data)
print(f"Mass flow for secondary producer {producer_index}: {mass_flow_slice}")
df_secondary_producer = pd.DataFrame(
index=time_steps, data={f"mdot_flow_kg_per_s_{producer_index}": mass_flow_slice}
)
data_source_secondary_producer = DFData(df_secondary_producer)
df_secondary_producer_flow_control = pd.DataFrame(
index=time_steps, data={f"controlled_mdot_kg_per_s_{producer_index}": mass_flow_slice}
)
data_source_secondary_producer_flow_control = DFData(df_secondary_producer_flow_control)
for ctrl in net.controller.object.values:
if (
isinstance(ctrl, ConstControl)
and ctrl.element == "circ_pump_mass"
and ctrl.variable == "mdot_flow_kg_per_s"
):
ctrl.data_source = data_source_secondary_producer
elif (
isinstance(ctrl, ConstControl)
and ctrl.element == "flow_control"
and ctrl.variable == "controlled_mdot_kg_per_s"
):
ctrl.data_source = data_source_secondary_producer_flow_control
[docs]
def update_heat_generator_supply_temperature_controller(
net, supply_temperature: np.ndarray, time_steps: range, start: int, end: int
) -> None:
"""
Update supply temperature controllers for heat generators with time-dependent profiles.
:param net: Pandapipes network object
:type net: pandapipes.pandapipesNet
:param supply_temperature: Supply temperature profile [°C]
:type supply_temperature: np.ndarray
:param time_steps: Time steps range
:type time_steps: range
:param start: Start index for slicing
:type start: int
:param end: End index for slicing
:type end: int
.. note::
Converts °C to K, updates both circ_pump_pressure and circ_pump_mass controllers.
"""
if np.isscalar(supply_temperature):
# If a single value is provided, repeat it for all time steps
supply_temperature = np.full(len(time_steps), supply_temperature)
# Create the DataFrame for the supply temperature
df_supply_temp = pd.DataFrame(
index=time_steps, data={"supply_temperature": supply_temperature[start:end] + KELVIN_OFFSET}
)
data_source_supply_temp = DFData(df_supply_temp)
for ctrl in net.controller.object.values:
if isinstance(ctrl, ConstControl) and ctrl.element == "circ_pump_pressure" and ctrl.variable == "t_flow_k":
ctrl.data_source = data_source_supply_temp
elif isinstance(ctrl, ConstControl) and ctrl.element == "circ_pump_mass" and ctrl.variable == "t_flow_k":
ctrl.data_source = data_source_supply_temp
[docs]
def create_log_variables(net) -> list[tuple[str, str]]:
"""
Create list of variables to log during time series simulation.
:param net: Pandapipes network to analyze for available components
:type net: pandapipes.pandapipesNet
:return: List of (table_name, variable_name) tuples to log
:rtype: List[Tuple[str, str]]
.. note::
Logs junction pressures/temperatures, heat consumer data, and conditionally circ_pump_mass.
"""
log_variables = [
("res_junction", "p_bar"),
("res_junction", "t_k"),
("heat_consumer", "qext_w"),
("res_heat_consumer", "vdot_m3_per_s"),
("res_heat_consumer", "t_from_k"),
("res_heat_consumer", "t_to_k"),
("res_heat_consumer", "mdot_from_kg_per_s"),
("res_circ_pump_pressure", "mdot_from_kg_per_s"),
("res_circ_pump_pressure", "p_to_bar"),
("res_circ_pump_pressure", "p_from_bar"),
("res_circ_pump_pressure", "t_to_k"),
("res_circ_pump_pressure", "t_from_k"),
]
if "circ_pump_mass" in net:
log_variables.extend(
[
("res_circ_pump_mass", "mdot_from_kg_per_s"),
("res_circ_pump_mass", "p_to_bar"),
("res_circ_pump_mass", "p_from_bar"),
("res_circ_pump_mass", "t_to_k"),
("res_circ_pump_mass", "t_from_k"),
]
)
return log_variables
[docs]
def time_series_preprocessing(NetworkGenerationData) -> Any:
"""
Preprocess time series data including temperature control and COP calculations.
:param NetworkGenerationData: Network data with simulation parameters and profiles
:type NetworkGenerationData: object
:return: Updated NetworkGenerationData with preprocessed time series
:rtype: Any
.. note::
Implements static/sliding temperature control, COP calculations for cold networks,
applies 2% minimum load, calculates secondary producer mass flows. Converts W to kW.
"""
print(f"Maximale Vorlauftemperatur Netz: {NetworkGenerationData.max_supply_temperature_heat_generator} °C")
print(f"Mindestvorlauftemperatur HAST: {NetworkGenerationData.min_supply_temperature_heat_consumer} °C")
print(f"Rücklauftemperatur HAST: {NetworkGenerationData.return_temperature_heat_consumer} °C")
print(f"Vorlauftemperatur Gebäude: {NetworkGenerationData.supply_temperature_buildings} °C")
print(f"Rücklauftemperatur Gebäude: {NetworkGenerationData.return_temperature_buildings} °C")
print(f"building_temperature_checked: {NetworkGenerationData.building_temperature_checked}")
print(f"Netconfiguration: {NetworkGenerationData.netconfiguration}")
COP_file_values = np.genfromtxt(NetworkGenerationData.COP_filename, delimiter=";")
# Supply temperature control strategy implementation
if NetworkGenerationData.supply_temperature_control == "Statisch":
NetworkGenerationData.supply_temperature_heat_generator = (
NetworkGenerationData.max_supply_temperature_heat_generator
) # Type: float in °C
if NetworkGenerationData.supply_temperature_control == "Gleitend":
air_temperature_data, _, _, _, _ = import_TRY(NetworkGenerationData.TRY_filename)
# Calculate slope of linear equation for sliding control
slope = (
NetworkGenerationData.max_supply_temperature_heat_generator
- NetworkGenerationData.min_supply_temperature_heat_generator
) / (
NetworkGenerationData.min_air_temperature_heat_generator
- NetworkGenerationData.max_air_temperature_heat_generator
)
air_temperature_data = np.array(air_temperature_data)
NetworkGenerationData.supply_temperature_heat_generator = np.where(
air_temperature_data <= NetworkGenerationData.min_air_temperature_heat_generator,
NetworkGenerationData.max_supply_temperature_heat_generator,
np.where(
air_temperature_data >= NetworkGenerationData.max_air_temperature_heat_generator,
NetworkGenerationData.min_supply_temperature_heat_generator,
NetworkGenerationData.max_supply_temperature_heat_generator
+ slope * (air_temperature_data - NetworkGenerationData.min_air_temperature_heat_generator),
),
)
print(f"Vorlauftemperatur Netz: {NetworkGenerationData.supply_temperature_heat_generator} °C")
# Temperature processing based on network configuration
### if building_temperature_checked is True, the time dependent building temperatures are used
### if netconfiguration is not "kaltes Netz", no changes are made to the heat demand and no power consumption is calculated
if NetworkGenerationData.building_temperature_checked and NetworkGenerationData.netconfiguration != "kaltes Netz":
NetworkGenerationData.min_supply_temperature_heat_consumer = (
NetworkGenerationData.supply_temperature_buildings_curve + NetworkGenerationData.dT_RL
)
NetworkGenerationData.return_temperature_heat_consumer = (
NetworkGenerationData.return_temperature_buildings_curve + NetworkGenerationData.dT_RL
)
### if building_temperature_checked is True, the time dependent building temperatures are used
### if netconfiguration is "kaltes Netz", the heat demand and power consumption are calculated using the COP calculation
elif NetworkGenerationData.building_temperature_checked and NetworkGenerationData.netconfiguration == "kaltes Netz":
NetworkGenerationData.min_supply_temperature_heat_consumer = (
NetworkGenerationData.return_temperature_heat_consumer + NetworkGenerationData.dT_RL
)
NetworkGenerationData.return_temperature_heat_consumer = (
NetworkGenerationData.return_temperature_buildings_curve + NetworkGenerationData.dT_RL
)
# Calculate COP, electricity consumption, and adjusted heat demand
cop, _ = COP_WP(
NetworkGenerationData.supply_temperature_buildings_curve,
NetworkGenerationData.return_temperature_heat_consumer,
COP_file_values,
)
strom_wp = NetworkGenerationData.waerme_hast_ges_W / cop
waerme_hast = NetworkGenerationData.waerme_hast_ges_W - strom_wp
NetworkGenerationData.waerme_hast_ges_W = waerme_hast
NetworkGenerationData.strombedarf_hast_ges_W = strom_wp
### if building_temperature_checked is False, the time dependent building temperatures are not used
### if netconfiguration is not "kaltes Netz", the heat demand and power consumption are calculated using the COP calculation
elif (
not NetworkGenerationData.building_temperature_checked
and NetworkGenerationData.netconfiguration == "kaltes Netz"
):
cop, _ = COP_WP(
NetworkGenerationData.supply_temperature_buildings,
NetworkGenerationData.return_temperature_heat_consumer,
COP_file_values,
)
strom_wp = NetworkGenerationData.waerme_hast_ges_W / cop
waerme_hast = NetworkGenerationData.waerme_hast_ges_W - strom_wp
NetworkGenerationData.waerme_hast_ges_W = waerme_hast
NetworkGenerationData.strombedarf_hast_ges_W = strom_wp
# Apply minimum load constraints (2% of maximum)
max_heat = np.max(NetworkGenerationData.waerme_hast_ges_W)
max_power = np.max(NetworkGenerationData.strombedarf_hast_ges_W)
NetworkGenerationData.waerme_hast_ges_W = np.where(
NetworkGenerationData.waerme_hast_ges_W < 0.02 * max_heat,
0.02 * max_heat,
NetworkGenerationData.waerme_hast_ges_W,
)
NetworkGenerationData.strombedarf_hast_ges_W = np.where(
NetworkGenerationData.waerme_hast_ges_W < 0.02 * max_heat,
0.02 * max_power,
NetworkGenerationData.strombedarf_hast_ges_W,
)
# Convert power values from W to kW
NetworkGenerationData.waerme_hast_ges_kW = np.where(
NetworkGenerationData.waerme_hast_ges_W == 0, 0, NetworkGenerationData.waerme_hast_ges_W / 1000
)
NetworkGenerationData.strombedarf_hast_ges_kW = np.where(
NetworkGenerationData.strombedarf_hast_ges_W == 0, 0, NetworkGenerationData.strombedarf_hast_ges_W / 1000
)
# Calculate total heat and electricity demand profiles
NetworkGenerationData.waerme_ges_kW = np.sum(NetworkGenerationData.waerme_hast_ges_kW, axis=0)
NetworkGenerationData.strombedarf_ges_kW = np.sum(NetworkGenerationData.strombedarf_hast_ges_kW, axis=0)
# Calculate mass flow for secondary producers
if NetworkGenerationData.secondary_producers:
# Calculate mass flow using the central water heat capacity
cp = CP_WATER_KJ_KGK # kJ/kgK
avg_return_temperature = np.mean(NetworkGenerationData.return_temperature_heat_consumer)
mass_flow = NetworkGenerationData.waerme_ges_kW / (
cp * (NetworkGenerationData.supply_temperature_heat_generator - avg_return_temperature)
) # kW / (kJ/kgK * K) = kg/s
print(f"Mass flow of main producer: {mass_flow} kg/s")
# Update each secondary producer's dictionary with calculated mass flow
for secondary_producer in NetworkGenerationData.secondary_producers:
secondary_producer.mass_flow = secondary_producer.load_percentage / 100 * mass_flow
print(f"Mass flow of secondary producer {secondary_producer.index}: {secondary_producer.mass_flow} kg/s")
return NetworkGenerationData
[docs]
def thermohydraulic_time_series_net(NetworkGenerationData) -> Any:
"""
Run thermohydraulic time series simulation with controller updates.
:param NetworkGenerationData: Network data with preprocessed model and parameters
:type NetworkGenerationData: object
:return: Updated NetworkGenerationData with simulation results and pump operations
:rtype: Any
.. note::
Runs bidirectional simulation with iter=100, alpha=0.5. Updates all controllers
(heat demand, temperatures, secondary producers). Logs junction, heat consumer,
and pump data.
"""
# Update the ConstControl
time_steps = range(
0,
len(
NetworkGenerationData.waerme_hast_ges_W[0][
NetworkGenerationData.start_time_step : NetworkGenerationData.end_time_step
]
),
)
update_heat_consumer_qext_controller(
NetworkGenerationData.net,
NetworkGenerationData.waerme_hast_ges_W,
time_steps,
NetworkGenerationData.start_time_step,
NetworkGenerationData.end_time_step,
)
# Update secondary producer controls
if NetworkGenerationData.secondary_producers:
update_secondary_producer_controller(
NetworkGenerationData.net,
NetworkGenerationData.secondary_producers,
time_steps,
NetworkGenerationData.start_time_step,
NetworkGenerationData.end_time_step,
)
# Update temperature controllers if applicable
if (
NetworkGenerationData.min_supply_temperature_heat_consumer is not None
and np.any(np.array(NetworkGenerationData.min_supply_temperature_heat_consumer) != 0)
and isinstance(NetworkGenerationData.min_supply_temperature_heat_consumer, np.ndarray)
):
print("Update TemperatureController")
update_heat_consumer_temperature_controller(
NetworkGenerationData.net,
NetworkGenerationData.min_supply_temperature_heat_consumer,
time_steps,
NetworkGenerationData.start_time_step,
NetworkGenerationData.end_time_step,
)
if NetworkGenerationData.return_temperature_heat_consumer is not None and isinstance(
NetworkGenerationData.return_temperature_heat_consumer, np.ndarray
):
print("Update Return Temperature Const Control")
update_heat_consumer_return_temperature_controller(
NetworkGenerationData.net,
NetworkGenerationData.return_temperature_heat_consumer,
time_steps,
NetworkGenerationData.start_time_step,
NetworkGenerationData.end_time_step,
)
# Always update supply temperature controller so its DFData covers all time_steps.
# update_heat_generator_supply_temperature_controller handles both scalars and arrays.
if NetworkGenerationData.supply_temperature_heat_generator is not None:
update_heat_generator_supply_temperature_controller(
NetworkGenerationData.net,
NetworkGenerationData.supply_temperature_heat_generator,
time_steps,
NetworkGenerationData.start_time_step,
NetworkGenerationData.end_time_step,
)
# Configure logging and run simulation
log_variables = create_log_variables(NetworkGenerationData.net)
ow = OutputWriter(NetworkGenerationData.net, time_steps, output_path=None, log_variables=log_variables)
try:
run_time_series.run_timeseries(NetworkGenerationData.net, time_steps, mode="bidirectional", iter=100, alpha=0.5)
except Exception as e:
raise RuntimeError(f"Thermohydraulic time-series simulation failed (bidirectional, iter=100): {e}") from e
NetworkGenerationData.net_results = ow.np_results
# Fail loudly on a non-converged / infeasible run instead of letting NaN/inf
# propagate into the heat and temperature post-processing (BACKLOG C2).
validate_simulation_results(NetworkGenerationData.net_results, context="thermohydraulic time series")
NetworkGenerationData.pump_results = calculate_results(NetworkGenerationData.net, NetworkGenerationData.net_results)
return NetworkGenerationData
[docs]
def simplified_time_series_net(NetworkGenerationData) -> Any:
"""
Run simplified time series by scaling design state with building heat demand.
:param NetworkGenerationData: Network data with design state from initialization
:type NetworkGenerationData: object
:return: Updated NetworkGenerationData with scaled load profiles
:rtype: Any
.. note::
No pipeflow calculation. Uses design state from net.res_*, scales with demand.
Constant temperatures/pressures from design. Losses scaled proportionally.
Much faster than thermohydraulic_time_series_net.
"""
print("Starte vereinfachte Zeitreihenberechnung (basierend auf Auslegung)...")
# Get time steps for selected simulation range
time_steps = range(
0,
len(
NetworkGenerationData.waerme_hast_ges_W[0][
NetworkGenerationData.start_time_step : NetworkGenerationData.end_time_step
]
),
)
n_steps = len(time_steps)
# Find design point (maximum load)
max_load_idx = np.argmax(NetworkGenerationData.waerme_ges_kW)
total_building_demand_design = NetworkGenerationData.waerme_ges_kW[max_load_idx]
print(f"Nutze Auslegungszustand bei max. Last: {total_building_demand_design:.1f} kW")
# Extract design state results from already calculated network
# (these were calculated during initialization)
design_results = {"Heizentrale Haupteinspeisung": {}, "weitere Einspeisung": {}}
# Get design state from pressure pumps (main generators)
if (
hasattr(NetworkGenerationData.net, "res_circ_pump_pressure")
and len(NetworkGenerationData.net.res_circ_pump_pressure) > 0
):
for idx in NetworkGenerationData.net.res_circ_pump_pressure.index:
res = NetworkGenerationData.net.res_circ_pump_pressure.loc[idx]
design_results["Heizentrale Haupteinspeisung"][idx] = {
"mass_flow_design": res["mdot_from_kg_per_s"],
"flow_pressure_design": res["p_to_bar"],
"return_pressure_design": res["p_from_bar"],
"deltap_design": res["p_to_bar"] - res["p_from_bar"],
"return_temp_design": res["t_from_k"] - KELVIN_OFFSET,
"flow_temp_design": res["t_to_k"] - KELVIN_OFFSET,
"qext_kW_design": res["mdot_from_kg_per_s"] * CP_WATER_KJ_KGK * (res["t_to_k"] - res["t_from_k"]),
}
print(
f" Haupteinspeisung {idx}: {design_results['Heizentrale Haupteinspeisung'][idx]['qext_kW_design']:.1f} kW Auslegungsleistung"
)
# Get design state from mass pumps (secondary producers)
if (
hasattr(NetworkGenerationData.net, "res_circ_pump_mass")
and len(NetworkGenerationData.net.res_circ_pump_mass) > 0
):
for idx in NetworkGenerationData.net.res_circ_pump_mass.index:
res = NetworkGenerationData.net.res_circ_pump_mass.loc[idx]
design_results["weitere Einspeisung"][idx] = {
"mass_flow_design": res["mdot_from_kg_per_s"],
"flow_pressure_design": res["p_to_bar"],
"return_pressure_design": res["p_from_bar"],
"deltap_design": res["p_to_bar"] - res["p_from_bar"],
"return_temp_design": res["t_from_k"] - KELVIN_OFFSET,
"flow_temp_design": res["t_to_k"] - KELVIN_OFFSET,
"qext_kW_design": res["mdot_from_kg_per_s"] * CP_WATER_KJ_KGK * (res["t_to_k"] - res["t_from_k"]),
}
print(
f" Weitere Einspeisung {idx}: {design_results['weitere Einspeisung'][idx]['qext_kW_design']:.1f} kW Auslegungsleistung"
)
# Fail loudly if the design state is NaN/inf (init pipeflow did not converge)
# rather than scaling NaN across every time step (BACKLOG C2).
validate_design_state(design_results, context="simplified time series")
# Calculate design losses (difference between generated and consumed heat)
total_generation_design = sum(
[data["qext_kW_design"] for pump_type in design_results.values() for data in pump_type.values()]
)
design_losses_kW = total_generation_design - total_building_demand_design
design_loss_factor = design_losses_kW / total_building_demand_design if total_building_demand_design > 0 else 0
print(f"Auslegungsverluste: {design_losses_kW:.1f} kW ({design_loss_factor * 100:.2f}%)")
# Create time series by scaling with building demand
NetworkGenerationData.pump_results = {"Heizentrale Haupteinspeisung": {}, "weitere Einspeisung": {}}
# Get building demand series for the selected time range
building_demand_series = NetworkGenerationData.waerme_ges_kW[
NetworkGenerationData.start_time_step : NetworkGenerationData.end_time_step
]
# Handle supply temperature (static or sliding)
if isinstance(NetworkGenerationData.supply_temperature_heat_generator, np.ndarray):
# Gleitende Vorlauftemperatur
supply_temp_series = NetworkGenerationData.supply_temperature_heat_generator[
NetworkGenerationData.start_time_step : NetworkGenerationData.end_time_step
]
print("Verwende gleitende Vorlauftemperatur")
else:
# Statische Vorlauftemperatur
supply_temp_series = np.full(n_steps, NetworkGenerationData.supply_temperature_heat_generator)
print(f"Verwende statische Vorlauftemperatur: {NetworkGenerationData.supply_temperature_heat_generator:.1f} °C")
# Scale results for each time step
for pump_type, pumps in design_results.items():
for idx, design_data in pumps.items():
# Calculate heat generation: building demand + constant losses
# Losses stay approximately constant (independent of load)
share_of_generation = (
design_data["qext_kW_design"] / total_generation_design if total_generation_design > 0 else 1.0
)
producer_losses = design_losses_kW * share_of_generation
qext_series = building_demand_series * share_of_generation + producer_losses
# Calculate mass flow from heat and temperature difference
# q = m * cp * dT => m = q / (cp * dT)
cp = CP_WATER_KJ_KGK # kJ/kgK
delta_T = supply_temp_series - design_data["return_temp_design"]
mass_flow_series = qext_series / (cp * delta_T)
# Temperature handling
if isinstance(NetworkGenerationData.supply_temperature_heat_generator, np.ndarray):
# Gleitende Vorlauftemperatur - verwende zeitabhängige Werte
flow_temp_series = supply_temp_series
else:
# Statische Vorlauftemperatur
flow_temp_series = np.full(n_steps, design_data["flow_temp_design"])
# Store results
NetworkGenerationData.pump_results[pump_type][idx] = {
"mass_flow": mass_flow_series,
"flow_pressure": np.full(n_steps, design_data["flow_pressure_design"]),
"return_pressure": np.full(n_steps, design_data["return_pressure_design"]),
"deltap": np.full(n_steps, design_data["deltap_design"]),
"return_temp": np.full(n_steps, design_data["return_temp_design"]),
"flow_temp": flow_temp_series,
"qext_kW": qext_series,
}
print(f"Vereinfachte Berechnung erfolgreich abgeschlossen ({n_steps} Zeitschritte).")
return NetworkGenerationData
[docs]
def calculate_results(
net, net_results: dict, cp_kJ_kgK: float = CP_WATER_KJ_KGK
) -> dict[str, dict[int, dict[str, np.ndarray]]]:
"""
Process and structure raw simulation results from pandapipes.
:param net: Pandapipes network with component definitions
:type net: pandapipes.pandapipesNet
:param net_results: Raw results dictionary from time series simulation
:type net_results: Dict
:param cp_kJ_kgK: Specific heat capacity of water [kJ/kg·K], defaults to CP_WATER_KJ_KGK (4.18)
:type cp_kJ_kgK: float
:return: Structured results dict: {producer_type: {index: {parameter: time_series}}}
:rtype: Dict[str, Dict[int, Dict[str, np.ndarray]]]
.. note::
Converts K→°C, calculates heat from mass flow and ΔT. Parameters: mass_flow,
flow_pressure, return_pressure, deltap, return_temp, flow_temp, qext_kW.
Handles circ_pump_pressure (main) and circ_pump_mass (secondary).
"""
# Prepare data structure
pump_results = {"Heizentrale Haupteinspeisung": {}, "weitere Einspeisung": {}}
# Add results for the Pressure Pump (main heat generator)
if "circ_pump_pressure" in net:
for idx, _row in net.circ_pump_pressure.iterrows():
pump_results["Heizentrale Haupteinspeisung"][idx] = {
"mass_flow": net_results["res_circ_pump_pressure.mdot_from_kg_per_s"][:, 0],
"flow_pressure": net_results["res_circ_pump_pressure.p_to_bar"][:, idx],
"return_pressure": net_results["res_circ_pump_pressure.p_from_bar"][:, idx],
"deltap": net_results["res_circ_pump_pressure.p_to_bar"][:, idx]
- net_results["res_circ_pump_pressure.p_from_bar"][:, idx],
"return_temp": net_results["res_circ_pump_pressure.t_from_k"][:, idx] - KELVIN_OFFSET,
"flow_temp": net_results["res_circ_pump_pressure.t_to_k"][:, idx] - KELVIN_OFFSET,
"qext_kW": net_results["res_circ_pump_pressure.mdot_from_kg_per_s"][:, idx]
* cp_kJ_kgK
* (
net_results["res_circ_pump_pressure.t_to_k"][:, idx]
- net_results["res_circ_pump_pressure.t_from_k"][:, idx]
),
}
# Add results for the Mass Pumps (secondary producers)
if "circ_pump_mass" in net:
for idx, _row in net.circ_pump_mass.iterrows():
pump_results["weitere Einspeisung"][idx] = {
"mass_flow": net_results["res_circ_pump_mass.mdot_from_kg_per_s"][:, idx],
"flow_pressure": net_results["res_circ_pump_mass.p_to_bar"][:, idx],
"return_pressure": net_results["res_circ_pump_mass.p_from_bar"][:, idx],
"deltap": net_results["res_circ_pump_mass.p_to_bar"][:, idx]
- net_results["res_circ_pump_mass.p_from_bar"][:, idx],
"return_temp": net_results["res_circ_pump_mass.t_from_k"][:, idx] - KELVIN_OFFSET,
"flow_temp": net_results["res_circ_pump_mass.t_to_k"][:, idx] - KELVIN_OFFSET,
"qext_kW": net_results["res_circ_pump_mass.mdot_from_kg_per_s"][:, idx]
* cp_kJ_kgK
* (
net_results["res_circ_pump_mass.t_to_k"][:, idx]
- net_results["res_circ_pump_mass.t_from_k"][:, idx]
),
}
return pump_results
[docs]
def save_results_csv(
time_steps: np.ndarray, total_heat_KW: np.ndarray, strom_wp_kW: np.ndarray, pump_results: dict, filename: str
) -> None:
"""
Export simulation results to CSV file with German column headers.
:param time_steps: Time step array
:type time_steps: np.ndarray
:param total_heat_KW: Building heat demand time series [kW]
:type total_heat_KW: np.ndarray
:param strom_wp_kW: Heat pump electrical consumption [kW]
:type strom_wp_kW: np.ndarray
:param pump_results: Structured pump results from calculate_results
:type pump_results: Dict
:param filename: Output CSV file path
:type filename: str
.. note::
Semicolon-separated CSV with German column names. Includes Zeit,
Gesamtwärmebedarf_Gebäude_kW, pump data (Wärmeerzeugung, Massenstrom,
Delta p, temperatures, pressures). UTF-8-sig encoding.
"""
# Convert arrays to pandas DataFrame
df = pd.DataFrame(
{
"Zeit": time_steps,
"Gesamtwärmebedarf_Gebäude_kW": total_heat_KW,
"Gesamtheizlast_Gebäude_kW": total_heat_KW + strom_wp_kW,
"Gesamtstrombedarf_Wärmepumpen_Gebäude_kW": strom_wp_kW,
}
)
# Add pump data columns
for pump_type, pumps in pump_results.items():
for idx, pump_data in pumps.items():
df[f"Wärmeerzeugung_{pump_type}_{idx + 1}_kW"] = pump_data["qext_kW"]
df[f"Massenstrom_{pump_type}_{idx + 1}_kg/s"] = pump_data["mass_flow"]
df[f"Delta p_{pump_type}_{idx + 1}_bar"] = pump_data["deltap"]
df[f"Vorlauftemperatur_{pump_type}_{idx + 1}_°C"] = pump_data["flow_temp"]
df[f"Rücklauftemperatur_{pump_type}_{idx + 1}_°C"] = pump_data["return_temp"]
df[f"Vorlaufdruck_{pump_type}_{idx + 1}_bar"] = pump_data["flow_pressure"]
df[f"Rücklaufdruck_{pump_type}_{idx + 1}_bar"] = pump_data["return_pressure"]
# Save DataFrame as CSV with German formatting
df.to_csv(filename, sep=";", date_format="%Y-%m-%d %H:%M:%S", index=False, encoding="utf-8-sig")
[docs]
def import_results_csv(filename: str) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict]:
"""
Import simulation results from CSV file created by save_results_csv.
:param filename: Input CSV file path
:type filename: str
:return: (time_steps, total_heat_KW, strom_wp_kW, pump_results)
:rtype: Tuple[np.ndarray, np.ndarray, np.ndarray, Dict]
:raises FileNotFoundError: If CSV file cannot be found
:raises pd.errors.ParserError: If CSV format is invalid
:raises KeyError: If required columns are missing
.. note::
Parses German semicolon-separated CSV. Reconstructs pump_results dict
structure. Converts dtypes to datetime64/float64. Returns time_steps as
datetime, heat/power as kW arrays, pump_results matching calculate_results.
"""
# Load data from CSV file
data = pd.read_csv(filename, sep=";", parse_dates=["Zeit"])
# Extract general time series and heat data
time_steps = data["Zeit"].values.astype("datetime64")
total_heat_KW = data["Gesamtwärmebedarf_Gebäude_kW"].values.astype("float64")
strom_wp_kW = data["Gesamtstrombedarf_Wärmepumpen_Gebäude_kW"].values.astype("float64")
# Create dictionary to store pump data
pump_results = {}
# Mapping of German column prefixes to result keys
pump_data = {
"Wärmeerzeugung": "qext_kW",
"Massenstrom": "mass_flow",
"Delta p": "deltap",
"Vorlauftemperatur": "flow_temp",
"Rücklauftemperatur": "return_temp",
"Vorlaufdruck": "flow_pressure",
"Rücklaufdruck": "return_pressure",
}
# Parse pump data columns
for column in data.columns:
if any(
prefix in column
for prefix in [
"Wärmeerzeugung",
"Massenstrom",
"Delta p",
"Vorlauftemperatur",
"Rücklauftemperatur",
"Vorlaufdruck",
"Rücklaufdruck",
]
):
parts = column.split("_")
if len(parts) >= 4:
# Expected structure: [prefix, pump_type, index, parameter]
prefix, pump_type, idx, _parameter = parts[0], parts[1], int(parts[2]) - 1, "_".join(parts[3:])
value = pump_data[prefix]
# Initialize nested dictionary structure
if pump_type not in pump_results:
pump_results[pump_type] = {}
if idx not in pump_results[pump_type]:
pump_results[pump_type][idx] = {}
# Add parameter data to corresponding pump
pump_results[pump_type][idx][value] = data[column].values.astype("float64")
else:
print(f"Warning: Column name '{column}' has an unexpected format and is ignored.")
return time_steps, total_heat_KW, strom_wp_kW, pump_results