7339456187

Committed 27 Dec 2023 03:01PM UTC coverage: 88.672% (+1.4%) from 87.264%

Build # 7339456187

Build Type

push

github

Committed by

web-flow

Commit Message

Replace shell script with call directly to docker (#607)

* fix(modelica_runner): Replace shell script with call directly to docker (#606)

* fix(modelica_runner): Replace shell script with call directly to docker

* fix(dockerfile): Update the base OM docker image

* use f-string for new docker command

* update dependencies

* comment to update MBL version in modelica-builder as well as Dockerfile when needed

* put exec call into list format for subprocess.Popen

* run simulations on Windows (shocked face!!)

* remove test for old om_docker_path

* remove oudated code from ModelicaRunner.init()

* remove outdated assert in modelica test file

* don't forget to remove the  dir from the model dir after simulation

* use new container that has OMv1.22.1, MBLv10, and MSLv4

* use pathlib for path component instead of os

* go back to only running simulations on Linux :(

* run_path has already been verified to be a Path. Make mypy happy.

* use GHA-style if-syntax

* use MBL tag instead of branch, and clean up call to install it

* change permissions in tmp folder to allow deletion

* temporarily run a very small CI config

* give myself even more permissions to try to delete tmp dir

* use .run method of subprocess instead of Popen

* move tmp dir deletion to the subprocess call method

* TMP: change CI to only run a single test for temporary testing

* chmod the offending dir

* move tmp deletion back to cleanup where it should be

* logging for the GHA runner

* ugh, different logging for GHA runner

* use exists() instead of is_dir(), just for grins

* grant write permissions to CI in the runner

* comment single dir permissions change

* another way of deleting tmp file/dir

* ugh, forgotten bit of text

* change permissions on tmp dir before deletion

* get more nuclear about deleting tmp dir

* clean up comments for removing tmp dir

* re-enable full test suite in CI

* cl... (continued)

Run Details

913 of 1091 branches covered (0.0%)

Branch coverage included in aggregate %.

10 of 10 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

2547 of 2811 relevant lines covered (90.61%)

1.81 hits per line

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

67.77

/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

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__)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s: %(message)s',
    datefmt='%d-%b-%y %H:%M:%S',
)


class ModelicaRunner(object):
    """
    Class to run Modelica models.
    """
    ACTION_LOG_MAP = {
        '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
        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": False if start_time is not None and stop_time is not None else True,
            "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')
        model_name = run_path.parts[-1]
        image_name = 'nrel/gmt-om-runner:v1.22.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,  # type: ignore
                stdout=stdout_log,
                stderr=subprocess.STDOUT,
                cwd=run_path
            )
            # 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')
        # Check the stdout.log file for errors
        with open(verified_run_path / 'stdout.log', 'r') as f:
            stdout_log = f.read()
            if 'Failed to build model' in stdout_log:
                logger.error('Model failed to build')
                exitcode = 1
            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

        logger.debug('removing temporary files')
        # 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>/install/dymola.mos file on Linux.

        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:
                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 not to_move == to_path:
                if (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",
        ]

        conditional_remove_files = [
            'compile_fmu.mos',
            'simulate.mos',
            'run.mos',
            'run_translate.mos',
        ]

        if not kwargs.get('debug', False):
            files_to_remove.extend(conditional_remove_files)

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

urbanopt / geojson-modelica-translator / 7339456187

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