Live LTMΒΆ
This notebook creates a model by using only the Python interface, with no file backing.
[1]:
import os
import numpy as np
from pyltmapi import LtmSession, LtmPlot
import logging
logging.basicConfig(level=logging.ERROR)
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")))
[2]:
from IPython.display import HTML, display, display_markdown
ltmapi_version = LtmSession.version()
display(f"pyltm version {ltmapi_version}")
'pyltm version PyLTM version: 0.21.0'
[3]:
import time
utc_offset = int(-(time.timezone / 3600))
print(utc_offset)
0
[4]:
# UTC-ish time
from datetime import datetime, timedelta, timezone, tzinfo
class LtmTimeZone(tzinfo):
def utcoffset(self, dt):
return timedelta(hours=15)
def dst(self, dt):
return timedelta(hours=16)
def tzname(self, dt):
return "LTM"
ltmtz = LtmTimeZone()
[5]:
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():
print(busbar)
if busbar.have_water_value_results():
# Water values
LtmPlot.make_water_value_plot(busbar.water_value_results(), 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():
print(busbar)
# Busbar reservoirs
rsvs = busbar.reservoirs()
for rsv in rsvs:
LtmPlot.make_generic_plot(rsv.reservoir(), f"Reservoir '{rsv.name}'")
for rsv in rsvs:
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}'")
[6]:
from datetime import datetime, timedelta
from pyltm import Txy
[7]:
def full_simulation_price_period(start: datetime):
"""Maximum number of price segments"""
timestamps = list()
values = list()
for n in range(168):
timestamps.append(start + timedelta(hours=n))
values.append(n + 1)
price_period = Txy()
price_period.timestamps = timestamps
price_period.scenarios = [values]
return price_period
def day_evening_night_weekend_price_period(start: datetime):
"""Price segments for weekend, day, night, evenings"""
timestamps = list()
values = list()
# Mon -> Fri
for n in range(5):
current_day = start + timedelta(days=n)
timestamps.append(current_day + timedelta(hours=0))
timestamps.append(current_day + timedelta(hours=8))
timestamps.append(current_day + timedelta(hours=14))
values.append(3)
values.append(1)
values.append(2)
# Weekend
timestamps.append(start + timedelta(days=5))
values.append(4)
price_period = Txy()
price_period.timestamps = timestamps
price_period.scenarios = [values]
return price_period
[8]:
def add_busbar(session, name):
bb = session.model.add(
"busbar",
name,
)
return bb
[9]:
def add_dclines(session, name):
# forward_capacity = Txy()
# forward_capacity.timestamps = [
# datetime.fromisoformat("1900-01-01T00:00:00"),
# datetime.fromisoformat("2024-06-01T00:00:00"),
# datetime.fromisoformat("2024-10-01T00:00:00"),
# ]
# forward_capacity.scenarios = [[200, 300, 500]]
# backward_capacity = Txy()
# backward_capacity.timestamps = [
# datetime.fromisoformat("1900-01-01T00:00:00"),
# datetime.fromisoformat("2024-06-01T00:00:00"),
# datetime.fromisoformat("2024-10-01T00:00:00"),
# ]
# backward_capacity.scenarios = [[1900, 1500, 500]]
dc = session.model.add(
"dcline",
name,
{
"forward_capacity": {
"timestamps": ["2023-01-02T00:00:00Z"],
"scenarios": [[200]],
},
"backward_capacity": {
"timestamps": ["2023-01-02T00:00:00Z"],
"scenarios": [[200]],
},
"loss_percentage": 2.0,
"forward_cost": 5.0,
"backward_cost": 5.0,
},
)
return dc
[10]:
def add_load(session, name):
load = session.model.add(
"load",
name,
{
"capacity": {
"timestamps": ["2023-01-01T00:00:00Z"],
"scenarios": [[10]],
},
},
)
return load
[11]:
def add_default_inflow_series(session):
session.model.add(
"inflow",
"small",
{
"series": {
"timestamps": ["1900-01-01T00:00:00Z"],
"scenarios": [[0.001]],
}
},
)
session.model.add(
"inflow",
"constant",
{
"series": {
"timestamps": ["1900-01-01T00:00:00Z"],
"scenarios": [[1.0]],
}
},
)
[12]:
def create_simple_topology(session):
upper1 = session.model.add(
"reservoir",
"upper1",
{
"average_spill_energy_equivalent": 1.6,
"degree_of_regulation": 3,
"regulated_inflow": {
"timestamps": ["1900-01-01T00:00:00Z"],
"scenarios": [[1.0]],
},
"average_regulated_inflow": 10,
"max_discharge": 50,
"reference_curve": {
"timestamps": [
"1900-W01"
],
"scenarios": [[450]],
},
"initial_volume": 5,
"volume_curve": {
"x": [1100.0, 1130.0],
"y": [0, 500],
},
},
)
lower1 = session.model.add(
"reservoir",
"lower1",
{
"average_spill_energy_equivalent": 0.6,
"degree_of_regulation": 0.5,
"regulated_inflow_name": "constant",
"average_regulated_inflow": 0,
"max_discharge": 200,
"reference_curve": {
"timestamps": ["1900-W01"],
"scenarios": [[1500]],
},
"initial_volume": 0
},
)
plant1 = session.model.add(
"plant",
"plant1",
{
"ownership": 100,
"discharge_energy_equivalent": {
"timestamps": ["2023-01-02T00:00:00Z"],
"scenarios": [[1.600]],
},
"pq_curves": {
"2023-01-02T00:00:00Z": {
"x": [0.0, 50.00],
"y": [0.0, 200.0],
}
},
"unregulated_inflow_name": "small",
"average_unregulated_inflow": 0.0,
"tailrace_elevation": 150.0,
"gross_head": 800.0,
},
)
session.connect(upper1, plant1)
session.connect(plant1, lower1)
return upper1
[13]:
def add_market_step(session, name):
ms = session.model.add(
"market_step",
name,
{
"price": {
"timestamps": ["2024-01-01T00:00:00Z", "2024-06-01T00:00:00Z"],
"scenarios": [[260, 200]],
},
"capacity": {
"timestamps": ["2023-01-01T00:00:00Z"],
"scenarios": [[150]],
},
},
)
return ms
[14]:
import pyltm
def add_wind(session, name, busbar, data_start, num_scenarios):
# wind = pyltm.wind() # session.model.add("wind", name)
timestamps = list()
values = list()
for n in range(1000):
timestamps.append(data_start + timedelta(days=n))
# values.append(Math.ran)
# y = np.sin(x) + np.random.random(1000) * 0.2
y = np.linspace(0, 1000 * num_scenarios, 1000 * num_scenarios)
scenarios = y.reshape(8, 1000)
# capacity = Txy()
# capacity.timestamps = timestamps
# # capacity.scenarios = scenarios
for scenario in range(num_scenarios):
data = list(scenarios[scenario])
values.append(data)
# capacity.scenarios = values
# wind.capacity = capacity
# {
# "capacity": {
# "timestamps": timestamps,
# "scenarios": values,
# },
# },
timestamps_str = [
"2024-W01",
"2024-W20",
"2024-W40",
]
scenarios_list = [
[1, 3, 0],
[2, 3, 0],
[3, 3, 0],
[4, 3, 0],
[5, 3, 0],
[6, 3, 0],
[7, 3, 0],
[8, 3, 0],
]
wind: pyltm.wind = session.model.add(
"wind",
name,
{
"capacity": {
"timestamps": timestamps_str,
"scenarios": scenarios_list,
}
},
)
session.connect(wind, busbar)
# wind2: pyltm.wind = session.model.add(
# "wind",
# f"{name}-2",
# {
# "capacity": {
# "timestamps": [
# "1930-01-01T00:00:00Z",
# "1950-01-01T00:00:00Z",
# "1970-01-01T00:00:00Z",
# ],
# "scenarios": [
# [1, 3, 0],
# ],
# }
# },
# )
# session.connect(wind2, busbar)
pass
[15]:
def create_basic_model(
name: str,
output_path: str,
wv_start: datetime,
wv_efi_years: int = 3,
hist_start: datetime = datetime(2000, 1, 1, utc_offset),
hist_end: datetime = datetime(2008, 1, 1, utc_offset),
):
session = LtmSession(name, ltm_core_path=ltm_core_path, overwrite_session=True)
sim = Txy()
sim.timestamps = [wv_start, wv_start + timedelta(weeks=wv_efi_years * 52)]
sim.scenarios = [[1, 0]]
hist = Txy()
hist.timestamps = [hist_start, hist_end]
hist.scenarios = [[1, 0]]
# Explicitly set license file
gs = session.model.global_settings
gs.name = name
gs.output_path = output_path
gs.ltm_license_file_path = license_file
gs.delete_output_dir = False
gs.generate_output_dir = True
gs.simulation_period = sim
gs.historical_period = hist
gs.timesteps_per_week = 168
gs.default_spill_cost = 0.01
gs.default_load_penalty = 900.0
return session
def build_model():
sim_start = datetime(2024, 1, 1, utc_offset, tzinfo=ltmtz)
session = create_basic_model(
"basic_model",
"testout_basic_model",
sim_start,
)
add_default_inflow_series(session)
# Busbars
tev = add_busbar(session, "tev")
no2 = add_busbar(session, "NO2")
# DCLines
dc1 = add_dclines(session, "DC_TEV_NO2")
session.connect(tev, dc1)
session.connect(dc1, no2)
# Loads
load_tev = add_load(session, "load_tev")
load_no2 = add_load(session, "load_no2")
session.connect(load_tev, tev)
session.connect(load_no2, tev)
# Market step
ms1 = add_market_step(session, "market1")
session.connect(ms1, no2)
# Wind
add_wind(session, "windy", no2, sim_start, 8)
# NO2 Topology
upper1 = create_simple_topology(session)
session.connect(upper1, no2)
return session
def run_model(session: LtmSession):
# with build_model() as session:
with session:
try:
# Write model to disk, and automatically generate an output directory.
session.write_model()
# 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"]
display_markdown(err)
else:
# Make plots from the results
generate_plots(session)
except Exception as e:
print(e)
raise (e)
[16]:
def debug_time():
from pyltm import util
pass
def go():
session = build_model()
run_model(session=session)
go()
busbar/tev
busbar/NO2
busbar/tev
busbar/NO2
