Design the construction of network on decision tree

mypy.ini

@@ -5,4 +5,7 @@ disallow_untyped_defs = true
 disallow_untyped_calls = true
 
 [mypy-ortools.*]
+ignore_missing_imports = true
+
+[mypy-anytree.*]
 ignore_missing_imports = true

src/andromede/model/model.py

@@ -19,7 +19,8 @@
 from dataclasses import dataclass, field
 from typing import Dict, Iterable, Optional
 
-from anytree import Node as TreeNode, LevelOrderIter
+from anytree import LevelOrderIter
+from anytree import Node as TreeNode
 
 from andromede.expression import (
     AdditionNode,

src/andromede/simulation/benders_decomposed.py

@@ -14,14 +14,17 @@
 The xpansion module extends the optimization module
 with Benders solver related functions
 """
-
 import json
 import os
 import pathlib
 import subprocess
 import sys
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
+
+from anytree import Node as TreeNode
 
+from andromede.model.model import Model
+from andromede.simulation.decision_tree import ConfiguredTree, create_master_network
 from andromede.simulation.optimization import (
     BlockBorderManagement,
     OptimizationProblem,
@@ -31,7 +34,7 @@
     InvestmentProblemStrategy,
     OperationalProblemStrategy,
 )
-from andromede.simulation.time_block import ConfiguredTree
+from andromede.simulation.time_block import TimeBlock
 from andromede.study.data import DataBase
 from andromede.study.network import Network
 from andromede.utils import serialize
@@ -178,10 +181,11 @@ def run(
 
 
 def build_benders_decomposed_problem(
-    network: Network,
+    network_on_tree: Dict[TreeNode, Network],
     database: DataBase,
     configured_tree: ConfiguredTree,
     *,
+    decision_coupling_model: Optional[Model] = None,
     border_management: BlockBorderManagement = BlockBorderManagement.CYCLE,
     solver_id: str = "GLOP",
 ) -> BendersDecomposedProblem:
@@ -191,31 +195,28 @@ def build_benders_decomposed_problem(
     Returns a Benders Decomposed problem
     """
 
+    master_network = create_master_network(network_on_tree, decision_coupling_model)
+
     # Benders Decomposed Master Problem
     master = build_problem(
-        network,
+        master_network,
         database,
-        configured_tree.root,  # Could be any node, given the implmentation of get_nodes()
-        configured_tree.node_to_config[configured_tree.root].blocks[
-            0
-        ],  # Probably useless arg ?
-        configured_tree.node_to_config[configured_tree.root].scenarios,
+        TimeBlock(
+            42, [27]
+        ),  # Probably useless arg as we have only investment variables ?
+        0,  # Useless for master ?
         problem_name="master",
         border_management=border_management,
         solver_id=solver_id,
         problem_strategy=InvestmentProblemStrategy(),
     )
 
-    for (
-        tree_node,
-        time_scenario_config,
-    ) in configured_tree.node_to_config.items():
+    for tree_node, time_scenario_config in configured_tree.node_to_config.items():
         for block in time_scenario_config.blocks:
             # Xpansion Sub-problems
             subproblem = build_problem(
-                network,
+                network_on_tree[tree_node],
                 database,
-                tree_node,
                 block,
                 time_scenario_config.scenarios,
                 problem_name="subproblem",

src/andromede/simulation/decision_tree.py

@@ -0,0 +1,169 @@
+# Copyright (c) 2024, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+
+import dataclasses
+from dataclasses import dataclass, field
+from typing import Dict, Iterable, List, Optional
+
+from anytree import LevelOrderIter
+from anytree import Node as TreeNode
+
+from andromede.expression.expression import ExpressionNode
+from andromede.model.constraint import Constraint
+from andromede.model.model import Model, PortFieldDefinition, PortFieldId, model
+from andromede.model.variable import Variable
+from andromede.simulation.time_block import TimeBlock
+from andromede.study.network import Component, Network, Node, create_component
+
+
+@dataclass(frozen=True)
+class InterDecisionTimeScenarioConfig:
+    blocks: List[TimeBlock]
+    scenarios: int
+
+
+@dataclass(frozen=True)
+class ConfiguredTree:
+    node_to_config: Dict[TreeNode, InterDecisionTimeScenarioConfig]
+    root: TreeNode = field(init=False)
+
+    def __post_init__(self) -> None:
+        # Stores the root, by getting it from any tree node
+        object.__setattr__(self, "root", next(iter(self.node_to_config.keys())).root)
+
+
+def create_single_node_decision_tree(
+    blocks: List[TimeBlock], scenarios: int
+) -> ConfiguredTree:
+    time_scenario_config = InterDecisionTimeScenarioConfig(blocks, scenarios)
+
+    root = TreeNode("root")
+    configured_tree = ConfiguredTree(
+        {
+            root: time_scenario_config,
+        },
+    )
+
+    return configured_tree
+
+
+def _generate_tree_variables(
+    variables: Dict[str, Variable], tree_node: TreeNode
+) -> Iterable[Variable]:
+    tree_variables = []
+    for variable in variables.values():
+        tree_variables.append(
+            dataclasses.replace(variable, name=f"{tree_node.name}_{variable.name}")
+        )
+    return tree_variables
+
+
+def _generate_tree_constraints(
+    constraints: Dict[str, Constraint], tree: TreeNode
+) -> Iterable[Constraint]:
+    raise NotImplementedError()
+
+
+def _generate_tree_expression(
+    expression: Optional[ExpressionNode], tree: TreeNode
+) -> ExpressionNode:
+    raise NotImplementedError()
+
+
+def _generate_tree_port_field_definition(
+    port_field_definition: Dict[PortFieldId, PortFieldDefinition], tree: TreeNode
+) -> Iterable[PortFieldDefinition]:
+    raise NotImplementedError()
+
+
+def _generate_tree_model(
+    tree_node: TreeNode, component: Component, network_id: str
+) -> Model:
+    variables = _generate_tree_variables(
+        component.model.variables,
+        tree_node,
+    )
+    constraints = _generate_tree_constraints(component.model.constraints, tree_node)
+    binding_constraints = _generate_tree_constraints(
+        component.model.binding_constraints, tree_node
+    )
+    objective_operational_contribution = _generate_tree_expression(
+        component.model.objective_operational_contribution, tree_node
+    )
+    objective_investment_contribution = _generate_tree_expression(
+        component.model.objective_investment_contribution, tree_node
+    )
+    port_fields_definitions = _generate_tree_port_field_definition(
+        component.model.port_fields_definitions, tree_node
+    )
+    tree_model = model(
+        id=f"{network_id}_{component.model.id}",
+        constraints=constraints,
+        binding_constraints=binding_constraints,
+        parameters=component.model.parameters.values(),
+        variables=variables,
+        objective_operational_contribution=objective_operational_contribution,
+        objective_investment_contribution=objective_investment_contribution,
+        inter_block_dyn=component.model.inter_block_dyn,
+        ports=component.model.ports.values(),
+        port_fields_definitions=port_fields_definitions,
+    )
+
+    return tree_model
+
+
+def _generate_network_on_node(network: Network, tree_node: TreeNode) -> Network:
+    network_id = tree_node.name
+    tree_node_network = Network(network_id)
+
+    for component in network.all_components:
+        tree_node_model = _generate_tree_model(
+            tree_node,
+            component,
+            network_id,
+        )
+
+        # It would be nice to have the same treatment for nodes and components as they are actually the same thing...
+        if isinstance(component, Node):
+            network_node = Node(tree_node_model, id=f"{network_id}_{component.id}")
+            tree_node_network.add_node(network_node)
+        else:
+            tree_node_component = create_component(
+                tree_node_model, id=f"{network_id}_{component.id}"
+            )
+            tree_node_network.add_component(tree_node_component)
+
+    for connection in network.connections:
+        tree_node_network.connect(connection.port1, connection.port2)
+    return tree_node_network
+
+
+def create_network_on_tree(network: Network, tree: TreeNode) -> Dict[TreeNode, Network]:
+    # On crée un gros modèle en dupliquant les variables; contraintes, etc à chaque noeud de l'arbre.
+    # Pour le master on peut :
+    #   - Utiliser uniquement les variables, contraintes, etc dont on va avoir besoin dans la construction du problème -> nécessite déjà d'avoir des infos sur la construction des problèmes alors qu'on agit au niveau modèle ici
+    #   - Dupliquer tout le modèle, permet de mutualiser du code avec la partie composant par noeud et plus lisible. Seul inconvénient, modèle master un peu trop riche, pas besoin des infos "opérationnelles". Mais les modèles ne sont pas très "lourds" donc on peut se le permettre. C'est l'option choisie ici.
+    if tree.size == 1:
+        return {tree: network}
+    else:
+        node_to_network = {}
+        for tree_node in LevelOrderIter(tree):
+            node_to_network[tree_node] = _generate_network_on_node(network, tree_node)
+        return node_to_network
+
+
+def create_master_network(
+    tree_node_to_network: Dict[TreeNode, Network],
+    decision_coupling_model: Optional[Model],
+) -> Network:
+    root = next(iter(tree_node_to_network.keys())).root
+    return tree_node_to_network[root]

src/andromede/simulation/optimization.py

@@ -18,7 +18,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from enum import Enum
-from typing import Dict, Iterable, List, Optional, Type
+from typing import Dict, Iterable, List, Optional
 
 import ortools.linear_solver.pywraplp as lp
 
@@ -809,7 +809,6 @@ def export_as_lp(self) -> str:
 def build_problem(
     network: Network,
     database: DataBase,
-    tree_node_name: str,
     block: TimeBlock,
     scenarios: int,
     *,

src/andromede/simulation/time_block.py

@@ -10,10 +10,8 @@
 #
 # This file is part of the Antares project.
 
-from dataclasses import dataclass, field
-from typing import Dict, List, Optional
-
-from anytree import Node as TreeNode
+from dataclasses import dataclass
+from typing import List, Optional
 
 
 # TODO: Move keys elsewhere as variables have no sense in this file
@@ -40,15 +38,3 @@ class TimeBlock:
 
     id: int
     timesteps: List[int]
-
-
-@dataclass(frozen=True)
-class InterDecisionTimeScenarioConfig:
-    blocks: List[TimeBlock]
-    scenarios: int
-
-
-@dataclass(frozen=True)
-class ConfiguredTree:
-    root: TreeNode  # Could be retrieved easily from any node with node.root, but clearer to identify it separately
-    node_to_config: Dict[TreeNode, InterDecisionTimeScenarioConfig]

src/andromede/study/data.py

@@ -35,7 +35,6 @@ class ScenarioIndex:
 
 @dataclass(frozen=True)
 class AbstractDataStructure(ABC):
-
     @abstractmethod
     def get_value(
         self, timestep: int, scenario: int, node_id: Optional[int] = None
@@ -81,7 +80,7 @@ class TimeSeriesData(AbstractDataStructure):
     can be defined by referencing one of those timeseries by its ID.
     """
 
-    time_series: Dict[TimeIndex, float]
+    time_series: Mapping[TimeIndex, float]
 
     def get_value(
         self, timestep: int, scenario: int, node_id: Optional[int] = None
@@ -103,7 +102,7 @@ class ScenarioSeriesData(AbstractDataStructure):
     can be defined by referencing one of those timeseries by its ID.
     """
 
-    scenario_series: Dict[ScenarioIndex, float]
+    scenario_series: Mapping[ScenarioIndex, float]
 
     def get_value(
         self, timestep: int, scenario: int, node_id: Optional[int] = None
@@ -125,7 +124,7 @@ class TimeScenarioSeriesData(AbstractDataStructure):
     can be defined by referencing one of those timeseries by its ID.
     """
 
-    time_scenario_series: Dict[TimeScenarioIndex, float]
+    time_scenario_series: Mapping[TimeScenarioIndex, float]
 
     def get_value(
         self, timestep: int, scenario: int, node_id: Optional[int] = None

tests/andromede/test_investment_pathway.py

@@ -19,12 +19,14 @@
     NODE_WITH_SPILL_AND_ENS,
     THERMAL_CANDIDATE_WITH_ALREADY_INSTALLED_CAPA,
 )
+from andromede.model.model import model
 from andromede.simulation.benders_decomposed import build_benders_decomposed_problem
-from andromede.simulation.time_block import (
+from andromede.simulation.decision_tree import (
     ConfiguredTree,
     InterDecisionTimeScenarioConfig,
-    TimeBlock,
+    create_network_on_tree,
 )
+from andromede.simulation.time_block import TimeBlock
 from andromede.study.data import ConstantData, DataBase, TreeData
 from andromede.study.network import Component, Network, Node, PortRef, create_component
 
@@ -197,16 +199,22 @@ def test_investment_pathway_on_a_tree_with_one_root_two_children(
     new_base = TreeNode("2040_new_base", parent=root)
     no_base = TreeNode("2040_no_base", parent=root)
     configured_tree = ConfiguredTree(
-        root,
         {
             root: time_scenario_config,
             new_base: time_scenario_config,
             no_base: time_scenario_config,
         },
     )
 
+    decision_coupling_model = model("DECISION_COUPLING")
+
+    tree_node_to_network = create_network_on_tree(network, configured_tree.root)
+
     problems = build_benders_decomposed_problem(
-        network, database, configured_tree
+        tree_node_to_network,
+        database,
+        configured_tree,
+        decision_coupling_model=decision_coupling_model,
     )
 
     # Réfléchir à la représentation des variables dans l'arbre