Multi period expansion planning

mypy.ini

@@ -7,6 +7,9 @@ disallow_untyped_calls = true
 [mypy-ortools.*]
 ignore_missing_imports = true
 
+[mypy-anytree.*]
+ignore_missing_imports = true
+
 [mypy-andromede.expression.parsing.antlr.*]
 ignore_errors = True
 

requirements.txt

@@ -1,6 +1,7 @@
 numpy==1.24.4
 ortools==9.9.3963
 scipy==1.10.1
+anytree==2.12.1
 antlr4-python3-runtime==4.13.1
 PyYAML~=6.0.1
 pydantic~=2.6.1

src/andromede/expression/expression.py

@@ -332,6 +332,14 @@ def literal(value: float) -> LiteralNode:
     return LiteralNode(value)
 
 
+def is_unbound(expr: ExpressionNode) -> bool:
+    return isinstance(expr, LiteralNode) and (abs(expr.value) == float("inf"))
+
+
+def is_non_negative(expr: ExpressionNode) -> bool:
+    return isinstance(expr, LiteralNode) and (expr.value >= 0)
+
+
 @dataclass(frozen=True, eq=False)
 class UnaryOperatorNode(ExpressionNode):
     operand: ExpressionNode

src/andromede/expression/parsing/parse_expression.py

@@ -12,7 +12,7 @@
 from dataclasses import dataclass
 from typing import Set
 
-from antlr4 import CommonTokenStream, DiagnosticErrorListener, InputStream
+from antlr4 import CommonTokenStream, InputStream
 from antlr4.error.ErrorStrategy import BailErrorStrategy
 
 from andromede.expression import ExpressionNode, literal, param, var

src/andromede/expression/port_resolver.py

@@ -19,12 +19,7 @@
     PortFieldAggregatorNode,
     PortFieldNode,
 )
-
-
-@dataclass(frozen=True)
-class PortFieldId:
-    port_name: str
-    field_name: str
+from andromede.model.port import PortFieldId
 
 
 @dataclass(eq=True, frozen=True)

src/andromede/libs/standard.py

@@ -16,10 +16,11 @@
 from andromede.expression import literal, param, var
 from andromede.expression.expression import port_field
 from andromede.expression.indexing_structure import IndexingStructure
+from andromede.model.common import ProblemContext
 from andromede.model.constraint import Constraint
-from andromede.model.model import ModelPort, PortFieldDefinition, PortFieldId, model
+from andromede.model.model import ModelPort, model
 from andromede.model.parameter import float_parameter, int_parameter
-from andromede.model.port import PortField, PortType
+from andromede.model.port import PortField, PortFieldDefinition, PortFieldId, PortType
 from andromede.model.variable import float_variable, int_variable
 
 CONSTANT = IndexingStructure(False, False)
@@ -41,7 +42,7 @@
     ],
 )
 
-NODE_WITH_SPILL_AND_ENS_MODEL = model(
+NODE_WITH_SPILL_AND_ENS = model(
     id="NODE_WITH_SPILL_AND_ENS_MODEL",
     parameters=[float_parameter("spillage_cost"), float_parameter("ens_cost")],
     variables=[
@@ -196,6 +197,39 @@
     .expec(),
 )
 
+"""
+A model for a linear cost generation limited by a maximum generation per time-step
+and total generation in whole period. It considers a full storage with no replenishing
+"""
+GENERATOR_MODEL_WITH_STORAGE = model(
+    id="GEN",
+    parameters=[
+        float_parameter("p_max", CONSTANT),
+        float_parameter("cost", CONSTANT),
+        float_parameter("full_storage", CONSTANT),
+    ],
+    variables=[float_variable("generation", lower_bound=literal(0))],
+    ports=[ModelPort(port_type=BALANCE_PORT_TYPE, port_name="balance_port")],
+    port_fields_definitions=[
+        PortFieldDefinition(
+            port_field=PortFieldId("balance_port", "flow"),
+            definition=var("generation"),
+        )
+    ],
+    constraints=[
+        Constraint(
+            name="Max generation", expression=var("generation") <= param("p_max")
+        ),
+        Constraint(
+            name="Total storage",
+            expression=var("generation").time_sum() <= param("full_storage"),
+        ),
+    ],
+    objective_operational_contribution=(param("cost") * var("generation"))
+    .time_sum()
+    .expec(),
+)
+
 # For now, no starting cost
 THERMAL_CLUSTER_MODEL_HD = model(
     id="GEN",
@@ -418,3 +452,76 @@
     ],
     objective_operational_contribution=literal(0),  # Implcitement nul ?
 )
+
+""" Simple thermal unit that can be invested on"""
+THERMAL_CANDIDATE = model(
+    id="GEN",
+    parameters=[
+        float_parameter("op_cost", CONSTANT),
+        float_parameter("invest_cost", CONSTANT),
+        float_parameter("max_invest", CONSTANT),
+    ],
+    variables=[
+        float_variable("generation", lower_bound=literal(0)),
+        float_variable(
+            "p_max",
+            lower_bound=literal(0),
+            upper_bound=param("max_invest"),
+            structure=CONSTANT,
+            context=ProblemContext.COUPLING,
+        ),
+    ],
+    ports=[ModelPort(port_type=BALANCE_PORT_TYPE, port_name="balance_port")],
+    port_fields_definitions=[
+        PortFieldDefinition(
+            port_field=PortFieldId("balance_port", "flow"),
+            definition=var("generation"),
+        )
+    ],
+    constraints=[
+        Constraint(name="Max generation", expression=var("generation") <= var("p_max"))
+    ],
+    objective_operational_contribution=(param("op_cost") * var("generation"))
+    .time_sum()
+    .expec(),
+    objective_investment_contribution=param("invest_cost") * var("p_max"),
+)
+
+""" Simple thermal unit that can be invested on and with already installed capacity"""
+THERMAL_CANDIDATE_WITH_ALREADY_INSTALLED_CAPA = model(
+    id="GEN",
+    parameters=[
+        float_parameter("op_cost", CONSTANT),
+        float_parameter("invest_cost", CONSTANT),
+        float_parameter("max_invest", CONSTANT),
+        float_parameter("already_installed_capa", CONSTANT),
+    ],
+    variables=[
+        float_variable("generation", lower_bound=literal(0)),
+        float_variable(
+            "invested_capa",
+            lower_bound=literal(0),
+            upper_bound=param("max_invest"),
+            structure=CONSTANT,
+            context=ProblemContext.COUPLING,
+        ),
+    ],
+    ports=[ModelPort(port_type=BALANCE_PORT_TYPE, port_name="balance_port")],
+    port_fields_definitions=[
+        PortFieldDefinition(
+            port_field=PortFieldId("balance_port", "flow"),
+            definition=var("generation"),
+        )
+    ],
+    constraints=[
+        Constraint(
+            name="Max generation",
+            expression=var("generation")
+            <= param("already_installed_capa") + var("invested_capa"),
+        )
+    ],
+    objective_operational_contribution=(param("op_cost") * var("generation"))
+    .time_sum()
+    .expec(),
+    objective_investment_contribution=param("invest_cost") * var("invested_capa"),
+)

src/andromede/libs/standard_sc.py

@@ -22,7 +22,7 @@
     float_variable,
     model,
 )
-from andromede.model.model import PortFieldDefinition, PortFieldId
+from andromede.model.port import PortFieldDefinition, PortFieldId
 
 """
 Simple Convertor model.
@@ -174,7 +174,7 @@
 )
 
 """
-Model of the CO² quota. 
+Model of the CO² quota.
 It takes a set a CO² emissions as input. It forces the sum of those emissions to be smaller than a predefined quota. 
 """
 QUOTA_CO2_MODEL = model(

src/andromede/model/__init__.py

@@ -14,5 +14,5 @@
 from .constraint import Constraint
 from .model import Model, ModelPort, model
 from .parameter import Parameter, float_parameter, int_parameter
-from .port import PortField, PortType
+from .port import PortField, PortFieldDefinition, PortFieldId, PortType
 from .variable import Variable, float_variable, int_variable

src/andromede/model/constraint.py

@@ -9,8 +9,9 @@
 # SPDX-License-Identifier: MPL-2.0
 #
 # This file is part of the Antares project.
-from dataclasses import dataclass
-from typing import Any, Optional
+
+from dataclasses import dataclass, field, replace
+from typing import Any
 
 from andromede.expression.degree import is_constant
 from andromede.expression.equality import (
@@ -21,6 +22,8 @@
     Comparator,
     ComparisonNode,
     ExpressionNode,
+    is_non_negative,
+    is_unbound,
     literal,
 )
 from andromede.expression.print import print_expr
@@ -37,58 +40,48 @@ class Constraint:
 
     name: str
     expression: ExpressionNode
-    lower_bound: ExpressionNode
-    upper_bound: ExpressionNode
-    context: ProblemContext
+    lower_bound: ExpressionNode = field(default=literal(-float("inf")))
+    upper_bound: ExpressionNode = field(default=literal(float("inf")))
+    context: ProblemContext = field(default=ProblemContext.OPERATIONAL)
 
-    def __init__(
+    def __post_init__(
         self,
-        name: str,
-        expression: ExpressionNode,
-        lower_bound: Optional[ExpressionNode] = None,
-        upper_bound: Optional[ExpressionNode] = None,
-        context: ProblemContext = ProblemContext.OPERATIONAL,
     ) -> None:
-        self.name = name
-        self.context = context
-
-        if isinstance(expression, ComparisonNode):
-            if lower_bound is not None or upper_bound is not None:
+        if isinstance(self.expression, ComparisonNode):
+            if not is_unbound(self.lower_bound) or not is_unbound(self.upper_bound):
                 raise ValueError(
                     "Both comparison between two expressions and a bound are specfied, set either only a comparison between expressions or a single linear expression with bounds."
                 )
 
-            merged_expr = expression.left - expression.right
-            self.expression = merged_expr
-
-            if expression.comparator == Comparator.LESS_THAN:
+            if self.expression.comparator == Comparator.LESS_THAN:
                 # lhs - rhs <= 0
                 self.upper_bound = literal(0)
                 self.lower_bound = literal(-float("inf"))
-            elif expression.comparator == Comparator.GREATER_THAN:
+            elif self.expression.comparator == Comparator.GREATER_THAN:
                 # lhs - rhs >= 0
                 self.lower_bound = literal(0)
                 self.upper_bound = literal(float("inf"))
             else:  # lhs - rhs == 0
                 self.lower_bound = literal(0)
                 self.upper_bound = literal(0)
+
+            self.expression = self.expression.left - self.expression.right
+
         else:
-            for bound in [lower_bound, upper_bound]:
-                if bound is not None and not is_constant(bound):
+            for bound in [self.lower_bound, self.upper_bound]:
+                if not is_constant(bound):
                     raise ValueError(
                         f"The bounds of a constraint should not contain variables, {print_expr(bound)} was given."
                     )
 
-            self.expression = expression
-            if lower_bound is not None:
-                self.lower_bound = lower_bound
-            else:
-                self.lower_bound = literal(-float("inf"))
+            if is_unbound(self.lower_bound) and is_non_negative(self.lower_bound):
+                raise ValueError("Lower bound should not be +Inf")
 
-            if upper_bound is not None:
-                self.upper_bound = upper_bound
-            else:
-                self.upper_bound = literal(float("inf"))
+            if is_unbound(self.upper_bound) and not is_non_negative(self.upper_bound):
+                raise ValueError("Upper bound should not be -Inf")
+
+    def replicate(self, /, **changes: Any) -> "Constraint":
+        return replace(self, **changes)
 
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, Constraint):