Skip to content

Simulation API

Simulation

A Facade for running a simulation from a defined Model.

This class hides the complexity of initializing the Rust engine and provides high-level methods to run simulations and retrieve results in convenient formats.

Source code in epimodel/api/simulation.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class Simulation:
    """
    A Facade for running a simulation from a defined Model.

    This class hides the complexity of initializing the Rust engine and provides
    high-level methods to run simulations and retrieve results in convenient formats.
    """

    def __init__(self, model: Model):
        """
        Initializes the simulation engine from a Pydantic Model definition.

        Parameters
        ----------
        model : Model
            A fully constructed and validated model object.
        """
        logging.info(f"Initializing Simulation with model: '{model.name}'")
        self.model_definition: Model = model
        self._engine: "DifferenceEquationsProtocol" = self._initialize_engine()

        self._compartments: list[str] = self._engine.compartments
        logging.info(
            f"Simulation engine ready. Total compartments: {len(self._compartments)}"
        )

    def _initialize_engine(self) -> "DifferenceEquationsProtocol":
        """Internal method to set up the Rust backend."""
        logging.info("Preparing model definition for Rust serialization...")
        model_json = self.model_definition.model_dump_json()

        try:
            from epimodel.epimodel_rs.epimodel_rs import core, difference
        except ImportError as e:
            raise ImportError(
                "Rust extension not available. Please compile the project."
            ) from e

        rust_model_instance: "RustModelProtocol" = core.Model.from_json(model_json)
        logging.info("Rust model instance created from JSON.")

        # This could be extended if you have more engine types
        if self.model_definition.dynamics.typology == ModelTypes.DIFFERENCE_EQUATIONS:
            logging.info("Initializing DifferenceEquations engine.")
            return difference.DifferenceEquations(rust_model_instance)

        raise NotImplementedError(
            (
                f"Engine for typology '{self.model_definition.dynamics.typology}' "
                f"not implemented."
            )
        )

    def _run_raw(self, num_steps: int) -> list[list[float]]:
        """
        Runs the simulation and returns the raw, high-performance output.
        This is the fastest method, returning a list of lists of floats.
        """
        logging.info(f"Running raw simulation for {num_steps} steps.")
        start = time.time()
        results = self._engine.run(num_steps)
        end = time.time()
        logging.info(f"Raw simulation complete. It tool {end - start} seconds.")
        return results

    @overload
    def run(
        self, num_steps: int, output_format: Literal["list_of_lists"]
    ) -> list[list[float]]: ...
    @overload
    def run(
        self, num_steps: int, output_format: Literal["dict_of_lists"]
    ) -> dict[str, list[float]]: ...
    @overload
    def run(self, num_steps: int) -> dict[str, list[float]]: ...
    def run(
        self,
        num_steps: int,
        output_format: Literal["dict_of_lists", "list_of_lists"] = "dict_of_lists",
    ) -> dict[str, list[float]] | list[list[float]]:
        """
        Runs the simulation and returns the output in the specified format.

        Parameters
        ----------
        num_steps : int
            The number of steps for the simulation.
        output_format : {'dict_of_lists', 'list_of_lists'}, default 'dict_of_lists'
            - 'dict_of_lists': Returns a dictionary of lists, with compartment names
                as keys.
            - 'list_of_lists': Returns a list of lists of floats, where the first level
                is the step and the second level the comptarment.

        Returns
        -------
        dict[str, list[float]] | list[list[float]]
            The simulation results in the specified format.
        """
        raw_results = self._run_raw(num_steps)
        if output_format == "list_of_lists":
            logging.info("Returning results in 'list_of_lists' format.")
            return raw_results

        elif output_format == "dict_of_lists":
            logging.info("Transposing raw results to 'dict_of_lists' format.")
            if not raw_results:
                return {c: [] for c in self._compartments}
            transposed_results = zip(*raw_results)
            return {
                compartment: list(values)
                for compartment, values in zip(self._compartments, transposed_results)
            }

        else:
            assert_never(output_format)

Functions

__init__ 

__init__(model: Model)

Initializes the simulation engine from a Pydantic Model definition.

Parameters:

Name Type Description Default
model Model

A fully constructed and validated model object.

 required
Source code in epimodel/api/simulation.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, model: Model):
    """
    Initializes the simulation engine from a Pydantic Model definition.

    Parameters
    ----------
    model : Model
        A fully constructed and validated model object.
    """
    logging.info(f"Initializing Simulation with model: '{model.name}'")
    self.model_definition: Model = model
    self._engine: "DifferenceEquationsProtocol" = self._initialize_engine()

    self._compartments: list[str] = self._engine.compartments
    logging.info(
        f"Simulation engine ready. Total compartments: {len(self._compartments)}"
    )

run 

run(
    num_steps: int, output_format: Literal["list_of_lists"]
) -> list[list[float]]

run(
    num_steps: int, output_format: Literal["dict_of_lists"]
) -> dict[str, list[float]]

run(num_steps: int) -> dict[str, list[float]]

run(
    num_steps: int,
    output_format: Literal[
        "dict_of_lists", "list_of_lists"
    ] = "dict_of_lists",
) -> dict[str, list[float]] | list[list[float]]

Runs the simulation and returns the output in the specified format.

Parameters:

Name Type Description Default
num_steps int

The number of steps for the simulation.

 required
output_format (dict_of_lists, list_of_lists)
  • 'dict_of_lists': Returns a dictionary of lists, with compartment names as keys.
  • 'list_of_lists': Returns a list of lists of floats, where the first level is the step and the second level the comptarment.
 'dict_of_lists'

Returns:

Type Description
dict[str, list[float]] | list[list[float]]

The simulation results in the specified format.
Source code in epimodel/api/simulation.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def run(
    self,
    num_steps: int,
    output_format: Literal["dict_of_lists", "list_of_lists"] = "dict_of_lists",
) -> dict[str, list[float]] | list[list[float]]:
    """
    Runs the simulation and returns the output in the specified format.

    Parameters
    ----------
    num_steps : int
        The number of steps for the simulation.
    output_format : {'dict_of_lists', 'list_of_lists'}, default 'dict_of_lists'
        - 'dict_of_lists': Returns a dictionary of lists, with compartment names
            as keys.
        - 'list_of_lists': Returns a list of lists of floats, where the first level
            is the step and the second level the comptarment.

    Returns
    -------
    dict[str, list[float]] | list[list[float]]
        The simulation results in the specified format.
    """
    raw_results = self._run_raw(num_steps)
    if output_format == "list_of_lists":
        logging.info("Returning results in 'list_of_lists' format.")
        return raw_results

    elif output_format == "dict_of_lists":
        logging.info("Transposing raw results to 'dict_of_lists' format.")
        if not raw_results:
            return {c: [] for c in self._compartments}
        transposed_results = zip(*raw_results)
        return {
            compartment: list(values)
            for compartment, values in zip(self._compartments, transposed_results)
        }

    else:
        assert_never(output_format)

options: show_root_heading: true show_source: true heading_level: 2 show_docstring_attributes: false