Skip to content

Simulation API

Simulation

A Facade for running a simulation from a defined Model.

Source code in commol/api/simulation.py 
 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
134
135
class Simulation:
    """
    A Facade for running a simulation from a defined Model.
    """

    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()

        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)

    @property
    def engine(self) -> "DifferenceEquationsProtocol":
        """Get the underlying simulation engine."""
        return self._engine

Attributes

engine property 

engine: DifferenceEquationsProtocol

Get the underlying simulation engine.

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 commol/api/simulation.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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 commol/api/simulation.py 
 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
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