# imports # standard import pytest import random from networkx import binomial_tree, MultiDiGraph # local from src.topupopt.problems.esipp.network import Arcs, Network from src.topupopt.problems.esipp.network import ArcsWithoutLosses from src.topupopt.problems.esipp.network import ArcsWithoutProportionalLosses from src.topupopt.problems.esipp.network import ArcsWithoutStaticLosses from src.topupopt.problems.esipp.resource import ResourcePrice from src.topupopt.data.misc.utils import generate_pseudo_unique_key # ***************************************************************************** # ***************************************************************************** class TestNetwork: # ************************************************************************* # ************************************************************************* def test_arc_technologies_static_losses(self): number_time_intervals = 3 number_scenarios = 2 number_options = 4 efficiency_dict = { (q, k): 0.95 for q in range(number_scenarios) for k in range(number_time_intervals) } static_loss_dict = { (h, q, k): 1 for h in range(number_options) for q in range(number_scenarios) for k in range(number_time_intervals) } for capacity_is_instantaneous in (True, False): arc_tech = Arcs( name="any", efficiency=efficiency_dict, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss=static_loss_dict, validate=True, ) assert arc_tech.has_proportional_losses() assert arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) # isotropic arc_tech = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss=static_loss_dict, validate=True, ) assert not arc_tech.has_proportional_losses() assert arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert arc_tech.is_isotropic(reverse_none_means_isotropic=False) # create arc technology with only one option arc_tech = Arcs( name="any", efficiency=efficiency_dict, efficiency_reverse=None, capacity=(1,), minimum_cost=(1,), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss={ (0, q, k): 1 # for h in range(number_options) for q in range(number_scenarios) for k in range(number_time_intervals) }, validate=True, ) assert arc_tech.has_proportional_losses() assert arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) # create arc technology for one time interval arc_tech = Arcs( name="any", efficiency={ (q, 0): 0.5 for q in range(number_scenarios) # for k in range(number_time_intervals) }, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss={ (h, q, 0): 1 for h in range(number_options) for q in range(number_scenarios) # for k in range(number_time_intervals) }, validate=True, ) assert arc_tech.has_proportional_losses() assert arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) # ********************************************************************* # TypeError: The static losses should be given as a dict or None. error_raised = False try: _ = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss=tuple( [k for k in range(number_time_intervals)] for o in range(number_options) ), validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('The static losses should be specified for each arc # option.') error_raised = False try: _ = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss={ ( h, q, ): 1 for h in range(number_options) for q in range(number_scenarios) }, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError('The static losses must be specified via a list of lists.') error_raised = False try: _ = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss=[ tuple(k for k in range(number_time_intervals)) for o in range(number_options) ], validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('The static loss values are inconsistent with the number ' # 'of options, scenarios and intervals.') error_raised = False try: arc_tech = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss={ (h, q, k): 1 for h in range(number_options) for q in range(number_scenarios) for k in range(number_time_intervals - 1) }, validate=True, ) arc_tech.validate_sizes( number_options=number_options, number_scenarios=number_scenarios, number_intervals=[ number_time_intervals for _ in range(number_scenarios) ], ) except ValueError: error_raised = True assert error_raised # TypeError('The static losses were not provided as numbers.') error_raised = False try: _ = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss={ (h, q, k): str(3.54) for h in range(number_options) for q in range(number_scenarios) for k in range(number_time_intervals) }, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('The static losses must be positive or zero.') error_raised = False try: _ = Arcs( name="any", efficiency=None, efficiency_reverse=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, static_loss={ (h, q, k): -random.randint(0, 1) * random.random() for h in range(number_options) for q in range(number_scenarios) for k in range(number_time_intervals) }, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError: The static loss dict keys must be tuples error_raised = False try: _ = Arcs( name="hey", efficiency=None, efficiency_reverse=None, static_loss={k: 1 for k in range(number_time_intervals)}, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError( 'The static loss dict keys must be tuples of size 3.') error_raised = False try: _ = Arcs( name="hey", efficiency=None, efficiency_reverse=None, static_loss={(k, 3): 1 for k in range(number_time_intervals)}, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError(The staticl osses should be given as a dict or None.') error_raised = False try: _ = Arcs( name="hey", efficiency=None, efficiency_reverse=None, static_loss=[1 for k in range(number_time_intervals)], capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError( # 'No static loss values were provided. There should be one'+ # ' value per option, scenario and time interval.') error_raised = False try: _ = Arcs( name="hey", efficiency=None, efficiency_reverse=None, static_loss={}, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # ************************************************************************* # ************************************************************************* def test_arc_technologies(self): # ********************************************************************* # create arc technology using instantaneous capacities number_scenarios = 2 number_options = 4 number_time_intervals = 3 efficiency_dict = { (q, k): 0.85 for q in range(number_scenarios) for k in range(number_time_intervals) } for capacity_is_instantaneous in (True, False): arc_tech = Arcs( name="any", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert arc_tech.has_constant_efficiency() # create arc technology with only one option arc_tech = Arcs( name="any", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=(1,), minimum_cost=(1,), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert arc_tech.has_constant_efficiency() # create arc technology for one time interval arc_tech = Arcs( name="any", efficiency={(0, 0): 0.95}, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert arc_tech.has_constant_efficiency() # create arc technology for one time interval and isotropic arc_tech = Arcs( name="any", efficiency={(0, 0): 0.95}, efficiency_reverse={(0, 0): 0.95}, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert arc_tech.has_constant_efficiency() # create arc technology for one time interval and anisotropic arc_tech = Arcs( name="any", efficiency={(0, 0): 0.95}, efficiency_reverse={(0, 0): 1}, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert not arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert not arc_tech.has_constant_efficiency() # create arc technology for one time interval and anisotropic arc_tech = Arcs( name="any", efficiency={(0, 0): 1}, efficiency_reverse={(0, 0): 0.95}, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert not arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert not arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert not arc_tech.has_constant_efficiency() # create arc technology for one time interval and anisotropic arc_tech = Arcs( name="any", efficiency={(0, 0): 0.95}, efficiency_reverse={(0, 0): 0.95}, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_been_selected() assert arc_tech.is_isotropic(reverse_none_means_isotropic=True) assert arc_tech.is_isotropic(reverse_none_means_isotropic=False) assert arc_tech.has_constant_efficiency() # ***************************************************************** # ***************************************************************** # trigger errors # TypeError('The name attribute is not hashable.') error_raised = False try: _ = Arcs( name=[1, 2, 3], efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # TypeError:The efficiency dict keys must be (scenario, interval) tuples error_raised = False try: _ = Arcs( name="hey", efficiency={k: 1 for k in range(number_time_intervals)}, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError( 'The efficiency dict keys must be tuples of size 2.') error_raised = False try: _ = Arcs( name="hey", efficiency={(k, 3, 4): 1 for k in range(number_time_intervals)}, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError(The efficiency should be given as a dict or None.') error_raised = False try: _ = Arcs( name="hey", efficiency=[1 for k in range(number_time_intervals)], efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # TypeError('The reverse efficiency has to match the nominal'+ # ' one when there are no proportional losses.') error_raised = False try: _ = Arcs( name="hey", efficiency=None, efficiency_reverse={}, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # TypeError:'The reverse efficiency should be given as a dict or None.' error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=[1 for k in range(number_time_intervals)], static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError( # 'No efficiency values were provided. There should be '+ # 'one value per scenario and time interval.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse={}, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # ValueError: The keys for the efficiency dicts do not match. error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse={ (key[1], key[0]): value for key, value in efficiency_dict.items() }, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError: Efficiency values must be provided as numeric types. error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse={ (key[0], key[1]): str(value) for key, value in efficiency_dict.items() }, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('Efficiency values must be positive.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse={ (key[0], key[1]): -1 for key, value in efficiency_dict.items() }, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError('The capacity should be given as a list or tuple.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity={o: 1 + o for o in range(number_options)}, minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # TypeError: The minimum cost values should be given as a list or tuple error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost={o: 1 + o for o in range(number_options)}, specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # TypeError: The specific capacity cost was not given as a numeric type error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=[1], capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError:The number of capacity and minimum cost entries must match error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options + 1)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # ValueError: No entries for capacity and minimum cost were provided. # At least one option should be provided. error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(), minimum_cost=tuple(), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # ValueError: No entries for efficiency were provided. There should be # one entry per time interval. error_raised = False try: _ = Arcs( name="hey", efficiency={}, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # ValueError('The number of efficiency values must match the number of # time intervals.') arc_tech = Arcs( name="hey", efficiency={ (q, k): 0.85 for q in range(number_scenarios) for k in range(number_time_intervals + 1) }, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) error_raised = False try: arc_tech.validate_sizes( number_options=number_options, number_scenarios=number_scenarios, number_intervals=[ number_time_intervals for _ in range(number_scenarios) ], ) except ValueError: error_raised = True assert error_raised # ValueError('The number of efficiency values must match the number of # time intervals.') error_raised = False try: arc_tech = Arcs( name="hey", efficiency={ (q, k): 0.85 for q in range(number_scenarios) for k in range(number_time_intervals) }, efficiency_reverse={ (q, k): 0.85 for q in range(number_scenarios) for k in range(number_time_intervals - 1) }, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) arc_tech.validate_sizes( number_options=number_options, number_scenarios=number_scenarios, number_intervals=[ number_time_intervals for _ in range(number_scenarios) ], ) except ValueError: error_raised = True assert error_raised # ValueError('The number of capacity values must match the number of # options.') arc_tech = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options + 1)), minimum_cost=tuple(1 + o for o in range(number_options + 1)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) error_raised = False try: arc_tech.validate_sizes( number_options=number_options, number_scenarios=number_scenarios, number_intervals=[ number_time_intervals for _ in range(number_scenarios) ], ) except ValueError: error_raised = True assert error_raised # ValueError: The minimum cost values are inconsistent with the number # of options. arc_tech = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options + 1)), minimum_cost=tuple(1 + o for o in range(number_options + 1)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) error_raised = False try: arc_tech.validate_sizes( number_options=number_options, number_scenarios=number_scenarios, number_intervals=[ number_time_intervals for _ in range(number_scenarios) ], ) except ValueError: error_raised = True assert error_raised # TypeError('Efficiency values must be provided as numeric types.') error_raised = False try: _ = Arcs( name="hey", efficiency={ key: str(value) for key, value in efficiency_dict.items() }, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('Efficiency values must be positive.') error_raised = False try: _ = Arcs( name="hey", efficiency={ key: -value * random.randint(0, 1) for key, value in efficiency_dict.items() }, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError('Capacity values must be provided as numeric types.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(str(1 + o) for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('Capacity values must be positive.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple( -random.randint(0, 1) for o in range(number_options) ), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError('Minimum cost values must be provided as numeric types.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(str(1 + o) for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except TypeError: error_raised = True assert error_raised # ValueError('Minimum cost values must be positive or zero.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(-1 for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=capacity_is_instantaneous, validate=True, ) except ValueError: error_raised = True assert error_raised # TypeError('The information about capacities being instantaneous or not # should be given as a boolean variable.') error_raised = False try: _ = Arcs( name="hey", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=1, validate=True, ) except TypeError: error_raised = True assert error_raised # ********************************************************************* # ********************************************************************* # Network arc_tech_AB = Arcs( name="AB", efficiency=efficiency_dict, efficiency_reverse=None, static_loss=None, capacity=tuple(1 + o for o in range(number_options)), minimum_cost=tuple(1 + o for o in range(number_options)), specific_capacity_cost=1, capacity_is_instantaneous=False, validate=True, ) arc_tech_AB.options_selected[0] = True assert arc_tech_AB.number_options() == number_options net = Network() # add undirected arc net.add_undirected_arc(node_key_a="A", node_key_b="B", arcs=arc_tech_AB) # add directed arc net.add_directed_arc(node_key_a="A", node_key_b="B", arcs=arc_tech_AB) # add infinite capacity arc net.add_infinite_capacity_arc( node_key_a="C", node_key_b="D", efficiency={(i, j): 1 for i in range(3) for j in range(4)}, static_loss=None, ) # add pre-existing directed arc net.add_preexisting_directed_arc( node_key_a="E", node_key_b="F", efficiency=efficiency_dict, static_loss=None, capacity=3, capacity_is_instantaneous=True, ) # add pre-existing undirected arc net.add_preexisting_undirected_arc( node_key_a="A", node_key_b="C", efficiency=efficiency_dict, efficiency_reverse=efficiency_dict, static_loss=None, capacity=3, capacity_is_instantaneous=True, ) net.modify_network_arc( node_key_a="A", node_key_b="C", arc_key_ab="AC", data_dict={net.KEY_ARC_TECH: arc_tech_AB, net.KEY_ARC_UND: False}, ) # ********************************************************************* # ********************************************************************* # add import node imp_resource_price = ResourcePrice( prices=[random.random() for k in range(number_time_intervals)], volumes=[ *[random.random() for k in range(number_time_intervals - 1)], None, ], ) net.add_import_node("G", prices={(0, 0, 0): imp_resource_price}) # add export node exp_resource_price = ResourcePrice( prices=[random.random() for k in range(number_time_intervals)], volumes=[ *[random.random() for k in range(number_time_intervals - 1)], None, ], ) net.add_export_node("H", prices={(0, 0, 0): exp_resource_price}) net.add_waypoint_node("Z") base_flow = {(i, j): random.random() for i in range(3) for j in range(4)} net.add_source_sink_node("Y", base_flow=base_flow) base_flow[(2, 3)] = random.random() net.modify_node("Y", **{net.KEY_NODE_BASE_FLOW: base_flow}) assert "Z" in net.waypoint_nodes assert "G" in net.import_nodes assert "H" in net.export_nodes assert "Y" in net.source_sink_nodes # ************************************************************************* # ************************************************************************* def test_arcs_without_losses(self): # test arc without (static and proportional) losses arc_tech = ArcsWithoutLosses( name="AB", capacity=(1, 2, 3), minimum_cost=(4, 5, 6), specific_capacity_cost=6, capacity_is_instantaneous=False, validate=True, ) assert not arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert arc_tech.has_constant_efficiency() # test arc without static losses arc_tech = ArcsWithoutStaticLosses( name="AB", efficiency={(0, 0): 1, (0, 1): 0.9, (0, 2): 0.8}, efficiency_reverse=None, capacity=(1, 2, 3), minimum_cost=(4, 5, 6), specific_capacity_cost=6, capacity_is_instantaneous=False, validate=True, ) assert arc_tech.has_proportional_losses() assert not arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert not arc_tech.has_constant_efficiency() # test arc without proportional losses arc_tech = ArcsWithoutProportionalLosses( name="AB", static_loss={ (0, 0, 0): 0.1, (0, 0, 1): 0.2, (0, 0, 2): 0.3, (1, 0, 0): 0.15, (1, 0, 1): 0.25, (1, 0, 2): 0.35, (2, 0, 0): 0.16, (2, 0, 1): 0.26, (2, 0, 2): 0.36, }, capacity=(1, 2, 3), minimum_cost=(4, 5, 6), specific_capacity_cost=6, capacity_is_instantaneous=False, validate=True, ) assert not arc_tech.has_proportional_losses() assert arc_tech.has_static_losses() assert not arc_tech.is_infinite_capacity() assert arc_tech.has_constant_efficiency() # ************************************************************************* # ************************************************************************* def test_modifying_nodes(self): # ********************************************************************* net = Network() number_intervals = 3 resource_price = ResourcePrice( prices=[random.random() for k in range(number_intervals)], volumes=[*[random.random() for k in range(number_intervals - 1)], None], ) base_flow = {(0, k): random.random() for k in range(number_intervals)} arc_tech = ArcsWithoutLosses( name="hello", capacity=[5], minimum_cost=[3], specific_capacity_cost=3, capacity_is_instantaneous=False, ) # add isolated import node net.add_import_node("I_iso", prices={(0, 0, 0): resource_price}) # add import node with outgoing arcs net.add_import_node("I", prices={(0, 0, 0): resource_price}) # add isolated export node net.add_import_node("E_iso", prices={(0, 0, 0): resource_price}) # add export node with incoming arcs net.add_export_node("E", prices={(0, 0, 0): resource_price}) # add isolated normal node net.add_source_sink_node("A_iso", base_flow=base_flow) # add normal node with incoming arcs net.add_source_sink_node("A_in", base_flow=base_flow) # add normal node with outgoing arcs net.add_source_sink_node("A_out", base_flow=base_flow) # add normal node with incoming and outgoing arcs net.add_source_sink_node("A", base_flow=base_flow) # ********************************************************************* # arcs net.add_directed_arc(node_key_a="I", node_key_b="A_in", arcs=arc_tech) net.add_directed_arc(node_key_a="I", node_key_b="A", arcs=arc_tech) net.add_directed_arc(node_key_a="A_out", node_key_b="E", arcs=arc_tech) net.add_directed_arc(node_key_a="A", node_key_b="E", arcs=arc_tech) # ********************************************************************* # change I_iso to regular: okay net.modify_node( "I_iso", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # reverse: okay net.modify_node( "I_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # change I_iso to export: okay net.modify_node( "I_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # reverse: okay net.modify_node( "I_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # change I_iso to waypoint: okay net.modify_node( "I_iso" ) # reverse: okay net.modify_node( "I_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # ********************************************************************* # change E_iso to regular: okay net.modify_node( "E_iso", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # reverse: okay net.modify_node( "E_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # change E_iso to import: okay net.modify_node( "E_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # reverse: okay net.modify_node( "E_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # change E_iso to waypoint: okay net.modify_node( "E_iso" ) # reverse: okay net.modify_node( "E_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # ********************************************************************* # change A_iso to export: okay net.modify_node( "A_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # reverse: okay net.modify_node( "A_iso", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # change A_iso to import: okay net.modify_node( "A_iso", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # reverse: okay net.modify_node( "A_iso", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # change A_iso to waypoint: okay net.modify_node( "A_iso" ) # reverse: okay net.modify_node( "A_iso", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # ********************************************************************* # change I to regular: okay net.modify_node( "I", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # reverse: okay net.modify_node( "I", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # change I to waypoint: okay net.modify_node( "I" ) # reverse: okay net.modify_node( "I", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # ********************************************************************* # change E to regular: okay net.modify_node( "E", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # reverse: okay net.modify_node( "E", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # change E to waypoint: okay net.modify_node( "E" ) # reverse: okay net.modify_node( "E", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # ********************************************************************* # change A_in to export: okay net.modify_node( "A_in", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) # reverse: okay net.modify_node( "A_in", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # change A_in to waypoint: okay net.modify_node( "A_in", **{net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_WAY} ) # reverse: okay net.modify_node( "A_in", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # ********************************************************************* # change A_out to import: okay net.modify_node( "A_out", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) # reverse: okay net.modify_node( "A_out", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # change A_out to waypoint: okay net.modify_node( "A_out", **{net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_WAY} ) # reverse: okay net.modify_node( "A_out", **{ net.KEY_NODE_BASE_FLOW: base_flow, }, ) # ********************************************************************* # change I to export: fail error_raised = False try: net.modify_node( "I", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) except ValueError: error_raised = True assert error_raised # change E to import: fail error_raised = False try: net.modify_node( "E", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) except ValueError: error_raised = True assert error_raised # change A_out to export: fail error_raised = False try: net.modify_node( "A_out", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) except ValueError: error_raised = True assert error_raised # change A_in to import: fail error_raised = False try: net.modify_node( "A_in", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) except ValueError: error_raised = True assert error_raised # change A to export: fail error_raised = False try: net.modify_node( "A", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP, net.KEY_NODE_PRICES: resource_price, }, ) except ValueError: error_raised = True assert error_raised # change A to import: fail error_raised = False try: net.modify_node( "A", **{ net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP, net.KEY_NODE_PRICES: resource_price, }, ) except ValueError: error_raised = True assert error_raised # ********************************************************************* # try to modify a non-existent node error_raised = False try: net.modify_node("ABCD") except ValueError: error_raised = True assert error_raised # ********************************************************************* # ************************************************************************* # ************************************************************************* def test_network_disallowed_cases(self): net = Network() number_intervals = 3 resource_price = ResourcePrice( prices=[random.random() for k in range(number_intervals)], volumes=[*[random.random() for k in range(number_intervals - 1)], None], ) base_flow = {(0, k): random.random() for k in range(number_intervals)} lossless_arcs = ArcsWithoutLosses( name="hello", capacity=[5], minimum_cost=[3], specific_capacity_cost=3, capacity_is_instantaneous=False, ) lossy_arcs = ArcsWithoutProportionalLosses( name="hello back", static_loss={(0, 0, k): random.random() for k in range(number_intervals)}, capacity=(1,), minimum_cost=(5,), specific_capacity_cost=0, capacity_is_instantaneous=False, ) # add import node I net.add_import_node("I", prices={(0, 0, 0): resource_price}) # add export node E net.add_export_node("E", prices={(0, 0, 0): resource_price}) # add regular node A net.add_source_sink_node("A", base_flow=base_flow) # add regular node B net.add_source_sink_node("B", base_flow=base_flow) # add a valid import-export arc net.add_directed_arc(node_key_a="I", node_key_b="E", arcs=lossless_arcs) # ********************************************************************* # ********************************************************************* # trigger errors using pre-identified nodes # directed arcs cannot start in an export node: E -> B error_raised = False try: net.add_directed_arc(node_key_a="E", node_key_b="B", arcs=lossless_arcs) except ValueError: error_raised = True assert error_raised # directed arcs cannot end on an import node: A -> I error_raised = False try: net.add_directed_arc(node_key_a="A", node_key_b="I", arcs=lossless_arcs) except ValueError: error_raised = True assert error_raised # import-export nodes cannot have static losses error_raised = False try: net.add_directed_arc(node_key_a="I", node_key_b="E", arcs=lossy_arcs) except ValueError: error_raised = True assert error_raised # undirected arcs cannot involve import nor export nodes error_raised = False try: net.add_undirected_arc(node_key_a="I", node_key_b="A", arcs=lossless_arcs) except ValueError: error_raised = True assert error_raised # undirected arcs cannot involve import nor export nodes error_raised = False try: net.add_undirected_arc(node_key_a="B", node_key_b="E", arcs=lossless_arcs) except ValueError: error_raised = True assert error_raised # undirected arcs cannot involve import nor export nodes error_raised = False try: net.add_undirected_arc(node_key_a="I", node_key_b="E", arcs=lossy_arcs) except ValueError: error_raised = True assert error_raised # ********************************************************************* # trigger errors using non-identified nodes # ********************************************************************* # create a new export node net.add_export_node("E1", prices={(0, 0, 0): resource_price}) # create an arc starting in that export node error_raised = False try: net.add_directed_arc(node_key_a="E1", node_key_b="B", arcs=lossless_arcs) except ValueError: error_raised = True assert error_raised # # remove the troublesome arc # net.remove_edge(u="E1", v="B") # ********************************************************************* # create a new import node net.add_import_node("I1", prices={(0, 0, 0): resource_price}) # create an arc ending in that import node error_raised = False try: net.add_directed_arc(node_key_a="A", node_key_b="I1", arcs=lossless_arcs) except ValueError: error_raised = True assert error_raised # # remove the troublesome arc # net.remove_edge(u="A", v="I1") # ********************************************************************* # check non-existent arc net.arc_is_undirected(("X", "Y", 1)) # ************************************************************************* # ************************************************************************* def test_undirected_arc_import_error(self): # network mynet = Network() # import node imp_node_key = generate_pseudo_unique_key(mynet.nodes()) mynet.add_import_node( imp_node_key, prices={ (0, 0, 0): ResourcePrice(prices=1+0.05, volumes=None) }, ) # other nodes node_A = generate_pseudo_unique_key(mynet.nodes()) mynet.add_source_sink_node( node_A, # base_flow=[1, -1, 0.5, -0.5] base_flow={(0, 0): 1, (0, 1): -1, (0, 2): 0.5, (0, 3): -0.5}, ) node_B = generate_pseudo_unique_key(mynet.nodes()) mynet.add_source_sink_node( node_B, # base_flow=[-1, 1, -0.5, 0.5] base_flow={(0, 0): -1, (0, 1): 1, (0, 2): -0.5, (0, 3): 0.5}, ) # add arcs # import arc arc_tech_IA = Arcs( name="any", # efficiency=[1, 1, 1, 1], efficiency={(0, 0): 1, (0, 1): 1, (0, 2): 1, (0, 3): 1}, capacity=[0.5, 0.75, 1.0, 1.25, 1.5, 2.0], minimum_cost=[10, 10.1, 10.2, 10.3, 10.4, 10.5], specific_capacity_cost=1, capacity_is_instantaneous=False, efficiency_reverse=None, static_loss=None, validate=False, ) error_raised = False try: # ValueError: Undirected arcs cannot involve import or export nodes. mynet.add_undirected_arc( node_key_a=imp_node_key, node_key_b=node_A, arcs=arc_tech_IA ) except ValueError: error_raised = True assert error_raised # ********************************************************************* # ********************************************************************* # ************************************************************************* # ************************************************************************* def test_undirected_arc_export_error(self): # 4 nodes: one import, one export, two supply/demand nodes mynet = Network() # export node exp_node_key = generate_pseudo_unique_key(mynet.nodes()) mynet.add_export_node( exp_node_key, prices={ (0, 0, 0): ResourcePrice(prices=0.1+0.05, volumes=None) }, ) # other nodes node_B = generate_pseudo_unique_key(mynet.nodes()) mynet.add_source_sink_node( node_B, # base_flow=[-1, 1, -0.5, 0.5] base_flow={(0, 0): -1, (0, 1): 1, (0, 2): -0.5, (0, 3): 0.5}, ) # export arc arc_tech_BE = Arcs( name="any", # efficiency=[1, 1, 1, 1], efficiency={(0, 0): 1, (0, 1): 1, (0, 2): 1, (0, 3): 1}, capacity=[0.5, 0.75, 1.0, 1.25, 1.5, 2.0], minimum_cost=[10, 10.1, 10.2, 10.3, 10.4, 10.5], specific_capacity_cost=1, capacity_is_instantaneous=False, efficiency_reverse=None, static_loss=None, validate=False, ) error_raised = False try: # ValueError: Undirected arcs cannot involve import or export nodes. mynet.add_undirected_arc( node_key_a=node_B, node_key_b=exp_node_key, arcs=arc_tech_BE ) except ValueError: error_raised = True assert error_raised # ************************************************************************* # ************************************************************************* def test_tree_topology(self): # create a network object with a tree topology tree_network = binomial_tree(3, create_using=MultiDiGraph) network = Network(network_type=Network.NET_TYPE_TREE, incoming_graph_data=tree_network) for edge_key in network.edges(keys=True): arc = ArcsWithoutLosses( name=str(edge_key), capacity=[5, 10], minimum_cost=[3, 6], specific_capacity_cost=0, capacity_is_instantaneous=False, ) network.add_edge(*edge_key, **{Network.KEY_ARC_TECH: arc}) # assert that it should have a tree topology assert network.should_be_tree_network() # assert that it does not have a tree topology assert not network.has_tree_topology() # select all the nodes for edge_key in network.edges(keys=True): network.edges[edge_key][Network.KEY_ARC_TECH].options_selected[0] = True # assert that it has a tree topology assert network.has_tree_topology() # ************************************************************************* # ************************************************************************* def test_pseudo_unique_key_generation(self): # create network network = Network() # add nodes A and B network.add_nodes_from(['A','B']) # add arcs key_list = [ "3e225573-4e78-48c8-bb08-efbeeb795c22", "f6d30428-15d1-41e9-a952-0742eaaa5a31", "8c29b906-2518-41c5-ada8-07b83508b5b8", "f9a72a39-1422-4a02-af97-906ce79c32a3", "b6941a48-10cc-465d-bf53-178bd2939bd1", ] for key in key_list: network.add_edge( u_for_edge="A", v_for_edge="B", key=key, **{network.KEY_ARC_UND: False, network.KEY_ARC_TECH: None} ) # use a seed number to trigger more iterations import uuid rand = random.Random() rand.seed(360) uuid.uuid4 = lambda: uuid.UUID(int=rand.getrandbits(128), version=4) error_raised = False try: _ = network.get_pseudo_unique_arc_key( node_key_start="A", node_key_end="B", max_iterations=len(key_list) - 1 ) except Exception: error_raised = True assert error_raised # ************************************************************************* # ************************************************************************* def test_imp_exp_static_losses(self): # assessment q = 0 # 4 nodes: one import, one export, two supply/demand nodes mynet = Network() # import node imp_node_key = generate_pseudo_unique_key(mynet.nodes()) imp_prices = { qpk: ResourcePrice( prices=0.5, volumes=None, ) for qpk in [(0,0,0),(0,0,1),(0,1,0),(0,1,1)] } mynet.add_import_node( imp_node_key, prices=imp_prices ) # export node exp_node_key = generate_pseudo_unique_key(mynet.nodes()) exp_prices = { qpk: ResourcePrice( prices=1.5, volumes=None, ) for qpk in [(0,0,0),(0,0,1),(0,1,0),(0,1,1)] } mynet.add_export_node( exp_node_key, prices=exp_prices, ) # add arc with fixed losses from import node to export arc_tech_IE_fix = Arcs( name="IE_fix", # efficiency=[1, 1, 1, 1], efficiency={(q, 0): 1, (q, 1): 1}, efficiency_reverse=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, # static_losses=[ # [0.10, 0.15, 0.20, 0.25], # [0.15, 0.20, 0.25, 0.30], # [0.20, 0.25, 0.30, 0.35]] static_loss={ (0, q, 0): 0.10, (0, q, 1): 0.15, (1, q, 0): 0.15, (1, q, 1): 0.20, (2, q, 0): 0.20, (2, q, 1): 0.25, }, ) error_raised = False try: # ValueError: Arcs between import and export nodes cannot have static losses. mynet.add_directed_arc( node_key_a=imp_node_key, node_key_b=exp_node_key, arcs=arc_tech_IE_fix ) except ValueError: error_raised = True assert error_raised # ************************************************************************* # ************************************************************************* def test_antiparallel_arcs(self): # create network net = Network() # add nodes node_a = 'A' # net.add_waypoint_node(node_a) node_b = 'B' # net.add_waypoint_node(node_b) node_c = 'C' # net.add_waypoint_node(node_c) net.add_nodes_from([node_a,node_b,node_c]) # add arcs node_pairs = ((node_a, node_b), (node_b, node_a),) # test network for node_pair in node_pairs: net.add_preexisting_directed_arc( *node_pair, efficiency=None, static_loss=None, capacity=1, capacity_is_instantaneous=False ) # assert that it can detected the selected antiparallel arcs assert net.has_selected_antiparallel_arcs() # check that it finds the right node pairs identified_node_pairs = net.find_selected_antiparallel_arcs() assert (node_a, node_b) in identified_node_pairs assert (node_b, node_a) in identified_node_pairs # ************************************************************************* # ************************************************************************* def test_add_nodes(self): # create network net = Network() # add nodes node_a = 'A' net.add_node(node_a) assert net.is_waypoint_node(node_a) # source node_b = 'B' net.add_node(node_b, **{net.KEY_NODE_BASE_FLOW: {(0,0):-1}}) assert net.is_source_sink_node(node_b) # sink node_c = 'C' net.add_node(node_c, **{net.KEY_NODE_BASE_FLOW: {(0,0):1}}) assert net.is_source_sink_node(node_c) # import node node_d = 'D' net.add_node(node_d, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[1, 2], volumes=[1,None])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP}) assert net.is_import_node(node_d) # export node node_e = 'E' net.add_node(node_e, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[2, 3], volumes=[4,None])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP}) assert net.is_export_node(node_e) # modify nodes # from waypoint to source/sink net.modify_node(node_a, **{net.KEY_NODE_BASE_FLOW: {(0,0):-2}}) assert not net.is_waypoint_node(node_a) assert net.is_source_sink_node(node_a) # from source/sink to waypoint net.modify_node(node_a) assert not net.is_source_sink_node(node_a) assert net.is_waypoint_node(node_a) # from waypoint to import node net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[5, 3.5], volumes=[2,4])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP}) assert not net.is_waypoint_node(node_a) assert net.is_import_node(node_a) # from import node to waypoint net.modify_node(node_a) assert not net.is_import_node(node_a) assert net.is_waypoint_node(node_a) # from waypoint node to export node net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP}) assert not net.is_waypoint_node(node_a) assert net.is_export_node(node_a) # from export node to sink/source net.modify_node(node_a, **{net.KEY_NODE_BASE_FLOW: {(0,0):-1}}) assert not net.is_export_node(node_a) assert net.is_source_sink_node(node_a) # from sink/source node to import node net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[5, 3.5], volumes=[2,4])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP}) assert not net.is_source_sink_node(node_a) assert net.is_import_node(node_a) # from import node to export node net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP}) assert not net.is_import_node(node_a) assert net.is_export_node(node_a) # from export node to waypoint node net.modify_node(node_a) assert not net.is_export_node(node_a) assert net.is_waypoint_node(node_a) # ********************************************************************* # test modifying nodes with preexisting arcs # add arcs # add arc between two waypoint nodes net.add_preexisting_directed_arc( node_key_a=node_a, node_key_b=node_b, efficiency=None, static_loss=None, capacity=3, capacity_is_instantaneous=False ) # modify nodes # try to change the start node to an export node with pytest.raises(ValueError): net.modify_node(node_a, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_EXP}) # try to change the end node to an import node with pytest.raises(ValueError): net.modify_node(node_b, **{net.KEY_NODE_PRICES: {(0,0): ResourcePrice(prices=[4, 1], volumes=[3,6])}, net.KEY_NODE_TYPE: net.KEY_NODE_TYPE_IMP}) # ************************************************************************* # ************************************************************************* # ***************************************************************************** # *****************************************************************************