5413684304

pending completion

Build # 5413684304

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Breakout package parser class (#566)

* move package_parser to its own file

* add cspell workspace dictionary

* add cspell workspace dictionary

* fix mypy errors

* skip 2 tests that fail using OM 1.20

* remove unnecessary python enumerate command

* refactor: swap os. for Path in package_parser.py

* refactor: transform self.path to be a Path no matter what it came in as

* Revert "refactor: transform self.path to be a Path no matter what it came in as"

This reverts commit 092e3863e.

* Revert "refactor: swap os. for Path in package_parser.py"

This reverts commit 137b8c163.

---------

Co-authored-by: Nathan Moore <nathan.moore@nrel.gov>

Run Details

888 of 1061 branches covered (83.69%)

Branch coverage included in aggregate %.

87 of 87 new or added lines in 14 files covered. (100.0%)

2460 of 2726 relevant lines covered (90.24%)

1.8 hits per line

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

83.24

/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 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, modelica_lib_path=None):
        """
        Initialize the runner with data needed for simulation

        :param modelica_lib_path: string, Path to the MBL to run against
        """
        # check if the user has defined a MODELICAPATH, is so, then use that.
        if os.environ.get('MODELICAPATH', None):
            print('Using predefined MODELICAPATH')
            self.modelica_lib_path = os.environ['MODELICAPATH']
            logger.debug(f'MODELICAPATH: {self.modelica_lib_path}')
        else:
            self.modelica_lib_path = modelica_lib_path
        local_path = Path(__file__).parent.resolve()
        self.om_docker_path = local_path / 'lib' / 'runner' / 'om_docker.sh'

        # Verify that docker is up and running
        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))

        # new om_docker.sh file name
        new_om_docker = run_path / self.om_docker_path.name
        shutil.copyfile(self.om_docker_path, new_om_docker)
        Path.chmod(new_om_docker, 0o775)

    def _subprocess_call_to_docker(self, run_path: Union[str, 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')
        try:
            # get the relative difference between the file to run and the path which everything is running in.
            # make sure to simulate at a directory above the project directory!

            # Use slashes for the location of the model to run. We can make these periods `.replace(os.sep, '.')`
            # but must strip off the .mo extension on the model to run
            # run_model = Path(file_to_run).relative_to(run_path)
            exec_call = ['./om_docker.sh', action]
            logger.debug(f"Calling {exec_call}")
            p = subprocess.Popen(
                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 inpsect 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: {p.args}")
            exitcode = p.wait()
        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
            image_name = 'nrel/gmt-om-runner'
            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 exitcode

    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 arugments 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
                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 '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 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 = [
            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 = [
            'om_docker.sh',
            'compile_fmu.mos',
            'simulate.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)

        # Note that the om.py script that runs within the container does cleanup
        # the 'tmp' folder including the 'tmp/temperatureResponseMatrix' folder

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 jinja2 import Environment, FileSystemLoader, StrictUndefined	2✔
12
13	from geojson_modelica_translator.jinja_filters import ALL_CUSTOM_FILTERS	2✔
14
15	logger = logging.getLogger(__name__)	2✔
16	logging.basicConfig(	2✔
17	level=logging.INFO,
18	format='%(asctime)s - %(levelname)s: %(message)s',
19	datefmt='%d-%b-%y %H:%M:%S',
20	)
21
22
23	class ModelicaRunner(object):	2✔
24	"""
25	Class to run Modelica models.
26	"""
27	ACTION_LOG_MAP = {	2✔
28	'compile': 'Compiling mo file', # creates an FMU
29	'compile_and_run': 'Compile and run the mo file',
30	'run': 'Running FMU',
31	}
32
33	def __init__(self, modelica_lib_path=None):	2✔
34	"""
35	Initialize the runner with data needed for simulation
36
37	:param modelica_lib_path: string, Path to the MBL to run against
38	"""
39	# check if the user has defined a MODELICAPATH, is so, then use that.
40	if os.environ.get('MODELICAPATH', None):	2!
41	print('Using predefined MODELICAPATH')	2✔
42	self.modelica_lib_path = os.environ['MODELICAPATH']	2✔
43	logger.debug(f'MODELICAPATH: {self.modelica_lib_path}')	2✔
44	else:
45	self.modelica_lib_path = modelica_lib_path	×
46	local_path = Path(__file__).parent.resolve()	2✔
47	self.om_docker_path = local_path / 'lib' / 'runner' / 'om_docker.sh'	2✔
48
49	# Verify that docker is up and running
50	r = subprocess.call(['docker', 'ps'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)	2✔
51	self.docker_configured = r == 0	2✔
52
53	def _verify_docker_run_capability(self, file_to_load: Union[str, Path, None]):	2✔
54	"""Verify that docker is configured on the host computer correctly before running
55
56	Args:
57	file_to_load (Union[str, Path]): Can be a file path or a modelica path
58
59	Raises:
60	SystemExit: _description_
61	SystemExit: _description_
62	"""
63	if not self.docker_configured:	2!
64	raise SystemExit('Docker not configured on host computer, unable to run')	×
65
66	# If there is a file to load (meaning that we aren't loading from the library),
67	# then check that it exists
68	if file_to_load and not Path(file_to_load).exists():	2✔
69	raise SystemExit(f'File not found to run {file_to_load}')	2✔
70
71	def _verify_run_path_for_docker(self, run_path: Union[str, Path, None], file_to_run: Union[str, Path, None]) -> Path:	2✔
72	"""If there is no run_path, then run it in the same directory as the
73	file being run. This works fine for simple Modelica projects but typically
74	the run_path needs to be a few levels higher in order to include other
75	project dependencies (e.g., multiple mo files).
76
77	Args:
78	run_path (str): directory of where to run the simulation
79	file_to_run (str): the name of the file to run. This should be the fully
80	qualified path to the file.
81
82	Raises:
83	SystemExit: Throw an exception if the run_path or file_to_run has spaces in it
84
85	Returns:
86	Path: Return the run_path as a Path object
87	"""
88	if not run_path:	2!
89	run_path = Path(file_to_run).parent # type: ignore	×
90	new_run_path = Path(run_path)	2✔
91
92	# Modelica can't handle spaces in project name or path
93	if (len(str(new_run_path).split()) > 1) or (len(str(file_to_run).split()) > 1):	2!
94	raise SystemExit(	×
95	f"\nModelica does not support spaces in project names or paths. "
96	f"You used '{new_run_path}' for run path and {file_to_run} for model project name. "
97	"Please update your directory path or model name to not include spaces anywhere.")
98	return new_run_path	2✔
99
100	def _copy_over_docker_resources(self, run_path: Path, filename: Union[str, Path, None], model_name: str, **kwargs) -> None:	2✔
101	"""Copy over files needed to run the simulation, this includes
102	the generation of the OpenModelica scripts to load and compile/run
103	the simulation.
104
105	Must pass start_time, stop_time, and either step_size or number_of_intervals.
106
107	Args:
108	run_path (Path): Path where the model will be run, this is where the files will be copied.
109	filename (str): name of the file that will be loaded (e.g., BouncingBall.mo, package.mo)
110	model_name (str): name of the model to run (e.g., BouncingBall, Districts.DistrictModel)
111	**kwargs: Arbitrary keyword arguments.
112	project_in_library (bool): If True, then the file_to_load is in the library, otherwise it is a file
113	on the local file system.
114	start_time (int): start time of the simulation, in seconds
115	stop_time (int): stop time of the simulation, in seconds
116	step_size (int): step size of the simulation, in seconds
117	number_of_intervals (int): number of intervals to run the simulation
118	"""
119	# read in the start, stop, and step times
120	project_in_library = kwargs.get('project_in_library', False)	2✔
121	start_time = kwargs.get('start_time', None)	2✔
122	stop_time = kwargs.get('stop_time', None)	2✔
123	step_size = kwargs.get('step_size', None)	2✔
124	number_of_intervals = kwargs.get('number_of_intervals', None)	2✔
125
126	# initialize the templating framework (Jinja2)
127	template_env = Environment(	2✔
128	loader=FileSystemLoader(searchpath=Path(__file__).parent.resolve() / 'lib' / 'runner'),
129	undefined=StrictUndefined
130	)
131	template_env.filters.update(ALL_CUSTOM_FILTERS)	2✔
132	template = template_env.get_template('simulate.most')	2✔
133	model_data = {	2✔
134	"project_in_library": project_in_library,
135	"file_to_load": Path(filename).name if filename else None,
136	"model_name": model_name,
137	"use_default_time_params": False if start_time is not None and stop_time is not None else True,
138	"start_time": start_time,
139	"stop_time": stop_time,
140	"step_size": step_size,
141	"number_of_intervals": number_of_intervals,
142	}
143	with open(run_path / 'simulate.mos', 'w') as f:	2✔
144	f.write(template.render(**model_data))	2✔
145	template = template_env.get_template('compile_fmu.most')	2✔
146	with open(run_path / 'compile_fmu.mos', 'w') as f:	2✔
147	f.write(template.render(**model_data))	2✔
148
149	# new om_docker.sh file name
150	new_om_docker = run_path / self.om_docker_path.name	2✔
151	shutil.copyfile(self.om_docker_path, new_om_docker)	2✔
152	Path.chmod(new_om_docker, 0o775)	2✔
153
154	def _subprocess_call_to_docker(self, run_path: Union[str, Path], action: str) -> int:	2✔
155	"""Call out to a subprocess to run the command in docker
156
157	Args:
158	run_path (Path): local path where the Modelica simulation or compilation will start
159	action (str): action to run either compile_and_run, compile, or run
160
161	Returns:
162	int: exit code of the subprocess
163	"""
164	# Set up the run content
165	curdir = Path.cwd()	2✔
166	os.chdir(run_path)	2✔
167	stdout_log = open('stdout.log', 'w')	2✔
168	try:	2✔
169	# get the relative difference between the file to run and the path which everything is running in.
170	# make sure to simulate at a directory above the project directory!
171
172	# Use slashes for the location of the model to run. We can make these periods `.replace(os.sep, '.')`
173	# but must strip off the .mo extension on the model to run
174	# run_model = Path(file_to_run).relative_to(run_path)
175	exec_call = ['./om_docker.sh', action]	2✔
176	logger.debug(f"Calling {exec_call}")	2✔
177	p = subprocess.Popen(	2✔
178	exec_call, # type: ignore
179	stdout=stdout_log,
180	stderr=subprocess.STDOUT,
181	cwd=run_path
182	)
183	# Uncomment this section and rebuild the container in order to pause the container
184	# to inpsect the container and test commands.
185	# import time
186	# time.sleep(10000) # wait for the subprocess to start
187	logger.debug(f"Subprocess command executed, waiting for completion... \nArgs used: {p.args}")	2✔
188	exitcode = p.wait()	2✔
189	except KeyboardInterrupt:	×
190	# List all containers and their images
191	docker_containers_cmd = ['docker', 'ps', '--format', '{{.ID}} {{.Image}}']	×
192	containers_list = subprocess.check_output(docker_containers_cmd, text=True).rstrip().split('\n')	×
193
194	# Find containers from our image
195	image_name = 'nrel/gmt-om-runner'	×
196	for container_line in containers_list:	×
197	container_id, container_image = container_line.split()	×
198	if container_image == image_name:	×
199	logger.debug(f"Killing container: {container_id} (Image: {container_image})")	×
200	# Kill the container
201	kill_command = f"docker kill {container_id}"	×
202	subprocess.run(kill_command.split(), check=True, text=True)	×
203	# Remind user why the simulation didn't complete
204	raise SystemExit("Simulation stopped by user KeyboardInterrupt")	×
205	finally:
206	os.chdir(curdir)	2✔
207	stdout_log.close()	2✔
208	logger.debug('Closed stdout.log')	2!
209
210	return exitcode	2✔
211
212	def run_in_docker(self, action: str, model_name: str, file_to_load: Union[str, Path, None] = None,	2✔
213	run_path: Union[str, Path, None] = None, **kwargs) -> tuple[bool, Union[str, Path]]:
214	"""Run the Modelica project in a docker-based environment. The action will determine
215	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.
216
217	stdout.log will store both stdout and stderr of the simulations
218
219	Args:
220	action (str): The action to run, must be one of compile_and_run, compile, or run
221	model_name (str): The name of the model to be simulated (this is the name within Modelica)
222	file_to_load (str, Path): The file path or a modelica path to be simulated
223	run_path (str, optional): location where the Modelica simulation will start. Defaults to None.
224	kwargs: additional arugments to pass to the runner which can include
225	project_in_library (bool): whether the project is in a library or not
226	start_time (float): start time of the simulation
227	stop_time (float): stop time of the simulation
228	step_size (float): step size of the simulation
229	debug (bool): whether to run in debug mode or not, prevents files from being deleted.
230
231	Returns:
232	tuple[bool, str]: success status and path to the results directory
233	"""
234	# Verify that the action is in the list of valid actions
235	if action not in ModelicaRunner.ACTION_LOG_MAP:	2✔
236	raise SystemExit(f'Invalid action {action}, must be one of {list(ModelicaRunner.ACTION_LOG_MAP.keys())}')	2✔
237
238	self._verify_docker_run_capability(file_to_load)	2✔
239	verified_run_path = self._verify_run_path_for_docker(run_path, file_to_load)	2✔
240
241	self._copy_over_docker_resources(	2✔
242	verified_run_path, file_to_load, model_name, **kwargs
243	)
244
245	exitcode = self._subprocess_call_to_docker(verified_run_path, action)	2✔
246
247	logger.debug('checking stdout.log for errors')	2✔
248	# Check the stdout.log file for errors
249	with open(verified_run_path / 'stdout.log', 'r') as f:	2✔
250	stdout_log = f.read()	2✔
251	if 'Failed to build model' in stdout_log:	2!
252	logger.error('Model failed to build')	×
253	exitcode = 1	×
254	elif 'The simulation finished successfully' in stdout_log:	2✔
255	logger.info('Model ran successfully')	2✔
256	exitcode = 0	2✔
257	elif action == 'compile':	2✔
258	logger.info('Model compiled successfully -- no errors')	2✔
259	exitcode = 0	2✔
260	else:
261	logger.error('Model failed to run -- unknown error')	2✔
262	exitcode = 1	2✔
263
264	logger.debug('removing temporary files')	2✔
265	# Cleanup all of the temporary files that get created
266	self.cleanup_path(verified_run_path, model_name, debug=kwargs.get('debug', False))	2✔
267
268	logger.debug('moving results to results directory')	2✔
269	# get the location of the results path
270	results_path = Path(verified_run_path / f'{model_name}_results')	2✔
271	self.move_results(verified_run_path, results_path, model_name)	2✔
272	return (exitcode == 0, results_path)	2✔
273
274	def move_results(self, from_path: Path, to_path: Path, model_name: Union[str, None] = None) -> None:	2✔
275	"""This method moves the results of the simulation that are known for now.
276	This method moves only specific files (stdout.log for now), plus all files and folders beginning
277	with the "{project_name}_" name.
278
279	If there are results, they will simply be overwritten (for now).
280
281	Args:
282	from_path (Path): where the files will move from
283	to_path (Path): where the files will be saved. Will be created if does not exist.
284	model_name (Union[str, None], optional): name of the project ran in run_in_docker method. Defaults to None.
285	"""
286	to_path.mkdir(parents=True, exist_ok=True)	2✔
287
288	files_to_move = [	2✔
289	'stdout.log',
290	]
291	if model_name is not None:	2!
292	files_to_move.append(f'{model_name}.log',)	2✔
293	files_to_move.append(f'{model_name}.fmu',)	2✔
294	files_to_move.append(f"{model_name.replace('.', '_')}.log",)	2✔
295	files_to_move.append(f"{model_name.replace('.', '_')}_FMU.log",)	2✔
296
297	for to_move in from_path.iterdir():	2✔
298	if not to_move == to_path:	2✔
299	if (to_move.name in files_to_move) or to_move.name.startswith(f'{model_name}_'):	2✔
300	# typecast back to strings for the shutil method.
301	shutil.move(str(to_move), str(to_path / to_move.name))	2✔
302
303	def cleanup_path(self, path: Path, model_name: str, **kwargs: dict) -> None:	2✔
304	"""Clean up the files in the path that was presumably used to run the simulation.
305	If debug is passed, then simulation running files will not be removed, but the
306	intermediate simulation files will be removed (e.g., .c, .h, .o, .bin)
307
308	Args:
309	path (Path): Path of the folder to clean
310	model_name (str): Name of the model, used to remove model-specific intermediate files
311	kwargs: additional arguments to pass to the runner which can include
312	debug (bool): whether to remove all files or not
313	"""
314	# list of files to always remove
315	files_to_remove = [	2✔
316	f'{model_name}',
317	f'{model_name}.makefile',
318	f'{model_name}.libs',
319	f"{model_name.replace('.', '_')}_info.json",
320	f"{model_name.replace('.', '_')}_FMU.makefile",
321	f"{model_name.replace('.', '_')}_FMU.libs",
322	]
323
324	conditional_remove_files = [	2✔
325	'om_docker.sh',
326	'compile_fmu.mos',
327	'simulate.mos',
328	]
329
330	if not kwargs.get('debug', False):	2!
331	files_to_remove.extend(conditional_remove_files)	2✔
332
333	for f in files_to_remove:	2✔
334	(path / f).unlink(missing_ok=True)	2✔
335
336	# The other files below will always be removed, debug or not
337
338	# glob for the .c, .h, .o, .bin files to remove
339	remove_files_glob = [	2✔
340	f'{model_name}*.c',
341	f'{model_name}*.h',
342	f'{model_name}*.o',
343	f'{model_name}*.bin',
344	]
345	for pattern in remove_files_glob:	2✔
346	for f in path.glob(pattern): # type: ignore	2✔
347	Path(f).unlink(missing_ok=True)	2✔
348
349	# Note that the om.py script that runs within the container does cleanup
350	# the 'tmp' folder including the 'tmp/temperatureResponseMatrix' folder

urbanopt / geojson-modelica-translator / 5413684304

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