Skip to content

Mosaik

flowcean.mosaik provides an active environment for energy-system co-simulation. See the energy-system example for setup and usage.

mosaik

Classes

EnergySystemActive 

EnergySystemActive(scenario_name: str, data_file: str, *, scenario_file: str | None = None, end: int = 86400, seed: int | None = None, reward_func: Callable | None = None)

Bases: ActiveEnvironment

Attributes
reward instance-attribute 
reward: Callable
rng instance-attribute 
rng: RandomState = RandomState(seed)
sensor_queue instance-attribute 
sensor_queue = queue.Queue(1)
actuator_queue instance-attribute 
actuator_queue = queue.Queue(1)
sync_finished instance-attribute 
sync_finished = threading.Event()
sync_terminate instance-attribute 
sync_terminate = threading.Event()
sim_finished instance-attribute 
sim_finished = threading.Event()
scenario instance-attribute 
scenario: Scenario = midas.run(scenario_name, params, scenario_file, no_build=True, no_run=True)
sensors instance-attribute 
sensors: dict[str, ActiveInterface] = create_interface(self.scenario.sensors)
actuators instance-attribute 
actuators: dict[str, ActiveInterface] = create_interface(self.scenario.actuators)
task instance-attribute 
task = threading.Thread(target=start_mosaik, args=(self.scenario,), kwargs={'sensors': list(self.sensors), 'actuators': list(self.actuators), 'sensor_queue': self.sensor_queue, 'actuator_queue': self.actuator_queue, 'sync_finished': self.sync_finished, 'sync_terminate': self.sync_terminate, 'sim_finished': self.sim_finished})
action instance-attribute 
action = Action(actuators=list(self.actuators.values()))
observation instance-attribute 
observation = Observation(sensors=list(self.sensors.values()), rewards=[])
Methods:
step 
step() -> None
act 
act(action: Action) -> None
shutdown 
shutdown() -> None