10794805970

Committed 10 Sep 2024 02:38PM UTC coverage: 88.037% (-0.2%) from 88.188%

Build # 10794805970

Build Type

Pull #648

github

Committed by

vtnate

Commit Message

skip removing intermediate file if dir, and add debug logging

Pull Request Pull Request #648: Update code & dependencies to address deprecation warnings

Run Details

955 of 1167 branches covered (81.83%)

Branch coverage included in aggregate %.

9 of 13 new or added lines in 5 files covered. (69.23%)

1 existing line in 1 file now uncovered.

2651 of 2929 relevant lines covered (90.51%)

1.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.44

/geojson_modelica_translator/modelica/modelica_runner.py

# :copyright (c) URBANopt, Alliance for Sustainable Energy, LLC, and other contributors.
# See also https://github.com/urbanopt/geojson-modelica-translator/blob/develop/LICENSE.md

# from __future__ import annotations
import logging
import os
import shutil
import subprocess
from pathlib import Path
from typing import Union

from buildingspy.simulate.Dymola import Simulator
from jinja2 import Environment, FileSystemLoader, StrictUndefined

from geojson_modelica_translator.jinja_filters import ALL_CUSTOM_FILTERS

logger = logging.getLogger(__name__)


class ModelicaRunner:
    """Class to run Modelica models."""

    ACTION_LOG_MAP = {  # noqa: RUF012
        "compile": "Compiling mo file",  # creates an FMU
        "compile_and_run": "Compile and run the mo file",
        "run": "Running FMU",
    }

    def __init__(self):
        """Initialize the runner with data needed for simulation"""

        # Verify that docker is up and running, if needed.
        r = subprocess.call(["docker", "ps"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        self.docker_configured = r == 0

    def _verify_docker_run_capability(self, file_to_load: Union[str, Path, None]):
        """Verify that docker is configured on the host computer correctly before running

        Args:
            file_to_load (Union[str, Path]): Can be a file path or a modelica path

        Raises:
            SystemExit: _description_
            SystemExit: _description_
        """
        if not self.docker_configured:
            raise SystemExit("Docker not configured on host computer, unable to run")

        # If there is a file to load (meaning that we aren't loading from the library),
        # then check that it exists
        if file_to_load and not Path(file_to_load).exists():
            raise SystemExit(f"File not found to run {file_to_load}")

    def _verify_run_path_for_docker(
        self, run_path: Union[str, Path, None], file_to_run: Union[str, Path, None]
    ) -> Path:
        """If there is no run_path, then run it in the same directory as the
        file being run. This works fine for simple Modelica projects but typically
        the run_path needs to be a few levels higher in order to include other
        project dependencies (e.g., multiple mo files).

        Args:
            run_path (str): directory of where to run the simulation
            file_to_run (str): the name of the file to run. This should be the fully
                               qualified path to the file.

        Raises:
            SystemExit: Throw an exception if the run_path or file_to_run has spaces in it

        Returns:
            Path: Return the run_path as a Path object
        """
        if not run_path:
            run_path = Path(file_to_run).parent  # type: ignore[arg-type]
        new_run_path = Path(run_path)

        # Modelica can't handle spaces in project name or path
        if (len(str(new_run_path).split()) > 1) or (len(str(file_to_run).split()) > 1):
            raise SystemExit(
                f"\nModelica does not support spaces in project names or paths. "
                f"You used '{new_run_path}' for run path and {file_to_run} for model project name. "
                "Please update your directory path or model name to not include spaces anywhere."
            )
        return new_run_path

    def _copy_over_docker_resources(
        self, run_path: Path, filename: Union[str, Path, None], model_name: str, **kwargs
    ) -> None:
        """Copy over files needed to run the simulation, this includes
        the generation of the OpenModelica scripts to load and compile/run
        the simulation.

        Must pass start_time, stop_time, and either step_size or number_of_intervals.

        Args:
            run_path (Path): Path where the model will be run, this is where the files will be copied.
            filename (str): name of the file that will be loaded (e.g., BouncingBall.mo, package.mo)
            model_name (str): name of the model to run (e.g., BouncingBall, Districts.DistrictModel)
            **kwargs: Arbitrary keyword arguments.
                project_in_library (bool): If True, then the file_to_load is in the library, otherwise it is a file
                                             on the local file system.
                start_time (int): start time of the simulation, in seconds
                stop_time (int): stop time of the simulation, in seconds
                step_size (int): step size of the simulation, in seconds
                number_of_intervals (int): number of intervals to run the simulation
        """
        # read in the start, stop, and step times
        project_in_library = kwargs.get("project_in_library", False)
        start_time = kwargs.get("start_time", None)
        stop_time = kwargs.get("stop_time", None)
        step_size = kwargs.get("step_size", None)
        number_of_intervals = kwargs.get("number_of_intervals", None)

        # initialize the templating framework (Jinja2)
        template_env = Environment(
            loader=FileSystemLoader(searchpath=Path(__file__).parent.resolve() / "lib" / "runner"),
            undefined=StrictUndefined,
        )
        template_env.filters.update(ALL_CUSTOM_FILTERS)
        template = template_env.get_template("simulate.most")
        model_data = {
            "project_in_library": project_in_library,
            "file_to_load": Path(filename).name if filename else None,
            "model_name": model_name,
            "use_default_time_params": not start_time
            and not stop_time,  # https://docs.astral.sh/ruff/rules/if-expr-with-false-true/#flake8-simplify-sim
            "start_time": start_time,
            "stop_time": stop_time,
            "step_size": step_size,
            "number_of_intervals": number_of_intervals,
        }
        with open(run_path / "simulate.mos", "w") as f:
            f.write(template.render(**model_data))
        template = template_env.get_template("compile_fmu.most")
        with open(run_path / "compile_fmu.mos", "w") as f:
            f.write(template.render(**model_data))

    def _subprocess_call_to_docker(self, run_path: Path, action: str) -> int:
        """Call out to a subprocess to run the command in docker

        Args:
            run_path (Path): local path where the Modelica simulation or compilation will start
            action (str):  action to run either compile_and_run, compile, or run

        Returns:
            int: exit code of the subprocess
        """
        # Set up the run content
        curdir = Path.cwd()
        os.chdir(run_path)
        stdout_log = open("stdout.log", "w")  # noqa: SIM115
        model_name = run_path.parts[-1]
        image_name = "nrel/gmt-om-runner:v2.0.1"
        mo_script = "compile_fmu" if action == "compile" else "simulate"
        try:
            # create the command to call the open modelica compiler inside the docker image
            exec_call = [
                "docker",
                "run",
                "-v",
                f"{run_path}:/mnt/shared/{model_name}",
                f"{image_name}",
                "/bin/bash",
                "-c",
                f"cd mnt/shared/{model_name} && omc {mo_script}.mos",
            ]
            # execute the command that calls docker
            logger.debug(f"Calling {exec_call}")
            completed_process = subprocess.run(
                exec_call,
                stdout=stdout_log,
                stderr=subprocess.STDOUT,
                cwd=run_path,
                check=False,
            )
            # Uncomment this section and rebuild the container in order to pause the container
            # to inspect the container and test commands.
            # import time
            # time.sleep(10000)  # wait for the subprocess to start
            logger.debug(
                f"Subprocess command executed, waiting for completion... \nArgs used: {completed_process.args}"
            )
        except KeyboardInterrupt:
            # List all containers and their images
            docker_containers_cmd = ["docker", "ps", "--format", "{{.ID}} {{.Image}}"]
            containers_list = subprocess.check_output(docker_containers_cmd, text=True).rstrip().split("\n")

            # Find containers from our image
            for container_line in containers_list:
                container_id, container_image = container_line.split()
                if container_image == image_name:
                    logger.debug(f"Killing container: {container_id} (Image: {container_image})")
                    # Kill the container
                    kill_command = f"docker kill {container_id}"
                    subprocess.run(kill_command.split(), check=True, text=True)
            # Remind user why the simulation didn't complete
            raise SystemExit("Simulation stopped by user KeyboardInterrupt")
        finally:
            os.chdir(curdir)
            stdout_log.close()
            logger.debug("Closed stdout.log")

        return completed_process.returncode

    def run_in_docker(
        self,
        action: str,
        model_name: str,
        file_to_load: Union[str, Path, None] = None,
        run_path: Union[str, Path, None] = None,
        **kwargs,
    ) -> tuple[bool, Union[str, Path]]:
        """Run the Modelica project in a docker-based environment.
        The action will determine what type of run will be conducted.
        This method supports either a file path pointing to the package to load, or a modelica path which is a
        period separated path. Results are saved into run_path.

        stdout.log will store both stdout and stderr of the simulations

        Args:
            action (str): The action to run, must be one of compile_and_run, compile, or run
            model_name (str): The name of the model to be simulated (this is the name within Modelica)
            file_to_load (str, Path): The file path or a modelica path to be simulated
            run_path (str, optional): location where the Modelica simulation will start. Defaults to None.
            kwargs: additional arguments to pass to the runner which can include
                project_in_library (bool): whether the project is in a library or not
                start_time (float): start time of the simulation
                stop_time (float): stop time of the simulation
                step_size (float): step size of the simulation
                number_of_intervals (int): number of intervals to run the simulation
                debug (bool): whether to run in debug mode or not, prevents files from being deleted.

        Returns:
            tuple[bool, str]: success status and path to the results directory
        """
        # Verify that the action is in the list of valid actions
        if action not in ModelicaRunner.ACTION_LOG_MAP:
            raise SystemExit(f"Invalid action {action}, must be one of {list(ModelicaRunner.ACTION_LOG_MAP.keys())}")

        self._verify_docker_run_capability(file_to_load)
        verified_run_path = self._verify_run_path_for_docker(run_path, file_to_load)

        self._copy_over_docker_resources(verified_run_path, file_to_load, model_name, **kwargs)

        exitcode = self._subprocess_call_to_docker(verified_run_path, action)

        logger.debug("Checking stdout.log for errors")
        with open(verified_run_path / "stdout.log") as f:
            stdout_log = f.read()
            if "Failed to build model" in stdout_log:
                logger.error("Model failed to build")
                exitcode = 1
            elif "Killed" in stdout_log:
                logger.error(
                    "Model is too large for the Docker resources available."
                    " Increase Docker resources, decrease number of buildings simulated, or both"
                )
            elif "division by zero at time" and "Simulation execution failed for model:" in stdout_log:
                logger.error("Model failed to run due to division by zero")
                exitcode = 1
            elif "The simulation finished successfully" in stdout_log:
                logger.info("Model ran successfully")
                exitcode = 0
            elif action == "compile":
                logger.info("Model compiled successfully -- no errors")
                exitcode = 0
            else:
                logger.error("Model failed to run -- unknown error")
                exitcode = 1

        # Cleanup all of the temporary files that get created
        self.cleanup_path(verified_run_path, model_name, debug=kwargs.get("debug", False))

        logger.debug("Moving results to results directory")
        # get the location of the results path
        results_path = Path(verified_run_path / f"{model_name}_results")
        self.move_results(verified_run_path, results_path, model_name)
        return (exitcode == 0, results_path)

    def run_in_dymola(
        self, action: str, model_name: str, file_to_load: Union[str, Path], run_path: Union[str, Path], **kwargs
    ) -> tuple[bool, Union[str, Path]]:
        """If running on Windows or Linux, you can run Dymola (assuming you have a license),
        using the BuildingsPy library. This is not supported on Mac.

        For using Dymola with the GMT, you need to ensure that MSL v4.0 are loaded correctly and that the
        Buildings library is in the MODELICAPATH. I added the MSL openModel via appending it to the Dymola's
        /opt/<install>/insert/dymola.mos file on Linux. The additions to the dymola.mos will look like the following:

            ```modelica
            // If using Dymola 2024, then the MSL v4.0.0 is already loaded
            // If needed, install the patch of the Modelica Standard Library v4.0.0 (Services)
            openModel("/home/username/Dymola/MSL_v4_ServicesPatch/ModelicaServices/package.mo", changeDirectory=false);
            // If needed, install MSL v4
            openModel("/home/username/Dymola/config/Modelica 4.0.0/package.mo", changeDirectory=false);

            // Open MBL
            openModel("/home/username/working/modelica-buildings/Buildings/package.mo", changeDirectory=false);

            // Set the home directory
            cd("/home/username/working")
            ```

        Args:
            action (str): compile (translate) or simulate
            model_name (str): Name of the model to translate or simulate
            package_path (Union[str, Path]): Name of the package to also load
            kwargs: additional arguments to pass to the runner which can include
                start_time (float): start time of the simulation
                stop_time (float): stop time of the simulation, in seconds
                step_size (float): step size of the simulation, in seconds
                debug (bool): whether to run in debug mode or not, prevents files from being deleted.

        Returns:
            tuple[bool, Union[str, Path]]:  success status and path to the results directory
        """
        run_path = str(run_path)
        current_dir = Path.cwd()
        try:
            os.chdir(run_path)
            print(run_path)
            if file_to_load is None:
                # This occurs when the model is already in a library that is loaded (e.g., MSL, Buildings)
                # Dymola does check the MODELICAPATH for any libraries that need to be loaded automatically.
                dymola_simulator = Simulator(
                    modelName=model_name,
                    outputDirectory=run_path,
                )
            else:
                file_to_load = str(file_to_load)
                dymola_simulator = Simulator(
                    modelName=model_name,
                    outputDirectory=run_path,
                    packagePath=file_to_load,
                )

            # TODO: add in passing of parameters
            # dymola_simulator.addParameters({'PI.k': 10.0, 'PI.Ti': 0.1})
            dymola_simulator.setSolver("dassl")

            start_time = kwargs.get("start_time", 0)
            stop_time = kwargs.get("stop_time", 300)
            step_size = kwargs.get("step_size", 5)

            # calculate the number of intervals based on the step size
            number_of_intervals = int((stop_time - start_time) / step_size)

            dymola_simulator.setStartTime(start_time)
            dymola_simulator.setStopTime(stop_time)
            dymola_simulator.setNumberOfIntervals(number_of_intervals)

            # Do not show progressbar! -- It will cause an "elapsed time used before assigned" error.
            # dymola_simulator.showProgressBar(show=True)

            if kwargs.get("debug", False):
                dymola_simulator.showGUI(show=True)
                dymola_simulator.exitSimulator(False)

            # BuildingPy throws an exception if the model errs
            try:
                if action == "compile":
                    dymola_simulator.translate()
                    # the results of this does not create an FMU, just
                    # the status of the translation/compilation.
                elif action == "simulate":
                    dymola_simulator.simulate()
            except Exception as e:  # noqa: BLE001
                logger.error(f"Exception running Dymola: {e}")
                return False, run_path

            # remove some of the unneeded results
            self.cleanup_path(Path(run_path), model_name, debug=kwargs.get("debug", False))
        finally:
            os.chdir(current_dir)

        return True, run_path

    def move_results(self, from_path: Path, to_path: Path, model_name: Union[str, None] = None) -> None:
        """This method moves the results of the simulation that are known for now.
        This method moves only specific files (stdout.log for now), plus all files and folders beginning
        with the "{project_name}_" name.

        If there are results, they will simply be overwritten (for now).

        Args:
            from_path (Path): where the files will move from
            to_path (Path): where the files will be saved. Will be created if does not exist.
            model_name (Union[str, None], optional): name of the project ran in run_in_docker method. Defaults to None.
        """
        to_path.mkdir(parents=True, exist_ok=True)

        files_to_move = [
            "stdout.log",
        ]
        if model_name is not None:
            files_to_move.append(
                f"{model_name}.log",
            )
            files_to_move.append(
                f"{model_name}.fmu",
            )
            files_to_move.append(
                f"{model_name.replace('.', '_')}.log",
            )
            files_to_move.append(
                f"{model_name.replace('.', '_')}_FMU.log",
            )

        for to_move in from_path.iterdir():
            if to_move != to_path and ((to_move.name in files_to_move) or to_move.name.startswith(f"{model_name}_")):
                # typecast back to strings for the shutil method.
                shutil.move(str(to_move), str(to_path / to_move.name))

    def cleanup_path(self, path: Path, model_name: str, **kwargs: dict) -> None:
        """Clean up the files in the path that was presumably used to run the simulation.
        If debug is passed, then simulation running files will not be removed, but the
        intermediate simulation files will be removed (e.g., .c, .h, .o, .bin)

        Args:
            path (Path): Path of the folder to clean
            model_name (str): Name of the model, used to remove model-specific intermediate files
            kwargs: additional arguments to pass to the runner which can include
                debug (bool): whether to remove all files or not
        """
        # list of files to always remove
        files_to_remove = [
            "dsin.txt",
            "dsmodel.c",
            "dymosim",
            "request.",
            f"{model_name}",
            f"{model_name}.makefile",
            f"{model_name}.libs",
            f"{model_name.replace('.', '_')}_info.json",
            f"{model_name.replace('.', '_')}_FMU.makefile",
            f"{model_name.replace('.', '_')}_FMU.libs",
        ]

        # keep these files if debug is passed
        conditional_remove_files = [
            "compile_fmu.mos",
            "simulate.mos",
            "run.mos",
            "run_translate.mos",
        ]

        logger.debug("Removing temporary files")

        if not kwargs.get("debug", False):
            logger.debug("...and removing scripts to run the simulation")
            files_to_remove.extend(conditional_remove_files)

        for f in files_to_remove:
            logger.debug(f"Removing {f}")
            if (path / f).is_dir():
                continue
            (path / f).unlink(missing_ok=True)

        # The other files below will always be removed, debug or not

        # glob for the .c, .h, .o, .bin files to remove
        remove_files_glob = [
            f"{model_name}*.c",
            f"{model_name}*.h",
            f"{model_name}*.o",
            f"{model_name}*.bin",
        ]
        for pattern in remove_files_glob:
            for f in path.glob(pattern):  # type: ignore[assignment]
                Path(f).unlink(missing_ok=True)

        # Remove the 'tmp' folder that was created by 5G simulations
        # Dir won't exist for 4G simulations, so ignoring errors
        shutil.rmtree(path / "tmp", ignore_errors=True)

1	# :copyright (c) URBANopt, Alliance for Sustainable Energy, LLC, and other contributors.
2	# See also https://github.com/urbanopt/geojson-modelica-translator/blob/develop/LICENSE.md
3
4	# from __future__ import annotations
5	import logging	2✔
6	import os	2✔
7	import shutil	2✔
8	import subprocess	2✔
9	from pathlib import Path	2✔
10	from typing import Union	2✔
11
12	from buildingspy.simulate.Dymola import Simulator	2✔
13	from jinja2 import Environment, FileSystemLoader, StrictUndefined	2✔
14
15	from geojson_modelica_translator.jinja_filters import ALL_CUSTOM_FILTERS	2✔
16
17	logger = logging.getLogger(__name__)	2✔
18
19
20	class ModelicaRunner:	2✔
21	"""Class to run Modelica models."""
22
23	ACTION_LOG_MAP = { # noqa: RUF012	2✔
24	"compile": "Compiling mo file", # creates an FMU
25	"compile_and_run": "Compile and run the mo file",
26	"run": "Running FMU",
27	}
28
29	def __init__(self):	2✔
30	"""Initialize the runner with data needed for simulation"""
31
32	# Verify that docker is up and running, if needed.
33	r = subprocess.call(["docker", "ps"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)	2✔
34	self.docker_configured = r == 0	2✔
35
36	def _verify_docker_run_capability(self, file_to_load: Union[str, Path, None]):	2✔
37	"""Verify that docker is configured on the host computer correctly before running
38
39	Args:
40	file_to_load (Union[str, Path]): Can be a file path or a modelica path
41
42	Raises:
43	SystemExit: _description_
44	SystemExit: _description_
45	"""
46	if not self.docker_configured:	2!
47	raise SystemExit("Docker not configured on host computer, unable to run")	×
48
49	# If there is a file to load (meaning that we aren't loading from the library),
50	# then check that it exists
51	if file_to_load and not Path(file_to_load).exists():	2✔
52	raise SystemExit(f"File not found to run {file_to_load}")	2✔
53
54	def _verify_run_path_for_docker(	2✔
55	self, run_path: Union[str, Path, None], file_to_run: Union[str, Path, None]
56	) -> Path:
57	"""If there is no run_path, then run it in the same directory as the
58	file being run. This works fine for simple Modelica projects but typically
59	the run_path needs to be a few levels higher in order to include other
60	project dependencies (e.g., multiple mo files).
61
62	Args:
63	run_path (str): directory of where to run the simulation
64	file_to_run (str): the name of the file to run. This should be the fully
65	qualified path to the file.
66
67	Raises:
68	SystemExit: Throw an exception if the run_path or file_to_run has spaces in it
69
70	Returns:
71	Path: Return the run_path as a Path object
72	"""
73	if not run_path:	2✔
74	run_path = Path(file_to_run).parent # type: ignore[arg-type]	2✔
75	new_run_path = Path(run_path)	2✔
76
77	# Modelica can't handle spaces in project name or path
78	if (len(str(new_run_path).split()) > 1) or (len(str(file_to_run).split()) > 1):	2!
79	raise SystemExit(	×
80	f"\nModelica does not support spaces in project names or paths. "
81	f"You used '{new_run_path}' for run path and {file_to_run} for model project name. "
82	"Please update your directory path or model name to not include spaces anywhere."
83	)
84	return new_run_path	2✔
85
86	def _copy_over_docker_resources(	2✔
87	self, run_path: Path, filename: Union[str, Path, None], model_name: str, **kwargs
88	) -> None:
89	"""Copy over files needed to run the simulation, this includes
90	the generation of the OpenModelica scripts to load and compile/run
91	the simulation.
92
93	Must pass start_time, stop_time, and either step_size or number_of_intervals.
94
95	Args:
96	run_path (Path): Path where the model will be run, this is where the files will be copied.
97	filename (str): name of the file that will be loaded (e.g., BouncingBall.mo, package.mo)
98	model_name (str): name of the model to run (e.g., BouncingBall, Districts.DistrictModel)
99	**kwargs: Arbitrary keyword arguments.
100	project_in_library (bool): If True, then the file_to_load is in the library, otherwise it is a file
101	on the local file system.
102	start_time (int): start time of the simulation, in seconds
103	stop_time (int): stop time of the simulation, in seconds
104	step_size (int): step size of the simulation, in seconds
105	number_of_intervals (int): number of intervals to run the simulation
106	"""
107	# read in the start, stop, and step times
108	project_in_library = kwargs.get("project_in_library", False)	2✔
109	start_time = kwargs.get("start_time", None)	2✔
110	stop_time = kwargs.get("stop_time", None)	2✔
111	step_size = kwargs.get("step_size", None)	2✔
112	number_of_intervals = kwargs.get("number_of_intervals", None)	2✔
113
114	# initialize the templating framework (Jinja2)
115	template_env = Environment(	2✔
116	loader=FileSystemLoader(searchpath=Path(__file__).parent.resolve() / "lib" / "runner"),
117	undefined=StrictUndefined,
118	)
119	template_env.filters.update(ALL_CUSTOM_FILTERS)	2✔
120	template = template_env.get_template("simulate.most")	2✔
121	model_data = {	2✔
122	"project_in_library": project_in_library,
123	"file_to_load": Path(filename).name if filename else None,
124	"model_name": model_name,
125	"use_default_time_params": not start_time
126	and not stop_time, # https://docs.astral.sh/ruff/rules/if-expr-with-false-true/#flake8-simplify-sim
127	"start_time": start_time,
128	"stop_time": stop_time,
129	"step_size": step_size,
130	"number_of_intervals": number_of_intervals,
131	}
132	with open(run_path / "simulate.mos", "w") as f:	2✔
133	f.write(template.render(**model_data))	2✔
134	template = template_env.get_template("compile_fmu.most")	2✔
135	with open(run_path / "compile_fmu.mos", "w") as f:	2✔
136	f.write(template.render(**model_data))	2✔
137
138	def _subprocess_call_to_docker(self, run_path: Path, action: str) -> int:	2✔
139	"""Call out to a subprocess to run the command in docker
140
141	Args:
142	run_path (Path): local path where the Modelica simulation or compilation will start
143	action (str): action to run either compile_and_run, compile, or run
144
145	Returns:
146	int: exit code of the subprocess
147	"""
148	# Set up the run content
149	curdir = Path.cwd()	2✔
150	os.chdir(run_path)	2✔
151	stdout_log = open("stdout.log", "w") # noqa: SIM115	2✔
152	model_name = run_path.parts[-1]	2✔
153	image_name = "nrel/gmt-om-runner:v2.0.1"	2✔
154	mo_script = "compile_fmu" if action == "compile" else "simulate"	2✔
155	try:	2✔
156	# create the command to call the open modelica compiler inside the docker image
157	exec_call = [	2✔
158	"docker",
159	"run",
160	"-v",
161	f"{run_path}:/mnt/shared/{model_name}",
162	f"{image_name}",
163	"/bin/bash",
164	"-c",
165	f"cd mnt/shared/{model_name} && omc {mo_script}.mos",
166	]
167	# execute the command that calls docker
168	logger.debug(f"Calling {exec_call}")	2✔
169	completed_process = subprocess.run(	2✔
170	exec_call,
171	stdout=stdout_log,
172	stderr=subprocess.STDOUT,
173	cwd=run_path,
174	check=False,
175	)
176	# Uncomment this section and rebuild the container in order to pause the container
177	# to inspect the container and test commands.
178	# import time
179	# time.sleep(10000) # wait for the subprocess to start
180	logger.debug(	2✔
181	f"Subprocess command executed, waiting for completion... \nArgs used: {completed_process.args}"
182	)
183	except KeyboardInterrupt:	×
184	# List all containers and their images
185	docker_containers_cmd = ["docker", "ps", "--format", "{{.ID}} {{.Image}}"]	×
186	containers_list = subprocess.check_output(docker_containers_cmd, text=True).rstrip().split("\n")	×
187
188	# Find containers from our image
189	for container_line in containers_list:	×
190	container_id, container_image = container_line.split()	×
191	if container_image == image_name:	×
192	logger.debug(f"Killing container: {container_id} (Image: {container_image})")	×
193	# Kill the container
194	kill_command = f"docker kill {container_id}"	×
195	subprocess.run(kill_command.split(), check=True, text=True)	×
196	# Remind user why the simulation didn't complete
197	raise SystemExit("Simulation stopped by user KeyboardInterrupt")	×
198	finally:
199	os.chdir(curdir)	2✔
200	stdout_log.close()	2✔
201	logger.debug("Closed stdout.log")	2!
202
203	return completed_process.returncode	2✔
204
205	def run_in_docker(	2✔
206	self,
207	action: str,
208	model_name: str,
209	file_to_load: Union[str, Path, None] = None,
210	run_path: Union[str, Path, None] = None,
211	**kwargs,
212	) -> tuple[bool, Union[str, Path]]:
213	"""Run the Modelica project in a docker-based environment.
214	The action will determine what type of run will be conducted.
215	This method supports either a file path pointing to the package to load, or a modelica path which is a
216	period separated path. Results are saved into run_path.
217
218	stdout.log will store both stdout and stderr of the simulations
219
220	Args:
221	action (str): The action to run, must be one of compile_and_run, compile, or run
222	model_name (str): The name of the model to be simulated (this is the name within Modelica)
223	file_to_load (str, Path): The file path or a modelica path to be simulated
224	run_path (str, optional): location where the Modelica simulation will start. Defaults to None.
225	kwargs: additional arguments to pass to the runner which can include
226	project_in_library (bool): whether the project is in a library or not
227	start_time (float): start time of the simulation
228	stop_time (float): stop time of the simulation
229	step_size (float): step size of the simulation
230	number_of_intervals (int): number of intervals to run the simulation
231	debug (bool): whether to run in debug mode or not, prevents files from being deleted.
232
233	Returns:
234	tuple[bool, str]: success status and path to the results directory
235	"""
236	# Verify that the action is in the list of valid actions
237	if action not in ModelicaRunner.ACTION_LOG_MAP:	2✔
238	raise SystemExit(f"Invalid action {action}, must be one of {list(ModelicaRunner.ACTION_LOG_MAP.keys())}")	2✔
239
240	self._verify_docker_run_capability(file_to_load)	2✔
241	verified_run_path = self._verify_run_path_for_docker(run_path, file_to_load)	2✔
242
243	self._copy_over_docker_resources(verified_run_path, file_to_load, model_name, **kwargs)	2✔
244
245	exitcode = self._subprocess_call_to_docker(verified_run_path, action)	2✔
246
247	logger.debug("Checking stdout.log for errors")	2✔
248	with open(verified_run_path / "stdout.log") as f:	2✔
249	stdout_log = f.read()	2✔
250	if "Failed to build model" in stdout_log:	2!
251	logger.error("Model failed to build")	×
252	exitcode = 1	×
253	elif "Killed" in stdout_log:	2!
254	logger.error(	×
255	"Model is too large for the Docker resources available."
256	" Increase Docker resources, decrease number of buildings simulated, or both"
257	)
258	elif "division by zero at time" and "Simulation execution failed for model:" in stdout_log:	2!
259	logger.error("Model failed to run due to division by zero")	×
260	exitcode = 1	×
261	elif "The simulation finished successfully" in stdout_log:	2✔
262	logger.info("Model ran successfully")	2✔
263	exitcode = 0	2✔
264	elif action == "compile":	2!
265	logger.info("Model compiled successfully -- no errors")	2✔
266	exitcode = 0	2✔
267	else:
268	logger.error("Model failed to run -- unknown error")	×
269	exitcode = 1	×
270
271	# Cleanup all of the temporary files that get created
272	self.cleanup_path(verified_run_path, model_name, debug=kwargs.get("debug", False))	2✔
273
274	logger.debug("Moving results to results directory")	2✔
275	# get the location of the results path
276	results_path = Path(verified_run_path / f"{model_name}_results")	2✔
277	self.move_results(verified_run_path, results_path, model_name)	2✔
278	return (exitcode == 0, results_path)	2✔
279
280	def run_in_dymola(	2✔
281	self, action: str, model_name: str, file_to_load: Union[str, Path], run_path: Union[str, Path], **kwargs
282	) -> tuple[bool, Union[str, Path]]:
283	"""If running on Windows or Linux, you can run Dymola (assuming you have a license),
284	using the BuildingsPy library. This is not supported on Mac.
285
286	For using Dymola with the GMT, you need to ensure that MSL v4.0 are loaded correctly and that the
287	Buildings library is in the MODELICAPATH. I added the MSL openModel via appending it to the Dymola's
288	/opt/<install>/insert/dymola.mos file on Linux. The additions to the dymola.mos will look like the following:
289
290	```modelica
291	// If using Dymola 2024, then the MSL v4.0.0 is already loaded
292	// If needed, install the patch of the Modelica Standard Library v4.0.0 (Services)
293	openModel("/home/username/Dymola/MSL_v4_ServicesPatch/ModelicaServices/package.mo", changeDirectory=false);
294	// If needed, install MSL v4
295	openModel("/home/username/Dymola/config/Modelica 4.0.0/package.mo", changeDirectory=false);
296
297	// Open MBL
298	openModel("/home/username/working/modelica-buildings/Buildings/package.mo", changeDirectory=false);
299
300	// Set the home directory
301	cd("/home/username/working")
302	```
303
304	Args:
305	action (str): compile (translate) or simulate
306	model_name (str): Name of the model to translate or simulate
307	package_path (Union[str, Path]): Name of the package to also load
308	kwargs: additional arguments to pass to the runner which can include
309	start_time (float): start time of the simulation
310	stop_time (float): stop time of the simulation, in seconds
311	step_size (float): step size of the simulation, in seconds
312	debug (bool): whether to run in debug mode or not, prevents files from being deleted.
313
314	Returns:
315	tuple[bool, Union[str, Path]]: success status and path to the results directory
316	"""
317	run_path = str(run_path)	×
318	current_dir = Path.cwd()	×
319	try:	×
320	os.chdir(run_path)	×
321	print(run_path)	×
322	if file_to_load is None:	×
323	# This occurs when the model is already in a library that is loaded (e.g., MSL, Buildings)
324	# Dymola does check the MODELICAPATH for any libraries that need to be loaded automatically.
325	dymola_simulator = Simulator(	×
326	modelName=model_name,
327	outputDirectory=run_path,
328	)
329	else:
330	file_to_load = str(file_to_load)	×
331	dymola_simulator = Simulator(	×
332	modelName=model_name,
333	outputDirectory=run_path,
334	packagePath=file_to_load,
335	)
336
337	# TODO: add in passing of parameters
338	# dymola_simulator.addParameters({'PI.k': 10.0, 'PI.Ti': 0.1})
339	dymola_simulator.setSolver("dassl")	×
340
341	start_time = kwargs.get("start_time", 0)	×
342	stop_time = kwargs.get("stop_time", 300)	×
343	step_size = kwargs.get("step_size", 5)	×
344
345	# calculate the number of intervals based on the step size
346	number_of_intervals = int((stop_time - start_time) / step_size)	×
347
348	dymola_simulator.setStartTime(start_time)	×
349	dymola_simulator.setStopTime(stop_time)	×
350	dymola_simulator.setNumberOfIntervals(number_of_intervals)	×
351
352	# Do not show progressbar! -- It will cause an "elapsed time used before assigned" error.
353	# dymola_simulator.showProgressBar(show=True)
354
355	if kwargs.get("debug", False):	×
356	dymola_simulator.showGUI(show=True)	×
357	dymola_simulator.exitSimulator(False)	×
358
359	# BuildingPy throws an exception if the model errs
360	try:	×
361	if action == "compile":	×
362	dymola_simulator.translate()	×
363	# the results of this does not create an FMU, just
364	# the status of the translation/compilation.
365	elif action == "simulate":	×
366	dymola_simulator.simulate()	×
367	except Exception as e: # noqa: BLE001	×
368	logger.error(f"Exception running Dymola: {e}")	×
369	return False, run_path	×
370
371	# remove some of the unneeded results
372	self.cleanup_path(Path(run_path), model_name, debug=kwargs.get("debug", False))	×
373	finally:
374	os.chdir(current_dir)	×
375
376	return True, run_path	×
377
378	def move_results(self, from_path: Path, to_path: Path, model_name: Union[str, None] = None) -> None:	2✔
379	"""This method moves the results of the simulation that are known for now.
380	This method moves only specific files (stdout.log for now), plus all files and folders beginning
381	with the "{project_name}_" name.
382
383	If there are results, they will simply be overwritten (for now).
384
385	Args:
386	from_path (Path): where the files will move from
387	to_path (Path): where the files will be saved. Will be created if does not exist.
388	model_name (Union[str, None], optional): name of the project ran in run_in_docker method. Defaults to None.
389	"""
390	to_path.mkdir(parents=True, exist_ok=True)	2✔
391
392	files_to_move = [	2✔
393	"stdout.log",
394	]
395	if model_name is not None:	2!
396	files_to_move.append(	2✔
397	f"{model_name}.log",
398	)
399	files_to_move.append(	2✔
400	f"{model_name}.fmu",
401	)
402	files_to_move.append(	2✔
403	f"{model_name.replace('.', '_')}.log",
404	)
405	files_to_move.append(	2✔
406	f"{model_name.replace('.', '_')}_FMU.log",
407	)
408
409	for to_move in from_path.iterdir():	2✔
410	if to_move != to_path and ((to_move.name in files_to_move) or to_move.name.startswith(f"{model_name}_")):	2✔
411	# typecast back to strings for the shutil method.
412	shutil.move(str(to_move), str(to_path / to_move.name))	2✔
413
414	def cleanup_path(self, path: Path, model_name: str, **kwargs: dict) -> None:	2✔
415	"""Clean up the files in the path that was presumably used to run the simulation.
416	If debug is passed, then simulation running files will not be removed, but the
417	intermediate simulation files will be removed (e.g., .c, .h, .o, .bin)
418
419	Args:
420	path (Path): Path of the folder to clean
421	model_name (str): Name of the model, used to remove model-specific intermediate files
422	kwargs: additional arguments to pass to the runner which can include
423	debug (bool): whether to remove all files or not
424	"""
425	# list of files to always remove
426	files_to_remove = [	2✔
427	"dsin.txt",
428	"dsmodel.c",
429	"dymosim",
430	"request.",
431	f"{model_name}",
432	f"{model_name}.makefile",
433	f"{model_name}.libs",
434	f"{model_name.replace('.', '_')}_info.json",
435	f"{model_name.replace('.', '_')}_FMU.makefile",
436	f"{model_name.replace('.', '_')}_FMU.libs",
437	]
438
439	# keep these files if debug is passed
440	conditional_remove_files = [	2✔
441	"compile_fmu.mos",
442	"simulate.mos",
443	"run.mos",
444	"run_translate.mos",
445	]
446
447	logger.debug("Removing temporary files")	2✔
448
449	if not kwargs.get("debug", False):	2!
450	logger.debug("...and removing scripts to run the simulation")	2✔
451	files_to_remove.extend(conditional_remove_files)	2✔
452
453	for f in files_to_remove:	2✔
454	logger.debug(f"Removing {f}")	2✔
455	if (path / f).is_dir():	2!
NEW 456	continue	×
457	(path / f).unlink(missing_ok=True)	2✔
458
459	# The other files below will always be removed, debug or not
460
461	# glob for the .c, .h, .o, .bin files to remove
462	remove_files_glob = [	2✔
463	f"{model_name}*.c",
464	f"{model_name}*.h",
465	f"{model_name}*.o",
466	f"{model_name}*.bin",
467	]
468	for pattern in remove_files_glob:	2✔
469	for f in path.glob(pattern): # type: ignore[assignment]	2✔
470	Path(f).unlink(missing_ok=True)	2✔
471
472	# Remove the 'tmp' folder that was created by 5G simulations
473	# Dir won't exist for 4G simulations, so ignoring errors
474	shutil.rmtree(path / "tmp", ignore_errors=True)	2✔

urbanopt / geojson-modelica-translator / 10794805970

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous