feat: basic FMI3 CS doStep implementation (#415)

Author: PeterMeisrimelModelon

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.
         """

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
 

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

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,19 +940,19 @@ 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.")
 
             time_res_init = timer()
             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, <termination_time>) 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
 

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)