Skip to content

Core

The dissmodel.core module provides the simulation clock and execution lifecycle.

All models and visualization components must be instantiated after the Environment — they register themselves automatically on creation.

Environment  →  Model  →  Visualization  →  env.run()
     ↑             ↑            ↑                ↑
  first         second        third           fourth

Usage

from dissmodel.core import Environment, Model

env = Environment(start_time=1, end_time=10)

class MyModel(Model):
    def setup(self):
        pass

    def execute(self):
        print(f"step {self.env.now()}")

MyModel()
env.run()

Object-Oriented Modeling

Object-oriented modeling is a core feature of DisSModel, inherited directly from Python's class system. Just as TerraME defines agents as objects with encapsulated attributes and behaviours, DisSModel uses class inheritance to build structured, reusable, and modular models.

Every model is a subclass of Model, which guarantees automatic registration with the active Environment. This means the simulation clock, the execution lifecycle, and any visualization components are wired together without any boilerplate.

from dissmodel.core import Model, Environment

class SIR(Model):

    def setup(self, susceptible=9998, infected=2, recovered=0,
              duration=2, contacts=6, probability=0.25):
        self.susceptible  = susceptible
        self.infected     = infected
        self.recovered    = recovered
        self.duration     = duration
        self.contacts     = contacts
        self.probability  = probability

    def execute(self):
        total       = self.susceptible + self.infected + self.recovered
        alpha       = self.contacts * self.probability
        new_inf     = self.infected * alpha * (self.susceptible / total)
        new_rec     = self.infected / self.duration
        self.susceptible -= new_inf
        self.infected    += new_inf - new_rec
        self.recovered   += new_rec

Instantiation is clean and parametric:

env = Environment(end_time=30)
SIR(susceptible=9998, infected=2, recovered=0,
    duration=2, contacts=6, probability=0.25)
env.run()

!!! tip "Why subclass Model?" - Automatic clock integration — self.env.now() is always available inside execute(). - Encapsulation — each model owns its state; multiple instances can run in the same environment independently. - Extensibility — override setup() to add parameters, execute() to define the transition rule. Nothing else is required. - Composability — models can read each other's state, enabling coupled CA + SysDyn simulations within a single env.run().

Each model can define its own start_time and end_time, independent of the environment interval. This allows different parts of a simulation to be active at different periods within the same run.

from dissmodel.core import Model, Environment

class ModelA(Model):
    def execute(self):
        print(f"[A] t={self.env.now()}")

class ModelB(Model):
    def execute(self):
        print(f"[B] t={self.env.now()}")

class ModelC(Model):
    def execute(self):
        print(f"[C] t={self.env.now()}")

env = Environment(start_time=2010, end_time=2016)

ModelA(start_time=2012)        # active from 2012 to end
ModelB(end_time=2013)          # active from start to 2013
ModelC()                       # active throughout

env.run()

Expected output:

Running from 2010 to 2016 (duration: 6)
[B] t=2010
[C] t=2010
[B] t=2011
[C] t=2011
[A] t=2012
[B] t=2012
[C] t=2012
[A] t=2013
[C] t=2013
[A] t=2014
[C] t=2014
[A] t=2015
[C] t=2015

!!! note Models with no start_time / end_time inherit the environment's interval. Models are synchronised — all active models execute at each time step before the clock advances.

API Reference

dissmodel.core.Environment

Simulation environment with support for a custom time window.

Manages the simulation clock and coordinates the execution of all registered :class:~dissmodel.core.Model instances.

Parameters:

Name Type Description Default
start_time float

Simulation start time, by default 0.

 0
end_time float

Simulation end time. Can also be set via till in :meth:run.

 None

Examples:

>>> env = Environment(start_time=0, end_time=10)
>>> env.start_time
0
>>> env.end_time
10
Source code in dissmodel/core/environment.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class Environment:
    """
    Simulation environment with support for a custom time window.

    Manages the simulation clock and coordinates the execution of all
    registered :class:`~dissmodel.core.Model` instances.

    Parameters
    ----------
    start_time : float, optional
        Simulation start time, by default 0.
    end_time : float, optional
        Simulation end time. Can also be set via ``till`` in :meth:`run`.

    Examples
    --------
    >>> env = Environment(start_time=0, end_time=10)
    >>> env.start_time
    0
    >>> env.end_time
    10
    """

    _current: ClassVar[Optional[Environment]] = None

    def __init__(
        self,
        start_time: float = 0,
        end_time: Optional[float] = None,
    ) -> None:
        self.start_time = start_time
        self.end_time = end_time
        self._now: float = start_time
        self._models: list[Any] = []
        self._plot_metadata: dict[str, Any] = {}
        Environment._current = self

    # ------------------------------------------------------------------
    # Clock
    # ------------------------------------------------------------------

    def now(self) -> float:
        """
        Return the current simulation time.

        Returns
        -------
        float
            Current simulation time.

        Examples
        --------
        >>> env = Environment(start_time=5)
        >>> env.now()
        5
        """
        return self._now

    # ------------------------------------------------------------------
    # Model registration
    # ------------------------------------------------------------------

    def _register(self, model: Any) -> None:
        """
        Register a model to be executed during the simulation.

        Called automatically by :class:`~dissmodel.core.Model.__init__`.

        Parameters
        ----------
        model : Model
            The model instance to register.
        """
        self._models.append(model)

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    def run(self, till: Optional[float] = None) -> None:
        """
        Run the simulation over the configured time window.

        Executes all registered models in time-step order. On each tick,
        every model whose next scheduled time is less than or equal to the
        current simulation time has its :meth:`~dissmodel.core.Model.execute`
        method called. The clock then advances to the nearest pending event.

        Parameters
        ----------
        till : float, optional
            Duration to run from ``start_time``. If provided, overrides
            ``end_time``. If omitted, ``end_time`` must be set.

        Raises
        ------
        ValueError
            If neither ``till`` nor ``end_time`` is defined.

        Examples
        --------
        >>> env = Environment(start_time=0, end_time=10)
        >>> env.run()
        Running from 0 to 10 (duration: 10)
        """
        self.reset()

        if till is not None:
            self.end_time = self.start_time + till
        elif self.end_time is not None:
            pass
        else:
            raise ValueError(
                "Provide 'till' or set 'end_time' before calling run()."
            )

        duration = self.end_time - self.start_time
        print(
            f"Running from {self.start_time} to {self.end_time} "
            f"(duration: {duration})"
        )

        # Initialise next-execution time for every registered model
        for model in self._models:
            model._next_time = model.start_time

        self._now = self.start_time

        while self._now < self.end_time:
            for model in self._models:
                if (
                    model._next_time <= self._now
                    and self._now < model.end_time
                ):
                    model.pre_execute()
                    model.execute()
                    model.post_execute()
                    model._next_time = self._now + model._step

            # Advance clock to the nearest pending event
            pending = [
                m._next_time
                for m in self._models
                if m._next_time < self.end_time
                and m._next_time < m.end_time
            ]
            if not pending:
                break
            self._now = min(pending)

    def reset(self) -> None:
        """
        Reset the clock and clear accumulated plot data.

        Called automatically at the start of :meth:`run` to ensure the
        environment starts fresh on each simulation run.

        Examples
        --------
        >>> env = Environment(start_time=0, end_time=10)
        >>> env._plot_metadata = {"x": {"data": [1, 2, 3]}}
        >>> env.reset()
        >>> env._plot_metadata["x"]["data"]
        []
        """
        self._now = self.start_time
        for item in self._plot_metadata.values():
            item["data"].clear()

now()

Return the current simulation time.

Returns:

Type Description
float

Current simulation time.

Examples:

>>> env = Environment(start_time=5)
>>> env.now()
5
Source code in dissmodel/core/environment.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def now(self) -> float:
    """
    Return the current simulation time.

    Returns
    -------
    float
        Current simulation time.

    Examples
    --------
    >>> env = Environment(start_time=5)
    >>> env.now()
    5
    """
    return self._now

reset()

Reset the clock and clear accumulated plot data.

Called automatically at the start of :meth:run to ensure the environment starts fresh on each simulation run.

Examples:

>>> env = Environment(start_time=0, end_time=10)
>>> env._plot_metadata = {"x": {"data": [1, 2, 3]}}
>>> env.reset()
>>> env._plot_metadata["x"]["data"]
[]
Source code in dissmodel/core/environment.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def reset(self) -> None:
    """
    Reset the clock and clear accumulated plot data.

    Called automatically at the start of :meth:`run` to ensure the
    environment starts fresh on each simulation run.

    Examples
    --------
    >>> env = Environment(start_time=0, end_time=10)
    >>> env._plot_metadata = {"x": {"data": [1, 2, 3]}}
    >>> env.reset()
    >>> env._plot_metadata["x"]["data"]
    []
    """
    self._now = self.start_time
    for item in self._plot_metadata.values():
        item["data"].clear()

run(till=None)

Run the simulation over the configured time window.

Executes all registered models in time-step order. On each tick, every model whose next scheduled time is less than or equal to the current simulation time has its :meth:~dissmodel.core.Model.execute method called. The clock then advances to the nearest pending event.

Parameters:

Name Type Description Default
till float

Duration to run from start_time. If provided, overrides end_time. If omitted, end_time must be set.

 None

Raises:

Type Description
ValueError

If neither till nor end_time is defined.

Examples:

>>> env = Environment(start_time=0, end_time=10)
>>> env.run()
Running from 0 to 10 (duration: 10)
Source code in dissmodel/core/environment.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def run(self, till: Optional[float] = None) -> None:
    """
    Run the simulation over the configured time window.

    Executes all registered models in time-step order. On each tick,
    every model whose next scheduled time is less than or equal to the
    current simulation time has its :meth:`~dissmodel.core.Model.execute`
    method called. The clock then advances to the nearest pending event.

    Parameters
    ----------
    till : float, optional
        Duration to run from ``start_time``. If provided, overrides
        ``end_time``. If omitted, ``end_time`` must be set.

    Raises
    ------
    ValueError
        If neither ``till`` nor ``end_time`` is defined.

    Examples
    --------
    >>> env = Environment(start_time=0, end_time=10)
    >>> env.run()
    Running from 0 to 10 (duration: 10)
    """
    self.reset()

    if till is not None:
        self.end_time = self.start_time + till
    elif self.end_time is not None:
        pass
    else:
        raise ValueError(
            "Provide 'till' or set 'end_time' before calling run()."
        )

    duration = self.end_time - self.start_time
    print(
        f"Running from {self.start_time} to {self.end_time} "
        f"(duration: {duration})"
    )

    # Initialise next-execution time for every registered model
    for model in self._models:
        model._next_time = model.start_time

    self._now = self.start_time

    while self._now < self.end_time:
        for model in self._models:
            if (
                model._next_time <= self._now
                and self._now < model.end_time
            ):
                model.pre_execute()
                model.execute()
                model.post_execute()
                model._next_time = self._now + model._step

        # Advance clock to the nearest pending event
        pending = [
            m._next_time
            for m in self._models
            if m._next_time < self.end_time
            and m._next_time < m.end_time
        ]
        if not pending:
            break
        self._now = min(pending)

dissmodel.core.Model

Base class for simulation models.

Provides a time-stepped execution loop and automatic tracking of attributes marked for plotting via the :func:~dissmodel.visualization.track_plot decorator.

Every Model instance auto-registers with the currently active :class:~dissmodel.core.Environment at construction time. An active environment must exist before instantiating any model.

Parameters:

Name Type Description Default
step float

Time increment between successive :meth:execute calls, by default 1.

 1
start_time float

Time at which the model starts executing, by default 0.

 0
end_time float

Time at which the model stops executing, by default math.inf.

 inf
name str

Human-readable model name, by default "".

 ''
**kwargs Any

Extra keyword arguments (ignored; kept for subclass compatibility).

 {}

Raises:

Type Description
RuntimeError

If no active :class:~dissmodel.core.Environment exists when the model is instantiated.

Examples:

>>> class MyModel(Model):
...     def execute(self):
...         print(self.env.now())
>>> env = Environment(start_time=0, end_time=5)
>>> model = MyModel(step=1)
>>> env.run()
Running from 0 to 5 (duration: 5)
0
1
2
3
4
Source code in dissmodel/core/model.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class Model:
    """
    Base class for simulation models.

    Provides a time-stepped execution loop and automatic tracking of
    attributes marked for plotting via the
    :func:`~dissmodel.visualization.track_plot` decorator.

    Every ``Model`` instance auto-registers with the currently active
    :class:`~dissmodel.core.Environment` at construction time. An active
    environment must exist before instantiating any model.

    Parameters
    ----------
    step : float, optional
        Time increment between successive :meth:`execute` calls, by default 1.
    start_time : float, optional
        Time at which the model starts executing, by default 0.
    end_time : float, optional
        Time at which the model stops executing, by default ``math.inf``.
    name : str, optional
        Human-readable model name, by default ``""``.
    **kwargs :
        Extra keyword arguments (ignored; kept for subclass compatibility).

    Raises
    ------
    RuntimeError
        If no active :class:`~dissmodel.core.Environment` exists when the
        model is instantiated.

    Examples
    --------
    >>> class MyModel(Model):
    ...     def execute(self):
    ...         print(self.env.now())
    >>> env = Environment(start_time=0, end_time=5)
    >>> model = MyModel(step=1)
    >>> env.run()
    Running from 0 to 5 (duration: 5)
    0
    1
    2
    3
    4
    """

    def __init__(
        self,
        step: float = 1,
        start_time: float = 0,
        end_time: float = math.inf,
        name: str = "",
        **kwargs: Any,
    ) -> None:
        env = Environment._current
        if env is None:
            raise RuntimeError(
                "No active Environment found. "
                "Create an Environment before instantiating a Model."
            )

        # Set internal attributes directly to avoid triggering __setattr__
        # plot-tracking logic before _plot_info is available.
        object.__setattr__(self, "name", name)
        object.__setattr__(self, "_step", step)
        object.__setattr__(self, "start_time", start_time)
        object.__setattr__(self, "end_time", end_time)
        object.__setattr__(self, "_next_time", start_time)
        object.__setattr__(self, "env", env)

        env._register(self)
        self.setup(**kwargs)

    def setup(self, **kwargs: Any) -> None:
        """
        Called once after instantiation, receiving any extra keyword
        arguments not consumed by ``__init__``.

        Override in subclasses to perform one-time setup such as building
        neighborhoods or initializing visualization state.
        """
        pass

    def pre_execute(self) -> None:
        """
        Called once before each :meth:`execute`.

        Override in subclasses to perform per-step setup, such as
        snapshotting state arrays before the transition rule runs.
        """
        pass

    def execute(self) -> None:
        """
        Called once per time step.

        Override in subclasses to define model behaviour.
        """
        pass

    def post_execute(self) -> None:
        """
        Called once after each :meth:`execute`.

        Override in subclasses to perform per-step cleanup or snapshotting
        after the transition rule runs.
        """
        pass

    # ------------------------------------------------------------------
    # Plot tracking
    # ------------------------------------------------------------------

    def __setattr__(self, name: str, value: Any) -> None:
        """
        Intercept attribute assignment to record values marked for plotting.

        If the class defines ``_plot_info`` (populated by the
        :func:`~dissmodel.visualization.track_plot` decorator) and ``name``
        matches a tracked attribute, the value is appended to the plot data
        buffer and registered in ``env._plot_metadata``.

        Parameters
        ----------
        name : str
            Attribute name being set.
        value : Any
            Value being assigned.
        """
        cls = self.__class__

        if hasattr(cls, "_plot_info") and name.lower() in cls._plot_info:
            plot_info: dict[str, Any] = cls._plot_info[name.lower()]
            plot_info["data"].append(value)

            env = Environment._current
            if env is not None and plot_info["label"] not in env._plot_metadata:
                env._plot_metadata[plot_info["label"]] = plot_info

        super().__setattr__(name, value)

__setattr__(name, value)

Intercept attribute assignment to record values marked for plotting.

If the class defines _plot_info (populated by the :func:~dissmodel.visualization.track_plot decorator) and name matches a tracked attribute, the value is appended to the plot data buffer and registered in env._plot_metadata.

Parameters:

Name Type Description Default
name str

Attribute name being set.

 required
value Any

Value being assigned.

 required
Source code in dissmodel/core/model.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def __setattr__(self, name: str, value: Any) -> None:
    """
    Intercept attribute assignment to record values marked for plotting.

    If the class defines ``_plot_info`` (populated by the
    :func:`~dissmodel.visualization.track_plot` decorator) and ``name``
    matches a tracked attribute, the value is appended to the plot data
    buffer and registered in ``env._plot_metadata``.

    Parameters
    ----------
    name : str
        Attribute name being set.
    value : Any
        Value being assigned.
    """
    cls = self.__class__

    if hasattr(cls, "_plot_info") and name.lower() in cls._plot_info:
        plot_info: dict[str, Any] = cls._plot_info[name.lower()]
        plot_info["data"].append(value)

        env = Environment._current
        if env is not None and plot_info["label"] not in env._plot_metadata:
            env._plot_metadata[plot_info["label"]] = plot_info

    super().__setattr__(name, value)

execute()

Called once per time step.

Override in subclasses to define model behaviour.

Source code in dissmodel/core/model.py 
102
103
104
105
106
107
108
def execute(self) -> None:
    """
    Called once per time step.

    Override in subclasses to define model behaviour.
    """
    pass

post_execute()

Called once after each :meth:execute.

Override in subclasses to perform per-step cleanup or snapshotting after the transition rule runs.

Source code in dissmodel/core/model.py 
110
111
112
113
114
115
116
117
def post_execute(self) -> None:
    """
    Called once after each :meth:`execute`.

    Override in subclasses to perform per-step cleanup or snapshotting
    after the transition rule runs.
    """
    pass

pre_execute()

Called once before each :meth:execute.

Override in subclasses to perform per-step setup, such as snapshotting state arrays before the transition rule runs.

Source code in dissmodel/core/model.py 
 93
 94
 95
 96
 97
 98
 99
100
def pre_execute(self) -> None:
    """
    Called once before each :meth:`execute`.

    Override in subclasses to perform per-step setup, such as
    snapshotting state arrays before the transition rule runs.
    """
    pass

setup(**kwargs)

Called once after instantiation, receiving any extra keyword arguments not consumed by __init__.

Override in subclasses to perform one-time setup such as building neighborhoods or initializing visualization state.

Source code in dissmodel/core/model.py 
83
84
85
86
87
88
89
90
91
def setup(self, **kwargs: Any) -> None:
    """
    Called once after instantiation, receiving any extra keyword
    arguments not consumed by ``__init__``.

    Override in subclasses to perform one-time setup such as building
    neighborhoods or initializing visualization state.
    """
    pass