Cyclic curves

This notebook is an example on how to repeat yearly curves with API help.

Supported curves (TxyLin) can be cyclical and most will repeat yearly. By defining a curve between years 1703-1704, it will be automatically repeated for each data year.

A definition of a minimum bypass curve that will repeat yearly may look like this:

"min_bypass_curve": {
    "timestamps": [
        "W05",
        "W20",
        "W35",
        "W50"
    ],
    "scenarios": [
        [
            1,
            2,
            2,
            1
        ]
    ]
}

It will repeat, and be linearized, giving this repeating pattern with a data period over 10 years.

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

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.27.0.post1.dev0'

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

        # Pumped energy
        if area.have_water_value_results():
            pumped_energy = area.sum_pumped_power()
            LtmPlot.make_generic_plot(pumped_energy, f"sum_pumped_power '{area.name}'")

            energy_pumping = area.sum_power_used_for_pumping()
            LtmPlot.make_generic_plot(
                energy_pumping, f"sum_power_used_for_pumping '{area.name}'"
            )

        if area.have_detailed_hydro_results() or area.have_aggregated_hydro_results():
            LtmPlot.make_generic_plot(
                area.sum_regulated_inflow(), f"Area sum regulated inflow '{area.name}'"
            )
            LtmPlot.make_generic_plot(
                area.sum_unregulated_inflow(),
                f"Area sum unregulated inflow '{area.name}'",
            )

        # Water values
        if area.have_water_value_results():
            LtmPlot.make_water_value_plot(area.water_value_results(), area.name)

        # Market results
        if area.have_market_results():
            LtmPlot.make_market_results_plot(area.market_result_price(), area.name)

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

        area: pyltm.Area

        LtmDot.display_dot_image(area.build_connection_tree())

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

from pyltm import ProgramInfo, ProgramInfos, ProgramResult, ProgramResults


def pre_model(programinfos: ProgramInfos, userdata: object):
    print(f"Executed before all programs are executed. Userdata: {userdata}")
    print(f"  Programs to be executed: {[x.program_name for x in programinfos]}")
    return True


def post_model(programresults: ProgramResults, userdata: object):
    print(f"Executed after all programs has been executed. Userdata: {userdata}")
    print(f"  Executed {[x.name for x in programresults]}")
    print(f"  Return codes: {[x.return_code for x in programresults]}")
    return True


def pre_execute(programinfo: ProgramInfo, userdata: object):
    print(f"Executed just before a program will be executed. Userdata: {userdata}")
    print(f"  Executing: {programinfo.program_name}")
    return True


def post_execute(programresult: ProgramResult, userdata: object):
    print(f"Executed just after a program has been executed. Userdata: {userdata}")
    print(f"  Executed: {programresult.name}")
    return True


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
    session.model.global_settings.simulation_type = pyltm.SimulationType.Parallel

    # Set execute callbacks
    session.set_pre_model_execute_callback(pre_model, 1)
    session.set_post_model_execute_callback(post_model, 2.0)
    session.set_pre_execute_callback(pre_execute, 42)
    session.set_post_execute_callback(post_execute, {"answer": "43"})

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

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

            # Run the model
            last_rc, results = session.run()

            # 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("cyclic-curves.json")

INFO:LtmApiModel:(ikernel) Loading model from file: cyclic-curves.json
INFO:LtmApiModel:(ikernel) Using license file '/home/sintef/ltm/ltm-license.dat'

INFO:LtmApiModel:(ikernel) LtmApiModel::maybe_generate_output_dir: output_path: /home/sintef/gitlab/pyltmapi/docs/ltm-api/guides/cyclic-curves/testout_cyclic_curve/2026-11-05-142337.246-EMPS-parallell
INFO:LtmApiModel:(ikernel) Using license file '/home/sintef/ltm/ltm-license.dat'
INFO:Validator:(ikernel) Model validation succeeded
INFO:LtmApiModel:(ikernel) Writing model to path /home/sintef/gitlab/pyltmapi/docs/ltm-api/guides/cyclic-curves/testout_cyclic_curve/2026-11-05-142337.246-EMPS-parallell
INFO:InflowHandler:(ikernel) Writing inflow series
INFO:Validator:(ikernel) Model validation succeeded

Executed before all programs are executed. Userdata: 1
  Programs to be executed: ['emps', 'emps']
Executed just before a program will be executed. Userdata: 42
  Executing: emps
Executed just after a program has been executed. Userdata: {'answer': '43'}
  Executed: emps LtmSystem_Step1_WV.XML
Executed just before a program will be executed. Userdata: 42
  Executing: emps

INFO:LtmApiModel:(ikernel) Model executed successfully

Executed just after a program has been executed. Userdata: {'answer': '43'}
  Executed: emps LtmSystem_Step2_Sim.XML
Executed after all programs has been executed. Userdata: 2.0
  Executed ['emps LtmSystem_Step1_WV.XML', 'emps LtmSystem_Step2_Sim.XML']
  Return codes: [0, 0]

area: area/midt

area: area/tev

area/midt

'Reservoir'

'Reservoir'

'Reservoir'

area/tev

INFO:LtmApiModel:(ikernel) Not deleting output dir (/home/sintef/gitlab/pyltmapi/docs/ltm-api/guides/cyclic-curves/testout_cyclic_curve/2026-11-05-142337.246-EMPS-parallell), as delete_output_dir: false, and has_generated_output_dir: true
INFO:LtmApiModel:(ikernel) Not deleting output dir (/home/sintef/gitlab/pyltmapi/docs/ltm-api/guides/cyclic-curves/testout_cyclic_curve/2026-11-05-142337.246-EMPS-parallell), as delete_output_dir: false, and has_generated_output_dir: true