Skip to content

Commit

Permalink
Fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
@killian-scalian
killian-scalian committed Jan 8, 2025
1 parent f0bd8f5 commit 8f0f786
Show file tree
Hide file tree
Showing 4 changed files with 287 additions and 337 deletions.
162 changes: 140 additions & 22 deletions src/andromede/input_converter/src/converter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,18 +12,12 @@
from pathlib import Path
from typing import Optional

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

from andromede.input_converter.src.utils import (
convert_area_to_component_list,
resolve_path,
convert_renewable_to_component_list,
convert_thermals_to_component_list,
)

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


class AntaresStudyConverter:
Expand All @@ -36,23 +30,147 @@ def __init__(self, study_path: Optional[Path]):
read_study_local(self.study_path) if self.study_path else None # type: ignore
)

def _convert_area_to_component_list(
self, areas: list[Area]
) -> list[InputComponent]:
components = []
for area in areas:
components.append(
InputComponent(
id=area.id,
model="area",
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
value=area.properties.energy_cost_unsupplied,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
value=area.properties.energy_cost_spilled,
),
],
)
)
return components

def _convert_renewable_to_component_list(
self, areas: list[Area], root_path: Path
) -> list[InputComponent]:
components = []
for area in areas:
renewables = area.read_renewables()
for renewable in renewables:
series_path = (
root_path
/ "input"
/ "renewables"
/ "series"
/ Path(area.id)
/ Path(renewable.id)
/ "series.txt"
)
components.append(
InputComponent(
id=renewable.id,
model="renewable",
parameters=[
InputComponentParameter(
name="unit_count",
type="constant",
value=renewable.properties.unit_count,
),
InputComponentParameter(
name="nominal_capacity",
type="constant",
value=renewable.properties.nominal_capacity,
),
InputComponentParameter(
name="generation",
type="timeseries",
timeseries=str(series_path),
),
],
)
)

return components

def _convert_thermals_to_component_list(
self, areas: list[Area], root_path: Path
) -> 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
/ "input"
/ "thermal"
/ "series"
/ Path(area.id)
/ Path(thermal.name)
/ "series.txt"
)
components.append(
InputComponent(
id=thermal.id,
model="thermal",
parameters=[
InputComponentParameter(
name="unit_count",
type="constant",
value=thermal.properties.unit_count,
),
InputComponentParameter(
name="efficiency",
type="constant",
value=thermal.properties.efficiency,
),
InputComponentParameter(
name="nominal_capacity",
type="constant",
value=thermal.properties.nominal_capacity,
),
InputComponentParameter(
name="marginal_cost",
type="constant",
value=thermal.properties.marginal_cost,
),
InputComponentParameter(
name="fixed_cost",
type="constant",
value=thermal.properties.fixed_cost,
),
InputComponentParameter(
name="startup_cost",
type="constant",
value=thermal.properties.startup_cost,
),
InputComponentParameter(
name="p_max_cluster",
type="timeseries",
timeseries=str(series_path),
),
],
)
)
return components

def convert_study_to_input_study(self) -> InputStudy:
areas = self.study.read_areas()
area_components = convert_area_to_component_list(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(convert_renewable_to_component_list(areas, root_path))
list_components.extend(convert_thermals_to_component_list(areas, root_path))
list_components.extend(
self._convert_renewable_to_component_list(areas, root_path)
)
list_components.extend(
self._convert_thermals_to_component_list(areas, root_path)
)
return InputStudy(nodes=area_components, components=list_components)

@staticmethod
def transform_to_yaml(model: BaseModel, output_path: str) -> None:
with open(output_path, "w", encoding="utf-8") as yaml_file:
yaml.dump(
{"study": model.model_dump(by_alias=True, exclude_unset=True)},
yaml_file,
allow_unicode=True,
)

def process_all(self) -> None:
raise NotImplementedError
131 changes: 8 additions & 123 deletions src/andromede/input_converter/src/utils.py
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,10 @@
from pathlib import Path

import yaml
from antares.craft.model.area import Area
from pydantic import BaseModel

from andromede.study.parsing import (
InputComponent,
InputComponentParameter,
)
from andromede.study.parsing import InputComponent, InputComponentParameter


def resolve_path(path_str: Path) -> Path:
Expand All @@ -17,123 +15,10 @@ def resolve_path(path_str: Path) -> Path:
absolute_path = path.resolve()
return absolute_path


def convert_area_to_component_list(areas: list[Area]) -> list[InputComponent]:
components = []
for area in areas:
components.append(
InputComponent(
id=area.id,
model="area",
parameters=[
InputComponentParameter(
name="energy_cost_unsupplied",
type="constant",
value=area.properties.energy_cost_unsupplied,
),
InputComponentParameter(
name="energy_cost_spilled",
type="constant",
value=area.properties.energy_cost_spilled,
),
],
)
def transform_to_yaml(model: BaseModel, output_path: str) -> None:
with open(output_path, "w", encoding="utf-8") as yaml_file:
yaml.dump(
{"study": model.model_dump(by_alias=True, exclude_unset=True)},
yaml_file,
allow_unicode=True,
)
return components


def convert_renewable_to_component_list(
areas: list[Area], root_path: Path
) -> list[InputComponent]:
components = []
for area in areas:
renewables = area.read_renewables()
for renewable in renewables:
series_path = (
root_path
/ "input"
/ "renewables"
/ "series"
/ Path(area.id)
/ Path(renewable.id)
/ "series.txt"
)
components.extend(
[
InputComponent(
id=renewable.id,
model="renewable",
parameters=[
InputComponentParameter(
name="unit_count",
type="constant",
value=renewable.properties.unit_count,
),
InputComponentParameter(
name="nominal_capacity",
type="constant",
value=renewable.properties.nominal_capacity,
),
InputComponentParameter(
name=renewable.id,
type="timeseries",
timeseries=str(series_path),
),
],
)
for renewable in renewables
]
)

return components


def convert_thermals_to_component_list(
areas: list[Area], root_path: Path
) -> list[InputComponent]:
components = []
# Ajouter les composants des thermals pour chaque area
for area in areas:
thermals = area.read_thermal_clusters()
for thermal in thermals:
series_path = (
root_path
/ "input"
/ "thermal"
/ "series"
/ Path(area.id)
/ Path(thermal.name)
/ "series.txt"
)
components.extend(
[
InputComponent(
id=thermal.id,
model="thermal",
parameters=[
InputComponentParameter(
name="unit_count",
type="constant",
value=thermal.properties.unit_count,
),
InputComponentParameter(
name="efficiency",
type="constant",
value=thermal.properties.efficiency,
),
InputComponentParameter(
name="nominal_capacity",
type="constant",
value=thermal.properties.nominal_capacity,
),
InputComponentParameter(
name="p_max_cluster",
type="timeseries",
timeseries=str(series_path),
),
],
)
for thermal in thermals
]
)
return components
4 changes: 2 additions & 2 deletions tests/input_converter/conftest.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ def local_study_w_links(tmp_path, local_study_w_areas):

@pytest.fixture
def local_study_w_thermal(tmp_path, local_study_w_links) -> Study:
thermal_name = "test thermal cluster"
thermal_name = "gaz"
local_study_w_links.get_areas()["fr"].create_thermal_cluster(thermal_name)
return local_study_w_links

Expand Down Expand Up @@ -140,7 +140,7 @@ def actual_adequacy_patch_ini(local_study_w_areas) -> IniFile:

@pytest.fixture
def local_study_with_renewable(local_study_w_thermal) -> Study:
renewable_cluster_name = "renewable cluster"
renewable_cluster_name = "generation"
time_serie = pd.DataFrame(
[
[-9999999980506447872, 0, 9999999980506447872],
Expand Down
Loading

0 comments on commit 8f0f786

Please sign in to comment.