From dd692e498dd165758f140ab21d4f2c75af1e0c82 Mon Sep 17 00:00:00 2001 From: petermeisrimelmodelon Date: Fri, 10 Jul 2026 13:58:28 +0000 Subject: [PATCH 1/5] more minor fixes for basic doStep implementation --- src/pyfmi/fmi3.pxd | 2 + src/pyfmi/fmi3.pyx | 175 +++++++++++++++++++++++++++++ src/pyfmi/fmi_algorithm_drivers.py | 8 +- src/pyfmi/fmil3_import.pxd | 11 ++ tests/test_fmi3.py | 8 ++ tests/test_fmi3_sim.py | 81 ++++++++++++- 6 files changed, 280 insertions(+), 5 deletions(-) diff --git a/src/pyfmi/fmi3.pxd b/src/pyfmi/fmi3.pxd index 2ed1b632..5dd02eeb 100644 --- a/src/pyfmi/fmi3.pxd +++ b/src/pyfmi/fmi3.pxd @@ -186,6 +186,8 @@ cdef class FMUModelCS3(FMUModelBase3): 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..7a91c69d 100644 --- a/src/pyfmi/fmi3.pyx +++ b/src/pyfmi/fmi3.pyx @@ -3888,6 +3888,168 @@ 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,FMI_PENDING... + + Calls the underlying low-level function fmi3DoStep. + """ + # TODO: Status docstring + 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() + + # On a fully completed step the reached time is current_t + step_size; + # lastSuccessfulTime is only meaningful when the FMU returns early. + if 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 +4096,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..84da0b78 100644 --- a/src/pyfmi/fmi_algorithm_drivers.py +++ b/src/pyfmi/fmi_algorithm_drivers.py @@ -26,7 +26,7 @@ from pyfmi.fmi1 import FMUModelME1, FMUModelCS1, 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") 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/test_fmi3.py b/tests/test_fmi3.py index c382ad27..07be2a1b 100644 --- a/tests/test_fmi3.py +++ b/tests/test_fmi3.py @@ -1668,6 +1668,14 @@ def test_free_instance_after_initialization(self, fmi3_cs_vanderpol): fmi3_cs_vanderpol.initialize() fmi3_cs_vanderpol.free_instance() + def test_do_step(self): + """Test basic call to doStep().""" + fmu_path = FMI3_REF_FMU_PATH / "VanDerPol.fmu" + fmu = FMUModelCS3(fmu_path) + fmu.initialize() + + fmu.do_step(0, 1) + class TestFMI3SE: # TODO: Unsupported for now pass diff --git a/tests/test_fmi3_sim.py b/tests/test_fmi3_sim.py index 274fb2a9..9c170765 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: @@ -310,8 +311,86 @@ def test_euler_with_interpolation(self): class TestSimulationCS: - pass + # Reference FMUs that can be simulated as CS FMUs + # 'Stair' is intentionally excluded, see test_simulate_stair_not_supported. + SIMULATABLE_REFERENCE_FMUS = ["VanDerPol", "Dahlquist", "BouncingBall", "Feedthrough", "Resource"] + def test_simulate(self): + """Test simulate VDP model and verify the integrity of the results. """ + fmu = load_fmu(FMI3_REF_FMU_PATH / "VanDerPol.fmu", kind = "CS") + results = fmu.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']))) + + @pytest.mark.parametrize("ref_fmu", SIMULATABLE_REFERENCE_FMUS) + def test_simulate_reference_fmus(self, ref_fmu): + """Test that the relevant reference FMUs simulate as Co-simulation. """ + fmu = load_fmu(FMI3_REF_FMU_PATH / (ref_fmu + ".fmu"), kind = "CS") + 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", SIMULATABLE_REFERENCE_FMUS) + 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): + """Test CS simulation with the supported result handlers. """ + fmu = load_fmu(FMI3_REF_FMU_PATH / "Feedthrough.fmu", kind = "CS") + fmu.set("Float64_continuous_input", 3.14) + res = fmu.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): + """Test CS simulation with result handling disabled. """ + fmu = load_fmu(FMI3_REF_FMU_PATH / "VanDerPol.fmu", kind = "CS") + # With result_handling = None no results are stored, but the + # simulation should still run through without raising. + fmu.simulate(options = {"result_handling": None}) + + @pytest.mark.parametrize("result_handling", ["file", "memory"]) + def test_simulate_unsupported_result_handler(self, result_handling): + """Verify unsupported result handlers raise an exception for CS FMUs. """ + fmu = load_fmu(FMI3_REF_FMU_PATH / "VanDerPol.fmu", kind = "CS") + 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): + fmu.simulate(options = {"result_handling": result_handling}) + + def test_simulate_stair_not_supported(self): + """The Stair reference FMU cannot yet be simulated as Co-simulation. + + Its doStep returns FMI_ERROR once the internal counter reaches its + maximum. Handling this gracefully (as FMI2 does via DISCARD and the + last successful time) requires additional master algorithm support, + which is out of scope for the current basic doStep implementation. + """ + fmu = load_fmu(FMI3_REF_FMU_PATH / "Stair.fmu", kind = "CS") + with pytest.raises(FMUException, match = "The simulation failed"): + fmu.simulate() class TestDynamicDiagnostics: """Tests involving simulation of FMI3 FMUs using 'dynamic_diagnostics' == True.""" From 88f55f2892c102121603467a4162edfc51932a7b Mon Sep 17 00:00:00 2001 From: petermeisrimelmodelon Date: Mon, 13 Jul 2026 11:54:35 +0000 Subject: [PATCH 2/5] cleanup --- src/pyfmi/fmi2.pyx | 4 ++-- src/pyfmi/fmi3.pxd | 1 + src/pyfmi/fmi3.pyx | 7 +++---- tests/test_fmi3.py | 3 ++- tests/test_fmi3_sim.py | 8 +------- 5 files changed, 9 insertions(+), 14 deletions(-) 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 5dd02eeb..19396537 100644 --- a/src/pyfmi/fmi3.pxd +++ b/src/pyfmi/fmi3.pxd @@ -183,6 +183,7 @@ 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 FMIL3.fmi3_boolean_t _instantiated_with_early_return cpdef _get_time(self) cpdef _set_time(self, FMIL3.fmi3_float64_t t) diff --git a/src/pyfmi/fmi3.pyx b/src/pyfmi/fmi3.pyx index 7a91c69d..eb8697ea 100644 --- a/src/pyfmi/fmi3.pyx +++ b/src/pyfmi/fmi3.pyx @@ -3861,6 +3861,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): @@ -3909,12 +3910,10 @@ cdef class FMUModelCS3(FMUModelBase3): 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 Calls the underlying low-level function fmi3DoStep. """ - # TODO: Status docstring cdef FMIL3.fmi3_status_t status cdef FMIL3.fmi3_boolean_t new_s cdef FMIL3.fmi3_boolean_t eventHandlingNeeded @@ -3949,7 +3948,7 @@ cdef class FMUModelCS3(FMUModelBase3): # On a fully completed step the reached time is current_t + step_size; # lastSuccessfulTime is only meaningful when the FMU returns early. - if earlyReturn: + if self._instantiated_with_early_return and earlyReturn: self.time = lastSuccessfulTime else: self.time = current_t + step_size diff --git a/tests/test_fmi3.py b/tests/test_fmi3.py index 07be2a1b..77c45d4e 100644 --- a/tests/test_fmi3.py +++ b/tests/test_fmi3.py @@ -30,6 +30,7 @@ from pyfmi.fmi import ( FMUModelME3, FMUModelCS3, + FMI_OK, ) from pyfmi.fmi3 import ( FMI3_Type, @@ -1674,7 +1675,7 @@ def test_do_step(self): fmu = FMUModelCS3(fmu_path) fmu.initialize() - fmu.do_step(0, 1) + assert fmu.do_step(0, 1) == FMI_OK class TestFMI3SE: # TODO: Unsupported for now diff --git a/tests/test_fmi3_sim.py b/tests/test_fmi3_sim.py index 9c170765..7a6a527b 100644 --- a/tests/test_fmi3_sim.py +++ b/tests/test_fmi3_sim.py @@ -381,13 +381,7 @@ def test_simulate_unsupported_result_handler(self, result_handling): fmu.simulate(options = {"result_handling": result_handling}) def test_simulate_stair_not_supported(self): - """The Stair reference FMU cannot yet be simulated as Co-simulation. - - Its doStep returns FMI_ERROR once the internal counter reaches its - maximum. Handling this gracefully (as FMI2 does via DISCARD and the - last successful time) requires additional master algorithm support, - which is out of scope for the current basic doStep implementation. - """ + """Stair reference FMU requires support for terminate with CS FMUs.""" fmu = load_fmu(FMI3_REF_FMU_PATH / "Stair.fmu", kind = "CS") with pytest.raises(FMUException, match = "The simulation failed"): fmu.simulate() From 35f8c07e70ec32bc06ee6181935fa65b5c1e8792 Mon Sep 17 00:00:00 2001 From: petermeisrimelmodelon Date: Thu, 6 Aug 2026 14:17:05 +0000 Subject: [PATCH 3/5] adding support for terminate --- src/pyfmi/fmi3.pxd | 1 + src/pyfmi/fmi3.pyx | 13 ++++++++++++- src/pyfmi/fmi_algorithm_drivers.py | 9 +++++++++ tests/test_fmi3.py | 10 ++++++++++ tests/test_fmi3_sim.py | 17 +++++++++-------- 5 files changed, 41 insertions(+), 9 deletions(-) diff --git a/src/pyfmi/fmi3.pxd b/src/pyfmi/fmi3.pxd index 19396537..088bdb17 100644 --- a/src/pyfmi/fmi3.pxd +++ b/src/pyfmi/fmi3.pxd @@ -183,6 +183,7 @@ 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) diff --git a/src/pyfmi/fmi3.pyx b/src/pyfmi/fmi3.pyx index eb8697ea..e097e6d1 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 @@ -3946,9 +3952,14 @@ cdef class FMUModelCS3(FMUModelBase3): 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 self._instantiated_with_early_return and earlyReturn: + 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 diff --git a/src/pyfmi/fmi_algorithm_drivers.py b/src/pyfmi/fmi_algorithm_drivers.py index 84da0b78..463751b7 100644 --- a/src/pyfmi/fmi_algorithm_drivers.py +++ b/src/pyfmi/fmi_algorithm_drivers.py @@ -1045,6 +1045,15 @@ def solve(self): status = self.model.do_step(t,h) self.status = status + if isinstance(self.model, FMUModelCS3): + if self.model.do_step_terminated: + final_time = self.model.time + + start_time_point = timer() + result_handler.integration_point() + self.timings["storing_result"] += timer() - start_time_point + break + if status != 0: if status == FMI_ERROR: diff --git a/tests/test_fmi3.py b/tests/test_fmi3.py index 77c45d4e..87652f9c 100644 --- a/tests/test_fmi3.py +++ b/tests/test_fmi3.py @@ -1677,6 +1677,16 @@ def test_do_step(self): assert fmu.do_step(0, 1) == FMI_OK + def test_do_step_terminated_resets(self): + fmu = load_fmu(FMI3_REF_FMU_PATH / "Stair.fmu", kind = "CS") + fmu.initialize() + assert fmu.do_step(0, 20) == FMI_OK + assert fmu.time == pytest.approx(9) + assert fmu.do_step_terminated + fmu.reset() + assert not fmu.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 7a6a527b..a6f92af6 100644 --- a/tests/test_fmi3_sim.py +++ b/tests/test_fmi3_sim.py @@ -312,8 +312,7 @@ def test_euler_with_interpolation(self): class TestSimulationCS: # Reference FMUs that can be simulated as CS FMUs - # 'Stair' is intentionally excluded, see test_simulate_stair_not_supported. - SIMULATABLE_REFERENCE_FMUS = ["VanDerPol", "Dahlquist", "BouncingBall", "Feedthrough", "Resource"] + REFERENCE_FMUS = ["VanDerPol", "Dahlquist", "BouncingBall", "Feedthrough", "Resource"] def test_simulate(self): """Test simulate VDP model and verify the integrity of the results. """ @@ -326,7 +325,7 @@ def test_simulate(self): assert results['x1'][-1] == pytest.approx(0.24419470751904407) np.testing.assert_equal(results['mu'], np.ones(len(results['x0']))) - @pytest.mark.parametrize("ref_fmu", SIMULATABLE_REFERENCE_FMUS) + @pytest.mark.parametrize("ref_fmu", REFERENCE_FMUS) def test_simulate_reference_fmus(self, ref_fmu): """Test that the relevant reference FMUs simulate as Co-simulation. """ fmu = load_fmu(FMI3_REF_FMU_PATH / (ref_fmu + ".fmu"), kind = "CS") @@ -335,7 +334,7 @@ def test_simulate_reference_fmus(self, ref_fmu): 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", SIMULATABLE_REFERENCE_FMUS) + @pytest.mark.parametrize("ref_fmu", REFERENCE_FMUS) 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 @@ -380,11 +379,13 @@ def test_simulate_unsupported_result_handler(self, result_handling): with pytest.raises(NotImplementedError, match = msg): fmu.simulate(options = {"result_handling": result_handling}) - def test_simulate_stair_not_supported(self): - """Stair reference FMU requires support for terminate with CS FMUs.""" + def test_stair_reference_fmu(self): + """Stair reference CS FMU, contains terminate usage.""" fmu = load_fmu(FMI3_REF_FMU_PATH / "Stair.fmu", kind = "CS") - with pytest.raises(FMUException, match = "The simulation failed"): - fmu.simulate() + res = fmu.simulate(0, 20) + assert res["time"][-1] == pytest.approx(9) + assert fmu.do_step_terminated + assert fmu.time == pytest.approx(9) class TestDynamicDiagnostics: """Tests involving simulation of FMI3 FMUs using 'dynamic_diagnostics' == True.""" From aee97c12c0472269879adc78b605472e013ff76e Mon Sep 17 00:00:00 2001 From: petermeisrimelmodelon Date: Mon, 10 Aug 2026 13:37:04 +0000 Subject: [PATCH 4/5] More test cleanup; re-using fixtures --- tests/conftest.py | 44 ++++++++++++++++++++++ tests/test_fmi3.py | 84 +++++++----------------------------------- tests/test_fmi3_sim.py | 44 ++++++++++------------ tests/utils.py | 58 +++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 95 deletions(-) create mode 100644 tests/utils.py 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 87652f9c..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 @@ -46,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 @@ -1542,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.""" @@ -1669,22 +1616,19 @@ def test_free_instance_after_initialization(self, fmi3_cs_vanderpol): fmi3_cs_vanderpol.initialize() fmi3_cs_vanderpol.free_instance() - def test_do_step(self): + def test_do_step(self, fmi3_cs_vanderpol): """Test basic call to doStep().""" - fmu_path = FMI3_REF_FMU_PATH / "VanDerPol.fmu" - fmu = FMUModelCS3(fmu_path) - fmu.initialize() - - assert fmu.do_step(0, 1) == FMI_OK - - def test_do_step_terminated_resets(self): - fmu = load_fmu(FMI3_REF_FMU_PATH / "Stair.fmu", kind = "CS") - fmu.initialize() - assert fmu.do_step(0, 20) == FMI_OK - assert fmu.time == pytest.approx(9) - assert fmu.do_step_terminated - fmu.reset() - assert not fmu.do_step_terminated + 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: diff --git a/tests/test_fmi3_sim.py b/tests/test_fmi3_sim.py index a6f92af6..4ae877ad 100644 --- a/tests/test_fmi3_sim.py +++ b/tests/test_fmi3_sim.py @@ -310,14 +310,12 @@ def test_euler_with_interpolation(self): assert len(res["time"]) == 101 + class TestSimulationCS: # Reference FMUs that can be simulated as CS FMUs - REFERENCE_FMUS = ["VanDerPol", "Dahlquist", "BouncingBall", "Feedthrough", "Resource"] - - def test_simulate(self): + def test_simulate(self, fmi3_cs_vanderpol): """Test simulate VDP model and verify the integrity of the results. """ - fmu = load_fmu(FMI3_REF_FMU_PATH / "VanDerPol.fmu", kind = "CS") - results = fmu.simulate() + results = fmi3_cs_vanderpol.simulate() assert results['x0'][0] == 2.0 assert results['x1'][0] == 0.0 @@ -325,16 +323,15 @@ def test_simulate(self): assert results['x1'][-1] == pytest.approx(0.24419470751904407) np.testing.assert_equal(results['mu'], np.ones(len(results['x0']))) - @pytest.mark.parametrize("ref_fmu", REFERENCE_FMUS) - def test_simulate_reference_fmus(self, ref_fmu): + def test_simulate_reference_fmus(self, fmi3_cs_reference_fmu_non_terminating): """Test that the relevant reference FMUs simulate as Co-simulation. """ - fmu = load_fmu(FMI3_REF_FMU_PATH / (ref_fmu + ".fmu"), kind = "CS") + 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", REFERENCE_FMUS) + @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 @@ -355,37 +352,34 @@ def test_simulate_identical_to_fmi2(self, ref_fmu, tmp_path): 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): - """Test CS simulation with the supported result handlers. """ - fmu = load_fmu(FMI3_REF_FMU_PATH / "Feedthrough.fmu", kind = "CS") - fmu.set("Float64_continuous_input", 3.14) - res = fmu.simulate(options = {"ncp": 2, + 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): + def test_simulate_result_handler_none(self, fmi3_cs_feedthrough): """Test CS simulation with result handling disabled. """ - fmu = load_fmu(FMI3_REF_FMU_PATH / "VanDerPol.fmu", kind = "CS") # With result_handling = None no results are stored, but the # simulation should still run through without raising. - fmu.simulate(options = {"result_handling": None}) + fmi3_cs_feedthrough.simulate(options = {"result_handling": None}) @pytest.mark.parametrize("result_handling", ["file", "memory"]) - def test_simulate_unsupported_result_handler(self, result_handling): + def test_simulate_unsupported_result_handler(self, result_handling, fmi3_cs_feedthrough): """Verify unsupported result handlers raise an exception for CS FMUs. """ - fmu = load_fmu(FMI3_REF_FMU_PATH / "VanDerPol.fmu", kind = "CS") 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): - fmu.simulate(options = {"result_handling": result_handling}) + fmi3_cs_feedthrough.simulate(options = {"result_handling": result_handling}) - def test_stair_reference_fmu(self): + def test_stair_reference_fmu(self, fmi3_cs_stair): """Stair reference CS FMU, contains terminate usage.""" - fmu = load_fmu(FMI3_REF_FMU_PATH / "Stair.fmu", kind = "CS") - res = fmu.simulate(0, 20) + res = fmi3_cs_stair.simulate(0, 20) assert res["time"][-1] == pytest.approx(9) - assert fmu.do_step_terminated - assert fmu.time == pytest.approx(9) + assert fmi3_cs_stair.do_step_terminated + assert fmi3_cs_stair.time == pytest.approx(9) + class TestDynamicDiagnostics: """Tests involving simulation of FMI3 FMUs using 'dynamic_diagnostics' == True.""" 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 + ) From 1738313abb045f7aa56da4d30aeae0a77f585160 Mon Sep 17 00:00:00 2001 From: petermeisrimelmodelon Date: Tue, 11 Aug 2026 11:42:39 +0000 Subject: [PATCH 5/5] review fixes: re-factor cs algorithm --- src/pyfmi/fmi3.pyx | 6 ++-- src/pyfmi/fmi_algorithm_drivers.py | 55 +++++++++++++++--------------- 2 files changed, 31 insertions(+), 30 deletions(-) diff --git a/src/pyfmi/fmi3.pyx b/src/pyfmi/fmi3.pyx index e097e6d1..2fff7df1 100644 --- a/src/pyfmi/fmi3.pyx +++ b/src/pyfmi/fmi3.pyx @@ -3916,9 +3916,11 @@ cdef class FMUModelCS3(FMUModelBase3): status -- The status of function which can be checked against - FMI_OK, FMI_WARNING. FMI_DISCARD, FMI_ERROR, FMI_FATAL + FMI_OK, FMI_WARNING, FMI_DISCARD, FMI_ERROR, FMI_FATAL. - Calls the underlying low-level function fmi3DoStep. + 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 diff --git a/src/pyfmi/fmi_algorithm_drivers.py b/src/pyfmi/fmi_algorithm_drivers.py index 463751b7..fb005ffb 100644 --- a/src/pyfmi/fmi_algorithm_drivers.py +++ b/src/pyfmi/fmi_algorithm_drivers.py @@ -24,7 +24,7 @@ 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, FMUModelCS3 from pyfmi.fmi_coupled import CoupledFMUModelME2 @@ -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,37 +1065,16 @@ def solve(self): status = self.model.do_step(t,h) self.status = status - if isinstance(self.model, FMUModelCS3): - if self.model.do_step_terminated: - final_time = self.model.time + 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 start_time_point = timer() result_handler.integration_point() self.timings["storing_result"] += timer() - start_time_point - break - - if status != 0: - - if status == FMI_ERROR: - raise FMUException("The simulation failed. See the log for more information. Return flag %d."%status) - - 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 + break # stop integration loop final_time = t+h