Coverage for pyrc\core\simulation.py: 12%
170 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-03 16:07 +0200
1# -------------------------------------------------------------------------------
2# Copyright (C) 2026 Joel Kimmich, Tim Jourdan
3# ------------------------------------------------------------------------------
4# License
5# This file is part of PyRC, distributed under GPL-3.0-or-later.
6# ------------------------------------------------------------------------------
8import multiprocessing as mp
9import os
10import time
11from copy import copy
12from typing import Any, Callable, Iterable
14from pyrc.core.components.templates import EquationItem, RCObjects, RCSolution
15from pyrc.core.network import RCNetwork
16from pyrc.core.settings import Settings
17from pyrc.core.settings import initial_settings as i_settings
18from pyrc.tools.functions import add_leading_underscore, subtract_seconds_from_string
21class Simulation:
22 def __init__(
23 self,
24 network_class: type[RCNetwork] | None = None,
25 network_keyword_arguments: dict[str, Any] | None = None,
26 pre_calculation_seconds: int | float = 36000,
27 t_span: tuple | None = None,
28 name_add_on: str = "",
29 settings: Settings = None,
30 pre_calculation_settings: Settings = None,
31 print_progress: bool = True,
32 time_dependent_tuple: tuple[Iterable, Callable] | None = None,
33 time_dependent_tuple_input: tuple[Iterable, Callable] | None = None,
34 ) -> None:
35 """
36 Handle one RC network simulation including pre-simulation.
38 The pre-simulation determines correct initial values for the network by calculating a time range before the real
39 simulation. The pre-simulation can be done with varying weather data or just static simulation of the initial
40 boundary conditions of the network. However, because of the dynamic values of all capacities it is
41 recommended to use the option with varying (realistic) boundary data.
43 The pre-simulation is saved as single initial values that can be loaded in. It is executed only once and then
44 loaded in from file.
46 All parameters can be also set in the run method. However, when you don't want to pass them in each run()
47 call you can initialize them directly and they will be used (if not overwritten).
49 Parameters
50 ----------
51 network_class : type[RCNetwork] | None, optional
52 The RCNetwork class that is used to create an object of it and run the simulation.
53 This is needed because when using parallelization the RCNetwork cannot be pickled to other CPU cores (on
54 Windows). Instead, first the network is build on each core.
55 network_keyword_arguments : dict[str, Any] | None, optional
56 The keyword arguments for the RCNetwork object that is created.
57 pre_calculation_seconds : int | float, default=36000
58 The length of the pre-simulation in seconds.
59 t_span : tuple | None, optional
60 The simulation start and end time tuple for the scipy.solve_ivp() in seconds.
61 name_add_on : str, default=""
62 settings
63 pre_calculation_settings
64 print_progress : bool, default=True
65 time_dependent_tuple : tuple[Iterable | Callable] | None
66 A ordered list with the time dependent symbols and the function that calculates their values.
67 The list represents the order of the output of the function.
68 The function calculates the value of the time dependent symbols in the order of the list. It gets passed
69 the time step, temperature vector and input vector (last one only if existing):
70 value1, value2, ... = my_function(time_step, temperature_vector, input_vector)
71 """
72 self.network_class = network_class
73 self.network_keyword_arguments: dict[str, Any] | None = network_keyword_arguments
74 self.pre_calculation_seconds: int | float = pre_calculation_seconds
75 self.t_span: tuple | None = t_span
76 self.name_add_on: str = name_add_on
77 self.settings: Settings | None = settings
78 self.pre_calculation_settings: Settings | None = pre_calculation_settings
79 self.print_progress: bool = print_progress
80 self.time_dependent_tuple: tuple[Iterable, Callable] | None = time_dependent_tuple
81 self.time_dependent_tuple_input: tuple[Iterable, Callable] | None = time_dependent_tuple_input
83 self.network = None
85 def run(
86 self,
87 network_class: type[RCNetwork] | None = None,
88 network_keyword_arguments: dict[str, Any] | None = None,
89 t_span: tuple | None = None,
90 name_add_on: str = "",
91 settings: Settings | None = None,
92 pre_calculation_settings: Settings | None = None,
93 print_progress: bool | None = None,
94 time_dependent_tuple: tuple[Iterable, Callable] | None = None,
95 time_dependent_tuple_input: tuple[Iterable, Callable] | None = None,
96 ):
97 """
98 Run the simulation including pre-simulation with the passed network type.
100 Parameters
101 ----------
102 network_class : type[RCNetwork] | None, optional
103 The RCNetwork that is created for the simulation.
104 network_keyword_arguments : dict[str, Any] | None, optional
105 The keyword arguments for the RCNetwork object that is created.
106 t_span : tuple | None, optional
107 The t_span for the simulation: (start, end) in seconds.
108 It should start at 0 (otherwise it can work but it's not tested).
109 name_add_on : str, optional
110 An add-on for the name to identify the name to the worker.
111 If None, a random five digits integer is used (with leading zeros).
112 settings : Settings | None, optional
113 The settings for the network and simulation.
114 If None, the initial settings dict from the network is used.
115 pre_calculation_settings : Settings | None, optional
116 The settings for the pre-calculation. Should only vary in the weather data start date.
117 If None, the initial settings dict from the network is used, but with static calculation.
118 print_progress : bool | None, optional
119 Whether to print some progress information during the simulation.
120 time_dependent_tuple : tuple[Iterable | Callable] | None, optional
121 A ordered list with the time dependent symbols and the function that calculates their values.
122 The list represents the order of the output of the function.
123 The function calculates the value of the time dependent symbols in the order of the list. It gets passed
124 the time step, temperature vector and input vector (last one only if existing):
125 value1, value2, ... = my_function(time_step, temperature_vector, input_vector)
126 """
127 self.network = self._worker(
128 network_class=network_class or self.network_class,
129 network_copy_dict=network_keyword_arguments or self.network_keyword_arguments or {},
130 pre_calculation_seconds=self.pre_calculation_seconds,
131 t_span_simulation=t_span or self.t_span,
132 name_add_on=name_add_on or self.name_add_on,
133 settings=settings or self.settings,
134 pre_calculation_settings=pre_calculation_settings or self.pre_calculation_settings,
135 print_progress=print_progress if print_progress is not None else self.print_progress,
136 return_network=True,
137 time_dependent_tuple=time_dependent_tuple or self.time_dependent_tuple,
138 time_dependent_tuple_input=time_dependent_tuple_input or self.time_dependent_tuple_input,
139 )
141 @staticmethod
142 def _worker(
143 network_class: type[RCNetwork],
144 network_copy_dict: dict,
145 pre_calculation_seconds: int | float,
146 t_span_simulation: tuple,
147 name_add_on: str = "",
148 settings: Settings | None = None,
149 pre_calculation_settings: Settings | None = None,
150 print_progress: bool = True,
151 return_network: bool = False,
152 time_dependent_tuple: tuple[Iterable, Callable] | None = None,
153 time_dependent_tuple_input: tuple[Iterable, Callable] | None = None,
154 ) -> RCNetwork | None:
155 """
156 Runs a single simulation.
158 Remember: The network shouldn't exist / built yet because it's not pickable. The network is created newly and if
159 the matrices are already created they are loaded from file.
160 Not only the unpickable state of the network forces the creation of the network within this method but also the
161 network dependency on both rc_objects and rc_solution objects, if the initial objects are used.
163 Parameters
164 ----------
165 network_class : type(RCNetwork)
166 The RCNetwork that is created for the simulation.
167 network_copy_dict : dict
168 A dictionary to give RCNetwork as keyword arguments for initializing.
169 This bypasses to move the unpickable RCNetwork object to the worker and instead creates it inside the
170 worker.
171 pre_calculation_seconds : int | float
172 See main class.
173 t_span_simulation : tuple
174 The t_span for the simulation: (start, end) in seconds.
175 It should start at 0 (otherwise it can work but it's not tested).
176 name_add_on : str, default=""
177 An add-on for the name to identify the name to the worker.
178 settings : Settings, optional
179 The settings for the network and simulation.
180 If None, the initial settings dict from the network is used.
181 pre_calculation_settings : Settings, optional
182 The settings for the pre-calculation. Should only vary in the weather data start date.
183 If None, the initial settings dict from the network is used, but with static calculation or a shifted
184 weather start date, if use_weather_data.
185 time_dependent_tuple : tuple[Iterable | Callable] | None, optional
186 A ordered list with the time dependent symbols and the function that calculates their values.
187 The list represents the order of the output of the function.
188 The function calculates the value of the time dependent symbols in the order of the list. It gets passed
189 the time step, temperature vector and input vector (last one only if existing):
190 value1, value2, ... = my_function(time_step, temperature_vector, input_vector)
191 """
192 # Just to be safe: Create new RCObjects and RCSolution instances that are not linked to other simulations.
193 rc_objects = RCObjects()
194 network_copy_dict.update(
195 {
196 "rc_objects": rc_objects,
197 "rc_solution": RCSolution(rc_objects=rc_objects),
198 }
199 )
200 network: RCNetwork = network_class(**network_copy_dict)
201 if settings is not None:
202 network.settings = copy(settings)
203 else:
204 settings = copy(network.settings)
205 network.create_network()
207 name_add_on = add_leading_underscore(name_add_on)
209 name_prefix = os.path.join(network.settings.save_folder_path, f"{network.hash}{name_add_on}")
210 single_solution_name = f"{name_prefix}_{pre_calculation_seconds}_single_solution.pickle"
212 use_time_dependent_system = False
213 time_dependent_symbols = None
214 time_dependent_function = None
215 if time_dependent_tuple is not None:
216 use_time_dependent_system = True
217 time_dependent_symbols = time_dependent_tuple[0]
218 time_dependent_function = time_dependent_tuple[1]
219 use_time_dependent_input = False
220 time_dependent_symbols_input = None
221 time_dependent_function_input = None
222 if time_dependent_tuple_input is not None:
223 use_time_dependent_input = True
224 time_dependent_symbols_input = time_dependent_tuple_input[0]
225 time_dependent_function_input = time_dependent_tuple_input[1]
227 if not network.load_initial_values(return_bool=True, pickle_path_single_solution=single_solution_name):
228 print(f"{network.hash}: starting pre-calculation")
230 static_time_dependent_function: Callable = time_dependent_function
231 static_time_dependent_function_input: Callable = time_dependent_function_input
232 if pre_calculation_settings is not None:
233 network.settings = pre_calculation_settings
234 else:
235 if network.settings.use_weather_data:
236 original_date = network.settings.start_date
237 network.settings.start_date = subtract_seconds_from_string(original_date, pre_calculation_seconds)
238 network.settings.calculate_static = False
239 print(f"Dynamic pre-calculation with weather start date: {network.settings.start_date}")
240 else:
241 print("Static pre-calculation.")
242 network.settings.calculate_static = True
243 if network.settings.calculate_static and use_time_dependent_system:
244 # always use the first value for the static calculation
245 def static_time_dependent_function(t, temp_vector, *args, **kwargs):
246 return time_dependent_function(0, temp_vector, *args, **kwargs)
248 if network.settings.calculate_static and use_time_dependent_input:
250 def static_time_dependent_function_input(t, temp_vector, *args, **kwargs):
251 return time_dependent_function_input(0, temp_vector, *args, **kwargs)
253 t_span = (0, pre_calculation_seconds)
254 network.solve_network(
255 t_span,
256 print_progress=print_progress,
257 name_add_on=name_add_on + "_pre_calculation",
258 time_dependent_tuple=(time_dependent_symbols, static_time_dependent_function),
259 time_dependent_tuple_input=(time_dependent_symbols_input, static_time_dependent_function_input),
260 )
262 # save last solution to load it back in later
263 network.rc_solution.save_last_step(single_solution_name)
265 # delete all static solutions in the solutions object and free space for dynamic solution
266 network.rc_solution.delete_solutions(confirm=True)
267 network.reset_properties()
269 assert network.load_initial_values(return_bool=True, pickle_path_single_solution=single_solution_name)
270 network.settings = settings # change back to original settings
271 print(f"{network.hash}: pre-calculation done.")
272 else:
273 print(f"{network.hash}: pre-calculation was loaded from file.")
275 t_span = t_span_simulation
276 network.solve_network(
277 t_span,
278 print_progress=print_progress,
279 name_add_on=name_add_on,
280 time_dependent_tuple=time_dependent_tuple,
281 time_dependent_tuple_input=time_dependent_tuple_input,
282 )
283 # result is saved in network.solve_network so it doesn't need to be saved in here.
284 # file_path = f"{name_prefix}_result.pickle"
285 # network.rc_solution.save_solution(file_path)
287 if return_network:
288 return network
289 return None
292class Parameterization(Simulation):
293 """
294 Class to handle `RCNetwork` calculations that are quite similar but differ in one settings parameter.
296 All calculations are run in parallel.
297 """
299 def __init__(
300 self,
301 parameters_tuples: list[tuple],
302 pre_calculation_seconds=36000,
303 initial_settings_dict: dict = i_settings,
304 max_core_number=0,
305 t_span=None,
306 ):
307 """
309 Parameters
310 ----------
311 parameters_tuples : list[tuple]
312 Defining the parameters for each simulation:
313 for parameters in parameters_tuples:
314 network_type: type = parameters[0]
315 settings_dict = parameters[1]
316 network_parameters = parameters[2]
317 t_span = parameters[3]
318 name_add_on = parameters[4]
319 pre_calculation_seconds : int | float, default=36000
320 How long the static calculation before the dynamic calculation should be.
321 initial_settings_dict : dict, default=i_settings
322 The initial settings to use for the calculations. If not given, the initial ones from ``core.settings`` is
323 used.
324 max_core_number : int, default=0
325 An optional limit how many cores can be used for the calculations.
326 0 for no limit.
327 """
328 super().__init__(pre_calculation_seconds=pre_calculation_seconds, settings=initial_settings_dict)
329 self.parameters_tuples: list[tuple] = parameters_tuples
331 if t_span is None:
332 t_span = (0, 8760 * 3600)
333 self.t_span = t_span
335 if max_core_number <= 0:
336 max_core_number = mp.cpu_count()
337 self.max_core_number: int = min(max_core_number, mp.cpu_count())
339 def get_parameters(self, parameters: tuple):
340 """
341 Returns the first few parameters for the worker method.
343 Parameters
344 ----------
345 parameters : tuple
346 The tuple out of self.parameters_tuples
348 Returns
349 -------
350 tuple :
351 The parameters as tuple.
352 """
353 network_type: type = parameters[0]
354 settings_dict = parameters[1]
355 network_parameters = parameters[2]
356 t_span = parameters[3]
357 name_add_on = parameters[4]
359 settings = Settings(**settings_dict)
360 network_parameters.update(
361 {
362 "settings": settings,
363 "load_from_pickle": True,
364 "save_to_pickle": True,
365 "num_cores_jacobian": 1,
366 "rc_objects": RCObjects(),
367 "rc_solution": RCSolution(),
368 }
369 )
371 pre_settings_dict = settings_dict.copy()
372 try:
373 original_date = settings_dict["start_date"]
374 except KeyError:
375 original_date = "2022-01-01T00:00:00"
376 pre_settings_dict.update(
377 {"start_date": subtract_seconds_from_string(original_date, self.pre_calculation_seconds)}
378 )
379 pre_settings = Settings(**pre_settings_dict)
381 return (
382 network_type,
383 network_parameters,
384 self.pre_calculation_seconds,
385 t_span,
386 name_add_on,
387 settings,
388 pre_settings,
389 False,
390 )
392 def pre_create_jacobians(self):
393 """
394 Pre-creates the jacobian matrices of all RCNetworks if not already found in pickle file.
396 This is useful to quickly generate all jacobian matrices outside the workers so that they only need to load
397 the matrices and lambdify them.
399 Returns
400 -------
402 """
403 for parameters in self.parameters_tuples:
404 # Reset class attribute from Equation item to get the same hash values for equal networks
405 EquationItem.item_counter = 0
406 params = self.get_parameters(parameters)
407 network_type, network_parameters = params[:2]
409 network_parameters = network_parameters.copy()
410 network_parameters.update(
411 {
412 "load_from_pickle": True,
413 "save_to_pickle": True,
414 "num_cores_jacobian": self.max_core_number,
415 "rc_objects": RCObjects(),
416 "rc_solution": RCSolution(),
417 }
418 )
419 # {"load_from_pickle": True, "save_to_pickle": True, "num_cores_jacobian": 1}) # for debugging
420 network: RCNetwork = network_type(**network_parameters)
422 network.create_network()
423 network.make_system_matrices()
424 print(f"{params[5]}: Jacobi pre-calculation done for hash: {network.hash}")
426 def run(self):
427 # first create the jacobian matrices for all networks using all cores
428 self.pre_create_jacobians()
430 if self.max_core_number == 1:
431 print("Using single core calculation.")
432 for parameters in self.parameters_tuples:
433 worker_parameters = self.get_parameters(parameters)
434 # check name_add_on
435 if worker_parameters[5] == "" or worker_parameters[5] is None:
436 worker_parameters = (*worker_parameters[:5], f"1", *worker_parameters[6:])
437 self._worker(*worker_parameters)
438 print(f"Starting process: {worker_parameters[5]}")
439 else:
440 # create single processes that run the parameter simulations
441 active = []
442 process_infos = []
443 process_id = 0
444 for parameters in self.parameters_tuples:
445 # Wait until a slot is free
446 while len(active) >= self.max_core_number:
447 skip_waiting = False
448 for idx, p in enumerate(active):
449 if not p.is_alive():
450 p.join()
451 proc_id, name_addon, p_name = process_infos[idx]
452 print(f"Process {proc_id} with name AddOn '{name_addon}' finished. Proc name {p_name}")
453 skip_waiting = True
454 if not skip_waiting:
455 time.sleep(0.1) # small delay to avoid busy-waiting
456 mask = [p.is_alive() for p in active]
457 active = [p for p, alive in zip(active, mask) if alive]
458 process_infos = [info for info, alive in zip(process_infos, mask) if alive]
460 worker_parameters = self.get_parameters(parameters)
461 # check name_add_on
462 if worker_parameters[5] == "" or worker_parameters[5] is None:
463 worker_parameters = (*worker_parameters[:5], f"{process_id}", *worker_parameters[6:])
464 process = mp.Process(target=self._worker, args=worker_parameters)
465 print(f"Starting process {process_id} - name AddOn: {worker_parameters[5]} - Proc name {process.name}")
466 process.start()
467 active.append(process)
468 process_infos.append((process_id, worker_parameters[5], process.name))
469 process_id += 1
471 while active:
472 skip_waiting = False
473 for idx, p in enumerate(active):
474 if not p.is_alive():
475 p.join()
476 proc_id, name_addon, p_name = process_infos[idx]
477 print(f"Process {proc_id} (name: {p_name}) with name AddOn '{name_addon}' finished.")
478 skip_waiting = True
479 if not skip_waiting:
480 time.sleep(0.1)
481 active_infos = [(p, info) for p, info in zip(active, process_infos) if p.is_alive()]
482 active, process_infos = zip(*active_infos) if active_infos else ([], [])
484 print("All processes finished.")