diff --git a/src/pyfmi/fmi2.pyx b/src/pyfmi/fmi2.pyx index 4014b734..c691de85 100644 --- a/src/pyfmi/fmi2.pyx +++ b/src/pyfmi/fmi2.pyx @@ -3700,8 +3700,8 @@ cdef class FMUModelCS2(FMUModelBase2): status -- The status of function which can be checked against - FMI_OK, FMI_WARNING. FMI_DISCARD, FMI_ERROR, - FMI_FATAL,FMI_PENDING... + FMI_OK, FMI_WARNING, FMI_DISCARD, FMI_ERROR, + FMI_FATAL, FMI_PENDING. Calls the underlying low-level function fmi2DoStep. """ diff --git a/src/pyfmi/fmi3.pxd b/src/pyfmi/fmi3.pxd index 2ed1b632..088bdb17 100644 --- a/src/pyfmi/fmi3.pxd +++ b/src/pyfmi/fmi3.pxd @@ -183,9 +183,13 @@ cdef class FMUModelME3(FMUModelBase3): cdef FMIL3.fmi3_status_t _get_nominal_continuous_states_fmil(self, FMIL3.fmi3_float64_t* xnominal, size_t nx) cdef class FMUModelCS3(FMUModelBase3): + cdef public bool do_step_terminated + cdef FMIL3.fmi3_boolean_t _instantiated_with_early_return cpdef _get_time(self) cpdef _set_time(self, FMIL3.fmi3_float64_t t) + cpdef FMIL3.fmi3_status_t do_step(self, FMIL3.fmi3_float64_t current_t, FMIL3.fmi3_float64_t step_size, new_step=*) + cdef class _WorkerClass3: cdef int _dim diff --git a/src/pyfmi/fmi3.pyx b/src/pyfmi/fmi3.pyx index 9c57882c..2fff7df1 100644 --- a/src/pyfmi/fmi3.pyx +++ b/src/pyfmi/fmi3.pyx @@ -3799,12 +3799,18 @@ cdef class FMUModelCS3(FMUModelBase3): FMUModelBase3.__init__(self, fmu, log_file_name, log_level, _unzipped_dir, _connect_dll, allow_unzipped_fmu) + self.do_step_terminated = False + if self.get_capability_flags().get('needsExecutionTool', False): raise FMUException("The FMU specifies 'needsExecutionTool=true' which implies that it requires an external execution tool to simulate, this is not supported.") if _connect_dll: self.instantiate() + def reset(self): + FMUModelBase3.reset(self) + self.do_step_terminated = False + def _get_fmu_kind(self): if self._fmu_kind & FMIL3.fmi3_fmu_kind_cs: return FMIL3.fmi3_fmu_kind_cs @@ -3861,6 +3867,7 @@ cdef class FMUModelCS3(FMUModelBase3): if status != FMIL.jm_status_success: raise FMUException('Failed to instantiate the model. See the log for possibly more information.') + self._instantiated_with_early_return = earlyReturnAllowed self._allocated_fmu = 1 cpdef _get_time(self): @@ -3888,6 +3895,173 @@ cdef class FMUModelCS3(FMUModelBase3): doc = "Property for accessing the current time of the simulation." ) + cpdef FMIL3.fmi3_status_t do_step(self, FMIL3.fmi3_float64_t current_t, FMIL3.fmi3_float64_t step_size, new_step=True): + """ + Performs an integrator step. + + Parameters:: + + current_t -- + The current communication point (current time) of + the master. + + step_size -- + The length of the step to be taken. + + new_step -- + True the last step was accepted by the master and + False if not. + + Returns:: + + status -- + The status of function which can be checked against + FMI_OK, FMI_WARNING, FMI_DISCARD, FMI_ERROR, FMI_FATAL. + + Calls the underlying low-level function fmi3DoStep. + The `do_step_terminated` class attribute tracks the fmi3DoStep return + for `terminateSimulation`. + """ + cdef FMIL3.fmi3_status_t status + cdef FMIL3.fmi3_boolean_t new_s + cdef FMIL3.fmi3_boolean_t eventHandlingNeeded + cdef FMIL3.fmi3_boolean_t terminate + cdef FMIL3.fmi3_boolean_t earlyReturn + cdef FMIL3.fmi3_float64_t lastSuccessfulTime + + if new_step: + new_s = FMIL3.fmi3_true + else: + new_s = FMIL3.fmi3_false + + log_open = self._log_open() + if not log_open and self.get_log_level() > 2: + self._open_log_file() + + self._log_handler.capi_start_callback(self._max_log_size_msg_sent, self._current_log_size) + status = FMIL3.fmi3_import_do_step( + self._fmu, + current_t, + step_size, + new_s, + &eventHandlingNeeded, + &terminate, + &earlyReturn, + &lastSuccessfulTime + ) + self._log_handler.capi_end_callback(self._max_log_size_msg_sent, self._current_log_size) + + if not log_open and self.get_log_level() > 2: + self._close_log_file() + + if status != FMIL3.fmi3_status_ok: + return status + # On a fully completed step the reached time is current_t + step_size; + # lastSuccessfulTime is only meaningful when the FMU returns early. + if terminate: + self.time = lastSuccessfulTime + self.do_step_terminated = True + elif self._instantiated_with_early_return and earlyReturn: + self.time = lastSuccessfulTime + else: + self.time = current_t + step_size + + return status + + def simulate(self, + start_time="Default", + final_time="Default", + input=(), + algorithm='FMICSAlg', + options={}): + """ + Compact function for model simulation. + + The simulation method depends on which algorithm is used, this can be + set with the function argument 'algorithm'. Options for the algorithm + are passed as option classes or as pure dicts. See + FMUModel.simulate_options for more details. + + The default algorithm for this function is FMICSAlg. + + Parameters:: + + start_time -- + Start time for the simulation. + Default: Start time defined in the default experiment from + the ModelDescription file. + + final_time -- + Final time for the simulation. + Default: Stop time defined in the default experiment from + the ModelDescription file. + + input -- + Input signal for the simulation. The input should be a 2-tuple + consisting of first the names of the input variable(s) and then + the data matrix. + Default: Empty tuple. + + algorithm -- + The algorithm which will be used for the simulation is specified + by passing the algorithm class as string or class object in this + argument. 'algorithm' can be any class which implements the + abstract class AlgorithmBase (found in algorithm_drivers.py). In + this way it is possible to write own algorithms and use them + with this function. + Default: 'FMICSAlg' + + options -- + The options that should be used in the algorithm. For details on + the options do: + + >> myModel = load_fmu(...) + >> opts = myModel.simulate_options() + >> opts? + + Valid values are: + - A dict which gives AssimuloFMIAlgOptions with + default values on all options except the ones + listed in the dict. Empty dict will thus give all + options with default values. + - An options object. + Default: Empty dict + + Returns:: + + Result object, subclass of common.algorithm_drivers.ResultBase. + """ + if start_time == "Default": + start_time = self.get_default_experiment_start_time() + if final_time == "Default": + final_time = self.get_default_experiment_stop_time() + + return self._exec_simulate_algorithm(start_time, + final_time, + input, + 'pyfmi.fmi_algorithm_drivers', + algorithm, + options) + + def simulate_options(self, algorithm='FMICSAlg'): + """ + Get an instance of the simulate options class, filled with default + values. If called without argument then the options class for the + default simulation algorithm will be returned. + + Parameters:: + + algorithm -- + The algorithm for which the options class should be fetched. + Possible values are: 'FMICSAlg'. + Default: 'FMICSAlg' + + Returns:: + + Options class for the algorithm specified with default values. + """ + return self._default_options('pyfmi.fmi_algorithm_drivers', algorithm) + def get_capability_flags(self) -> dict: """ Returns a dictionary with the capability flags of the FMU. @@ -3934,6 +4108,19 @@ cdef class FMUModelCS3(FMUModelBase3): return capabilities + def _provides_directional_derivatives(self) -> bool: + """ + Check capability to provide directional derivatives. + """ + return bool(FMIL3.fmi3_import_get_capability(self._fmu, FMIL3.fmi3_cs_providesDirectionalDerivatives)) + + def _supports_get_set_FMU_state(self) -> bool: + """ + Check support for getting and setting the FMU-state. + """ + return bool(FMIL3.fmi3_import_get_capability(self._fmu, FMIL3.fmi3_cs_canGetAndSetFMUState)) + + cdef class FMUModelME3(FMUModelBase3): """ FMI3 ModelExchange model loaded from a dll diff --git a/src/pyfmi/fmi_algorithm_drivers.py b/src/pyfmi/fmi_algorithm_drivers.py index 6b04d18a..fb005ffb 100644 --- a/src/pyfmi/fmi_algorithm_drivers.py +++ b/src/pyfmi/fmi_algorithm_drivers.py @@ -24,9 +24,9 @@ import numpy as np import scipy.optimize as spopt -from pyfmi.fmi1 import FMUModelME1, FMUModelCS1, FMI_ERROR, FMI_DISCARD, FMI1_LAST_SUCCESSFUL_TIME # TODO +from pyfmi.fmi1 import FMUModelME1, FMUModelCS1, FMI_OK, FMI_ERROR, FMI_DISCARD, FMI1_LAST_SUCCESSFUL_TIME # TODO from pyfmi.fmi2 import FMUModelME2, FMUModelCS2, FMI2_INPUT, FMI2_LAST_SUCCESSFUL_TIME -from pyfmi.fmi3 import FMUModelME3 +from pyfmi.fmi3 import FMUModelME3, FMUModelCS3 from pyfmi.fmi_coupled import CoupledFMUModelME2 from pyfmi.fmi_extended import FMUModelME1Extended from pyfmi.fmi_util import parameter_estimation_f @@ -940,11 +940,11 @@ def __init__(self, if self.options['initialize']: if isinstance(self.model, (FMUModelCS1, FMUModelME1Extended)): self.model.initialize(start_time, final_time, stop_time_defined=self.options["stop_time_defined"]) - elif isinstance(self.model, FMUModelCS2): self.model.setup_experiment(start_time=start_time, stop_time_defined=self.options["stop_time_defined"], stop_time=final_time) self.model.initialize() - + elif isinstance(self.model, FMUModelCS3): + self.model.initialize(start_time=start_time, stop_time_defined=self.options["stop_time_defined"], stop_time=final_time) else: raise FMUException("Unknown model.") @@ -952,7 +952,7 @@ def __init__(self, self.result_handler.initialize_complete() time_res_init = timer() - time_res_init - elif self.model.time is None and isinstance(self.model, FMUModelCS2): + elif self.model.time is None and isinstance(self.model, (FMUModelCS2, FMUModelCS3)): raise FMUException("Setup Experiment has not been called, this has to be called prior to the initialization call.") elif self.model.time is None: raise FMUException("The model need to be initialized prior to calling the simulate method if the option 'initialize' is set to False") @@ -1015,6 +1015,26 @@ def _set_solver_options(self): """ pass #No solver options + def _check_do_step_status_and_terminated(self, status) -> tuple[bool, float]: + """Return (true, ) if terminated, (False, 0) else. + Raise exception in case of error returns.""" + if status != FMI_OK: + if status == FMI_DISCARD and isinstance(self.model, (FMUModelCS1, FMUModelCS2)): + try: + if isinstance(self.model, FMUModelCS1): + last_time = self.model.get_real_status(FMI1_LAST_SUCCESSFUL_TIME) + else: + last_time = self.model.get_real_status(FMI2_LAST_SUCCESSFUL_TIME) + return True, last_time + except FMUException: + pass + else: # status = error || fatal || (discard && FMI3) + raise FMUException("The simulation failed. See the log for more information. Return flag %d."%status) + elif isinstance(self.model, FMUModelCS3): + if self.model.do_step_terminated: + return True, self.model.time + return False, 0 + def solve(self): """ Runs the simulation. @@ -1045,28 +1065,16 @@ def solve(self): status = self.model.do_step(t,h) self.status = status - if status != 0: - - if status == FMI_ERROR: - raise FMUException("The simulation failed. See the log for more information. Return flag %d."%status) + terminated, terminated_time = self._check_do_step_status_and_terminated(status) + if terminated: + if terminated_time > t: # only store additional point if time advanced + self.model.time = terminated_time + final_time = terminated_time - elif status == FMI_DISCARD and isinstance(self.model, (FMUModelCS1, FMUModelCS2)): - - try: - if isinstance(self.model, FMUModelCS1): - last_time = self.model.get_real_status(FMI1_LAST_SUCCESSFUL_TIME) - else: - last_time = self.model.get_real_status(FMI2_LAST_SUCCESSFUL_TIME) - if last_time > t: #Solver succeeded in taken a step a little further than the last time - self.model.time = last_time - final_time = last_time - - start_time_point = timer() - result_handler.integration_point() - self.timings["storing_result"] += timer() - start_time_point - except FMUException: - pass - break + start_time_point = timer() + result_handler.integration_point() + self.timings["storing_result"] += timer() - start_time_point + break # stop integration loop final_time = t+h diff --git a/src/pyfmi/fmil3_import.pxd b/src/pyfmi/fmil3_import.pxd index ebdaec66..7913e499 100644 --- a/src/pyfmi/fmil3_import.pxd +++ b/src/pyfmi/fmil3_import.pxd @@ -385,6 +385,17 @@ cdef extern from 'fmilib.h': fmi3_status_t fmi3_import_serialize_fmu_state(fmi3_import_t*, fmi3_FMU_state_t, fmi3_byte_t*, size_t) fmi3_status_t fmi3_import_de_serialize_fmu_state(fmi3_import_t*, fmi3_byte_t*, size_t, fmi3_FMU_state_t*) + # CS CAPI methods + fmi3_status_t fmi3_import_do_step( + fmi3_import_t* fmu, + fmi3_float64_t currentCommunicationPoint, + fmi3_float64_t communicationStepSize, + fmi3_boolean_t noSetFMUStatePriorToCurrentPoint, + fmi3_boolean_t* eventHandlingNeeded, + fmi3_boolean_t* terminate, + fmi3_boolean_t* earlyReturn, + fmi3_float64_t* lastSuccessfulTime) + # FMI HELPER METHODS (3.0) fmi3_fmu_kind_enu_t fmi3_import_get_fmu_kind(fmi3_import_t*) char* fmi3_fmu_kind_to_string(fmi3_fmu_kind_enu_t) diff --git a/tests/conftest.py b/tests/conftest.py index 94abb741..eee7a001 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,6 +7,9 @@ from typing import Callable import pytest +from pyfmi.fmi3 import FMUModelCS3 +from tests.utils import get_fmi3_reference_fmu + files_directory = Path(__file__).parent / 'files' @pytest.fixture(autouse=True, scope="session") @@ -96,3 +99,44 @@ def setup_test_fmus(): zip_file_name = 'FMUs.zip', unzip_to = files_directory / 'test_fmus' ) + + +@pytest.fixture(params = [ + "BouncingBall", + "Dahlquist", + "Resource", + "StateSpace", + "Feedthrough", + "Stair", + "VanDerPol", +]) +def fmi3_cs_reference_fmu(request) -> FMUModelCS3: + fmu_name = request.param + return get_fmi3_reference_fmu(fmu_name, model_class = FMUModelCS3) + +@pytest.fixture(params = [ + "BouncingBall", + "Dahlquist", + "Resource", + "Feedthrough", + "VanDerPol", +]) +def fmi3_cs_reference_fmu_non_terminating(request) -> FMUModelCS3: + fmu_name = request.param + return get_fmi3_reference_fmu(fmu_name, model_class = FMUModelCS3) + +@pytest.fixture +def fmi3_cs_vanderpol() -> FMUModelCS3: + return get_fmi3_reference_fmu("VanDerPol", model_class = FMUModelCS3) + +@pytest.fixture +def fmi3_cs_bouncingball() -> FMUModelCS3: + return get_fmi3_reference_fmu("BouncingBall", model_class = FMUModelCS3) + +@pytest.fixture +def fmi3_cs_stair() -> FMUModelCS3: + return get_fmi3_reference_fmu("Stair", model_class = FMUModelCS3) + +@pytest.fixture +def fmi3_cs_feedthrough() -> FMUModelCS3: + return get_fmi3_reference_fmu("Feedthrough", model_class = FMUModelCS3) diff --git a/tests/test_fmi3.py b/tests/test_fmi3.py index c382ad27..739123fc 100644 --- a/tests/test_fmi3.py +++ b/tests/test_fmi3.py @@ -20,7 +20,6 @@ from io import StringIO from pathlib import Path import contextlib -import functools import pytest import numpy as np @@ -30,6 +29,7 @@ from pyfmi.fmi import ( FMUModelME3, FMUModelCS3, + FMI_OK, ) from pyfmi.fmi3 import ( FMI3_Type, @@ -45,42 +45,11 @@ InvalidVersionException ) +from tests.utils import _get_fmu, get_fmi3_reference_fmu + this_dir = Path(__file__).parent FMI3_REF_FMU_PATH = Path(this_dir) / 'files' / 'reference_fmus' / '3.0' -# possibly move to some util function and use more widely for all PyFMI testing -@functools.cache -def _fmu_cached(fmu_path, model_class = FMUModelME3, allow_unzipped_fmu = False, _connect_dll = True, **kwargs): - return model_class( - fmu = fmu_path, - allow_unzipped_fmu = allow_unzipped_fmu, - _connect_dll = _connect_dll, - **kwargs - ) - -def _get_fmu(fmu_path, model_class = FMUModelME3, allow_unzipped_fmu = False, _connect_dll = True, **kwargs): - fmu = _fmu_cached( - fmu_path = fmu_path, - model_class = model_class, - allow_unzipped_fmu = allow_unzipped_fmu, - _connect_dll = _connect_dll, - **kwargs - ) - if _connect_dll: - fmu.instantiate() - fmu.reset() - return fmu - -def get_fmi3_reference_fmu(name, model_class = FMUModelME3, allow_unzipped_fmu = False, _connect_dll = True, **kwargs): - fmu_path = FMI3_REF_FMU_PATH / (name + ".fmu") - return _get_fmu( - fmu_path = fmu_path, - model_class = model_class, - allow_unzipped_fmu = allow_unzipped_fmu, - _connect_dll = _connect_dll, - **kwargs - ) - # TODO: A lot of the tests here could be parameterized with the tests in test_fmi.py # This would however require one of the following: # a) Changing the tests in test_fmi.py to use the FMI1/2 reference FMUs @@ -1541,27 +1510,6 @@ def test_set_enum_case_sensitivity(self): fmu.set_enum(["Enumeration_input"], ["option 1"]) assert "not in the list of allowed enumeration items" in str(exc_info.value) -@pytest.fixture(params = [ - "BouncingBall", - "Dahlquist", - "Resource", - "StateSpace", - "Feedthrough", - "Stair", - "VanDerPol", -]) -def fmi3_cs_reference_fmu(request) -> FMUModelCS3: - fmu_name = request.param - return get_fmi3_reference_fmu(fmu_name, model_class = FMUModelCS3) - -@pytest.fixture -def fmi3_cs_vanderpol() -> FMUModelCS3: - return get_fmi3_reference_fmu("VanDerPol", model_class = FMUModelCS3) - -@pytest.fixture -def fmi3_cs_bouncingball() -> FMUModelCS3: - return get_fmi3_reference_fmu("BouncingBall", model_class = FMUModelCS3) - class Test_FMI3CS: """Basic unit tests for FMI3 import directly via the FMUModelCS3 class.""" @@ -1668,6 +1616,21 @@ def test_free_instance_after_initialization(self, fmi3_cs_vanderpol): fmi3_cs_vanderpol.initialize() fmi3_cs_vanderpol.free_instance() + def test_do_step(self, fmi3_cs_vanderpol): + """Test basic call to doStep().""" + fmi3_cs_vanderpol.initialize() + assert fmi3_cs_vanderpol.do_step(0, 1) == FMI_OK + + def test_do_step_terminated_resets(self, fmi3_cs_stair): + """Test a basic FMU that invokes terminate.""" + fmi3_cs_stair.initialize() + assert fmi3_cs_stair.do_step(0, 20) == FMI_OK + assert fmi3_cs_stair.time == pytest.approx(9) + assert fmi3_cs_stair.do_step_terminated + fmi3_cs_stair.reset() + assert not fmi3_cs_stair.do_step_terminated + + class TestFMI3SE: # TODO: Unsupported for now pass diff --git a/tests/test_fmi3_sim.py b/tests/test_fmi3_sim.py index 274fb2a9..4ae877ad 100644 --- a/tests/test_fmi3_sim.py +++ b/tests/test_fmi3_sim.py @@ -28,6 +28,7 @@ from pyfmi.exceptions import FMUException this_dir = Path(__file__).parent +FMI2_REF_FMU_PATH = Path(this_dir) / 'files' / 'reference_fmus' / '2.0' FMI3_REF_FMU_PATH = Path(this_dir) / 'files' / 'reference_fmus' / '3.0' class TestSimulationME: @@ -309,8 +310,75 @@ def test_euler_with_interpolation(self): assert len(res["time"]) == 101 + class TestSimulationCS: - pass + # Reference FMUs that can be simulated as CS FMUs + def test_simulate(self, fmi3_cs_vanderpol): + """Test simulate VDP model and verify the integrity of the results. """ + results = fmi3_cs_vanderpol.simulate() + + assert results['x0'][0] == 2.0 + assert results['x1'][0] == 0.0 + assert results['x0'][-1] == pytest.approx(2.0148418861546133) + assert results['x1'][-1] == pytest.approx(0.24419470751904407) + np.testing.assert_equal(results['mu'], np.ones(len(results['x0']))) + + def test_simulate_reference_fmus(self, fmi3_cs_reference_fmu_non_terminating): + """Test that the relevant reference FMUs simulate as Co-simulation. """ + fmu = fmi3_cs_reference_fmu_non_terminating + results = fmu.simulate() + # The result should at least cover the default experiment interval. + assert results['time'][0] == fmu.get_default_experiment_start_time() + assert results['time'][-1] == pytest.approx(fmu.get_default_experiment_stop_time()) + + @pytest.mark.parametrize("ref_fmu", ["VanDerPol", "Dahlquist", "BouncingBall", "Feedthrough", "Resource"]) + def test_simulate_identical_to_fmi2(self, ref_fmu, tmp_path): + """Test that CS simulation results are numerically identical to FMI2. """ + # Distinct result files, otherwise the (lazy) binary result readers + # collide since both versions share the same model name. + res2 = load_fmu(FMI2_REF_FMU_PATH / (ref_fmu + ".fmu"), kind = "CS").simulate( + options = {"result_handling": "binary", + "result_file_name": str(tmp_path / f"{ref_fmu}_fmi2.mat")}) + res3 = load_fmu(FMI3_REF_FMU_PATH / (ref_fmu + ".fmu"), kind = "CS").simulate( + options = {"result_handling": "binary", + "result_file_name": str(tmp_path / f"{ref_fmu}_fmi3.mat")}) + + # All variables the two versions have in common should match exactly. + common_variables = set(res2.keys()) & set(res3.keys()) + assert "time" in common_variables + for var in common_variables: + np.testing.assert_array_equal( + np.asarray(res3[var]), np.asarray(res2[var]), + err_msg = f"Mismatch between FMI3 and FMI2 for variable '{var}'") + + @pytest.mark.parametrize("result_handling", ["binary", "csv"]) + def test_simulate_result_handlers(self, result_handling, fmi3_cs_feedthrough): + """Test CS simulation with the supported result handlers.""" + fmi3_cs_feedthrough.set("Float64_continuous_input", 3.14) + res = fmi3_cs_feedthrough.simulate(options = {"ncp": 2, + "result_handling": result_handling}) + assert all(v == 3.14 for v in res["Float64_continuous_output"]) + + def test_simulate_result_handler_none(self, fmi3_cs_feedthrough): + """Test CS simulation with result handling disabled. """ + # With result_handling = None no results are stored, but the + # simulation should still run through without raising. + fmi3_cs_feedthrough.simulate(options = {"result_handling": None}) + + @pytest.mark.parametrize("result_handling", ["file", "memory"]) + def test_simulate_unsupported_result_handler(self, result_handling, fmi3_cs_feedthrough): + """Verify unsupported result handlers raise an exception for CS FMUs. """ + msg = f"For FMI3: 'result_handling' set to '{result_handling}' is not supported. " + \ + "Consider setting this option to 'binary', 'custom' or None to continue." + with pytest.raises(NotImplementedError, match = msg): + fmi3_cs_feedthrough.simulate(options = {"result_handling": result_handling}) + + def test_stair_reference_fmu(self, fmi3_cs_stair): + """Stair reference CS FMU, contains terminate usage.""" + res = fmi3_cs_stair.simulate(0, 20) + assert res["time"][-1] == pytest.approx(9) + assert fmi3_cs_stair.do_step_terminated + assert fmi3_cs_stair.time == pytest.approx(9) class TestDynamicDiagnostics: diff --git a/tests/utils.py b/tests/utils.py new file mode 100644 index 00000000..88ae74c8 --- /dev/null +++ b/tests/utils.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +# Copyright (C) 2026 Modelon AB +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, version 3 of the License. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import functools +from pathlib import Path + +from pyfmi.fmi3 import FMUModelME3 + +this_dir = Path(__file__).parent +FMI3_REF_FMU_PATH = Path(this_dir) / 'files' / 'reference_fmus' / '3.0' + +# possibly move to some util function and use more widely for all PyFMI testing +@functools.cache +def _fmu_cached(fmu_path, model_class = FMUModelME3, allow_unzipped_fmu = False, _connect_dll = True, **kwargs): + return model_class( + fmu = fmu_path, + allow_unzipped_fmu = allow_unzipped_fmu, + _connect_dll = _connect_dll, + **kwargs + ) + +def _get_fmu(fmu_path, model_class = FMUModelME3, allow_unzipped_fmu = False, _connect_dll = True, **kwargs): + fmu = _fmu_cached( + fmu_path = fmu_path, + model_class = model_class, + allow_unzipped_fmu = allow_unzipped_fmu, + _connect_dll = _connect_dll, + **kwargs + ) + if _connect_dll: + fmu.free_instance() + fmu.instantiate() + fmu.reset() + return fmu + +def get_fmi3_reference_fmu(name, model_class = FMUModelME3, allow_unzipped_fmu = False, _connect_dll = True, **kwargs): + fmu_path = FMI3_REF_FMU_PATH / (name + ".fmu") + return _get_fmu( + fmu_path = fmu_path, + model_class = model_class, + allow_unzipped_fmu = allow_unzipped_fmu, + _connect_dll = _connect_dll, + **kwargs + )