Input format for components

requirements-dev.txt

@@ -6,3 +6,5 @@ pytest-cov
 pre-commit~=3.5.0
 types-PyYAML~=6.0.12.12
 antlr4-tools~=0.2.1
+pandas~=2.0.3
+pandas-stubs<=2.0.3

src/andromede/study/data.py

@@ -9,10 +9,13 @@
 # SPDX-License-Identifier: MPL-2.0
 #
 # This file is part of the Antares project.
-
+import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Dict
+from pathlib import Path
+from typing import Dict, Optional
+
+import pandas as pd
 
 from andromede.study.network import Network
 
@@ -101,6 +104,18 @@ def check_requirement(self, time: bool, scenario: bool) -> bool:
         return scenario
 
 
+def load_ts_from_txt(
+    timeseries_name: Optional[str], path_to_file: Optional[Path]
+) -> pd.DataFrame:
+    if path_to_file is not None and timeseries_name is not None:
+        timeseries_with_extension = timeseries_name + ".txt"
+        ts_path = path_to_file / timeseries_with_extension
+    try:
+        return pd.read_csv(ts_path, header=None, sep="\s+")
+    except Exception:
+        raise Exception(f"An error has arrived when processing '{ts_path}'")
+
+
 @dataclass(frozen=True)
 class TimeScenarioSeriesData(AbstractDataStructure):
     """
@@ -109,10 +124,11 @@ class TimeScenarioSeriesData(AbstractDataStructure):
     can be defined by referencing one of those timeseries by its ID.
     """
 
-    time_scenario_series: Dict[TimeScenarioIndex, float]
+    time_scenario_series: pd.DataFrame
 
     def get_value(self, timestep: int, scenario: int) -> float:
-        return self.time_scenario_series[TimeScenarioIndex(timestep, scenario)]
+        value = str(self.time_scenario_series.iloc[timestep, scenario])
+        return float(value)
 
     def check_requirement(self, time: bool, scenario: bool) -> bool:
         if not isinstance(self, TimeScenarioSeriesData):

src/andromede/study/parsing.py

@@ -0,0 +1,55 @@
+# 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 typing
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+from yaml import safe_load
+
+
+def parse_yaml_components(input_components: typing.TextIO) -> "InputComponents":
+    tree = safe_load(input_components)
+    return InputComponents.model_validate(tree["study"])
+
+
+# Design note: actual parsing and validation is delegated to pydantic models
+def _to_kebab(snake: str) -> str:
+    return snake.replace("_", "-")
+
+
+class InputPortConnections(BaseModel):
+    component1: str
+    port_1: str
+    component2: str
+    port_2: str
+
+
+class InputComponentParameter(BaseModel):
+    name: str
+    type: str
+    value: Optional[float] = None
+    timeseries: Optional[str] = None
+
+
+class InputComponent(BaseModel):
+    id: str
+    model: str
+    parameters: Optional[List[InputComponentParameter]] = None
+
+
+class InputComponents(BaseModel):
+    nodes: List[InputComponent] = Field(default_factory=list)
+    components: List[InputComponent] = Field(default_factory=list)
+    connections: List[InputPortConnections] = Field(default_factory=list)
+
+    class Config:
+        alias_generator = _to_kebab

src/andromede/study/resolve_components.py

@@ -0,0 +1,170 @@
+# 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.
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Iterable, List, Optional
+
+import pandas as pd
+
+from andromede.model import Model
+from andromede.model.library import Library
+from andromede.study import (
+    Component,
+    ConstantData,
+    DataBase,
+    Network,
+    Node,
+    PortRef,
+    PortsConnection,
+)
+from andromede.study.data import (
+    AbstractDataStructure,
+    TimeScenarioIndex,
+    TimeScenarioSeriesData,
+    load_ts_from_txt,
+)
+from andromede.study.parsing import (
+    InputComponent,
+    InputComponents,
+    InputPortConnections,
+)
+
+
+@dataclass(frozen=True)
+class NetworkComponents:
+    components: Dict[str, Component]
+    nodes: Dict[str, Component]
+    connections: List[PortsConnection]
+
+
+def network_components(
+    components_list: Iterable[Component],
+    nodes: Iterable[Component],
+    connections: Iterable[PortsConnection],
+) -> NetworkComponents:
+    return NetworkComponents(
+        components=dict((m.id, m) for m in components_list),
+        nodes=dict((n.id, n) for n in nodes),
+        connections=list(connections),
+    )
+
+
+def resolve_components_and_cnx(
+    input_comp: InputComponents, library: Library
+) -> NetworkComponents:
+    """
+    Resolves:
+    - components to be used for study
+    - connections between components"""
+    components_list = [_resolve_component(library, m) for m in input_comp.components]
+    nodes = [_resolve_component(library, n) for n in input_comp.nodes]
+    all_components: List[Component] = components_list + nodes
+    connections = []
+    for cnx in input_comp.connections:
+        resolved_cnx = _resolve_connections(cnx, all_components)
+        connections.append(resolved_cnx)
+
+    return network_components(components_list, nodes, connections)
+
+
+def _resolve_component(library: Library, component: InputComponent) -> Component:
+    model = library.models[component.model]
+
+    return Component(
+        model=model,
+        id=component.id,
+    )
+
+
+def _resolve_connections(
+    connection: InputPortConnections,
+    all_components: List[Component],
+) -> PortsConnection:
+    cnx_component1 = connection.component1
+    cnx_component2 = connection.component2
+    port1 = connection.port_1
+    port2 = connection.port_2
+
+    component_1 = _get_component_by_id(all_components, cnx_component1)
+    component_2 = _get_component_by_id(all_components, cnx_component2)
+    assert component_1 is not None and component_2 is not None
+    port_ref_1 = PortRef(component_1, port1)
+    port_ref_2 = PortRef(component_2, port2)
+
+    return PortsConnection(port_ref_1, port_ref_2)
+
+
+def _get_component_by_id(
+    all_components: List[Component], component_id: str
+) -> Optional[Component]:
+    components_dict = {component.id: component for component in all_components}
+    return components_dict.get(component_id)
+
+
+def consistency_check(
+    input_components: Dict[str, Component], input_models: Dict[str, Model]
+) -> bool:
+    """
+    Checks if all components in the Components instances have a valid model from the library.
+    Returns True if all components are consistent, raises ValueError otherwise.
+    """
+    model_ids_set = input_models.keys()
+    for component_id, component in input_components.items():
+        if component.model.id not in model_ids_set:
+            raise ValueError(
+                f"Error: Component {component_id} has invalid model ID: {component.model.id}"
+            )
+    return True
+
+
+def build_network(comp_network: NetworkComponents) -> Network:
+    network = Network("study")
+
+    for node_id, node in comp_network.nodes.items():
+        node = Node(model=node.model, id=node_id)
+        network.add_node(node)
+
+    for component_id, component in comp_network.components.items():
+        network.add_component(component)
+
+    for connection in comp_network.connections:
+        network.connect(connection.port1, connection.port2)
+    return network
+
+
+def build_data_base(
+    input_comp: InputComponents, timeseries_dir: Optional[Path]
+) -> DataBase:
+    database = DataBase()
+    for comp in input_comp.components:
+        for param in comp.parameters or []:
+            param_value = _evaluate_param_type(
+                param.type, param.value, param.timeseries, timeseries_dir
+            )
+            database.add_data(comp.id, param.name, param_value)
+
+    return database
+
+
+def _evaluate_param_type(
+    param_type: str,
+    param_value: Optional[float],
+    timeseries_name: Optional[str],
+    timeseries_dir: Optional[Path],
+) -> AbstractDataStructure:
+    if param_type == "constant" and param_value is not None:
+        return ConstantData(float(param_value))
+
+    elif param_type == "timeseries":
+        return TimeScenarioSeriesData(load_ts_from_txt(timeseries_name, timeseries_dir))
+
+    raise ValueError(f"Data should be either constant or timeseries ")

tests/functional/test_andromede.py

@@ -10,6 +10,7 @@
 #
 # This file is part of the Antares project.
 
+import pandas as pd
 import pytest
 
 from andromede.expression import literal, param, var
@@ -215,11 +216,17 @@ def test_timeseries() -> None:
 
     database.add_data("G", "p_max", ConstantData(100))
     database.add_data("G", "cost", ConstantData(30))
-
-    demand_data = TimeScenarioSeriesData(
-        {TimeScenarioIndex(0, 0): 100, TimeScenarioIndex(1, 0): 50}
+    demand_data = pd.DataFrame(
+        [
+            [100],
+            [50],
+        ],
+        index=[0, 1],
+        columns=[0],
     )
-    database.add_data("D", "demand", demand_data)
+
+    demand_time_scenario_series = TimeScenarioSeriesData(demand_data)
+    database.add_data("D", "demand", demand_time_scenario_series)
 
     node = Node(model=NODE_BALANCE_MODEL, id="1")
     demand = create_component(
@@ -401,14 +408,18 @@ def test_min_up_down_times() -> None:
     database.add_data("U", "cost", ConstantData(3000))
     database.add_data("S", "cost", ConstantData(10))
 
-    demand_data = TimeScenarioSeriesData(
-        {
-            TimeScenarioIndex(0, 0): 500,
-            TimeScenarioIndex(1, 0): 0,
-            TimeScenarioIndex(2, 0): 0,
-        }
+    demand_data = pd.DataFrame(
+        [
+            [500],
+            [0],
+            [0],
+        ],
+        index=[0, 1, 2],
+        columns=[0],
     )
-    database.add_data("D", "demand", demand_data)
+
+    demand_time_scenario_series = TimeScenarioSeriesData(demand_data)
+    database.add_data("D", "demand", demand_time_scenario_series)
 
     time_block = TimeBlock(1, [0, 1, 2])
     scenarios = 1
@@ -473,7 +484,10 @@ def generate_data(
                 data[TimeScenarioIndex(absolute_timestep, scenario)] = -18
             else:
                 data[TimeScenarioIndex(absolute_timestep, scenario)] = 2 * efficiency
-    return TimeScenarioSeriesData(time_scenario_series=data)
+
+    values = [value for value in data.values()]
+    data_df = pd.DataFrame(values, columns=["Value"])
+    return TimeScenarioSeriesData(data_df)
 
 
 def short_term_storage_base(efficiency: float, horizon: int) -> None:

tests/functional/test_xpansion.py

@@ -10,6 +10,7 @@
 #
 # This file is part of the Antares project.
 
+import pandas as pd
 import pytest
 
 from andromede.expression.expression import literal, param, port_field, var
@@ -47,7 +48,6 @@
     Network,
     Node,
     PortRef,
-    TimeScenarioIndex,
     TimeScenarioSeriesData,
     create_component,
 )
@@ -347,11 +347,7 @@ def test_generation_xpansion_two_time_steps_two_scenarios(
     horizon = 2
     time_block = TimeBlock(1, list(range(horizon)))
 
-    data = {}
-    data[TimeScenarioIndex(0, 0)] = 300
-    data[TimeScenarioIndex(1, 0)] = 500
-    data[TimeScenarioIndex(0, 1)] = 200
-    data[TimeScenarioIndex(1, 1)] = 400
+    data = pd.DataFrame([[300, 200], [500, 400]], index=[0, 1], columns=[0, 1])
 
     demand_data = TimeScenarioSeriesData(time_scenario_series=data)