# imports
# standard
import math
import pytest
# local
# import numpy as np
# import networkx as nx
import pyomo.environ as pyo
# import src.topupopt.problems.esipp.utils as utils
from src.topupopt.data.misc.utils import generate_pseudo_unique_key
from src.topupopt.problems.esipp.problem import InfrastructurePlanningProblem
from src.topupopt.problems.esipp.network import Arcs, Network
from src.topupopt.problems.esipp.resource import ResourcePrice
# from src.topupopt.problems.esipp.utils import compute_cost_volume_metrics
from src.topupopt.problems.esipp.utils import statistics
from src.topupopt.problems.esipp.time import EconomicTimeFrame
# from src.topupopt.problems.esipp.converter import Converter
from test_esipp import build_solve_ipp
# *****************************************************************************
# *****************************************************************************
class TestESIPPProblem:
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_increasing_imp_prices(self, use_prices_block):
# assessment
q = 0
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,)},
time_interval_durations={q: (1,)},
)
# 2 nodes: one import, one regular
mynet = Network()
# import node
node_IMP = 'I'
mynet.add_import_node(
node_key=node_IMP,
prices={
qpk: ResourcePrice(prices=[1.0, 2.0], volumes=[0.5, None])
for qpk in tf.qpk()
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(node_key=node_A, base_flow={(q, 0): 1.0})
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_IMP, node_key_b=node_A, arcs=arc_tech_IA)
# no sos, regular time intervals
ipp = build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
use_prices_block=use_prices_block
)
assert not ipp.has_peak_total_assessments()
# *********************************************************************
# *********************************************************************
# validation
# the arc should be installed since it is required for feasibility
assert (
True
in ipp.networks["mynet"]
.edges[(node_IMP, node_A, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# the flows should be 1.0, 0.0 and 2.0
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_IMP, node_A, 0, q, 0)]),
2.0,
abs_tol=1e-6,
)
# arc amplitude should be two
assert math.isclose(
pyo.value(ipp.instance.var_v_amp_gllj[("mynet", node_IMP, node_A, 0)]),
2.0,
abs_tol=0.01,
)
# capex should be four
assert math.isclose(pyo.value(ipp.instance.var_capex), 4.0, abs_tol=1e-3)
# sdncf should be -3.5
assert math.isclose(pyo.value(ipp.instance.var_sdncf_q[q]), -3.5, abs_tol=1e-3)
# the objective function should be -7.5
assert math.isclose(pyo.value(ipp.instance.obj_f), -7.5, abs_tol=1e-3)
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_decreasing_imp_prices(self, use_prices_block):
# assessment
q = 0
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,)},
time_interval_durations={q: (1,)},
)
# 2 nodes: one import, one regular
mynet = Network()
# import node
node_IMP = 'I'
mynet.add_import_node(
node_key=node_IMP,
prices={
qpk: ResourcePrice(prices=[2.0, 1.0], volumes=[0.5, 3.0])
for qpk in tf.qpk()
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(node_key=node_A, base_flow={(q, 0): 1.0})
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_IMP, node_key_b=node_A, arcs=arc_tech_IA)
# no sos, regular time intervals
ipp = build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
use_prices_block=use_prices_block
)
assert not ipp.has_peak_total_assessments()
# *********************************************************************
# *********************************************************************
# validation
# the arc should be installed since it is required for feasibility
assert (
True
in ipp.networks["mynet"]
.edges[(node_IMP, node_A, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# the flows should be 1.0, 0.0 and 2.0
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_IMP, node_A, 0, q, 0)]),
2.0,
abs_tol=1e-6,
)
# arc amplitude should be two
assert math.isclose(
pyo.value(ipp.instance.var_v_amp_gllj[("mynet", node_IMP, node_A, 0)]),
2.0,
abs_tol=0.01,
)
# capex should be four
assert math.isclose(pyo.value(ipp.instance.var_capex), 4.0, abs_tol=1e-3)
# sdncf should be -2.5
assert math.isclose(pyo.value(ipp.instance.var_sdncf_q[q]), -2.5, abs_tol=1e-3)
# the objective function should be -7.5
assert math.isclose(pyo.value(ipp.instance.obj_f), -6.5, abs_tol=1e-3)
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_decreasing_imp_prices_infinite_capacity(self, use_prices_block):
# assessment
q = 0
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,)},
time_interval_durations={q: (1,)},
)
# 2 nodes: one import, one regular
mynet = Network()
# import node
node_IMP = 'I'
mynet.add_import_node(
node_key=node_IMP,
prices={
qpk: ResourcePrice(prices=[2.0, 1.0], volumes=[0.5, None])
for qpk in tf.qpk()
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(node_key=node_A, base_flow={(q, 0): 1.0})
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_IMP, node_key_b=node_A, arcs=arc_tech_IA)
# trigger the error
error_raised = False
try:
# no sos, regular time intervals
build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
)
except Exception:
error_raised = True
assert error_raised
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_decreasing_exp_prices(self, use_prices_block):
# assessment
q = 0
# time
number_intervals = 1
# periods
number_periods = 1
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,)},
time_interval_durations={q: (1,)},
)
# 2 nodes: one export, one regular
mynet = Network()
# import node
node_EXP = generate_pseudo_unique_key(mynet.nodes())
mynet.add_export_node(
node_key=node_EXP,
prices={
(q, p, k): ResourcePrice(prices=[2.0, 1.0], volumes=[0.5, None])
for p in range(number_periods)
for k in range(number_intervals)
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(node_key=node_A, base_flow={(q, 0): -1.0})
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_A, node_key_b=node_EXP, arcs=arc_tech_IA)
# no sos, regular time intervals
ipp = build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
)
assert not ipp.has_peak_total_assessments()
# *********************************************************************
# *********************************************************************
# validation
# the arc should be installed since it is required for feasibility
assert (
True
in ipp.networks["mynet"]
.edges[(node_A, node_EXP, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# the flows should be 1.0, 0.0 and 2.0
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_A, node_EXP, 0, q, 0)]),
1.0,
abs_tol=1e-6,
)
# arc amplitude should be two
assert math.isclose(
pyo.value(ipp.instance.var_v_amp_gllj[("mynet", node_A, node_EXP, 0)]),
1.0,
abs_tol=0.01,
)
# capex should be four
assert math.isclose(pyo.value(ipp.instance.var_capex), 3.0, abs_tol=1e-3)
# sdncf should be 1.0
assert math.isclose(pyo.value(ipp.instance.var_sdncf_q[q]), 1.0, abs_tol=1e-3)
# the objective function should be -7.5
assert math.isclose(pyo.value(ipp.instance.obj_f), -2.0, abs_tol=1e-3)
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_increasing_exp_prices(self, use_prices_block):
# assessment
q = 0
# time
number_intervals = 1
# periods
number_periods = 1
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,)},
time_interval_durations={q: (1,)},
)
# 2 nodes: one export, one regular
mynet = Network()
# import node
node_EXP = generate_pseudo_unique_key(mynet.nodes())
mynet.add_export_node(
node_key=node_EXP,
prices={
(q, p, k): ResourcePrice(prices=[1.0, 2.0], volumes=[0.25, 3.0])
for p in range(number_periods)
for k in range(number_intervals)
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(node_key=node_A, base_flow={(q, 0): -1.0})
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_A, node_key_b=node_EXP, arcs=arc_tech_IA)
# no sos, regular time intervals
ipp = build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
)
assert not ipp.has_peak_total_assessments()
# *********************************************************************
# *********************************************************************
# validation
# the arc should be installed since it is required for feasibility
assert (
True
in ipp.networks["mynet"]
.edges[(node_A, node_EXP, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# the flows should be 1.0, 0.0 and 2.0
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_A, node_EXP, 0, q, 0)]),
1.0,
abs_tol=1e-6,
)
# arc amplitude should be two
assert math.isclose(
pyo.value(ipp.instance.var_v_amp_gllj[("mynet", node_A, node_EXP, 0)]),
1.0,
abs_tol=0.01,
)
# capex should be four
assert math.isclose(pyo.value(ipp.instance.var_capex), 3.0, abs_tol=1e-3)
# sdncf should be 0.75
assert math.isclose(pyo.value(ipp.instance.var_sdncf_q[q]), 0.75, abs_tol=1e-3)
# the objective function should be -2.25
assert math.isclose(pyo.value(ipp.instance.obj_f), -2.25, abs_tol=1e-3)
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_increasing_exp_prices_infinite_capacity(self, use_prices_block):
# assessment
q = 0
# time
number_intervals = 1
# periods
number_periods = 1
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,)},
time_interval_durations={q: (1,)},
)
# 2 nodes: one export, one regular
mynet = Network()
# import node
node_EXP = generate_pseudo_unique_key(mynet.nodes())
mynet.add_export_node(
node_key=node_EXP,
prices={
(q, p, k): ResourcePrice(prices=[1.0, 2.0], volumes=[0.25, None])
for p in range(number_periods)
for k in range(number_intervals)
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(node_key=node_A, base_flow={(q, 0): -1.0})
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_A, node_key_b=node_EXP, arcs=arc_tech_IA)
# trigger the error
error_raised = False
try:
# no sos, regular time intervals
build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
use_prices_block=use_prices_block
)
except Exception:
error_raised = True
assert error_raised
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_problem_increasing_imp_decreasing_exp_prices(self, use_prices_block):
# scenario
q = 0
# time
number_intervals = 2
# periods
number_periods = 1
tf = EconomicTimeFrame(
discount_rate=0.0,
reporting_periods={q: (0,)},
reporting_period_durations={q: (365 * 24 * 3600,)},
time_intervals={q: (0,1)},
time_interval_durations={q: (1,1)},
)
# 3 nodes: one import, one export, one regular
mynet = Network()
# import node
node_IMP = 'I'
mynet.add_import_node(
node_key=node_IMP,
prices={
(q, p, k): ResourcePrice(prices=[1.0, 2.0], volumes=[0.5, None])
for p in range(number_periods)
for k in range(number_intervals)
},
)
# export node
node_EXP = generate_pseudo_unique_key(mynet.nodes())
mynet.add_export_node(
node_key=node_EXP,
prices={
(q, p, k): ResourcePrice(prices=[2.0, 1.0], volumes=[0.5, None])
for p in range(number_periods)
for k in range(number_intervals)
},
)
# other nodes
node_A = 'A'
mynet.add_source_sink_node(
node_key=node_A, base_flow={(q, 0): 1.0, (q, 1): -1.0}
)
# arc IA
arc_tech_IA = Arcs(
name="any",
efficiency={(q, 0): 0.5, (q, 1): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_IMP, node_key_b=node_A, arcs=arc_tech_IA)
# arc AE
arc_tech_AE = Arcs(
name="any",
efficiency={(q, 0): 0.5, (q, 1): 0.5},
efficiency_reverse=None,
static_loss=None,
capacity=[3],
minimum_cost=[2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
validate=False,
)
mynet.add_directed_arc(node_key_a=node_A, node_key_b=node_EXP, arcs=arc_tech_AE)
# no sos, regular time intervals
ipp = build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
time_frame=tf,
networks={"mynet": mynet},
static_losses_mode=True, # just to reach a line,
mandatory_arcs=[],
max_number_parallel_arcs={},
simplify_problem=False,
# discount_rates={0: (0.0,)},
use_prices_block=use_prices_block
)
assert not ipp.has_peak_total_assessments()
# *********************************************************************
# *********************************************************************
# validation
# the arc should be installed since it is required for feasibility
assert (
True
in ipp.networks["mynet"]
.edges[(node_IMP, node_A, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# the arc should be installed since it is required for feasibility
assert (
True
in ipp.networks["mynet"]
.edges[(node_A, node_EXP, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# interval 0: import only
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_IMP, node_A, 0, q, 0)]),
2.0,
abs_tol=1e-6,
)
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_A, node_EXP, 0, q, 0)]),
0.0,
abs_tol=1e-6,
)
# interval 1: export only
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_IMP, node_A, 0, q, 1)]),
0.0,
abs_tol=1e-6,
)
assert math.isclose(
pyo.value(ipp.instance.var_v_glljqk[("mynet", node_A, node_EXP, 0, q, 1)]),
1.0,
abs_tol=1e-6,
)
# IA amplitude
assert math.isclose(
pyo.value(ipp.instance.var_v_amp_gllj[("mynet", node_IMP, node_A, 0)]),
2.0,
abs_tol=0.01,
)
# AE amplitude
assert math.isclose(
pyo.value(ipp.instance.var_v_amp_gllj[("mynet", node_A, node_EXP, 0)]),
1.0,
abs_tol=0.01,
)
# capex should be 7.0: 4+3
assert math.isclose(pyo.value(ipp.instance.var_capex), 7.0, abs_tol=1e-3)
# sdncf should be -2.5: -3.5+1.0
assert math.isclose(pyo.value(ipp.instance.var_sdncf_q[q]), -2.5, abs_tol=1e-3)
# the objective function should be -9.5: -7.5-2.5
assert math.isclose(pyo.value(ipp.instance.obj_f), -9.5, abs_tol=1e-3)
# *************************************************************************
# *************************************************************************
@pytest.mark.parametrize("use_prices_block", [True, False])
def test_direct_imp_exp_network_higher_exp_prices(self, use_prices_block):
# time frame
q = 0
tf = EconomicTimeFrame(
discount_rate=3.5/100,
reporting_periods={q: (0,1)},
reporting_period_durations={q: (365 * 24 * 3600,365 * 24 * 3600)},
time_intervals={q: (0,1)},
time_interval_durations={q: (1,1)},
)
# 4 nodes: one import, one export, two supply/demand nodes
mynet = Network()
# import node
imp_node_key = 'thatimpnode'
imp_prices = {
qpk: ResourcePrice(
prices=0.5,
volumes=None,
)
for qpk in tf.qpk()
}
mynet.add_import_node(
node_key=imp_node_key,
prices=imp_prices
)
# export node
exp_node_key = 'thatexpnode'
exp_prices = {
qpk: ResourcePrice(
prices=1.5,
volumes=None,
)
for qpk in tf.qpk()
}
mynet.add_export_node(
node_key=exp_node_key,
prices=exp_prices,
)
# add arc without fixed losses from import node to export
arc_tech_IE = Arcs(
name="IE",
# efficiency=[1, 1, 1, 1],
efficiency={(0, 0): 1, (0, 1): 1, (0, 2): 1, (0, 3): 1},
efficiency_reverse=None,
static_loss=None,
validate=False,
capacity=[0.5, 1.0, 2.0],
minimum_cost=[5, 5.1, 5.2],
specific_capacity_cost=1,
capacity_is_instantaneous=False,
)
mynet.add_directed_arc(
node_key_a=imp_node_key, node_key_b=exp_node_key, arcs=arc_tech_IE
)
# no sos, regular time intervals
ipp = build_solve_ipp(
solver_options={},
perform_analysis=False,
plot_results=False, # True,
print_solver_output=False,
networks={"mynet": mynet},
time_frame=tf,
static_losses_mode=InfrastructurePlanningProblem.STATIC_LOSS_MODE_DEP,
mandatory_arcs=[],
max_number_parallel_arcs={},
use_prices_block=use_prices_block
)
# export prices are higher: it makes sense to install the arc since the
# revenue (@ max. cap.) exceeds the cost of installing the arc
assert (
True
in ipp.networks["mynet"]
.edges[(imp_node_key, exp_node_key, 0)][Network.KEY_ARC_TECH]
.options_selected
)
# overview
(imports_qpk,
exports_qpk,
balance_qpk,
import_costs_qpk,
export_revenue_qpk,
ncf_qpk,
aggregate_static_demand_qpk,
aggregate_static_supply_qpk,
aggregate_static_balance_qpk) = statistics(ipp)
# there should be no imports
abs_tol = 1e-6
abs_tol = 1e-3
imports_qp = sum(imports_qpk[qpk] for qpk in tf.qpk() if qpk[1] == 0)
assert imports_qp > 0.0 - abs_tol
abs_tol = 1e-3
import_costs_qp = sum(import_costs_qpk[qpk] for qpk in tf.qpk() if qpk[1] == 0)
assert import_costs_qp > 0.0 - abs_tol
# there should be no exports
abs_tol = 1e-2
exports_qp = sum(exports_qpk[(q, 0, k)] for k in tf.time_intervals[q])
export_revenue_qp = sum(export_revenue_qpk[(q, 0, k)] for k in tf.time_intervals[q])
assert exports_qp > 0.0 - abs_tol
assert export_revenue_qp > 0.0 - abs_tol
# the revenue should exceed the costs
abs_tol = 1e-2
assert (
export_revenue_qp > import_costs_qp - abs_tol
)
# the capex should be positive
abs_tol = 1e-6
assert pyo.value(ipp.instance.var_capex) > 0 - abs_tol
# *************************************************************************
# *************************************************************************
# *****************************************************************************
# *****************************************************************************