Coverage for ebcpy/simulationapi/dymola_api.py: 64%
608 statements
« prev ^ index » next coverage.py v7.4.4, created at 2026-05-27 10:55 +0000
1"""Module containing the DymolaAPI used for simulation
2of Modelica-Models."""
4import sys
5import os
6import shutil
7import uuid
8import warnings
9import atexit
10import json
11import time
12import socket
13from pathlib import Path
14from contextlib import closing
15from typing import Union, List
17from pydantic import Field, BaseModel
18import pandas as pd
20from ebcpy.modelica import manipulate_ds
21from ebcpy.simulationapi import SimulationSetup, SimulationAPI, \
22 SimulationSetupClass, Variable
23from ebcpy.utils.conversion import convert_tsd_to_modelica_txt
26class DymolaSimulationSetup(SimulationSetup):
27 """
28 Adds ``tolerance`` to the list of possible
29 setup fields.
30 """
31 tolerance: float = Field(
32 title="tolerance",
33 default=0.0001,
34 description="Tolerance of integration"
35 )
37 _default_solver = "Dassl"
38 _allowed_solvers = ["Dassl", "Euler", "Cerk23", "Cerk34", "Cerk45",
39 "Esdirk23a", "Esdirk34a", "Esdirk45a", "Cvode",
40 "Rkfix2", "Rkfix3", "Rkfix4", "Lsodar",
41 "Radau", "Dopri45", "Dopri853", "Sdirk34hw"]
44class ExperimentSetupOutput(BaseModel):
45 """
46 Experiment setup output data model with
47 defaults equal to those in Dymola
48 """
49 states: bool = True
50 derivatives: bool = True
51 inputs: bool = True
52 outputs: bool = True
53 auxiliaries: bool = True
54 equidistant: bool = False
55 events: bool = True
57 class Config:
58 """
59 Pydantic internal model settings
60 """
61 # pylint: disable=too-few-public-methods
62 extra = "forbid"
65class DymolaAPI(SimulationAPI):
66 """
67 API to a Dymola instance.
69 :param str,Path working_directory:
70 Dirpath for the current working directory of dymola
71 :param str model_name:
72 Name of the model to be simulated.
73 If None, it has to be provided prior to or when calling simulate().
74 :param list packages:
75 List with path's to the packages needed to simulate the model
76 :keyword Boolean show_window:
77 True to show the Dymola window. Default is False
78 :keyword Boolean modify_structural_parameters:
79 .. deprecated::
80 Will be removed in the next major release.
81 Use model name modifiers directly instead.
82 True to automatically set the structural parameters of the
83 simulation model via Modelica modifiers. Default is True.
84 See also the keyword ``structural_parameters``
85 of the ``simulate`` function.
86 :keyword Boolean equidistant_output:
87 If True (Default), Dymola stores variables in an
88 equisdistant output and does not store variables at events.
89 :keyword dict[str,bool] variables_to_save:
90 A dictionary to select which variables are going
91 to be stored if the simulation creates .mat files.
92 Options (with the default being all True):
93 - states=True
94 - derivatives=True
95 - inputs=True
96 - outputs=True
97 - auxiliaries=False
98 :keyword int n_restart:
99 Number of iterations after which Dymola should restart.
100 This is done to free memory. Default value -1. For values
101 below 1 Dymola does not restart.
102 :keyword bool extract_variables:
103 If True (the default), all variables of the model will be extracted
104 on init of this class.
105 This required translating the model.
106 :keyword bool debug:
107 If True (not the default), the dymola instance is not closed
108 on exit of the python script. This allows further debugging in
109 dymola itself if API-functions cause a python error.
110 :keyword str mos_script_pre:
111 Path to a valid mos-script for Modelica/Dymola.
112 If given, the script is executed **prior** to laoding any
113 package specified in this API.
114 May be relevant for handling version conflicts.
115 :keyword str mos_script_post:
116 Path to a valid mos-script for Modelica/Dymola.
117 If given, the script is executed before closing Dymola.
118 :keyword str dymola_version:
119 Version of Dymola to use.
120 If not given, newest version will be used.
121 If given, the Version needs to be equal to the folder name
122 of your installation.
124 **Example:** If you have two versions installed at
126 - ``C://Program Files//Dymola 2021`` and
127 - ``C://Program Files//Dymola 2020x``
129 and you want to use Dymola 2020x, specify
130 ``dymola_version='Dymola 2020x'``.
132 This parameter is overwritten if ``dymola_path`` is specified.
133 :keyword str dymola_path:
134 Path to the dymola installation on the device. Necessary
135 e.g. on linux, if we can't find the path automatically.
136 Example: ``dymola_path="C://Program Files//Dymola 2020x"``
137 :keyword str dymola_interface_path:
138 Direct path to the .egg-file of the dymola interface.
139 Only relevant when the dymola_path
140 differs from the interface path.
141 :keyword str dymola_exe_path:
142 Direct path to the dymola executable.
143 Only relevant if the dymola installation do not follow
144 the official guideline.
145 :keyword float time_delay_between_starts:
146 If starting multiple Dymola instances on multiple
147 cores, a time delay between each start avoids weird
148 behaviour, such as requiring to set the C-Compiler again
149 as Dymola overrides the default .dymx setup file.
150 If you start e.g. 20 instances and specify `time_delay_between_starts=5`,
151 each 5 seconds one instance will start, taking in total
152 100 seconds. Default is no delay.
154 Example:
156 >>> import os
157 >>> from ebcpy import DymolaAPI
158 >>> # Specify the model name
159 >>> model_name = "Modelica.Thermal.FluidHeatFlow.Examples.PumpAndValve"
160 >>> dym_api = DymolaAPI(working_directory=os.getcwd(),
161 >>> model_name=model_name,
162 >>> packages=[],
163 >>> show_window=True)
164 >>> dym_api.sim_setup = {"start_time": 100,
165 >>> "stop_time": 200}
166 >>> dym_api.simulate()
167 >>> dym_api.close()
169 """
170 _sim_setup_class: SimulationSetupClass = DymolaSimulationSetup
171 _items_to_drop = ["pool", "dymola", "_dummy_dymola_instance"]
172 dymola = None
173 # Default simulation setup
174 _supported_kwargs = [
175 "show_window",
176 "modify_structural_parameters",
177 "dymola_path",
178 "equidistant_output",
179 "variables_to_save",
180 "n_restart",
181 "debug",
182 "mos_script_pre",
183 "mos_script_post",
184 "dymola_version",
185 "dymola_interface_path",
186 "dymola_exe_path",
187 "time_delay_between_starts"
188 ]
190 def __init__(
191 self,
192 working_directory: Union[Path, str],
193 model_name: str = None,
194 packages: List[Union[Path, str]] = None,
195 **kwargs
196 ):
197 """Instantiate class objects."""
198 self.dymola = None # Avoid key-error in get-state. Instance attribute needs to be there.
199 # Update kwargs with regard to what kwargs are supported.
200 self.extract_variables = kwargs.pop("extract_variables", True)
201 self.fully_initialized = False
202 self.debug = kwargs.pop("debug", False)
203 self.show_window = kwargs.pop("show_window", False)
204 _modify_structural_parameters = kwargs.pop("modify_structural_parameters", True)
205 if _modify_structural_parameters is not True:
206 warnings.warn(
207 "'modify_structural_parameters' is deprecated and will be removed "
208 "in the next major release. Use model name modifiers directly instead.",
209 FutureWarning,
210 stacklevel=2,
211 )
212 self.modify_structural_parameters = _modify_structural_parameters
213 self.equidistant_output = kwargs.pop("equidistant_output", True)
214 _variables_to_save = kwargs.pop("variables_to_save", {})
215 self.experiment_setup_output = ExperimentSetupOutput(**_variables_to_save)
217 self.mos_script_pre = kwargs.pop("mos_script_pre", None)
218 self.mos_script_post = kwargs.pop("mos_script_post", None)
219 self.dymola_version = kwargs.pop("dymola_version", None)
220 self.dymola_interface_path = kwargs.pop("dymola_interface_path", None)
221 self.dymola_exe_path = kwargs.pop("dymola_exe_path", None)
222 _time_delay_between_starts = kwargs.pop("time_delay_between_starts", 0)
223 for mos_script in [self.mos_script_pre, self.mos_script_post]:
224 if mos_script is not None:
225 if not os.path.isfile(mos_script):
226 raise FileNotFoundError(
227 f"Given mos_script '{mos_script}' does "
228 f"not exist."
229 )
230 if not str(mos_script).endswith(".mos"):
231 raise TypeError(
232 f"Given mos_script '{mos_script}' "
233 f"is not a valid .mos file."
234 )
236 # Convert to modelica path
237 if self.mos_script_pre is not None:
238 self.mos_script_pre = self._make_modelica_normpath(self.mos_script_pre)
239 if self.mos_script_post is not None:
240 self.mos_script_post = self._make_modelica_normpath(self.mos_script_post)
242 super().__init__(working_directory=working_directory,
243 model_name=model_name,
244 n_cpu=kwargs.pop("n_cpu", 1),
245 save_logs=kwargs.pop("save_logs", True))
247 # First import the dymola-interface
248 dymola_path = kwargs.pop("dymola_path", None)
249 if dymola_path is not None:
250 if not os.path.exists(dymola_path):
251 raise FileNotFoundError(f"Given path '{dymola_path}' can not be found on "
252 "your machine.")
253 else:
254 # Get the dymola-install-path:
255 _dym_installations = self.get_dymola_install_paths()
256 if _dym_installations:
257 if self.dymola_version:
258 dymola_path = _get_dymola_path_of_version(
259 dymola_installations=_dym_installations,
260 dymola_version=self.dymola_version
261 )
262 else:
263 dymola_path = _dym_installations[0] # 0 is the newest
264 self.logger.info("Using dymola installation at %s", dymola_path)
265 else:
266 if self.dymola_exe_path is None or self.dymola_interface_path is None:
267 raise FileNotFoundError(
268 "Could not find dymola on your machine. "
269 "Thus, not able to find the `dymola_exe_path` and `dymola_interface_path`. "
270 "Either specify both or pass an existing `dymola_path`."
271 )
272 self.dymola_path = dymola_path
273 if self.dymola_exe_path is None:
274 self.dymola_exe_path = self.get_dymola_exe_path(dymola_path)
275 self.logger.info("Using dymola.exe: %s", self.dymola_exe_path)
276 if self.dymola_interface_path is None:
277 self.dymola_interface_path = self.get_dymola_interface_path(dymola_path)
278 self.logger.info("Using dymola interface: %s", self.dymola_interface_path)
280 self.packages = []
281 if packages is not None:
282 for package in packages:
283 if isinstance(package, Path):
284 self.packages.append(str(package))
285 elif isinstance(package, str):
286 self.packages.append(package)
287 else:
288 raise TypeError(f"Given package is of type {type(package)}"
289 f" but should be any valid path.")
291 # Import n_restart
292 self.sim_counter = 0
293 self.n_restart = kwargs.pop("n_restart", -1)
294 if not isinstance(self.n_restart, int):
295 raise TypeError(f"n_restart has to be type int but "
296 f"is of type {type(self.n_restart)}")
298 self._dummy_dymola_instance = None # Ensure self._close_dummy gets the attribute.
299 if self.n_restart > 0:
300 self.logger.info("Open blank placeholder Dymola instance to ensure"
301 " a licence during Dymola restarts")
302 # Use standard port allocation, should always work
303 self._dummy_dymola_instance = self._open_dymola_interface(port=-1)
304 atexit.register(self._close_dummy)
306 # List storing structural parameters for later modifying the simulation-name.
307 # Parameter for raising a warning if to many dymola-instances are running
308 self._critical_number_instances = 10 + self.n_cpu
309 # Register the function now in case of an error.
310 if not self.debug:
311 atexit.register(self.close)
312 if self.use_mp:
313 ports = _get_n_available_ports(n_ports=self.n_cpu)
314 self.pool.map(
315 self._setup_dymola_interface,
316 [dict(use_mp=True, port=port, time_delay=i * _time_delay_between_starts)
317 for i, port in enumerate(ports)]
318 )
319 # For translation etc. always setup a default dymola instance
320 self.dymola = self._setup_dymola_interface(dict(use_mp=False))
321 if not self.license_is_available():
322 warnings.warn("You have no licence to use Dymola. "
323 "Hence you can only simulate models with 8 or less equations.")
324 # Update experiment setup output
325 self.update_experiment_setup_output(self.experiment_setup_output)
326 self.fully_initialized = True
327 # Trigger on init.
328 if model_name is not None:
329 self._update_model()
330 # Set result_names to output variables.
331 self.result_names = list(self.outputs.keys())
333 # Check if some kwargs are still present. If so, inform the user about
334 # false usage of kwargs:
335 if kwargs:
336 self.logger.error(
337 "You passed the following kwargs which "
338 "are not part of the supported kwargs and "
339 "have thus no effect: %s.", " ,".join(list(kwargs.keys())))
341 def _update_model(self):
342 # Translate the model and extract all variables,
343 # if the user wants to:
344 if self.extract_variables and self.fully_initialized:
345 self.extract_model_variables()
347 def simulate(self,
348 parameters: Union[dict, List[dict]] = None,
349 return_option: str = "time_series",
350 **kwargs):
351 """
352 Simulate the given parameters.
354 Additional settings:
356 :keyword List[str] model_names:
357 List of Dymola model-names to simulate. Should be either the size
358 of parameters or parameters needs to be sized 1.
359 Keep in mind that different models may use different parameters!
360 :keyword Boolean show_eventlog:
361 Default False. True to show evenlog of simulation (advanced)
362 :keyword Boolean squeeze:
363 Default True. If only one set of initialValues is provided,
364 a DataFrame is returned directly instead of a list.
365 :keyword str table_name:
366 If inputs are given, you have to specify the name of the table
367 in the instance of CombiTimeTable. In order for the inputs to
368 work the value should be equal to the value of 'tableName' in Modelica.
369 :keyword str file_name:
370 If inputs are given, you have to specify the file_name of the table
371 in the instance of CombiTimeTable. In order for the inputs to
372 work the value should be equal to the value of 'fileName' in Modelica.
373 :keyword callable postprocess_mat_result:
374 When choosing return_option savepath and no equidistant output, the mat files may take up
375 a lot of disk space while you are only interested in some variables or parts
376 of the simulation results. This features enables you to pass any function which
377 gets the mat-path as an input and returns some result you are interested in.
378 The function signature is `foo(mat_result_file, **kwargs_postprocessing) -> Any`.
379 Be sure to define the function in a global scope to allow multiprocessing.
380 :keyword dict kwargs_postprocessing:
381 Keyword arguments used in the function `postprocess_mat_result`.
382 :keyword List[str] structural_parameters:
383 .. deprecated::
384 Will be removed in the next major release.
385 Write modifiers directly in the model_name instead, e.g.
386 ``model_name='MyModel(myParam=newValue)'``
387 A list containing all parameter names which are structural in Modelica.
388 This means a modifier has to be created in order to change
389 the value of this parameter. Internally, the given list
390 is added to the known states of the model. Hence, you only have to
391 specify this keyword argument if your structural parameter
392 does not appear in the dsin.txt file created during translation.
394 Example:
395 Changing a record in a model:
397 >>> sim_api.simulate(
398 >>> parameters={"parameterPipe": "AixLib.DataBase.Pipes.PE_X.DIN_16893_SDR11_d160()"},
399 >>> structural_parameters=["parameterPipe"])
401 """
402 # Handle special case for structural_parameters
403 if "structural_parameters" in kwargs:
404 warnings.warn(
405 "The 'structural_parameters' keyword argument is deprecated and will "
406 "be removed in the next major release. Instead, write modifiers directly "
407 "in the model_name, e.g.:\n"
408 " model_name='MyModel(myParam=newValue)'\n"
409 "or pass modified model names via the 'model_names' keyword argument of simulate().",
410 FutureWarning,
411 stacklevel=2,
412 )
413 _struc_params = kwargs["structural_parameters"]
414 # Check if input is 2-dimensional for multiprocessing.
415 # If not, make it 2-dimensional to avoid list flattening in
416 # the super method.
417 if not isinstance(_struc_params[0], list):
418 kwargs["structural_parameters"] = [_struc_params]
419 if "model_names" in kwargs:
420 model_names = kwargs["model_names"]
421 if not isinstance(model_names, list):
422 raise TypeError("model_names needs to be a list.")
423 if isinstance(parameters, dict):
424 # Make an array of parameters to enable correct use of super function.
425 parameters = [parameters] * len(model_names)
426 if parameters is None:
427 parameters = [{}] * len(model_names)
428 return super().simulate(parameters=parameters, return_option=return_option, **kwargs)
430 def _single_simulation(self, kwargs):
431 # Unpack kwargs
432 show_eventlog = kwargs.pop("show_eventlog", False)
433 squeeze = kwargs.pop("squeeze", True)
434 result_file_name = kwargs.pop("result_file_name", 'resultFile')
435 if not isinstance(result_file_name, str):
436 raise TypeError(f"result_file_name has to be type str but is of type {type(result_file_name)}")
437 parameters = kwargs.pop("parameters")
438 return_option = kwargs.pop("return_option")
439 model_names = kwargs.pop("model_names", None)
440 inputs = kwargs.pop("inputs", None)
441 fail_on_error = kwargs.pop("fail_on_error", True)
442 structural_parameters = kwargs.pop("structural_parameters", [])
443 table_name = kwargs.pop("table_name", None)
444 file_name = kwargs.pop("file_name", None)
445 savepath = kwargs.pop("savepath", None)
447 def empty_postprocessing(mat_result, **_kwargs):
448 return mat_result
450 postprocess_mat_result = kwargs.pop("postprocess_mat_result", empty_postprocessing)
451 kwargs_postprocessing = kwargs.pop("kwargs_postprocessing", {})
452 if kwargs:
453 self.logger.error(
454 "You passed the following kwargs which "
455 "are not part of the supported kwargs and "
456 "have thus no effect: %s.", " ,".join(list(kwargs.keys())))
458 # Handle multiprocessing
459 if self.use_mp:
460 if self.dymola is None:
461 # This should not affect #119, as this rarely happens. Thus, the
462 # method used in the DymolaInterface should work.
463 self._setup_dymola_interface(dict(use_mp=True))
465 # Re-set the dymola experiment output if API is newly started
466 self.dymola.experimentSetupOutput(**self.experiment_setup_output.model_dump())
468 # Handle eventlog
469 if show_eventlog:
470 if not self.experiment_setup_output.events:
471 raise ValueError("You can't log events and have an "
472 "equidistant output, set equidistant output=False")
473 self.dymola.ExecuteCommand("Advanced.Debug.LogEvents = true")
474 self.dymola.ExecuteCommand("Advanced.Debug.LogEventsInitialization = true")
476 # Restart Dymola after n_restart iterations
477 self._check_restart()
479 # Handle custom model_names
480 if model_names is not None:
481 # Custom model_name setting
482 _res_names = self.result_names.copy()
483 self._model_name = model_names
484 self._update_model_variables()
485 if _res_names != self.result_names:
486 self.logger.info(
487 "Result names changed due to setting the new model. "
488 "If you do not expect custom result names, ignore this warning."
489 "If you do expect them, please raise an issue to add the "
490 "option when using the model_names keyword.")
491 self.logger.info(
492 "Difference: %s",
493 " ,".join(list(set(_res_names).difference(self.result_names)))
494 )
496 if self.model_name is None:
497 raise ValueError(
498 "You neither passed a model_name when "
499 "starting DymolaAPI, nor when calling simulate. "
500 "Can't simulate no model."
501 )
503 # Handle parameters:
504 if parameters is None:
505 parameters = {}
506 unsupported_parameters = False
507 else:
508 unsupported_parameters = self.check_unsupported_variables(
509 variables=list(parameters.keys()),
510 type_of_var="parameters"
511 )
513 # Handle structural parameters
515 if (unsupported_parameters and
516 (self.modify_structural_parameters or
517 structural_parameters)):
518 # Alter the model_name for the next simulation
519 model_name, parameters_new = self._alter_model_name(
520 parameters=parameters,
521 model_name=self.model_name,
522 structural_params=list(self.states.keys()) + structural_parameters
523 )
524 # Trigger translation only if something changed
525 if model_name != self.model_name:
526 _res_names = self.result_names.copy()
527 self.model_name = model_name
528 self.result_names = _res_names # Restore previous result names
529 self.logger.warning(
530 "Warning: Currently, the model is re-translating "
531 "for each simulation. You should add to your Modelica "
532 "parameters \"annotation(Evaluate=false)\".\n "
533 "Check for these parameters: %s",
534 ', '.join(set(parameters.keys()).difference(parameters_new.keys()))
535 )
536 parameters = parameters_new
537 # Check again
538 unsupported_parameters = self.check_unsupported_variables(
539 variables=list(parameters.keys()),
540 type_of_var="parameters"
541 )
543 initial_names = list(parameters.keys())
544 initial_values = list(parameters.values())
545 # Convert to float for Boolean and integer types:
546 try:
547 initial_values = [float(v) for v in initial_values]
548 except (ValueError, TypeError) as err:
549 raise TypeError("Dymola only accepts float values. "
550 "Could bot automatically convert the given "
551 "parameter values to float.") from err
553 # Handle inputs
554 if inputs is not None:
555 # Unpack additional kwargs
556 if table_name is None or file_name is None:
557 raise KeyError("For inputs to be used by DymolaAPI.simulate, you "
558 "have to specify the 'table_name' and the 'file_name' "
559 "as keyword arguments of the function. These must match"
560 "the values 'tableName' and 'fileName' in the CombiTimeTable"
561 " model in your modelica code.") from err
562 # Generate the input in the correct format
563 offset = self.sim_setup.start_time - inputs.index[0]
564 filepath = convert_tsd_to_modelica_txt(
565 tsd=inputs,
566 table_name=table_name,
567 save_path_file=file_name,
568 offset=offset
569 )
570 self.logger.info("Successfully created Dymola input file at %s", filepath)
572 if return_option == "savepath":
573 if unsupported_parameters:
574 raise KeyError("Dymola does not accept invalid parameter "
575 "names for option return_type='savepath'. "
576 "To use this option, delete unsupported "
577 "parameters from your setup.")
578 res = self.dymola.simulateExtendedModel(
579 self.model_name,
580 startTime=self.sim_setup.start_time,
581 stopTime=self.sim_setup.stop_time,
582 numberOfIntervals=0,
583 outputInterval=self.sim_setup.output_interval,
584 method=self.sim_setup.solver,
585 tolerance=self.sim_setup.tolerance,
586 fixedstepsize=self.sim_setup.fixedstepsize,
587 resultFile=result_file_name,
588 initialNames=initial_names,
589 initialValues=initial_values)
590 else:
591 if not parameters and not self.parameters:
592 raise ValueError(
593 "Sadly, simulating a model in Dymola "
594 "with no parameters returns no result. "
595 "Call this function using return_option='savepath' to get the results."
596 )
597 if not parameters:
598 random_name = list(self.parameters.keys())[0]
599 initial_values = [self.parameters[random_name].value]
600 initial_names = [random_name]
602 # Handle 1 and 2 D initial names:
603 # Convert a 1D list to 2D list
604 if initial_values and isinstance(initial_values[0], (float, int)):
605 initial_values = [initial_values]
607 # Handle the time of the simulation:
608 res_names = self.result_names.copy()
609 if "Time" not in res_names:
610 res_names.append("Time")
612 # Internally convert output Interval to number of intervals
613 # (Required by function simulateMultiResultsModel
614 number_of_intervals = (self.sim_setup.stop_time - self.sim_setup.start_time) / \
615 self.sim_setup.output_interval
616 if int(number_of_intervals) != number_of_intervals:
617 raise ValueError(
618 "Given output_interval and time interval did not yield "
619 "an integer numberOfIntervals. To use this functions "
620 "without savepaths, you have to provide either a "
621 "numberOfIntervals or a value for output_interval "
622 "which can be converted to numberOfIntervals.")
624 res = self.dymola.simulateMultiResultsModel(
625 self.model_name,
626 startTime=self.sim_setup.start_time,
627 stopTime=self.sim_setup.stop_time,
628 numberOfIntervals=int(number_of_intervals),
629 method=self.sim_setup.solver,
630 tolerance=self.sim_setup.tolerance,
631 fixedstepsize=self.sim_setup.fixedstepsize,
632 resultFile=None,
633 initialNames=initial_names,
634 initialValues=initial_values,
635 resultNames=res_names)
637 if not res[0]:
638 self.logger.error("Simulation failed!")
639 self.logger.error("The last error log from Dymola:")
640 dslog_path = self._get_worker_directory(use_mp=self.use_mp).joinpath('dslog.txt')
641 try:
642 with open(dslog_path, "r") as dslog_file:
643 dslog_content = dslog_file.read()
644 self.logger.error(dslog_content)
645 except Exception:
646 log = self.dymola.getLastErrorLog()
647 # Only print last part as output is sometimes to verbose and the error is at the bottom
648 self.logger.error(log[-10000:])
649 dslog_content = "Not retreivable. Open it yourself."
650 msg = f"Simulation failed: Reason according " \
651 f"to dslog, located at '{dslog_path}': {dslog_content}"
652 if fail_on_error:
653 raise Exception(msg)
654 # Don't raise and return None
655 self.logger.error(msg)
656 return None
658 if return_option == "savepath":
659 _save_name_dsres = f"{result_file_name}.mat"
660 # Get the working_directory of the current dymola instance
661 self.dymola.cd()
662 # Get the value and convert it to a 100 % fitting str-path
663 cd_log = self.dymola.getLastErrorLog()
664 if cd_log.endswith("\n"): # Typically ends with \n
665 cd_log = cd_log[:-1]
666 # Sometimes, the other logs are also included. Only use the last line.
667 dymola_working_directory = Path(cd_log.split("\n")[-1])
668 if dymola_working_directory != self._get_worker_directory(use_mp=self.use_mp):
669 self.logger.warning(
670 "The working directory set by ebcpy and the one with the result does not match: "
671 "%s (dymola) vs. %s (ebcpy). This will inhibit correct error "
672 "messages upon failed simulations.",
673 dymola_working_directory,
674 self._get_worker_directory(use_mp=self.use_mp)
675 )
677 mat_working_directory = dymola_working_directory.joinpath(_save_name_dsres).as_posix()
678 if savepath is None or str(savepath) == str(dymola_working_directory):
679 mat_result_file = mat_working_directory
680 else:
681 mat_save_path = os.path.join(savepath, _save_name_dsres)
682 os.makedirs(savepath, exist_ok=True)
683 # Copying dslogs and dsfinals can lead to errors,
684 # as the names are not unique
685 # for filename in [_save_name_dsres, "dslog.txt", "dsfinal.txt"]:
686 # Delete existing files
687 try:
688 os.remove(mat_save_path)
689 except OSError:
690 pass
691 # Move files
692 shutil.copy(mat_working_directory, mat_save_path)
693 os.remove(mat_working_directory)
694 mat_result_file = mat_save_path
695 result_file = postprocess_mat_result(mat_result_file, **kwargs_postprocessing)
696 return result_file
698 data = res[1] # Get data
700 # Sometimes, Dymola adds a last row with all 0 values, even the time
701 data_clean = []
702 for ini_val_set in data:
703 if all(res[-1] == 0 for res in ini_val_set):
704 data_clean.append([res[:-1] for res in ini_val_set])
705 else:
706 data_clean.append(ini_val_set)
708 if return_option == "last_point":
709 results = []
710 for ini_val_set in data_clean:
711 results.append({result_name: ini_val_set[idx][-1] for idx, result_name
712 in enumerate(res_names)})
713 if len(results) == 1 and squeeze:
714 return results[0]
715 return results
716 # Else return as dataframe.
717 dfs = []
718 for ini_val_set in data_clean:
719 df = pd.DataFrame({result_name: ini_val_set[idx] for idx, result_name
720 in enumerate(res_names)})
721 # Set time index
722 df = df.set_index("Time")
723 # Convert it to float
724 df.index = df.index.astype("float64")
726 dfs.append(df)
727 # Most of the cases, only one set is provided. In that case, avoid
728 if len(dfs) == 1 and squeeze:
729 return dfs[0]
730 return dfs
732 def translate(self):
733 """
734 Translates the current model using dymola.translateModel()
735 and checks if erros occur.
736 """
737 res = self.dymola.translateModel(self.model_name)
738 if not res:
739 self.logger.error("Translation failed!")
740 self.logger.error("The last error log from Dymola:")
741 self.logger.error(self.dymola.getLastErrorLog())
742 raise Exception("Translation failed - Aborting")
744 def set_compiler(self, name, path, dll=False, dde=False, opc=False):
745 """
746 Set up the compiler and compiler options on Windows.
747 Optional: Specify if you want to enable dll, dde or opc.
749 :param str name:
750 Name of the compiler, avaiable options:
751 - 'vs': Visual Studio
752 - 'gcc': GCC
753 :param str,os.path.normpath path:
754 Path to the compiler files.
755 Example for name='vs': path='C:/Program Files (x86)/Microsoft Visual Studio 10.0/Vc'
756 Example for name='gcc': path='C:/MinGW/bin/gcc'
757 :param Boolean dll:
758 Set option for dll support. Check Dymolas Manual on what this exactly does.
759 :param Boolean dde:
760 Set option for dde support. Check Dymolas Manual on what this exactly does.
761 :param Boolean opc:
762 Set option for opc support. Check Dymolas Manual on what this exactly does.
763 :return: True, on success.
764 """
765 # Lookup dict for internal name of CCompiler-Variable
766 _name_int = {"vs": "MSVC",
767 "gcc": "GCC"}
769 if "win" not in sys.platform:
770 raise OSError(f"set_compiler function only implemented "
771 f"for windows systems, you are using {sys.platform}")
772 # Manually check correct input as Dymola's error are not a help
773 name = name.lower()
774 if name not in ["vs", "gcc"]:
775 raise ValueError(f"Given compiler name {name} not supported.")
776 if not os.path.exists(path):
777 raise FileNotFoundError(f"Given compiler path {path} does not exist on your machine.")
778 # Convert path for correct input
779 path = self._make_modelica_normpath(path)
780 if self.use_mp:
781 raise ValueError("Given function is not yet supported for multiprocessing")
783 res = self.dymola.SetDymolaCompiler(name.lower(),
784 [f"CCompiler={_name_int[name]}",
785 f"{_name_int[name]}DIR={path}",
786 f"DLL={int(dll)}",
787 f"DDE={int(dde)}",
788 f"OPC={int(opc)}"])
790 return res
792 def import_initial(self, filepath):
793 """
794 Load given dsfinal.txt into dymola
796 :param str,os.path.normpath filepath:
797 Path to the dsfinal.txt to be loaded
798 """
799 if not os.path.isfile(filepath):
800 raise FileNotFoundError(f"Given filepath {filepath} does not exist")
801 if not os.path.splitext(filepath)[1] == ".txt":
802 raise TypeError('File is not of type .txt')
803 if self.use_mp:
804 raise ValueError("Given function is not yet supported for multiprocessing")
805 res = self.dymola.importInitial(dsName=filepath)
806 if res:
807 self.logger.info("Successfully loaded dsfinal.txt")
808 else:
809 raise Exception("Could not load dsfinal into Dymola.")
811 @SimulationAPI.working_directory.setter
812 def working_directory(self, working_directory: Union[Path, str]):
813 """Set the working directory to the given path"""
814 if isinstance(working_directory, str):
815 working_directory = Path(working_directory)
816 self._working_directory = working_directory
817 if self.dymola is None: # Not yet started
818 return
819 # Also set the working_directory in the dymola api
820 self.set_dymola_working_directory(dymola=self.dymola,
821 working_directory=working_directory)
822 if self.use_mp:
823 self.logger.warning("Won't set the working_directory for all workers, "
824 "not yet implemented.")
826 def set_dymola_working_directory(self, dymola, working_directory):
827 """
828 Set the working directory of the Dymola Instance.
829 Before calling the Function, create the path and
830 convert to a modelica-normpath.
831 """
832 os.makedirs(working_directory, exist_ok=True)
833 modelica_working_directory = self._make_modelica_normpath(path=working_directory)
834 res = dymola.cd(modelica_working_directory)
835 if not res:
836 raise OSError(f"Could not change working directory to {working_directory}")
838 def close(self):
839 """Closes dymola."""
840 # Close MP of super class
841 super().close()
842 # Always close main instance
843 self._single_close(dymola=self.dymola)
845 def _close_multiprocessing(self, _):
846 self._single_close()
847 DymolaAPI.dymola = None
849 def _single_close(self, **kwargs):
850 """Closes a single dymola instance"""
851 if self.dymola is None:
852 return # Already closed prior
853 # Execute the mos-script if given:
854 if self.mos_script_post is not None:
855 self.logger.info("Executing given mos_script_post "
856 "prior to closing.")
857 self.dymola.RunScript(self.mos_script_post)
858 self.logger.info("Output of mos_script_post: %s", self.dymola.getLastErrorLog())
859 self.logger.info('Closing Dymola')
860 self.dymola.close()
861 self.logger.info('Successfully closed Dymola')
862 self.dymola = None
864 def _close_dummy(self):
865 """
866 Closes dummy instance at the end of the execution
867 """
868 if self._dummy_dymola_instance is not None:
869 self.logger.info('Closing dummy Dymola instance')
870 self._dummy_dymola_instance.close()
871 self.logger.info('Successfully closed dummy Dymola instance')
873 def extract_model_variables(self):
874 """
875 Extract all variables of the model by
876 translating it and then processing the dsin
877 using the manipulate_ds module.
878 """
879 # Translate model
880 self.logger.info("Translating model '%s' to extract model variables ",
881 self.model_name)
882 self.translate()
883 # Get dsin:
884 df = manipulate_ds.convert_ds_file_to_dataframe(
885 self._get_worker_directory(use_mp=self.use_mp).joinpath("dsin.txt")
886 )
887 # Convert and return all parameters of dsin to initial values and names
888 for idx, row in df.iterrows():
889 _max = float(row["4"])
890 _min = float(row["3"])
891 if _min >= _max:
892 _var_ebcpy = Variable(value=float(row["2"]))
893 else:
894 _var_ebcpy = Variable(
895 min=_min,
896 max=_max,
897 value=float(row["2"])
898 )
899 if row["5"] == "1":
900 self.parameters[idx] = _var_ebcpy
901 elif row["5"] == "5":
902 self.inputs[idx] = _var_ebcpy
903 elif row["5"] == "4":
904 self.outputs[idx] = _var_ebcpy
905 else:
906 self.states[idx] = _var_ebcpy
908 def _setup_dymola_interface(self, kwargs: dict):
909 """Load all packages and change the current working directory"""
910 use_mp = kwargs["use_mp"]
911 port = kwargs.get("port", -1)
912 time_delay = kwargs.get("time_delay", 0)
913 time.sleep(time_delay)
914 dymola = self._open_dymola_interface(port=port)
915 self._check_dymola_instances()
917 # Execute the mos-script if given:
918 if self.mos_script_pre is not None:
919 self.logger.info("Executing given mos_script_pre "
920 "prior to loading packages.")
921 dymola.RunScript(self.mos_script_pre)
922 self.logger.info("Output of mos_script_pre: %s", dymola.getLastErrorLog())
924 # Set the cd in the dymola api
925 self.set_dymola_working_directory(dymola=dymola, working_directory=self._get_worker_directory(use_mp))
927 for package in self.packages:
928 self.logger.info("Loading Model %s", os.path.dirname(package).split("\\")[-1])
929 res = dymola.openModel(package, changeDirectory=False)
930 if not res:
931 raise ImportError(dymola.getLastErrorLog())
932 self.logger.info("Loaded modules")
934 dymola.experimentSetupOutput(**self.experiment_setup_output.model_dump())
935 if use_mp:
936 DymolaAPI.dymola = dymola
937 return None
938 return dymola
940 def update_experiment_setup_output(self, experiment_setup_output: Union[ExperimentSetupOutput, dict]):
941 """
942 Function to update the ExperimentSetupOutput in Dymola for selection
943 of which variables are going to be saved. The options
944 `events` and `equidistant` are overridden if equidistant output is required.
946 :param (ExperimentSetupOutput, dict) experiment_setup_output:
947 An instance of ExperimentSetupOutput or a dict with valid keys for it.
948 """
949 if isinstance(experiment_setup_output, dict):
950 self.experiment_setup_output = ExperimentSetupOutput(**experiment_setup_output)
951 else:
952 self.experiment_setup_output = experiment_setup_output
953 if self.equidistant_output:
954 # Change the Simulation Output, to ensure all
955 # simulation results have the same array shape.
956 # Events can also cause errors in the shape.
957 self.experiment_setup_output.equidistant = True
958 self.experiment_setup_output.events = False
959 if self.dymola is None:
960 return
961 self.dymola.experimentSetupOutput(**self.experiment_setup_output.model_dump())
963 def license_is_available(self, option: str = "Standard"):
964 """Check if license is available"""
965 if self.dymola is None:
966 warnings.warn("You want to check the license before starting dymola, this is not supported.")
967 return False
968 return self.dymola.RequestOption(option)
970 def _open_dymola_interface(self, port):
971 """Open an instance of dymola and return the API-Object"""
972 if self.dymola_interface_path not in sys.path:
973 sys.path.insert(0, self.dymola_interface_path)
974 try:
975 from dymola.dymola_interface import DymolaInterface
976 from dymola.dymola_exception import DymolaConnectionException
977 return DymolaInterface(showwindow=self.show_window,
978 dymolapath=self.dymola_exe_path,
979 port=port)
980 except ImportError as error:
981 raise ImportError("Given dymola-interface could not be "
982 "loaded:\n %s" % self.dymola_interface_path) from error
983 except DymolaConnectionException as error:
984 raise ConnectionError(error) from error
986 def to_dict(self):
987 """
988 Store the most relevant information of this class
989 into a dictionary. This may be used for future configuration.
991 :return: dict config:
992 Dictionary with keys to re-init this class.
993 """
994 # Convert Path to str to enable json-dumping
995 config = {"working_directory": str(self.working_directory),
996 "packages": [str(pack) for pack in self.packages],
997 "model_name": self.model_name,
998 "type": "DymolaAPI",
999 }
1000 # Update kwargs
1001 config.update({kwarg: self.__dict__.get(kwarg, None)
1002 for kwarg in self._supported_kwargs})
1004 return config
1006 def get_packages(self):
1007 """
1008 Get the currently loaded packages of Dymola
1009 """
1010 packages = self.dymola.ExecuteCommand(
1011 'ModelManagement.Structure.AST.Misc.ClassesInPackage("")'
1012 )
1013 if packages is None:
1014 self.logger.error("Could not load packages from Dymola, using self.packages")
1015 packages = []
1016 for pack in self.packages:
1017 pack = Path(pack)
1018 if pack.name == "package.mo":
1019 packages.append(pack.parent.name)
1020 valid_packages = []
1021 for pack in packages:
1022 current_package = f"modelica://{pack}/package.order"
1023 pack_path = self.dymola.ExecuteCommand(
1024 f'Modelica.Utilities.Files.loadResource("{current_package}")'
1025 )
1026 if not isinstance(pack_path, str):
1027 self.logger.error("Could not load model resource for package %s", pack)
1028 if os.path.isfile(pack_path):
1029 valid_packages.append(Path(pack_path).parent)
1030 return valid_packages
1032 def save_for_reproduction(
1033 self,
1034 title: str,
1035 path: Path = None,
1036 files: list = None,
1037 save_total_model: bool = True,
1038 export_fmu: bool = True,
1039 **kwargs
1040 ):
1041 """
1042 Additionally to the basic reproduction, add info
1043 for Dymola packages.
1045 Content which is saved:
1046 - DymolaAPI configuration
1047 - Information on Dymola: Version, flags
1048 - All loaded packages
1049 - Total model, if save_total_model = True
1050 - FMU, if export_fmu = True
1052 :param bool save_total_model:
1053 True to save the total model
1054 :param bool export_fmu:
1055 True to export the FMU of the current model.
1056 """
1057 # Local import to require git-package only when called
1058 from ebcpy.utils.reproduction import ReproductionFile, CopyFile, get_git_information
1060 if files is None:
1061 files = []
1062 # DymolaAPI Info:
1063 files.append(ReproductionFile(
1064 filename="Dymola/DymolaAPI_config.json",
1065 content=json.dumps(self.to_dict(), indent=2)
1066 ))
1067 # Dymola info:
1068 self.dymola.ExecuteCommand("list();")
1069 _flags = self.dymola.getLastErrorLog()
1070 dymola_info = [
1071 self.dymola.ExecuteCommand("DymolaVersion()"),
1072 str(self.dymola.ExecuteCommand("DymolaVersionNumber()")),
1073 "\n\n"
1074 ]
1075 files.append(ReproductionFile(
1076 filename="Dymola/DymolaInfo.txt",
1077 content="\n".join(dymola_info) + _flags
1078 ))
1080 # Packages
1081 packages = self.get_packages()
1082 package_infos = []
1083 for pack_path in packages:
1085 for pack_dir_parent in [pack_path] + list(pack_path.parents):
1086 repo_info = get_git_information(
1087 path=pack_dir_parent,
1088 zip_folder_path="Dymola"
1089 )
1090 if not repo_info:
1091 continue
1093 files.extend(repo_info.pop("difference_files"))
1094 pack_path = str(pack_path) + "; " + "; ".join([f"{key}: {value}" for key, value in repo_info.items()])
1095 break
1096 package_infos.append(str(pack_path))
1097 files.append(ReproductionFile(
1098 filename="Dymola/Modelica_packages.txt",
1099 content="\n".join(package_infos)
1100 ))
1101 # Total model
1102 if save_total_model and self.model_name is not None:
1103 # split ( catches model_names with modifiers. Dots are replaced as they indicate a file suffix.
1104 _total_model_name = f"Dymola/{self.model_name.split('(')[0].replace('.', '_')}_total.mo"
1105 _total_model = Path(self.working_directory).joinpath(_total_model_name)
1106 os.makedirs(_total_model.parent, exist_ok=True) # Create to ensure model can be saved.
1107 if "(" in self.model_name:
1108 # Create temporary model:
1109 temp_model_file = Path(self.working_directory).joinpath(f"temp_total_model_{uuid.uuid4()}.mo")
1110 temp_mode_name = f"{self.model_name.split('(')[0].split('.')[-1]}WithModifier"
1111 with open(temp_model_file, "w") as file:
1112 file.write(f"model {temp_mode_name}\n extends {self.model_name};\nend {temp_mode_name};")
1113 res = self.dymola.openModel(str(temp_model_file), changeDirectory=False)
1114 if not res:
1115 self.logger.error(
1116 "Could not create separate model for model with modifiers: %s",
1117 self.model_name
1118 )
1119 model_name_to_save = self.model_name
1120 else:
1121 model_name_to_save = temp_mode_name
1122 os.remove(temp_model_file)
1123 else:
1124 model_name_to_save = self.model_name
1125 res = self.dymola.saveTotalModel(
1126 fileName=str(_total_model),
1127 modelName=model_name_to_save
1128 )
1129 if res:
1130 files.append(ReproductionFile(
1131 filename=_total_model_name,
1132 content=_total_model.read_text()
1133 ))
1134 os.remove(_total_model)
1135 else:
1136 self.logger.error("Could not save total model: %s",
1137 self.dymola.getLastErrorLog())
1138 # FMU
1139 if export_fmu:
1140 _fmu_path = self._save_to_fmu(fail_on_error=False)
1141 if _fmu_path is not None:
1142 files.append(CopyFile(
1143 sourcepath=_fmu_path,
1144 filename="Dymola/" + _fmu_path.name,
1145 remove=True
1146 ))
1148 return super().save_for_reproduction(
1149 title=title,
1150 path=path,
1151 files=files,
1152 **kwargs
1153 )
1155 def _save_to_fmu(self, fail_on_error):
1156 """Save model as an FMU"""
1157 res = self.dymola.translateModelFMU(
1158 modelToOpen=self.model_name,
1159 storeResult=False,
1160 modelName='',
1161 fmiVersion='2',
1162 fmiType='all',
1163 includeSource=False,
1164 includeImage=0
1165 )
1166 if not res:
1167 msg = "Could not export fmu: %s" % self.dymola.getLastErrorLog()
1168 self.logger.error(msg)
1169 if fail_on_error:
1170 raise Exception(msg)
1171 else:
1172 path = Path(self.working_directory).joinpath(res + ".fmu")
1173 return path
1175 @staticmethod
1176 def _make_modelica_normpath(path):
1177 """
1178 Convert given path to a path readable in dymola.
1179 If the base path does not exist, create it.
1181 :param str,os.path.normpath path:
1182 Either a file or a folder path. The base to this
1183 path is created in non existent.
1184 :return: str
1185 Path readable in dymola
1186 """
1187 if isinstance(path, Path):
1188 path = str(path)
1190 path = path.replace("\\", "/")
1191 # Search for e.g. "D:testzone" and replace it with D:/testzone
1192 loc = path.find(":")
1193 if path[loc + 1] != "/" and loc != -1:
1194 path = path.replace(":", ":/")
1195 return path
1197 @staticmethod
1198 def get_dymola_interface_path(dymola_install_dir):
1199 """
1200 Function to get the path of the newest dymola interface
1201 installment on the used machine
1203 :param str dymola_install_dir:
1204 The dymola installation folder. Example:
1205 "C://Program Files//Dymola 2020"
1206 :return: str
1207 Path to the dymola.egg-file or .whl file (for 2024 refresh 1 or newer versions)
1208 """
1209 path_to_interface = os.path.join(dymola_install_dir, "Modelica", "Library", "python_interface")
1210 path_to_egg_file = os.path.join(path_to_interface, "dymola.egg")
1211 if os.path.isfile(path_to_egg_file):
1212 return path_to_egg_file
1213 # Try to find .whl file:
1214 for file in os.listdir(path_to_interface):
1215 if file.endswith(".whl"):
1216 return os.path.join(path_to_interface, file)
1217 # If still here, no .egg or .whl was found
1218 raise FileNotFoundError(f"The given dymola installation directory "
1219 f"'{dymola_install_dir}' has no "
1220 f"dymola-interface .egg or .whl-file.")
1222 @staticmethod
1223 def get_dymola_exe_path(dymola_install_dir, dymola_name=None):
1224 """
1225 Function to get the path of the dymola exe-file
1226 on the current used machine.
1228 :param str dymola_install_dir:
1229 The dymola installation folder. Example:
1230 "C://Program Files//Dymola 2020"
1231 :param str dymola_name:
1232 Name of the executable. On Windows it is always Dymola.exe, on
1233 linux just dymola.
1234 :return: str
1235 Path to the dymola-exe-file.
1236 """
1237 if dymola_name is None:
1238 if "linux" in sys.platform:
1239 dymola_name = "dymola"
1240 elif "win" in sys.platform:
1241 dymola_name = "Dymola.exe"
1242 else:
1243 raise OSError(f"Your operating system {sys.platform} has no default dymola-name."
1244 f"Please provide one.")
1246 bin_64 = os.path.join(dymola_install_dir, "bin64", dymola_name)
1247 bin_32 = os.path.join(dymola_install_dir, "bin", dymola_name)
1248 if os.path.isfile(bin_64): # First check for 64bit installation
1249 dym_file = bin_64
1250 elif os.path.isfile(bin_32): # Else use the 32bit version
1251 dym_file = bin_32
1252 else:
1253 raise FileNotFoundError(
1254 f"The given dymola installation has not executable at '{bin_32}'. "
1255 f"If your dymola_path exists, please raise an issue."
1256 )
1258 return dym_file
1260 @staticmethod
1261 def get_dymola_install_paths(basedir=None):
1262 """
1263 Function to get all paths of dymola installations
1264 on the used machine. Supported platforms are:
1265 * Windows
1266 * Linux
1267 * Mac OS X
1268 If multiple installation of Dymola are found, the newest version will be returned.
1269 This assumes the names are sortable, e.g. Dymola 2020, Dymola 2019 etc.
1271 :param str basedir:
1272 The base-directory to search for the dymola-installation.
1273 The default value depends on the platform one is using.
1274 On Windows it is "C://Program Files" or "C://Program Files (x86)" (for 64 bit)
1275 On Linux it is "/opt" (based on our ci-Docker configuration
1276 On Mac OS X "/Application" (based on the default)
1277 :return: str
1278 Path to the dymola-installation
1279 """
1281 if basedir is None:
1282 if "linux" in sys.platform:
1283 basedir = os.path.normpath("/opt")
1284 elif "win" in sys.platform:
1285 basedir = os.path.normpath("C:/Program Files")
1286 elif "darwin" in sys.platform:
1287 basedir = os.path.normpath("/Applications")
1288 else:
1289 raise OSError(f"Your operating system ({sys.platform})does not support "
1290 f"a default basedir. Please provide one.")
1292 syspaths = [basedir]
1293 # Check if 64bit is installed (Windows only)
1294 systempath_64 = os.path.normpath("C://Program Files (x86)")
1295 if os.path.exists(systempath_64):
1296 syspaths.append(systempath_64)
1297 # Get all folders in both path's
1298 temp_list = []
1299 for systempath in syspaths:
1300 temp_list += os.listdir(systempath)
1301 # Filter programs that are not Dymola
1302 dym_versions = []
1303 for folder_name in temp_list:
1304 # Catch both Dymola and dymola folder-names
1305 if "dymola" in folder_name.lower():
1306 dym_versions.append(folder_name)
1307 del temp_list
1308 # Find the newest version and return the egg-file
1309 # This sorting only works with a good Folder structure, eg. Dymola 2020, Dymola 2019 etc.
1310 dym_versions.sort()
1311 valid_paths = []
1312 for dym_version in reversed(dym_versions):
1313 for system_path in syspaths:
1314 full_path = os.path.join(system_path, dym_version)
1315 if os.path.isdir(full_path):
1316 valid_paths.append(full_path)
1317 return valid_paths
1319 def _check_dymola_instances(self):
1320 """
1321 Check how many dymola instances are running on the machine.
1322 Raise a warning if the number exceeds a certain amount.
1323 """
1324 # The option may be useful. However the explicit requirement leads to
1325 # Problems on linux, therefore the feature is not worth the trouble.
1326 # pylint: disable=import-outside-toplevel
1327 try:
1328 import psutil
1329 except ImportError:
1330 return
1331 counter = 0
1332 for proc in psutil.process_iter():
1333 try:
1334 if "Dymola" in proc.name():
1335 counter += 1
1336 except (psutil.AccessDenied, psutil.NoSuchProcess):
1337 continue
1338 if counter >= self._critical_number_instances:
1339 warnings.warn("There are currently %s Dymola-Instances "
1340 "running on your machine!" % counter)
1342 @staticmethod
1343 def _alter_model_name(parameters, model_name, structural_params):
1344 """
1345 Creates a modifier for all structural parameters,
1346 based on the modelname and the initalNames and values.
1348 :param dict parameters:
1349 Parameters of the simulation
1350 :param str model_name:
1351 Name of the model to be modified
1352 :param list structural_params:
1353 List of strings with structural parameters
1354 :return: str altered_modelName:
1355 modified model name
1356 """
1357 # the structural parameter needs to be removed from paramters dict
1358 new_parameters = parameters.copy()
1359 model_name = model_name.split("(")[0] # Trim old modifier
1360 if parameters == {}:
1361 return model_name
1362 all_modifiers = []
1363 for var_name, value in parameters.items():
1364 # Check if the variable is in the
1365 # given list of structural parameters
1366 if var_name in structural_params:
1367 all_modifiers.append(f"{var_name}={value}")
1368 # removal of the structural parameter
1369 new_parameters.pop(var_name)
1370 altered_model_name = f"{model_name}({','.join(all_modifiers)})"
1371 return altered_model_name, new_parameters
1373 def _check_restart(self):
1374 """Restart Dymola every n_restart iterations in order to free memory"""
1376 if self.sim_counter == self.n_restart:
1377 self.logger.info("Closing and restarting Dymola to free memory")
1378 self.close()
1379 self._dummy_dymola_instance = self._setup_dymola_interface(dict(use_mp=False))
1380 self.sim_counter = 1
1381 else:
1382 self.sim_counter += 1
1385def _get_dymola_path_of_version(dymola_installations: list, dymola_version: str):
1386 """
1387 Helper function to get the path associated to the dymola_version
1388 from the list of all installations
1389 """
1390 for dymola_path in dymola_installations:
1391 if dymola_path.endswith(dymola_version):
1392 return dymola_path
1393 # If still here, version was not found
1394 raise ValueError(
1395 f"Given dymola_version '{dymola_version}' not found in "
1396 f"the list of dymola installations {dymola_installations}"
1397 )
1400def _get_n_available_ports(n_ports: int, start_range: int = 44000, end_range: int = 44400):
1401 """
1402 Get a specified number of available network ports within a given range.
1404 This function uses socket connections to check the availability of ports within the specified range.
1405 If the required number of open ports is found, it returns a list of those ports. If not, it raises
1406 a ConnectionError with a descriptive message indicating the failure to find the necessary ports.
1408 Parameters:
1409 - n_ports (int): The number of open ports to find.
1410 - start_range (int, optional):
1411 The starting port of the range to check (inclusive).
1412 Default is 44000.
1413 - end_range (int, optional):
1414 The ending port of the range to check (exclusive).
1415 Default is 44400.
1417 Returns:
1418 - list of int:
1419 A list containing the available ports.
1420 The length of the list is equal to 'n_ports'.
1422 Raises:
1423 - ConnectionError:
1424 If the required number of open ports cannot
1425 be found within the specified range.
1427 Example:
1429 ```
1430 try:
1431 open_ports = _get_n_available_ports(3, start_range=50000, end_range=50500)
1432 print(f"Found open ports: {open_ports}")
1433 except ConnectionError as e:
1434 print(f"Error: {e}")
1435 ```
1436 """
1437 ports = []
1438 for port in range(start_range, end_range):
1439 try:
1440 with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
1441 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1442 sock.bind(("127.0.0.1", port))
1443 ports.append(port)
1444 except OSError:
1445 pass
1446 if len(ports) == n_ports:
1447 return ports
1448 raise ConnectionError(
1449 f"Could not find {n_ports} open ports in range {start_range}-{end_range}."
1450 f"Can't open {n_ports} Dymola instances"
1451 )