Merge branch 'main' into stock_pipeline

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/libs/standard.py

@@ -63,32 +63,6 @@
     .sum()
     .expec(),
 )
-"""
-A standard model for a linear cost generation, limited by a maximum generation.
-"""
-GENERATOR_MODEL = model(
-    id="GEN",
-    parameters=[
-        float_parameter("p_max", CONSTANT),
-        float_parameter("cost", 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")
-        ),
-    ],
-    objective_operational_contribution=(param("cost") * var("generation"))
-    .sum()
-    .expec(),
-)
 
 """
 Basic link model using ports
@@ -132,6 +106,33 @@
     ],
 )
 
+"""
+A standard model for a linear cost generation, limited by a maximum generation.
+"""
+GENERATOR_MODEL = model(
+    id="GEN",
+    parameters=[
+        float_parameter("p_max", CONSTANT),
+        float_parameter("cost", 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")
+        ),
+    ],
+    objective_operational_contribution=(param("cost") * var("generation"))
+    .sum()
+    .expec(),
+)
+
 GENERATOR_MODEL_WITH_PMIN = model(
     id="GEN",
     parameters=[
@@ -162,6 +163,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").sum() <= param("full_storage"),
+        ),
+    ],
+    objective_operational_contribution=(param("cost") * var("generation"))
+    .sum()
+    .expec(),
+)
+
 # For now, no starting cost
 THERMAL_CLUSTER_MODEL_HD = model(
     id="GEN",

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 ")