Coverage for pyrc\core\simulation.py: 12%
172 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-29 15:57 +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 if self.use_analytic_pre_calculation:
254 network.solve_stationary()
255 else:
256 t_span = (0, pre_calculation_seconds)
257 network.solve_network(
258 t_span,
259 print_progress=print_progress,
260 name_add_on=name_add_on + "_pre_calculation",
261 time_dependent_tuple=(time_dependent_symbols, static_time_dependent_function),
262 time_dependent_tuple_input=(time_dependent_symbols_input, static_time_dependent_function_input),
263 )
265 # save last solution to load it back in later
266 network.rc_solution.save_last_step(single_solution_name)
268 # delete all static solutions in the solutions object and free space for dynamic solution
269 network.rc_solution.delete_solutions(confirm=True)
270 network.reset_properties()
272 assert network.load_initial_values(return_bool=True, pickle_path_single_solution=single_solution_name)
273 network.settings = settings # change back to original settings
274 print(f"{network.hash}: pre-calculation done.")
275 else:
276 print(f"{network.hash}: pre-calculation was loaded from file.")
278 t_span = t_span_simulation
279 network.solve_network(
280 t_span,
281 print_progress=print_progress,
282 name_add_on=name_add_on,
283 time_dependent_tuple=time_dependent_tuple,
284 time_dependent_tuple_input=time_dependent_tuple_input,
285 )
286 # result is saved in network.solve_network so it doesn't need to be saved in here.
287 # file_path = f"{name_prefix}_result.pickle"
288 # network.rc_solution.save_solution(file_path)
290 if return_network:
291 return network
292 return None
295class Parameterization(Simulation):
296 """
297 Class to handle `RCNetwork` calculations that are quite similar but differ in one settings parameter.
299 All calculations are run in parallel.
300 """
302 def __init__(
303 self,
304 parameters_tuples: list[tuple],
305 pre_calculation_seconds=36000,
306 initial_settings_dict: dict = i_settings,
307 max_core_number=0,
308 t_span=None,
309 ):
310 """
312 Parameters
313 ----------
314 parameters_tuples : list[tuple]
315 Defining the parameters for each simulation:
316 for parameters in parameters_tuples:
317 network_type: type = parameters[0]
318 settings_dict = parameters[1]
319 network_parameters = parameters[2]
320 t_span = parameters[3]
321 name_add_on = parameters[4]
322 pre_calculation_seconds : int | float, default=36000
323 How long the static calculation before the dynamic calculation should be.
324 initial_settings_dict : dict, default=i_settings
325 The initial settings to use for the calculations. If not given, the initial ones from ``core.settings`` is
326 used.
327 max_core_number : int, default=0
328 An optional limit how many cores can be used for the calculations.
329 0 for no limit.
330 """
331 super().__init__(pre_calculation_seconds=pre_calculation_seconds, settings=initial_settings_dict)
332 self.parameters_tuples: list[tuple] = parameters_tuples
334 if t_span is None:
335 t_span = (0, 8760 * 3600)
336 self.t_span = t_span
338 if max_core_number <= 0:
339 max_core_number = mp.cpu_count()
340 self.max_core_number: int = min(max_core_number, mp.cpu_count())
342 def get_parameters(self, parameters: tuple):
343 """
344 Returns the first few parameters for the worker method.
346 Parameters
347 ----------
348 parameters : tuple
349 The tuple out of self.parameters_tuples
351 Returns
352 -------
353 tuple :
354 The parameters as tuple.
355 """
356 network_type: type = parameters[0]
357 settings_dict = parameters[1]
358 network_parameters = parameters[2]
359 t_span = parameters[3]
360 name_add_on = parameters[4]
362 settings = Settings(**settings_dict)
363 network_parameters.update(
364 {
365 "settings": settings,
366 "load_from_pickle": True,
367 "save_to_pickle": True,
368 "num_cores_jacobian": 1,
369 "rc_objects": RCObjects(),
370 "rc_solution": RCSolution(),
371 }
372 )
374 pre_settings_dict = settings_dict.copy()
375 try:
376 original_date = settings_dict["start_date"]
377 except KeyError:
378 original_date = "2022-01-01T00:00:00"
379 pre_settings_dict.update(
380 {"start_date": subtract_seconds_from_string(original_date, self.pre_calculation_seconds)}
381 )
382 pre_settings = Settings(**pre_settings_dict)
384 return (
385 network_type,
386 network_parameters,
387 self.pre_calculation_seconds,
388 t_span,
389 name_add_on,
390 settings,
391 pre_settings,
392 False,
393 )
395 def pre_create_jacobians(self):
396 """
397 Pre-creates the jacobian matrices of all RCNetworks if not already found in pickle file.
399 This is useful to quickly generate all jacobian matrices outside the workers so that they only need to load
400 the matrices and lambdify them.
402 Returns
403 -------
405 """
406 for parameters in self.parameters_tuples:
407 # Reset class attribute from Equation item to get the same hash values for equal networks
408 EquationItem.item_counter = 0
409 params = self.get_parameters(parameters)
410 network_type, network_parameters = params[:2]
412 network_parameters = network_parameters.copy()
413 network_parameters.update(
414 {
415 "load_from_pickle": True,
416 "save_to_pickle": True,
417 "num_cores_jacobian": self.max_core_number,
418 "rc_objects": RCObjects(),
419 "rc_solution": RCSolution(),
420 }
421 )
422 # {"load_from_pickle": True, "save_to_pickle": True, "num_cores_jacobian": 1}) # for debugging
423 network: RCNetwork = network_type(**network_parameters)
425 network.create_network()
426 network.make_system_matrices()
427 print(f"{params[5]}: Jacobi pre-calculation done for hash: {network.hash}")
429 def run(self):
430 # first create the jacobian matrices for all networks using all cores
431 self.pre_create_jacobians()
433 if self.max_core_number == 1:
434 print("Using single core calculation.")
435 for parameters in self.parameters_tuples:
436 worker_parameters = self.get_parameters(parameters)
437 # check name_add_on
438 if worker_parameters[5] == "" or worker_parameters[5] is None:
439 worker_parameters = (*worker_parameters[:5], f"1", *worker_parameters[6:])
440 self._worker(*worker_parameters)
441 print(f"Starting process: {worker_parameters[5]}")
442 else:
443 # create single processes that run the parameter simulations
444 active = []
445 process_infos = []
446 process_id = 0
447 for parameters in self.parameters_tuples:
448 # Wait until a slot is free
449 while len(active) >= self.max_core_number:
450 skip_waiting = False
451 for idx, p in enumerate(active):
452 if not p.is_alive():
453 p.join()
454 proc_id, name_addon, p_name = process_infos[idx]
455 print(f"Process {proc_id} with name AddOn '{name_addon}' finished. Proc name {p_name}")
456 skip_waiting = True
457 if not skip_waiting:
458 time.sleep(0.1) # small delay to avoid busy-waiting
459 mask = [p.is_alive() for p in active]
460 active = [p for p, alive in zip(active, mask) if alive]
461 process_infos = [info for info, alive in zip(process_infos, mask) if alive]
463 worker_parameters = self.get_parameters(parameters)
464 # check name_add_on
465 if worker_parameters[5] == "" or worker_parameters[5] is None:
466 worker_parameters = (*worker_parameters[:5], f"{process_id}", *worker_parameters[6:])
467 process = mp.Process(target=self._worker, args=worker_parameters)
468 print(f"Starting process {process_id} - name AddOn: {worker_parameters[5]} - Proc name {process.name}")
469 process.start()
470 active.append(process)
471 process_infos.append((process_id, worker_parameters[5], process.name))
472 process_id += 1
474 while active:
475 skip_waiting = False
476 for idx, p in enumerate(active):
477 if not p.is_alive():
478 p.join()
479 proc_id, name_addon, p_name = process_infos[idx]
480 print(f"Process {proc_id} (name: {p_name}) with name AddOn '{name_addon}' finished.")
481 skip_waiting = True
482 if not skip_waiting:
483 time.sleep(0.1)
484 active_infos = [(p, info) for p, info in zip(active, process_infos) if p.is_alive()]
485 active, process_infos = zip(*active_infos) if active_infos else ([], [])
487 print("All processes finished.")