Skip to content

Environment

Environment

Environment for a Vessim co-simulation.

This class manages the simulation time, the interaction between different components, and the execution of the Mosaik co-simulation.

Two modes are supported:

  • Simulated — Environment(sim_start=..., step_size=...). The simulation clock advances as fast as possible, anchored at the explicit sim_start.
  • Live — Environment.live(step_size=...). The simulation clock tracks wall-clock time and sim_start is captured when run() is called (i.e. defaults to "now at the moment of running"). Use this when mixing in SilSignals.

Parameters:

Name Type Description Default
sim_start Optional[str | datetime]

The start time of the simulation. Can be a datetime object or a string in the format "YYYY-MM-DD HH:MM:SS".

 None
step_size int

The step size of the simulation in seconds. Defaults to 1.

 1
Source code in vessim/environment.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
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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class Environment:
    """Environment for a Vessim co-simulation.

    This class manages the simulation time, the interaction between different components,
    and the execution of the [Mosaik](https://mosaik.offis.de/) co-simulation.

    Two modes are supported:

    - **Simulated** — `Environment(sim_start=..., step_size=...)`. The simulation
      clock advances as fast as possible, anchored at the explicit `sim_start`.
    - **Live** — `Environment.live(step_size=...)`. The simulation clock tracks
      wall-clock time and `sim_start` is captured when `run()` is called (i.e.
      defaults to "now at the moment of running"). Use this when mixing in
      `SilSignal`s.

    Args:
        sim_start: The start time of the simulation. Can be a `datetime` object or a
            string in the format "YYYY-MM-DD HH:MM:SS".
        step_size: The step size of the simulation in seconds. Defaults to 1.
    """

    COSIM_CONFIG: mosaik.SimConfig = {
        "Actor": {"python": "vessim.actor:_ActorSim"},
        "Controller": {"python": "vessim.controller:_ControllerSim"},
        "Microgrid": {"python": "vessim.microgrid:_MicrogridSim"},
    }

    def __init__(
        self,
        sim_start: Optional[str | datetime] = None,
        step_size: int = 1,
        name: Optional[str] = None,
        _live: bool = False,
        _behind_threshold: float = 5.0,
    ):
        if not _live and sim_start is None:
            raise ValueError(
                "sim_start is required for simulated mode. "
                "Use Environment.live(...) for real-time experiments."
            )
        # In live mode, sim_start is purely a label (for logging / metadata).
        # If omitted, run() will fall back to datetime.now() at start.
        self.sim_start: pd.Timestamp | None = (
            pd.to_datetime(sim_start) if sim_start is not None else None
        )
        self.step_size = step_size
        self.name = name
        self.microgrids: dict[str, Microgrid] = {}
        self.controllers: list[Controller] = []
        self.world = mosaik.World(self.COSIM_CONFIG, skip_greetings=True)
        self._live = _live
        self._behind_threshold = _behind_threshold

    @classmethod
    def live(
        cls,
        sim_start: Optional[str | datetime] = None,
        step_size: int = 1,
        behind_threshold: float = 5.0,
        name: Optional[str] = None,
    ) -> "Environment":
        """Create an environment that advances in real-time (1× wall-clock).

        The simulation clock advances 1:1 with wall-clock time, so traces start
        replaying from `sim_start` and SiL signals stay in sync with the system
        clock.

        Args:
            sim_start: Optional label used only for logging and result metadata.
                Does not affect signal queries (which are always indexed by
                elapsed time since `run()`). Defaults to `datetime.now()` at
                the moment `run()` is called.
            step_size: Step size in seconds. Defaults to 1.
            behind_threshold: Seconds the simulation may fall behind real-time before
                a warning is logged. Defaults to 5.
            name: Optional name for the environment.
        """
        return cls(
            sim_start=sim_start,
            step_size=step_size,
            name=name,
            _live=True,
            _behind_threshold=behind_threshold,
        )

    def add_microgrid(
        self,
        actors: list[Actor],
        dispatchables: Optional[list[Dispatchable]] = None,
        policy: Optional[DispatchPolicy] = None,
        grid_signals: Optional[dict[str, Signal]] = None,
        name: Optional[str] = None,
        coords: Optional[tuple[float, float]] = None,
    ) -> Microgrid:
        """Add a microgrid to the environment.

        Args:
            actors: A list of exogenous actors (consumers/producers) in the microgrid.
            dispatchables: Optional list of dispatchable resources (e.g., batteries,
                generators).
            policy: The dispatch policy that controls energy management. If None, a
                `DefaultDispatchPolicy` is used.
            grid_signals: Optional signals from the public grid (e.g., carbon intensity).
            name: An optional name for the microgrid.
            coords: Optional coordinates (latitude, longitude) for the microgrid.

        Returns:
            The created `Microgrid` instance.
        """
        if not actors:
            raise ValueError("There should be at least one actor in the Microgrid.")

        microgrid = Microgrid(
            world=self.world,
            step_size=self.step_size,
            actors=actors,
            dispatchables=dispatchables or [],
            policy=policy if policy is not None else DefaultDispatchPolicy(),
            grid_signals=grid_signals,
            name=name,
            coords=coords,
        )
        if microgrid.name in self.microgrids:
            raise ValueError(
                f"A microgrid named '{microgrid.name}' already exists in this environment."
            )
        self.microgrids[microgrid.name] = microgrid
        return microgrid

    def add_controller(self, controller: Controller):
        """Add a controller to the environment.

        Args:
            controller: The controller instance.
        """
        self.controllers.append(controller)

    def _initialize_controllers(self):
        """Initialize all controllers after all microgrids have been added."""
        if not self.controllers:
            return

        # Execute start() method on all controllers
        for controller in self.controllers:
            controller.start(self)

        # Create one global controller simulator
        controller_sim = self.world.start(
            "Controller",
            sim_start=self.sim_start,
            step_size=self.step_size,
        )
        controller_entity = controller_sim.Controller(controllers=self.controllers)

        # Connect global controller to all microgrids
        for microgrid in self.microgrids.values():
            self.world.connect(microgrid.entity, controller_entity, "p_delta")
            self.world.connect(microgrid.entity, controller_entity, "grid_signals")
            self.world.connect(microgrid.entity, controller_entity, "p_grid")
            self.world.connect(microgrid.entity, controller_entity, "policy_state")
            if microgrid.dispatchables:
                self.world.connect(
                    microgrid.entity, controller_entity, "dispatch_states"
                )

            # Connect actors for state
            for actor_entity in microgrid.actor_entities.values():
                self.world.connect(
                    actor_entity,
                    controller_entity,
                    ("state", "actor_states"),
                )

    def run(
        self,
        until: Optional[timedelta | datetime | int | float] = None,
        print_progress: bool | Literal["individual"] = True,
    ):
        """Run the simulation.

        Args:
            until: When the simulation should end. Accepts:

                - `int` or `float` — elapsed seconds since `sim_start`.
                - `timedelta` — elapsed time since `sim_start`.
                - `datetime` — absolute wall-clock end (resolved against `sim_start`).
                - `None` — run indefinitely.
            print_progress: Whether to print a progress bar.
        """
        # Live mode: anchor sim_start to "now" if the user didn't pin one explicitly.
        if self._live and self.sim_start is None:
            self.sim_start = pd.to_datetime(datetime.now())

        if until is None:
            until = float("inf")  # type: ignore
        elif isinstance(until, timedelta):
            until = until.total_seconds()
        elif isinstance(until, datetime):
            assert self.sim_start is not None
            until = (pd.to_datetime(until) - self.sim_start).total_seconds()
            if until < 0:
                raise ValueError("`until` must be after `sim_start`.")
        assert until is not None

        # Initialize controllers before running simulation
        if self.name:
            logger.info(f"Experiment: {self.name}")
        self._initialize_controllers()

        # SiL signals require live mode (otherwise they'd be polled out of sync
        # with simulated time).
        if self._contains_sil_signals() and not self._live:
            raise RuntimeError(
                "SiL signals detected but not running in live mode. "
                "Use Environment.live(...) instead of Environment(...)."
            )

        rt_factor = 1.0 if self._live else None
        if rt_factor:
            disable_rt_warnings(self._behind_threshold)
        try:
            self.world.run(until=until, rt_factor=rt_factor, print_progress=print_progress)  # type: ignore
        except Exception:
            for microgrid in self.microgrids.values():
                microgrid.finalize()
            raise

    def _contains_sil_signals(self) -> bool:
        """Check if any microgrid contains SiL signals."""
        for microgrid in self.microgrids.values():
            for actor in microgrid.actors:
                if isinstance(actor.signal, SilSignal):
                    return True
        return False

live classmethod

Create an environment that advances in real-time (1× wall-clock).

The simulation clock advances 1:1 with wall-clock time, so traces start replaying from sim_start and SiL signals stay in sync with the system clock.

Parameters:

Name Type Description Default
sim_start Optional[str | datetime]

Optional label used only for logging and result metadata. Does not affect signal queries (which are always indexed by elapsed time since run()). Defaults to datetime.now() at the moment run() is called.

 None
step_size int

Step size in seconds. Defaults to 1.

 1
behind_threshold float

Seconds the simulation may fall behind real-time before a warning is logged. Defaults to 5.

 5.0
name Optional[str]

Optional name for the environment.

 None
Source code in vessim/environment.py 
 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
@classmethod
def live(
    cls,
    sim_start: Optional[str | datetime] = None,
    step_size: int = 1,
    behind_threshold: float = 5.0,
    name: Optional[str] = None,
) -> "Environment":
    """Create an environment that advances in real-time (1× wall-clock).

    The simulation clock advances 1:1 with wall-clock time, so traces start
    replaying from `sim_start` and SiL signals stay in sync with the system
    clock.

    Args:
        sim_start: Optional label used only for logging and result metadata.
            Does not affect signal queries (which are always indexed by
            elapsed time since `run()`). Defaults to `datetime.now()` at
            the moment `run()` is called.
        step_size: Step size in seconds. Defaults to 1.
        behind_threshold: Seconds the simulation may fall behind real-time before
            a warning is logged. Defaults to 5.
        name: Optional name for the environment.
    """
    return cls(
        sim_start=sim_start,
        step_size=step_size,
        name=name,
        _live=True,
        _behind_threshold=behind_threshold,
    )

add_microgrid

Add a microgrid to the environment.

Parameters:

Name Type Description Default
actors list[Actor]

A list of exogenous actors (consumers/producers) in the microgrid.

 required
dispatchables Optional[list[Dispatchable]]

Optional list of dispatchable resources (e.g., batteries, generators).

 None
policy Optional[DispatchPolicy]

The dispatch policy that controls energy management. If None, a DefaultDispatchPolicy is used.

 None
grid_signals Optional[dict[str, Signal]]

Optional signals from the public grid (e.g., carbon intensity).

 None
name Optional[str]

An optional name for the microgrid.

 None
coords Optional[tuple[float, float]]

Optional coordinates (latitude, longitude) for the microgrid.

 None

Returns:

Type Description
Microgrid

The created Microgrid instance.
Source code in vessim/environment.py 
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
def add_microgrid(
    self,
    actors: list[Actor],
    dispatchables: Optional[list[Dispatchable]] = None,
    policy: Optional[DispatchPolicy] = None,
    grid_signals: Optional[dict[str, Signal]] = None,
    name: Optional[str] = None,
    coords: Optional[tuple[float, float]] = None,
) -> Microgrid:
    """Add a microgrid to the environment.

    Args:
        actors: A list of exogenous actors (consumers/producers) in the microgrid.
        dispatchables: Optional list of dispatchable resources (e.g., batteries,
            generators).
        policy: The dispatch policy that controls energy management. If None, a
            `DefaultDispatchPolicy` is used.
        grid_signals: Optional signals from the public grid (e.g., carbon intensity).
        name: An optional name for the microgrid.
        coords: Optional coordinates (latitude, longitude) for the microgrid.

    Returns:
        The created `Microgrid` instance.
    """
    if not actors:
        raise ValueError("There should be at least one actor in the Microgrid.")

    microgrid = Microgrid(
        world=self.world,
        step_size=self.step_size,
        actors=actors,
        dispatchables=dispatchables or [],
        policy=policy if policy is not None else DefaultDispatchPolicy(),
        grid_signals=grid_signals,
        name=name,
        coords=coords,
    )
    if microgrid.name in self.microgrids:
        raise ValueError(
            f"A microgrid named '{microgrid.name}' already exists in this environment."
        )
    self.microgrids[microgrid.name] = microgrid
    return microgrid

add_controller

Add a controller to the environment.

Parameters:

Name Type Description Default
controller Controller

The controller instance.

 required
Source code in vessim/environment.py 
148
149
150
151
152
153
154
def add_controller(self, controller: Controller):
    """Add a controller to the environment.

    Args:
        controller: The controller instance.
    """
    self.controllers.append(controller)

run

Run the simulation.

Parameters:

Name Type Description Default
until Optional[timedelta | datetime | int | float]

When the simulation should end. Accepts:

  • int or float — elapsed seconds since sim_start.
  • timedelta — elapsed time since sim_start.
  • datetime — absolute wall-clock end (resolved against sim_start).
  • None — run indefinitely.
 None
print_progress bool | Literal['individual']

Whether to print a progress bar.

 True
Source code in vessim/environment.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def run(
    self,
    until: Optional[timedelta | datetime | int | float] = None,
    print_progress: bool | Literal["individual"] = True,
):
    """Run the simulation.

    Args:
        until: When the simulation should end. Accepts:

            - `int` or `float` — elapsed seconds since `sim_start`.
            - `timedelta` — elapsed time since `sim_start`.
            - `datetime` — absolute wall-clock end (resolved against `sim_start`).
            - `None` — run indefinitely.
        print_progress: Whether to print a progress bar.
    """
    # Live mode: anchor sim_start to "now" if the user didn't pin one explicitly.
    if self._live and self.sim_start is None:
        self.sim_start = pd.to_datetime(datetime.now())

    if until is None:
        until = float("inf")  # type: ignore
    elif isinstance(until, timedelta):
        until = until.total_seconds()
    elif isinstance(until, datetime):
        assert self.sim_start is not None
        until = (pd.to_datetime(until) - self.sim_start).total_seconds()
        if until < 0:
            raise ValueError("`until` must be after `sim_start`.")
    assert until is not None

    # Initialize controllers before running simulation
    if self.name:
        logger.info(f"Experiment: {self.name}")
    self._initialize_controllers()

    # SiL signals require live mode (otherwise they'd be polled out of sync
    # with simulated time).
    if self._contains_sil_signals() and not self._live:
        raise RuntimeError(
            "SiL signals detected but not running in live mode. "
            "Use Environment.live(...) instead of Environment(...)."
        )

    rt_factor = 1.0 if self._live else None
    if rt_factor:
        disable_rt_warnings(self._behind_threshold)
    try:
        self.world.run(until=until, rt_factor=rt_factor, print_progress=print_progress)  # type: ignore
    except Exception:
        for microgrid in self.microgrids.values():
            microgrid.finalize()
        raise