energy_system

EnergySystemOffline(scenario_name, data_file, *, scenario_file=None, end=0, seed=None)

Bases: DataFrame

Offline version of the energy system environment.

Runs the simulation without interaction and stores the results in a csv file. The file is read into polars and can be used for offline learning.

Source code in src/flowcean/mosaik/energy_system.py

def __init__(
    self,
    scenario_name: str,
    data_file: str,
    *,
    scenario_file: str | None = None,
    end: int = 0,
    seed: int | None = None,
) -> None:
    if "DatabaseCSV" not in META["models"]:
        msg = (
            "Wrong version of midas_store is used. "
            "At least required is 2.1.0a1"
        )
        raise ValueError(msg)
    params = {
        "seed": seed,
        "store_params": {
            "path": ".",
            "filename": data_file,
            # "buffer_size": 10,
        },
    }
    if end > 0:
        params["end"] = end

    midas.api.run(scenario_name, params, scenario_file)
    data = pl.scan_csv(data_file)
    super().__init__(data)

start_mosaik(scenario, *, sensors, actuators, sensor_queue, actuator_queue, sync_finished, sync_terminate, sim_finished)

Start the mosaik simulation process.

Source code in src/flowcean/mosaik/energy_system.py

def start_mosaik(
    scenario: Scenario,
    *,
    sensors: list[str],
    actuators: list[str],
    sensor_queue: queue.Queue,
    actuator_queue: queue.Queue,
    sync_finished: threading.Event,
    sync_terminate: threading.Event,
    sim_finished: threading.Event,
) -> None:
    """Start the mosaik simulation process."""
    scenario.makelists.sim_config["SyncSimulator"] = {
        "python": "flowcean.mosaik.simulator:SyncSimulator",
    }
    scenario.build()
    world = scenario.world
    if not isinstance(world, mosaik.World):
        sim_finished.set()
        logger.error(
            "Malformed world object: %s (%s)",
            str(world),
            type(world),
        )
        return

    logger.debug("Starting SyncSimulator ...")
    sync_sim = world.start(
        "SyncSimulator",
        step_size=scenario.base.step_size,
        start_date=scenario.base.start_date,
        sensor_queue=sensor_queue,
        actuator_queue=actuator_queue,
        sync_finished=sync_finished,
        sync_terminate=sync_terminate,
        end=scenario.base.end,
    )

    logger.debug("Connecting sensor entities ...")
    for uid in sensors:
        sid, eid, attr = uid.split(".")
        full_id = f"{sid}.{eid}"
        sensor_model = sync_sim.Sensor(uid=uid)
        logger.debug("Connecting %s ...", full_id)
        world.connect(
            scenario.entities[full_id],
            sensor_model,
            (attr, "reading"),
        )

    logger.debug("Connecting actuator entities ...")
    for uid in actuators:
        sid, eid, attr = uid.split(".")
        full_id = f"{sid}.{eid}"
        actuator_model = sync_sim.Actuator(uid=uid)
        world.connect(
            actuator_model,
            scenario.entities[full_id],
            ("setpoint", attr),
            time_shifted=True,
            initial_data={"setpoint": None},
        )

    logger.info("Starting mosaik run ...")
    try:
        world.run(until=scenario.base.end)
    except SimulationError:
        logger.info("Simulation finished non-regular.")
        world.shutdown()
    else:
        logger.info("Simulation finished.")
    sim_finished.set()