EMPS - serial simulation

How it works

EMPS/serial simulation is a way to run scenarios (historical years) in sequence. The output from year 1 is input for year 2, until the reqested number of scenarios has been reached.

Results from detailed hydro has been made into a “continuous” line showing the inputs matches the outputs from previous year.

In this notebook, real inflow series from NVE has been used.

How to use

EMPS/serial simulation is enabled by settings keys "model_type": "EMPS" and "simulation_type": "serial" under global_settings.

{
    "model": {
        "global_settings": {
            "model_type": "EMPS",
            "simulation_type": "serial"
        }
    }
}

The simulation must be exactly 52 weeks. The API enforce that limit and will refuse to run simulations with other durations.

"simulation_period": {
    "timestamps": [
        "2024-W01",
        "2025-W01"
    ],
    "scenarios": [
        [
            1,
            0
        ]
    ]
}

For the special case EMPS/serial, it is allowed to only provide 1 timestamp for simulation_period. The API will then use a 52 week duration from the start.

"simulation_period": {
    "timestamps": [
        "2024-W01"
    ],
    "scenarios": [
        [
            1
        ]
    ]
}

import os
import numpy as np
from pyltmapi import LtmSession, LtmPlot, LtmDot

import logging
logging.basicConfig(level=logging.INFO)

from pathlib import Path

ltm_core_path =  os.environ.get("LTM_CORE_PATH", str(Path("~").expanduser().joinpath("ltm/release/bin/")))
license_file = os.environ.get("LTM_CORE_LICENSE_FILE", str(Path("~").expanduser().joinpath("ltm/ltm-license.dat")))

from IPython.display import HTML, Image, display, display_markdown

ltmapi_version = LtmSession.version()
display(f"pyltm version {ltmapi_version}")

'pyltm version PyLTM version: 0.21.0'

def usercallback(program_info: dict, userdata: any):

    print(userdata)
    print(program_info)

    return True


def generate_plots(ltm):
    # Water values and price series
    for busbar in ltm.model.busbars():
        display_markdown(f"# Busbar: {busbar}", raw=True)

        # Water values
        try:
            LtmPlot.make_water_value_plot(busbar.water_value_results(), busbar.name)
        except:
            display(f"Could not get water value results for {busbar.name}")

        # Market results
        LtmPlot.make_market_results_plot(busbar.market_result_price(), busbar.name)

    # Detailed hydro results from
    for busbar in ltm.model.busbars():
        display_markdown(f"## {busbar}", raw=True)

        # Busbar reservoirs
        for rsv in busbar.reservoirs():
            display("Reservoir")
            LtmPlot.make_continuous_generic_plot(rsv.reservoir(), f"Reservoir '{rsv.name}'")
            LtmPlot.make_continuous_generic_plot(rsv.discharge(), f"Discharge '{rsv.name}'")
            LtmPlot.make_continuous_generic_plot(rsv.inflow(), f"Inflow '{rsv.name}'")
            LtmPlot.make_continuous_generic_plot(rsv.production(), f"Production '{rsv.name}'")
            LtmPlot.make_continuous_generic_plot(rsv.bypass(), f"Bypass '{rsv.name}'")
            LtmPlot.make_continuous_generic_plot(rsv.spill(), f"Spill '{rsv.name}'")

def open_and_write_model(filename: str):
    session = LtmSession("ikernel", ltm_core_path=ltm_core_path, overwrite_session=True)

    # Explicitly set license file
    session.model.global_settings.ltm_license_file_path = license_file

    with session:
        try:
            # Load model from file.
            session.load(filename=filename)
            print(vars(session.model.global_settings))

            # Write model to disk, and automatically generate an output directory.
            session.write_model()

            # return

            # Display model graph
            LtmDot.display_dot_image(session.build_connection_tree())

            # return

            # Execute/run LTM/EMPS on the model
            last_rc, results = session.execute_model()

            # If last return code is not 0, then there was an error.
            if last_rc != 0:
                err = results[0]["log_file_contents"]
                LtmDot.print(err)
            else:
                # Make plots from the results
                generate_plots(session)

        except Exception as e:
            print(e)
            raise (e)

open_and_write_model("emps_serial.json")

INFO:LtmApiModel:(ikernel) Loading model from file: emps_serial.json
INFO:LtmApiModel:(ikernel) LtmApiModel::maybe_generate_output_dir: output_path: /builds/energy/ltm/pyltmapi/docs/ltm-api/guides/emps_serial/testout_emps_serial/2026-03-24-133937.788-EMPS-series
INFO:LtmApiModel:(ikernel) Using license file '/builds/energy/ltm/pyltmapi.tmp/CI_LTM_LICENSE_FILE'
INFO:LtmApiModel:(ikernel) Using license file '/builds/energy/ltm/pyltmapi.tmp/CI_LTM_LICENSE_FILE'
INFO:Validator:(ikernel) Model validation succeeded
INFO:LtmApiModel:(ikernel) Writing model to path /builds/energy/ltm/pyltmapi/docs/ltm-api/guides/emps_serial/testout_emps_serial/2026-03-24-133937.788-EMPS-series
INFO:Validator:(ikernel) Model validation succeeded

{}

INFO:LtmApiModel:(ikernel) Model executed successfully

Busbar: busbar/detailed_hydro

Busbar: busbar/tev

busbar/detailed_hydro

'Reservoir'

busbar/tev

INFO:LtmApiModel:(ikernel) Not deleting output dir (/builds/energy/ltm/pyltmapi/docs/ltm-api/guides/emps_serial/testout_emps_serial/2026-03-24-133937.788-EMPS-series), as delete_output_dir: false, and has_generated_output_dir: true
INFO:LtmApiModel:(ikernel) Not deleting output dir (/builds/energy/ltm/pyltmapi/docs/ltm-api/guides/emps_serial/testout_emps_serial/2026-03-24-133937.788-EMPS-series), as delete_output_dir: false, and has_generated_output_dir: true