Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Loading
Loading status checks…
Final fixes
Browse files Browse the repository at this point in the history
@killian-scalian
killian-scalian committed Jan 9, 2025
1 parent 946a0cf commit 9c7801e
Showing 2 changed files with 229 additions and 61 deletions.
37 changes: 18 additions & 19 deletions src/andromede/input_converter/src/converter.py
View file
Original file line number Diff line number Diff line change
@@ -10,24 +10,27 @@
#
# This file is part of the Antares project.
from pathlib import Path
from typing import Optional
from typing import Union

from antares.craft.model.area import Area
from antares.craft.model.study import Study, read_study_local

from andromede.input_converter.src.utils import resolve_path
from andromede.study.parsing import InputComponent, InputComponentParameter, InputStudy
from andromede.study.parsing import (InputComponent, InputComponentParameter,
InputStudy)


class AntaresStudyConverter:
def __init__(self, study_path: Optional[Path]):
def __init__(self, study_input: Union[Path, Study]):
"""
Initialize processor
"""
self.study_path = resolve_path(study_path) if study_path else None
self.study: Study = (
read_study_local(self.study_path) if self.study_path else None # type: ignore
)
if isinstance(study_input, Study):
self.study = study_input
self.study_path = study_input.service.config.study_path # type: ignore
else:
self.study = read_study_local(self.study_path)
self.study_path = resolve_path(study_input)

def _convert_area_to_component_list(
self, areas: list[Area]
@@ -55,14 +58,14 @@ def _convert_area_to_component_list(
return components

def _convert_renewable_to_component_list(
self, areas: list[Area], root_path: Path
self, areas: list[Area]
) -> list[InputComponent]:
components = []
for area in areas:
renewables = area.read_renewables()
for renewable in renewables:
series_path = (
root_path
self.study_path
/ "input"
/ "renewables"
/ "series"
@@ -96,16 +99,16 @@ def _convert_renewable_to_component_list(

return components

def _convert_thermals_to_component_list(
self, areas: list[Area], root_path: Path
def _convert_thermal_to_component_list(
self, areas: list[Area]
) -> list[InputComponent]:
components = []
# Add thermal components for each area
for area in areas:
thermals = area.read_thermal_clusters()
for thermal in thermals:
series_path = (
root_path
self.study_path
/ "input"
/ "thermal"
/ "series"
@@ -160,15 +163,11 @@ def _convert_thermals_to_component_list(

def convert_study_to_input_study(self) -> InputStudy:
areas = self.study.read_areas()
root_path = self.study.service.config.study_path # type: ignore
area_components = self._convert_area_to_component_list(areas)
list_components = []
list_components.extend(
self._convert_renewable_to_component_list(areas, root_path)
)
list_components.extend(
self._convert_thermals_to_component_list(areas, root_path)
)
list_components.extend(self._convert_renewable_to_component_list(areas))
list_components.extend(self._convert_thermal_to_component_list(areas))

return InputStudy(nodes=area_components, components=list_components)

def process_all(self) -> None:
253 changes: 211 additions & 42 deletions tests/input_converter/test_converter.py
View file
Original file line number Diff line number Diff line change
@@ -12,34 +12,30 @@

from andromede.input_converter.src.converter import AntaresStudyConverter
from andromede.input_converter.src.utils import transform_to_yaml
from andromede.study.parsing import (
InputComponent,
InputComponentParameter,
InputStudy,
parse_yaml_components,
)
from andromede.study.parsing import (InputComponent, InputComponentParameter,
InputStudy, parse_yaml_components)


class TestConverter:
def _init_area_reading(self, local_study):
converter = AntaresStudyConverter(study_path=None)
converter.study = local_study
converter = AntaresStudyConverter(study_input=local_study)
areas = converter.study.read_areas()
area_components = converter._convert_area_to_component_list(areas)
return areas, converter, area_components

def test_convert_study_to_input_study(self, local_study_w_areas):
converter = AntaresStudyConverter(study_path=None)
converter.study = local_study_w_areas
study_component = converter.convert_study_to_input_study()
return areas, converter

assert isinstance(study_component, InputStudy)
def test_convert_study_to_input_study(self, local_study_w_thermal):
converter = AntaresStudyConverter(study_input=local_study_w_thermal)
input_study = converter.convert_study_to_input_study()

def test_convert_area_to_component(self, local_study_w_areas):
_, _, area_components = self._init_area_reading(local_study_w_areas)

input_study = InputStudy(nodes=area_components)
expected_area_components = InputStudy(
p_max_thermal_timeserie = str(
converter.study_path
/ "input"
/ "thermal"
/ "series"
/ "fr"
/ "gaz"
/ "series.txt"
)
expected_input_study = InputStudy(
nodes=[
InputComponent(
id="fr",
@@ -81,26 +77,156 @@ def test_convert_area_to_component(self, local_study_w_areas):
),
],
),
InputComponent(
id="at",
model="area",
scenario_group=None,
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
scenario_group=None,
value=0.0,
timeseries=None,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
scenario_group=None,
value=0.0,
timeseries=None,
),
],
),
],
components=[
InputComponent(
id="gaz",
model="thermal",
scenario_group=None,
parameters=[
InputComponentParameter(
name="unit_count",
type="constant",
scenario_group=None,
value=1.0,
timeseries=None,
),
InputComponentParameter(
name="efficiency",
type="constant",
scenario_group=None,
value=100.0,
timeseries=None,
),
InputComponentParameter(
name="nominal_capacity",
type="constant",
scenario_group=None,
value=0.0,
timeseries=None,
),
InputComponentParameter(
name="marginal_cost",
type="constant",
scenario_group=None,
value=0.0,
timeseries=None,
),
InputComponentParameter(
name="fixed_cost",
type="constant",
scenario_group=None,
value=0.0,
timeseries=None,
),
InputComponentParameter(
name="startup_cost",
type="constant",
scenario_group=None,
value=0.0,
timeseries=None,
),
InputComponentParameter(
name="p_max_cluster",
type="timeseries",
scenario_group=None,
value=None,
timeseries=f"{p_max_thermal_timeserie}",
),
],
)
],
components=[],
connections=[],
)

# To ensure that the comparison between the actual and expected results is not affected by the order of the nodes,
# both the area_components.nodes and expected_area_components.nodes lists are sorted by the id attribute of each node.
# This sorting step ensures that the test checks only the presence and validity of the nodes, not their order.
input_study.nodes.sort(key=lambda x: x.id)
expected_area_components.nodes.sort(key=lambda x: x.id)
expected_input_study.nodes.sort(key=lambda x: x.id)

assert input_study == expected_input_study

def test_convert_area_to_component(self, local_study_w_areas):
converter, areas = self._init_area_reading(local_study_w_areas)
area_components = converter._convert_area_to_component_list(areas)

expected_area_components = [
InputComponent(
id="fr",
model="area",
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
scenario_group=None,
value=1.5,
timeseries=None,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
scenario_group=None,
value=1.0,
timeseries=None,
),
],
),
InputComponent(
id="it",
model="area",
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
scenario_group=None,
value=1.5,
timeseries=None,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
scenario_group=None,
value=1.0,
timeseries=None,
),
],
),
]

assert input_study == expected_area_components
# To ensure that the comparison between the actual and expected results is not affected by the order of the nodes,
# both the area_components.nodes and expected_area_components.nodes lists are sorted by the id attribute of each node.
# This sorting step ensures that the test checks only the presence and validity of the nodes, not their order.
expected_area_components.sort(key=lambda x: x.id)
area_components.sort(key=lambda x: x.id)
assert area_components == expected_area_components

def test_convert_renewables_to_input_study(self, local_study_with_renewable):
areas, converter, _ = self._init_area_reading(local_study_with_renewable)
study_path = local_study_with_renewable.service.config.study_path
areas, converter = self._init_area_reading(local_study_with_renewable)
study_path = converter.study_path
renewables_components = converter._convert_renewable_to_component_list(areas)

renewables_components = converter._convert_renewable_to_component_list(
areas, root_path=study_path
)
timeserie_path = str(
study_path
/ "input"
@@ -142,14 +268,12 @@ def test_convert_renewables_to_input_study(self, local_study_with_renewable):
]
assert renewables_components == expected_renewable_component

def test_convert_thermals_to_input_study(self, local_study_with_renewable):
areas, converter, _ = self._init_area_reading(local_study_with_renewable)
study_path = local_study_with_renewable.service.config.study_path
def test_convert_thermals_to_input_study(self, local_study_w_thermal):
areas, converter = self._init_area_reading(local_study_w_thermal)

thermals_components = converter._convert_thermals_to_component_list(
areas, study_path
)
thermals_components = converter._convert_thermal_to_component_list(areas)

study_path = converter.study_path
p_max_timeserie = str(
study_path / "input" / "thermal" / "series" / "fr" / "gaz" / "series.txt"
)
@@ -214,19 +338,64 @@ def test_convert_thermals_to_input_study(self, local_study_with_renewable):
assert thermals_components == expected_thermals_components

def test_convert_area_to_yaml(self, local_study_w_areas):
_, _, area_components = self._init_area_reading(local_study_w_areas)

areas, converter = self._init_area_reading(local_study_w_areas)
area_components = converter._convert_area_to_component_list(areas)
input_study = InputStudy(nodes=area_components)

# Dump model into yaml file
yaml_path = local_study_w_areas.service.config.study_path / "study_path.yaml"
yaml_path = converter.study_path / "study_path.yaml"
transform_to_yaml(model=input_study, output_path=yaml_path)

# Open yaml file to validate
with open(yaml_path, "r", encoding="utf-8") as yaml_file:
validated_data = parse_yaml_components(yaml_file)

assert isinstance(
validated_data, InputStudy
), "Validated datas don't match InputStudy model"
assert len(validated_data.nodes) == 2, "Yaml file must contain 2 nodes"
expected_validated_data = InputStudy(
nodes=[
InputComponent(
id="it",
model="area",
scenario_group=None,
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
scenario_group=None,
value=1.5,
timeseries=None,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
scenario_group=None,
value=1.0,
timeseries=None,
),
],
),
InputComponent(
id="fr",
model="area",
scenario_group=None,
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
scenario_group=None,
value=1.5,
timeseries=None,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
scenario_group=None,
value=1.0,
timeseries=None,
),
],
),
],
components=[],
connections=[],
)
assert validated_data == expected_validated_data

0 comments on commit 9c7801e

Please sign in to comment.