diff --git a/antarest/study/dao/api/common.py b/antarest/study/dao/api/common.py
index 7a7324d57e..abeb4c7466 100644
--- a/antarest/study/dao/api/common.py
+++ b/antarest/study/dao/api/common.py
@@ -39,7 +39,7 @@ def check_thermal_symmetries_integrity(study_dao: "StudyDao", new_symmetries: Th
for area_id, value in new_symmetries.items():
# Handle the case where no symmetries are given. Means we only want to clear them all.
- if all(symmetries == [[]] for symmetries in value.values()):
+ if not (any(symmetry for symmetry in value.values())):
continue
if area_id not in existing_certifications:
@@ -77,7 +77,7 @@ def check_st_storage_symmetries_integrity(
for area_id, value in new_symmetries.items():
# Handle the case where no symmetries are given. Means we only want to clear them all.
- if all(symmetries == [[]] for symmetries in value.values()):
+ if not (any(symmetry for symmetry in value.values())):
continue
if area_id not in existing_certifications:
@@ -102,7 +102,8 @@ def check_st_storage_symmetries_integrity(
def remove_reserve_symmetries_by_cascade(
- symmetries_dict: dict[str, ReserveSymmetries], reserve_ids_to_remove: set[ReserveDefinitionId]
+ symmetries_dict: dict[str, ReserveSymmetries],
+ reserves_to_remove: dict[str, set[ReserveDefinitionId]] | set[ReserveDefinitionId],
) -> dict[str, ReserveSymmetries] | None:
"""
When removing a reserve, we should also remove it from the symmetries.
@@ -111,9 +112,17 @@ def remove_reserve_symmetries_by_cascade(
The updated symmetries dictionary or None if no symmetries were updated.
"""
should_update_symmetries = False
- for symmetries in symmetries_dict.values():
+ for object_id, symmetries in symmetries_dict.items():
for i, symmetry in enumerate(symmetries):
- symmetries[i] = [reserve_id for reserve_id in symmetry if reserve_id not in reserve_ids_to_remove]
+ new_symmetry = []
+ for reserve_id in symmetry:
+ if isinstance(reserves_to_remove, set):
+ if reserve_id not in reserves_to_remove:
+ new_symmetry.append(reserve_id)
+ else:
+ if reserve_id not in reserves_to_remove.get(object_id, []):
+ new_symmetry.append(reserve_id)
+ symmetries[i] = new_symmetry
if len(symmetries[i]) != len(symmetry):
should_update_symmetries = True
if len(symmetries[i]) == 1:
diff --git a/antarest/study/dao/database/common.py b/antarest/study/dao/database/common.py
index f180d09314..634933a012 100644
--- a/antarest/study/dao/database/common.py
+++ b/antarest/study/dao/database/common.py
@@ -9,17 +9,28 @@
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.
-from typing import TYPE_CHECKING
+import json
+from enum import StrEnum
+from typing import TYPE_CHECKING, Any, Sequence, cast
-from sqlalchemy import Table, select
+from sqlalchemy import Row, Table, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from antarest.core.exceptions import AreaNotFound
from antarest.core.utils.sql_utils import upsert_multiple
+from antarest.dbmodel import get_row_representation_as_dict
from antarest.study.business.model.area_properties_model import FILTER_OPTIONS, FrequencyFilter, sort_filter_options
-from antarest.study.dao.common import AreaSeriesMapping
+from antarest.study.business.model.reserve_certification_model import (
+ ReserveCertification,
+)
+from antarest.study.business.model.reserve_symmetries_model import ReserveSymmetries
+from antarest.study.dao.common import AreaSeriesMapping, ReserveSymmetriesMapping
from antarest.study.dao.database.models.area import AREA_TABLE
+from antarest.study.dao.database.models.st_storage_reserve_certification import ST_STORAGE_RESERVE_CERTIFICATION_TABLE
+from antarest.study.dao.database.models.st_storage_reserve_symmetries import ST_STORAGE_RESERVE_SYMMETRIES_TABLE
+from antarest.study.dao.database.models.thermal_reserve_certification import THERMAL_RESERVE_CERTIFICATION_TABLE
+from antarest.study.dao.database.models.thermal_reserve_symmetries import THERMAL_RESERVE_SYMMETRIES_TABLE
if TYPE_CHECKING:
from antarest.study.dao.database.database_study_dao import DatabaseStudyDao
@@ -37,6 +48,14 @@ def area_exists(session: Session, study_data_id: int, area_id: str) -> bool:
return session.execute(stmt).fetchone() is not None
+def validate_areas_exist(session: Session, study_data_id: int, area_ids: set[str]) -> None:
+ stmt = select(AREA_TABLE.c.area_id).where((AREA_TABLE.c.study_data_id == study_data_id))
+ rows = session.execute(stmt).fetchall()
+ existing_area_ids = {row.area_id for row in rows}
+ if invalid_areas := area_ids - existing_area_ids:
+ raise AreaNotFound(*invalid_areas)
+
+
def save_area_matrix(dao: "DatabaseStudyDao", series: AreaSeriesMapping, table: Table) -> None:
session = dao._db_session
study_data_id = dao._study_data_id
@@ -87,3 +106,76 @@ def serialize_frequency_filters(encoded_value: set[FrequencyFilter]) -> str:
if isinstance(encoded_value, str):
return encoded_value
return ", ".join(sort_filter_options(encoded_value))
+
+
+"""
+Reserve types
+"""
+
+
+def _convert_row_to_symmetries(row: Row[Any]) -> ReserveSymmetries:
+ return cast(ReserveSymmetries, json.loads(row.symmetries))
+
+
+class ReserveObjectType(StrEnum):
+ THERMAL = "thermal"
+ ST_STORAGE = "st_storage"
+
+ def _db_key(self) -> str:
+ if self == ReserveObjectType.THERMAL:
+ return "thermal_id"
+ else:
+ return "st_storage_id"
+
+ def db_symmetry_table(self) -> Table:
+ if self == ReserveObjectType.THERMAL:
+ return THERMAL_RESERVE_SYMMETRIES_TABLE
+ else:
+ return ST_STORAGE_RESERVE_SYMMETRIES_TABLE
+
+ def db_certification_table(self) -> Table:
+ if self == ReserveObjectType.THERMAL:
+ return THERMAL_RESERVE_CERTIFICATION_TABLE
+ else:
+ return ST_STORAGE_RESERVE_CERTIFICATION_TABLE
+
+ def convert_symmetry_to_row(
+ self, study_data_id: int, area_id: str, object_id: str, symmetries: ReserveSymmetries
+ ) -> dict[str, Any]:
+ return {
+ "study_data_id": study_data_id,
+ "area_id": area_id,
+ "symmetries": json.dumps([symmetry for symmetry in symmetries if symmetry]),
+ self._db_key(): object_id,
+ }
+
+ def convert_all_rows_to_symmetries(self, rows: Sequence[Row[Any]]) -> dict[str, ReserveSymmetries]:
+ result = {}
+ for row in rows:
+ row_as_dict = get_row_representation_as_dict(row)
+ result[row_as_dict[self._db_key()]] = _convert_row_to_symmetries(row)
+ return result
+
+ def convert_all_rows_to_dict_of_symmetries(self, rows: Sequence[Row[Any]]) -> ReserveSymmetriesMapping:
+ result: ReserveSymmetriesMapping = {}
+ for row in rows:
+ row_as_dict = get_row_representation_as_dict(row)
+ result.setdefault(row.area_id, {})[row_as_dict[self._db_key()]] = _convert_row_to_symmetries(row)
+ return result
+
+ def convert_certification_to_row(
+ self, study_data_id: int, area_id: str, object_id: str, reserve_id: str, certification: ReserveCertification
+ ) -> dict[str, Any]:
+ return {
+ "study_data_id": study_data_id,
+ "area_id": area_id,
+ "reserve_id": reserve_id,
+ self._db_key(): object_id,
+ **certification.model_dump(),
+ }
+
+ def convert_row_to_mapping(self, row: Row[Any]) -> dict[str, Any]:
+ data = get_row_representation_as_dict(row)
+ for key in ("study_data_id", "area_id", self._db_key(), "reserve_id"):
+ del data[key]
+ return data
diff --git a/antarest/study/dao/database/database_renewable_dao.py b/antarest/study/dao/database/database_renewable_dao.py
index 60e667ab44..6424d8277e 100644
--- a/antarest/study/dao/database/database_renewable_dao.py
+++ b/antarest/study/dao/database/database_renewable_dao.py
@@ -23,7 +23,6 @@
from typing_extensions import override
from antarest.core.exceptions import (
- AreaNotFound,
RenewableClusterNotFound,
RenewableClustersNotFound,
)
@@ -35,7 +34,7 @@
)
from antarest.study.dao.api.renewable_dao import RenewableDao
from antarest.study.dao.common import AreaId, RenewableId, RenewableSeriesMapping
-from antarest.study.dao.database.common import validate_area_exists
+from antarest.study.dao.database.common import validate_area_exists, validate_areas_exist
from antarest.study.dao.database.dao_context import DatabaseDaoBase
from antarest.study.dao.database.models.renewable import RENEWABLE_CLUSTER_TABLE, RENEWABLE_SERIES_TABLE
from antarest.study.storage.rawstudy.model.filesystem.matrix.simulator_default import default_scenario_hourly
@@ -63,9 +62,7 @@ def _raise_the_right_renewable_exception(
self, data: dict[AreaId, list[RenewableId]], exc: IntegrityError | None = None
) -> NoReturn:
# Checks if some areas are missing
- existing_ids = set(self.get_impl().get_all_area_ids())
- if invalid_areas := set(data) - existing_ids:
- raise AreaNotFound(*invalid_areas)
+ validate_areas_exist(self._db_session, self._study_data_id, set(data))
# Means the issue lies in the renewables
all_existing_renewables = self.get_all_renewables()
diff --git a/antarest/study/dao/database/database_reserve_certification_dao.py b/antarest/study/dao/database/database_reserve_certification_dao.py
index a1000a0bb9..798cb628eb 100644
--- a/antarest/study/dao/database/database_reserve_certification_dao.py
+++ b/antarest/study/dao/database/database_reserve_certification_dao.py
@@ -12,17 +12,15 @@
from collections.abc import Mapping
from typing import Any, NoReturn
-from sqlalchemy import Row, Select, Table, delete, insert, select
+from sqlalchemy import delete, insert, select
from sqlalchemy.exc import IntegrityError
from typing_extensions import override
from antarest.core.exceptions import (
- AreaNotFound,
ReserveDefinitionsNotFound,
STStoragesNotFound,
ThermalClustersNotFound,
)
-from antarest.dbmodel import get_row_representation_as_dict
from antarest.study.business.model.reserve_certification_model import (
ReserveCertification,
StorageId,
@@ -34,80 +32,35 @@
from antarest.study.business.model.reserve_definition_model import ReserveDefinitionId
from antarest.study.dao.api.reserve_certification_dao import ReserveCertificationDao
from antarest.study.dao.common import AreaId, ThermalId
+from antarest.study.dao.database.common import ReserveObjectType, validate_areas_exist
from antarest.study.dao.database.dao_context import DatabaseDaoBase
-from antarest.study.dao.database.models.st_storage_reserve_certification import ST_STORAGE_RESERVE_CERTIFICATION_TABLE
-from antarest.study.dao.database.models.thermal_reserve_certification import THERMAL_RESERVE_CERTIFICATION_TABLE
-
-_THERMAL_TABLE = THERMAL_RESERVE_CERTIFICATION_TABLE
-_ST_STORAGE_TABLE = ST_STORAGE_RESERVE_CERTIFICATION_TABLE
-
-
-def _convert_thermal_row_to_model(row: Row[Any]) -> ThermalReserveCertification:
- data = get_row_representation_as_dict(row)
- for key in ("study_data_id", "area_id", "thermal_id", "reserve_id"):
- del data[key]
- return ThermalReserveCertification.model_validate(data)
-
-
-def _convert_thermal_model_to_row(
- study_data_id: int, area_id: str, thermal_id: str, reserve_id: str, certification: ThermalReserveCertification
-) -> dict[str, Any]:
- values = certification.model_dump()
- values["reserve_id"] = reserve_id
- values["study_data_id"] = study_data_id
- values["area_id"] = area_id
- values["thermal_id"] = thermal_id
- return values
-
-
-def _convert_st_storage_row_to_model(row: Row[Any]) -> StorageReserveCertification:
- data = get_row_representation_as_dict(row)
- for key in ("study_data_id", "area_id", "st_storage_id", "reserve_id"):
- del data[key]
- return StorageReserveCertification.model_validate(data)
-
-
-def _convert_st_storage_model_to_row(
- study_data_id: int, area_id: str, storage_id: str, reserve_id: str, certification: StorageReserveCertification
-) -> dict[str, Any]:
- values = certification.model_dump()
- values["study_data_id"] = study_data_id
- values["area_id"] = area_id
- values["st_storage_id"] = storage_id
- values["reserve_id"] = reserve_id
- return values
class DatabaseReserveCertificationDao(ReserveCertificationDao, DatabaseDaoBase):
"""Database implementation of ReserveCertificationDao."""
- def _select_one(self, area_id: str, thermal_id: str, reserve_id: str) -> Select[Any]:
- return select(_THERMAL_TABLE).where(
- (_THERMAL_TABLE.c.study_data_id == self._study_data_id)
- & (_THERMAL_TABLE.c.area_id == area_id)
- & (_THERMAL_TABLE.c.thermal_id == thermal_id)
- & (_THERMAL_TABLE.c.reserve_id == reserve_id)
- )
-
@override
def get_all_thermal_reserve_certifications(self) -> dict[AreaId, ThermalReserveCertificationMapping]:
- stmt = select(_THERMAL_TABLE).where(_THERMAL_TABLE.c.study_data_id == self._study_data_id)
+ reserve_type = ReserveObjectType.THERMAL
+ table = reserve_type.db_certification_table()
+ stmt = select(table).where(table.c.study_data_id == self._study_data_id)
rows = self._db_session.execute(stmt).fetchall()
result: dict[AreaId, ThermalReserveCertificationMapping] = {}
for row in rows:
- certification = _convert_thermal_row_to_model(row)
+ certification = ThermalReserveCertification.model_validate(reserve_type.convert_row_to_mapping(row))
result.setdefault(row.area_id, {}).setdefault(row.reserve_id, {})[row.thermal_id] = certification
return result
@override
def get_thermal_reserve_certifications(self, area_id: AreaId) -> ThermalReserveCertificationMapping:
- stmt = select(_THERMAL_TABLE).where(
- (_THERMAL_TABLE.c.study_data_id == self._study_data_id) & (_THERMAL_TABLE.c.area_id == area_id)
- )
+ reserve_type = ReserveObjectType.THERMAL
+ table = reserve_type.db_certification_table()
+ stmt = select(table).where((table.c.study_data_id == self._study_data_id) & (table.c.area_id == area_id))
rows = self._db_session.execute(stmt).fetchall()
result: ThermalReserveCertificationMapping = {}
for row in rows:
- result.setdefault(row.reserve_id, {})[row.thermal_id] = _convert_thermal_row_to_model(row)
+ certification = ThermalReserveCertification.model_validate(reserve_type.convert_row_to_mapping(row))
+ result.setdefault(row.reserve_id, {})[row.thermal_id] = certification
return result
@override
@@ -116,113 +69,118 @@ def save_thermal_reserve_certifications(
) -> None:
if not new_certifications:
return
- values = []
- for area_id, reserves_dict in new_certifications.items():
- for reserve_id, thermal_dict in reserves_dict.items():
- for thermal_id, certification in thermal_dict.items():
- values.append(
- _convert_thermal_model_to_row(
- self._study_data_id, area_id, thermal_id, reserve_id, certification
- )
- )
+
+ old_certifications = self.get_all_thermal_reserve_certifications()
+
try:
- self._clean_db(_THERMAL_TABLE, new_certifications)
- self._insert_data_to_table(_THERMAL_TABLE, values)
+ self._save_certifications(ReserveObjectType.THERMAL, old_certifications, new_certifications)
except IntegrityError as e:
self._db_session.rollback()
self._raise_the_right_thermal_reserve_exception(new_certifications, exc=e)
self._db_session.commit()
- def _raise_the_right_thermal_reserve_exception(
- self,
- data: dict[AreaId, ThermalReserveCertificationMapping],
- exc: IntegrityError | None = None,
- ) -> NoReturn:
- self._raise_exception_if_missing_area(data)
- self._raise_exception_if_missing_reserve(data)
-
- # Checks if some thermals are missing
- all_existing_thermals = self.get_impl().get_all_thermals()
- invalid_thermal_dict: dict[AreaId, set[ThermalId]] = {}
- for area_id, reserves_dict in data.items():
- for thermal_ids in reserves_dict.values():
- if invalid_thermals := set(thermal_ids) - set(all_existing_thermals.get(area_id, [])):
- invalid_thermal_dict.setdefault(area_id, set())
- invalid_thermal_dict[area_id] |= invalid_thermals
-
- if invalid_thermal_dict:
- raise ThermalClustersNotFound(invalid_thermal_dict) from exc
-
- # All objects exist. It means that the DB table does not contain the information.
- raise ValueError("One of the thermal reserve certification table is not filled as it should") from exc
-
@override
def get_all_st_storage_reserve_certifications(self) -> dict[AreaId, StorageReserveCertificationMapping]:
- stmt = select(_ST_STORAGE_TABLE).where(_ST_STORAGE_TABLE.c.study_data_id == self._study_data_id)
+ reserve_type = ReserveObjectType.ST_STORAGE
+ table = reserve_type.db_certification_table()
+ stmt = select(table).where(table.c.study_data_id == self._study_data_id)
rows = self._db_session.execute(stmt).fetchall()
result: dict[AreaId, StorageReserveCertificationMapping] = {}
for row in rows:
- certification = _convert_st_storage_row_to_model(row)
+ certification = StorageReserveCertification.model_validate(reserve_type.convert_row_to_mapping(row))
result.setdefault(row.area_id, {}).setdefault(row.reserve_id, {})[row.st_storage_id] = certification
return result
@override
def get_st_storage_reserve_certifications(self, area_id: AreaId) -> StorageReserveCertificationMapping:
- stmt = select(_ST_STORAGE_TABLE).where(
- (_ST_STORAGE_TABLE.c.study_data_id == self._study_data_id) & (_ST_STORAGE_TABLE.c.area_id == area_id)
- )
+ reserve_type = ReserveObjectType.ST_STORAGE
+ table = reserve_type.db_certification_table()
+ stmt = select(table).where((table.c.study_data_id == self._study_data_id) & (table.c.area_id == area_id))
rows = self._db_session.execute(stmt).fetchall()
result: StorageReserveCertificationMapping = {}
for row in rows:
- result.setdefault(row.reserve_id, {})[row.st_storage_id] = _convert_st_storage_row_to_model(row)
+ certification = StorageReserveCertification.model_validate(reserve_type.convert_row_to_mapping(row))
+ result.setdefault(row.reserve_id, {})[row.st_storage_id] = certification
return result
@override
def save_st_storage_reserve_certifications(
self, new_certifications: dict[AreaId, StorageReserveCertificationMapping]
) -> None:
- if not new_certifications:
- return
- values = self._convert_st_storages_models_to_rows(new_certifications)
+ old_certifications = self.get_all_st_storage_reserve_certifications()
+
try:
- self._clean_db(_ST_STORAGE_TABLE, new_certifications)
- self._insert_data_to_table(_ST_STORAGE_TABLE, values)
+ self._save_certifications(ReserveObjectType.ST_STORAGE, old_certifications, new_certifications)
except IntegrityError as e:
self._db_session.rollback()
self._raise_the_right_st_storage_reserve_exception(new_certifications, exc=e)
+
self._db_session.commit()
- def _convert_st_storages_models_to_rows(
- self, data: dict[str, dict[ReserveDefinitionId, dict[str, StorageReserveCertification]]]
- ) -> list[Any]:
+ def _save_certifications(
+ self,
+ reserve_type: ReserveObjectType,
+ old_certifications: dict[AreaId, dict[ReserveDefinitionId, dict[str, Any]]],
+ new_certifications: dict[AreaId, dict[ReserveDefinitionId, dict[str, Any]]],
+ ) -> None:
values = []
- for area_id, reserves_dict in data.items():
- for reserve_id, storage_dict in reserves_dict.items():
- for storage_id, certification in storage_dict.items():
+ for area_id, reserves_dict in new_certifications.items():
+ for reserve_id, value in reserves_dict.items():
+ for object_id, certification in value.items():
values.append(
- _convert_st_storage_model_to_row(
- self._study_data_id, area_id, storage_id, reserve_id, certification
+ reserve_type.convert_certification_to_row(
+ self._study_data_id, area_id, object_id, reserve_id, certification
)
)
- return values
-
- def _clean_db(
- self, table: Table, data: Mapping[str, Mapping[ReserveDefinitionId, Mapping[str, ReserveCertification]]]
- ) -> None:
- area_ids = set(data)
+ table = reserve_type.db_certification_table()
+ area_ids = set(new_certifications)
stmt = delete(table).where((table.c.study_data_id == self._study_data_id) & (table.c.area_id.in_(area_ids)))
self._db_session.execute(stmt)
-
- def _insert_data_to_table(self, table: Table, values: list[Any]) -> None:
if values:
self._db_session.execute(insert(table), values)
+ # Clean orphan symmetries
+ for area_id in area_ids:
+ missing_reserves: dict[str, set[ReserveDefinitionId]] = {}
+ for reserve_id, v in old_certifications.get(area_id, {}).items():
+ for object_id in v:
+ if object_id not in new_certifications.get(area_id, {}).get(reserve_id, {}):
+ missing_reserves.setdefault(object_id, set()).add(reserve_id)
+ if missing_reserves:
+ if reserve_type == ReserveObjectType.THERMAL:
+ self.get_impl().delete_orphan_thermal_symmetries(area_id, missing_reserves)
+ else:
+ self.get_impl().delete_orphan_st_storage_symmetries(area_id, missing_reserves)
+
+ def _raise_the_right_thermal_reserve_exception(
+ self,
+ data: dict[AreaId, ThermalReserveCertificationMapping],
+ exc: IntegrityError | None = None,
+ ) -> NoReturn:
+ validate_areas_exist(self._db_session, self._study_data_id, set(data))
+ self._raise_exception_if_missing_reserve(data)
+
+ # Checks if some thermals are missing
+ all_existing_thermals = self.get_impl().get_all_thermals()
+ invalid_thermal_dict: dict[AreaId, set[ThermalId]] = {}
+ for area_id, reserves_dict in data.items():
+ for thermal_ids in reserves_dict.values():
+ if invalid_thermals := set(thermal_ids) - set(all_existing_thermals.get(area_id, [])):
+ invalid_thermal_dict.setdefault(area_id, set())
+ invalid_thermal_dict[area_id] |= invalid_thermals
+
+ if invalid_thermal_dict:
+ raise ThermalClustersNotFound(invalid_thermal_dict) from exc
+
+ # All objects exist. It means that the DB table does not contain the information.
+ raise ValueError("One of the thermal reserve certification table is not filled as it should") from exc
+
def _raise_the_right_st_storage_reserve_exception(
self,
data: dict[AreaId, StorageReserveCertificationMapping],
exc: IntegrityError | None = None,
) -> NoReturn:
- self._raise_exception_if_missing_area(data)
+ validate_areas_exist(self._db_session, self._study_data_id, set(data))
self._raise_exception_if_missing_reserve(data)
all_existing_st_storage = self.get_impl().get_all_st_storages()
@@ -241,13 +199,6 @@ def _raise_the_right_st_storage_reserve_exception(
"One of the short-term storage reserve certification table is not filled as it should"
) from exc
- def _raise_exception_if_missing_area(
- self, data: Mapping[str, Mapping[ReserveDefinitionId, Mapping[str, ReserveCertification]]]
- ) -> None:
- existing_ids = set(self.get_impl().get_all_area_ids())
- if invalid_areas := set(data) - existing_ids:
- raise AreaNotFound(*invalid_areas)
-
def _raise_exception_if_missing_reserve(
self, data: Mapping[str, Mapping[ReserveDefinitionId, Mapping[str, ReserveCertification]]]
) -> None:
diff --git a/antarest/study/dao/database/database_reserve_definition_dao.py b/antarest/study/dao/database/database_reserve_definition_dao.py
index ceee360390..740bd5033f 100644
--- a/antarest/study/dao/database/database_reserve_definition_dao.py
+++ b/antarest/study/dao/database/database_reserve_definition_dao.py
@@ -130,21 +130,24 @@ def delete_reserve_definitions(self, area_id: AreaId, reserve_ids: Sequence[Rese
if invalid_ids := set(reserve_ids) - existing:
raise ReserveDefinitionsNotFound({area_id: invalid_ids}) # type: ignore
- self._delete_symmetries_associated_with_deleted_reserve(area_id, reserve_ids)
+ reserve_ids_to_delete = set(reserve_ids)
+ self.delete_orphan_thermal_symmetries(area_id, reserve_ids_to_delete)
+ self.delete_orphan_st_storage_symmetries(area_id, reserve_ids_to_delete)
self._db_session.commit()
- def _delete_symmetries_associated_with_deleted_reserve(
- self, area_id: str, reserve_ids: Sequence[ReserveDefinitionId]
+ def delete_orphan_thermal_symmetries(
+ self, area_id: str, reserves: dict[str, set[ReserveDefinitionId]] | set[ReserveDefinitionId]
) -> None:
- reserve_ids_set = set(reserve_ids)
-
thermal_symmetries_dict = self.get_impl().get_thermal_reserve_symmetries(area_id)
- new_thermal_symmetries = remove_reserve_symmetries_by_cascade(thermal_symmetries_dict, reserve_ids_set)
+ new_thermal_symmetries = remove_reserve_symmetries_by_cascade(thermal_symmetries_dict, reserves)
if new_thermal_symmetries is not None:
self.get_impl().save_thermal_reserve_symmetries({area_id: new_thermal_symmetries})
+ def delete_orphan_st_storage_symmetries(
+ self, area_id: str, reserves: dict[str, set[ReserveDefinitionId]] | set[ReserveDefinitionId]
+ ) -> None:
st_storage_symmetries_dict = self.get_impl().get_st_storage_reserve_symmetries(area_id)
- new_st_storage_symmetries = remove_reserve_symmetries_by_cascade(st_storage_symmetries_dict, reserve_ids_set)
+ new_st_storage_symmetries = remove_reserve_symmetries_by_cascade(st_storage_symmetries_dict, reserves)
if new_st_storage_symmetries is not None:
self.get_impl().save_st_storage_reserve_symmetries({area_id: new_st_storage_symmetries})
diff --git a/antarest/study/dao/database/database_reserve_symmetries_dao.py b/antarest/study/dao/database/database_reserve_symmetries_dao.py
index fb20780851..7b96baf717 100644
--- a/antarest/study/dao/database/database_reserve_symmetries_dao.py
+++ b/antarest/study/dao/database/database_reserve_symmetries_dao.py
@@ -9,15 +9,12 @@
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.
-import json
-from enum import StrEnum
-from typing import Any, Sequence, cast
+from typing import Any
-from sqlalchemy import Row, Table, delete, insert, select
+from sqlalchemy import Table, delete, insert, select
from sqlalchemy.exc import IntegrityError
from typing_extensions import override
-from antarest.dbmodel import get_row_representation_as_dict
from antarest.study.business.model.reserve_symmetries_model import ReserveSymmetries
from antarest.study.dao.api.common import check_st_storage_symmetries_integrity, check_thermal_symmetries_integrity
from antarest.study.dao.api.reserve_symmetries_dao import ReserveSymmetriesDao
@@ -29,54 +26,8 @@
ThermalId,
ThermalReserveSymmetriesMapping,
)
+from antarest.study.dao.database.common import ReserveObjectType
from antarest.study.dao.database.dao_context import DatabaseDaoBase
-from antarest.study.dao.database.models.st_storage_reserve_symmetries import ST_STORAGE_RESERVE_SYMMETRIES_TABLE
-from antarest.study.dao.database.models.thermal_reserve_symmetries import THERMAL_RESERVE_SYMMETRIES_TABLE
-
-
-def _convert_row_to_model(row: Row[Any]) -> ReserveSymmetries:
- return cast(ReserveSymmetries, json.loads(row.symmetries))
-
-
-class SymmetryType(StrEnum):
- THERMAL = "thermal"
- ST_STORAGE = "st_storage"
-
- def _db_key(self) -> str:
- if self == SymmetryType.THERMAL:
- return "thermal_id"
- else:
- return "st_storage_id"
-
- def db_table(self) -> Table:
- if self == SymmetryType.THERMAL:
- return THERMAL_RESERVE_SYMMETRIES_TABLE
- else:
- return ST_STORAGE_RESERVE_SYMMETRIES_TABLE
-
- def convert_to_row(
- self, study_data_id: int, area_id: str, object_id: str, symmetries: ReserveSymmetries
- ) -> dict[str, Any]:
- return {
- "study_data_id": study_data_id,
- "area_id": area_id,
- "symmetries": json.dumps(symmetries),
- self._db_key(): object_id,
- }
-
- def convert_all_rows_to_model(self, rows: Sequence[Row[Any]]) -> dict[str, ReserveSymmetries]:
- result = {}
- for row in rows:
- row_as_dict = get_row_representation_as_dict(row)
- result[row_as_dict[self._db_key()]] = _convert_row_to_model(row)
- return result
-
- def convert_all_rows_to_dict_of_models(self, rows: Sequence[Row[Any]]) -> ReserveSymmetriesMapping:
- result: ReserveSymmetriesMapping = {}
- for row in rows:
- row_as_dict = get_row_representation_as_dict(row)
- result.setdefault(row.area_id, {})[row_as_dict[self._db_key()]] = _convert_row_to_model(row)
- return result
class DatabaseReserveSymmetriesDao(ReserveSymmetriesDao, DatabaseDaoBase):
@@ -84,40 +35,46 @@ class DatabaseReserveSymmetriesDao(ReserveSymmetriesDao, DatabaseDaoBase):
@override
def get_all_thermal_reserve_symmetries(self) -> ThermalReserveSymmetriesMapping:
- return self._get_all_symmetries(SymmetryType.THERMAL)
+ return self._get_all_symmetries(ReserveObjectType.THERMAL)
@override
def get_all_st_storage_reserve_symmetries(self) -> STStorageReserveSymmetriesMapping:
- return self._get_all_symmetries(SymmetryType.ST_STORAGE)
+ return self._get_all_symmetries(ReserveObjectType.ST_STORAGE)
- def _get_all_symmetries(self, symmetry_type: SymmetryType) -> ReserveSymmetriesMapping:
- table = symmetry_type.db_table()
+ def _get_all_symmetries(self, reserve_type: ReserveObjectType) -> ReserveSymmetriesMapping:
+ table = reserve_type.db_symmetry_table()
stmt = select(table).where(table.c.study_data_id == self._study_data_id)
rows = self._db_session.execute(stmt).fetchall()
- return symmetry_type.convert_all_rows_to_dict_of_models(rows)
+ return reserve_type.convert_all_rows_to_dict_of_symmetries(rows)
@override
def get_thermal_reserve_symmetries(self, area_id: AreaId) -> dict[ThermalId, ReserveSymmetries]:
- return self._get_all_symmetries_for_area(area_id, SymmetryType.THERMAL)
+ return self._get_all_symmetries_for_area(area_id, ReserveObjectType.THERMAL)
@override
def get_st_storage_reserve_symmetries(self, area_id: AreaId) -> dict[StStorageId, ReserveSymmetries]:
- return self._get_all_symmetries_for_area(area_id, SymmetryType.ST_STORAGE)
+ return self._get_all_symmetries_for_area(area_id, ReserveObjectType.ST_STORAGE)
- def _get_all_symmetries_for_area(self, area_id: str, symmetry_type: SymmetryType) -> dict[str, ReserveSymmetries]:
- table = symmetry_type.db_table()
+ def _get_all_symmetries_for_area(
+ self, area_id: str, reserve_type: ReserveObjectType
+ ) -> dict[str, ReserveSymmetries]:
+ table = reserve_type.db_symmetry_table()
stmt = select(table).where((table.c.study_data_id == self._study_data_id) & (table.c.area_id == area_id))
rows = self._db_session.execute(stmt).fetchall()
- return symmetry_type.convert_all_rows_to_model(rows)
+ return reserve_type.convert_all_rows_to_symmetries(rows)
@override
def save_thermal_reserve_symmetries(self, data: ThermalReserveSymmetriesMapping) -> None:
- # Check foreign keys integrity
- check_thermal_symmetries_integrity(self.get_impl(), data)
+ reserve_type = ReserveObjectType.THERMAL
+ values = self._build_symmetry_rows(data, reserve_type)
+
+ if values:
+ # Check foreign keys integrity for values to insert
+ check_thermal_symmetries_integrity(self.get_impl(), data)
# Save the new values
try:
- self._save_reserve_symmetries(data, SymmetryType.THERMAL)
+ self._save_reserve_symmetries(set(data), reserve_type.db_symmetry_table(), values)
except IntegrityError as e:
self._db_session.rollback()
thermals = {area_id: list(thermal_dict) for area_id, thermal_dict in data.items()}
@@ -126,28 +83,35 @@ def save_thermal_reserve_symmetries(self, data: ThermalReserveSymmetriesMapping)
@override
def save_st_storage_reserve_symmetries(self, data: STStorageReserveSymmetriesMapping) -> None:
- # Check foreign keys integrity
- check_st_storage_symmetries_integrity(self.get_impl(), data)
+ reserve_type = ReserveObjectType.ST_STORAGE
+ values = self._build_symmetry_rows(data, reserve_type)
+
+ if values:
+ # Check foreign keys integrity for values to insert
+ check_st_storage_symmetries_integrity(self.get_impl(), data)
# Save the new values
try:
- self._save_reserve_symmetries(data, SymmetryType.ST_STORAGE)
+ self._save_reserve_symmetries(set(data), reserve_type.db_symmetry_table(), values)
except IntegrityError as e:
self._db_session.rollback()
st_storages = {area_id: list(st_storage_dict) for area_id, st_storage_dict in data.items()}
self.get_impl().raise_the_right_storage_exception(st_storages, exc=e)
self._db_session.commit()
- def _save_reserve_symmetries(self, data: ReserveSymmetriesMapping, symmetry_type: SymmetryType) -> None:
+ def _build_symmetry_rows(
+ self, data: ReserveSymmetriesMapping, reserve_type: ReserveObjectType
+ ) -> list[dict[str, Any]]:
values = []
for area_id, value in data.items():
for object_id, symmetries in value.items():
if not (any(symmetry for symmetry in symmetries)):
continue
- values.append(symmetry_type.convert_to_row(self._study_data_id, area_id, object_id, symmetries))
+ values.append(reserve_type.convert_symmetry_to_row(self._study_data_id, area_id, object_id, symmetries))
+ return values
- table = symmetry_type.db_table()
- stmt = delete(table).where((table.c.study_data_id == self._study_data_id) & (table.c.area_id.in_(set(data))))
+ def _save_reserve_symmetries(self, area_ids: set[str], table: Table, values: list[dict[str, Any]]) -> None:
+ stmt = delete(table).where((table.c.study_data_id == self._study_data_id) & (table.c.area_id.in_(area_ids)))
self._db_session.execute(stmt)
if values:
self._db_session.execute(insert(table), values)
diff --git a/antarest/study/dao/database/database_st_storage_dao.py b/antarest/study/dao/database/database_st_storage_dao.py
index 3e38fec1c3..df0ae540e4 100644
--- a/antarest/study/dao/database/database_st_storage_dao.py
+++ b/antarest/study/dao/database/database_st_storage_dao.py
@@ -45,7 +45,7 @@
StStorageId,
StStorageSeriesMapping,
)
-from antarest.study.dao.database.common import validate_area_exists
+from antarest.study.dao.database.common import validate_area_exists, validate_areas_exist
from antarest.study.dao.database.dao_context import DatabaseDaoBase
from antarest.study.dao.database.models.st_storage import (
COST_INJECTION_TABLE,
@@ -108,9 +108,7 @@ def _convert_db_row_to_constraint(self, row: Row[Any]) -> STStorageAdditionalCon
def _raise_the_right_exc(self, data: dict[AreaId, list[StStorageId]], exc: IntegrityError | None = None) -> None:
# Checks if some areas are missing
- existing_ids = set(self.get_impl().get_all_area_ids())
- if invalid_areas := set(data) - existing_ids:
- raise AreaNotFound(*invalid_areas)
+ validate_areas_exist(self._db_session, self._study_data_id, set(data))
# Means the issue lies in the short-term storages
all_existing_storages = self.get_all_st_storages()
diff --git a/antarest/study/dao/database/database_thermal_dao.py b/antarest/study/dao/database/database_thermal_dao.py
index c2604c3e86..98690c7b29 100644
--- a/antarest/study/dao/database/database_thermal_dao.py
+++ b/antarest/study/dao/database/database_thermal_dao.py
@@ -22,7 +22,7 @@
from sqlalchemy.exc import IntegrityError
from typing_extensions import override
-from antarest.core.exceptions import AreaNotFound, ThermalClusterNotFound, ThermalClustersNotFound
+from antarest.core.exceptions import ThermalClusterNotFound, ThermalClustersNotFound
from antarest.core.utils.sql_utils import upsert_multiple
from antarest.dbmodel import get_row_representation_as_dict
from antarest.study.business.model.thermal_cluster_model import (
@@ -32,7 +32,7 @@
)
from antarest.study.dao.api.thermal_dao import ThermalDao
from antarest.study.dao.common import AreaId, SeriesId, ThermalId, ThermalSeriesMapping
-from antarest.study.dao.database.common import validate_area_exists
+from antarest.study.dao.database.common import validate_area_exists, validate_areas_exist
from antarest.study.dao.database.dao_context import DatabaseDaoBase
from antarest.study.dao.database.models.thermal import (
THERMAL_CLUSTER_TABLE,
@@ -113,9 +113,7 @@ def raise_the_right_thermal_exception(
self, data: dict[AreaId, list[ThermalId]], exc: IntegrityError | None = None
) -> NoReturn:
# Checks if some areas are missing
- existing_ids = set(self.get_impl().get_all_area_ids())
- if invalid_areas := set(data) - existing_ids:
- raise AreaNotFound(*invalid_areas)
+ validate_areas_exist(self._db_session, self._study_data_id, set(data))
# Means the issue lies in the thermals
all_existing_thermals = self.get_all_thermals()
diff --git a/antarest/study/dao/file/file_study_st_storage_dao.py b/antarest/study/dao/file/file_study_st_storage_dao.py
index 9cb083ba88..43cde40458 100644
--- a/antarest/study/dao/file/file_study_st_storage_dao.py
+++ b/antarest/study/dao/file/file_study_st_storage_dao.py
@@ -25,7 +25,11 @@
)
from antarest.study.dao.api.st_storage_dao import STStorageDao
from antarest.study.dao.common import AreaId, StStorageConstraintSeriesMapping, StStorageId, StStorageSeriesMapping
-from antarest.study.dao.file.common import check_area_exists
+from antarest.study.dao.file.common import (
+ check_area_exists,
+ get_st_storage_reserve_participations_as_yaml_content,
+ get_st_storage_reserve_path,
+)
from antarest.study.model import STUDY_VERSION_9_2, STUDY_VERSION_10_2
from antarest.study.storage.rawstudy.model.filesystem.config.st_storage import (
parse_st_storage,
@@ -560,19 +564,19 @@ def _remove_st_storage_reserve_certifications(self, area_id: str, storage_id: st
# Cascade: Remove any reserve certification attached to the deleted storage.
# Avoids leaving orphan sections in `input/st-storage/clusters//reserve-participations.yml`.
"""
- if self.get_file_study().config.version < STUDY_VERSION_10_2:
+ file_study = self.get_file_study()
+ if file_study.config.version < STUDY_VERSION_10_2:
# Reserves only exist in version 10.2+
return
- storage_exists = False
- all_area_certifications = self.get_impl().get_st_storage_reserve_certifications(area_id)
- for reserve_id, thermal_dict in all_area_certifications.items():
- for current_storage_id in thermal_dict:
- if current_storage_id == storage_id:
- del all_area_certifications[reserve_id][storage_id]
- storage_exists = True
- break
-
- if storage_exists:
- # Avoid performing an empty save if there are no certifications to remove
- self.get_impl().save_st_storage_reserve_certifications({area_id: all_area_certifications})
+ st_storage_exists = False
+ yaml_content = get_st_storage_reserve_participations_as_yaml_content(area_id, file_study)
+ for k, participation in enumerate(yaml_content["participations"]):
+ if participation["storage"] == storage_id:
+ st_storage_exists = True
+ del yaml_content["participations"][k]
+ break
+
+ if st_storage_exists:
+ # Avoid performing an empty save if there is no st-storage to remove
+ file_study.tree.save(yaml_content, get_st_storage_reserve_path(area_id))
diff --git a/antarest/study/dao/file/file_study_thermal_dao.py b/antarest/study/dao/file/file_study_thermal_dao.py
index 97e0ce3a30..ce4bc39ccc 100644
--- a/antarest/study/dao/file/file_study_thermal_dao.py
+++ b/antarest/study/dao/file/file_study_thermal_dao.py
@@ -26,7 +26,11 @@
from antarest.study.business.model.thermal_cluster_model import ThermalCluster, initialize_thermal_cluster
from antarest.study.dao.api.thermal_dao import ThermalDao
from antarest.study.dao.common import AreaId, ThermalId, ThermalSeriesMapping
-from antarest.study.dao.file.common import check_area_exists
+from antarest.study.dao.file.common import (
+ check_area_exists,
+ get_thermal_reserve_participations_as_yaml_content,
+ get_thermal_reserve_path,
+)
from antarest.study.model import STUDY_VERSION_10_2
from antarest.study.storage.rawstudy.model.filesystem.config.model import FileStudyTreeConfig
from antarest.study.storage.rawstudy.model.filesystem.config.thermal import (
@@ -309,19 +313,19 @@ def _remove_thermal_reserve_certifications(self, area_id: str, thermal_id: str)
# Cascade: Remove any reserve certification attached to the deleted cluster.
# Avoids leaving orphan sections in `input/thermal/clusters//reserve-participations.yml`.
"""
- if self.get_file_study().config.version < STUDY_VERSION_10_2:
+ file_study = self.get_file_study()
+ if file_study.config.version < STUDY_VERSION_10_2:
# Reserves only exist in version 10.2+
return
thermal_exists = False
- all_area_certifications = self.get_impl().get_thermal_reserve_certifications(area_id)
- for reserve_id, thermal_dict in all_area_certifications.items():
- for cluster_id in thermal_dict:
- if cluster_id == thermal_id:
- del all_area_certifications[reserve_id][thermal_id]
- thermal_exists = True
- break
+ yaml_content = get_thermal_reserve_participations_as_yaml_content(area_id, file_study)
+ for k, participation in enumerate(yaml_content["participations"]):
+ if participation["cluster"] == thermal_id:
+ thermal_exists = True
+ del yaml_content["participations"][k]
+ break
if thermal_exists:
- # Avoid performing an empty save if there are no certifications to remove
- self.get_impl().save_thermal_reserve_certifications({area_id: all_area_certifications})
+ # Avoid performing an empty save if there is no thermal to remove
+ file_study.tree.save(yaml_content, get_thermal_reserve_path(area_id))
diff --git a/antarest/study/storage/rawstudy/model/filesystem/config/reserve_participations.py b/antarest/study/storage/rawstudy/model/filesystem/config/reserve_participations.py
index b31ffc44cf..dfb8fade4b 100644
--- a/antarest/study/storage/rawstudy/model/filesystem/config/reserve_participations.py
+++ b/antarest/study/storage/rawstudy/model/filesystem/config/reserve_participations.py
@@ -200,14 +200,23 @@ def _build_participations_from_symmetries(
) -> None:
"""
Builds a participation entry for every asset that has symmetries, appending it to `participations`.
+
+ This method is in charge of silently ignoring symmetries that have no associated certifications.
"""
for asset_id, reserve_symmetries in symmetries.items():
- certifs = certifications.pop(asset_id, {})
participation: dict[str, Any] = cls.initialize_participation(asset_id)
- if certifs:
- participation["certifications"] = [{"reserve": r_id, **c.model_dump()} for r_id, c in certifs.items()]
- if reserve_symmetries != [[]]:
- participation["symmetries"] = [{"reserves": s} for s in reserve_symmetries]
+
+ if asset_id in certifications:
+ if certifs := certifications.pop(asset_id):
+ certification = [{"reserve": r_id, **c.model_dump()} for r_id, c in certifs.items()]
+ participation["certifications"] = certification
+
+ if any(symmetry for symmetry in reserve_symmetries):
+ symmetries_with_certification = [
+ [reserve_id for reserve_id in symmetry if reserve_id in certifs]
+ for symmetry in reserve_symmetries
+ ]
+ participation["symmetries"] = [{"reserves": s} for s in symmetries_with_certification if len(s) > 1]
participations.append(participation)
diff --git a/tests/study/dao/test_st_storage_reserves_dao.py b/tests/study/dao/test_st_storage_reserves_dao.py
index e32d0f77e9..372eb34450 100644
--- a/tests/study/dao/test_st_storage_reserves_dao.py
+++ b/tests/study/dao/test_st_storage_reserves_dao.py
@@ -49,22 +49,23 @@ def test_symmetries_and_certifications_do_not_overwrite_each_other(dao_10_2: Stu
}
)
- # Save 2 symmetries. Then 1 certification. Ensures the certification writing didn't affect the symmetries.
- dao.save_st_storage_reserve_symmetries({"fr": {"sts1": [["r1", "r2"], ["r3", "r4"]]}})
+ # Save 2 symmetries. Then 1 certification.
+ # As we've removed the certification from `r4` and from `r3` for `sts1`, the relative symmetry should be removed.
+ dao.save_st_storage_reserve_symmetries({"fr": {"sts1": [["r1", "r2"], ["r3", "r4"]], "sts2": [["r1", "r3"]]}})
dao.save_st_storage_reserve_certifications(
{
"fr": {
"r1": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
- "r2": {"sts2": StorageReserveCertification()},
+ "r2": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
"r3": {"sts2": StorageReserveCertification()},
}
}
)
- assert dao.get_st_storage_reserve_symmetries("fr") == {"sts1": [["r1", "r2"], ["r3", "r4"]]}
+ assert dao.get_st_storage_reserve_symmetries("fr") == {"sts1": [["r1", "r2"]], "sts2": [["r1", "r3"]]}
assert dao.get_st_storage_reserve_certifications("fr") == {
"r1": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
- "r2": {"sts2": StorageReserveCertification()},
+ "r2": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
"r3": {"sts2": StorageReserveCertification()},
}
@@ -73,7 +74,7 @@ def test_symmetries_and_certifications_do_not_overwrite_each_other(dao_10_2: Stu
assert dao.get_st_storage_reserve_certifications("fr") == {
"r1": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
- "r2": {"sts2": StorageReserveCertification()},
+ "r2": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
"r3": {"sts2": StorageReserveCertification()},
}
# The symmetry should also be overwritten by the new value.
@@ -119,3 +120,36 @@ def test_clearing_symmetries(dao_10_2: StudyDao, symmetries: dict[str, ReserveSy
# Ensures it's now empty
assert dao.get_st_storage_reserve_symmetries("fr") == {}
+
+
+def test_symmetries_removal_when_deleting_st_storage_or_certification(dao_10_2: StudyDao) -> None:
+ dao = dao_10_2
+ _set_up(dao)
+
+ # Both st-storages are certified on both reserves and symmetric on the r1/r2 pair.
+ dao.save_st_storage_reserve_certifications(
+ {
+ "fr": {
+ "r1": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
+ "r2": {"sts1": StorageReserveCertification(), "sts2": StorageReserveCertification()},
+ }
+ }
+ )
+ dao.save_st_storage_reserve_symmetries({"fr": {"sts1": [["r1", "r2"]], "sts2": [["r1", "r2"]]}})
+
+ # Removes the short-term storage `sts1`.
+ dao.delete_st_storage("fr", STStorage(name="sts1"))
+
+ # The certifications of the deleted st-storage are gone, sts2 is untouched.
+ assert dao.get_st_storage_reserve_certifications("fr") == {
+ "r1": {"sts2": StorageReserveCertification()},
+ "r2": {"sts2": StorageReserveCertification()},
+ }
+
+ # Same for the symmetries.
+ assert dao.get_st_storage_reserve_symmetries("fr") == {"sts2": [["r1", "r2"]]}
+
+ # Removing a certification should also clean the symmetries.
+ dao.save_st_storage_reserve_certifications({"fr": {}})
+
+ assert dao.get_st_storage_reserve_symmetries("fr") == {}
diff --git a/tests/study/dao/test_thermal_reserves_dao.py b/tests/study/dao/test_thermal_reserves_dao.py
index cc1b4e7951..40ecb05a9a 100644
--- a/tests/study/dao/test_thermal_reserves_dao.py
+++ b/tests/study/dao/test_thermal_reserves_dao.py
@@ -9,6 +9,9 @@
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.
+import pytest
+
+from antarest.core.exceptions import ReserveCertificationsNotFound
from antarest.study.business.model.area_properties_model import AreaProperties
from antarest.study.business.model.reserve_certification_model import ThermalReserveCertification
from antarest.study.business.model.reserve_definition_model import ReserveDefinition, ReserveType
@@ -37,7 +40,10 @@ def test_symmetries_and_certifications_do_not_overwrite_each_other(dao_10_2: Stu
# A cluster can only be symmetric on reserves it is certified for, so certify both clusters first.
certification = ThermalReserveCertification()
certifications = {
- reserve_id: {"th1": certification, "th2": certification} for reserve_id in ["r1", "r2", "r3", "r4"]
+ "r1": {"th1": certification, "th2": certification},
+ "r2": {"th1": certification, "th2": certification},
+ "r3": {"th1": certification, "th2": certification},
+ "r4": {"th1": certification},
}
dao.save_thermal_reserve_certifications({"fr": certifications})
@@ -56,6 +62,10 @@ def test_symmetries_and_certifications_do_not_overwrite_each_other(dao_10_2: Stu
# The symmetry should also be overwritten by the new value.
assert dao.get_thermal_reserve_symmetries("fr") == {"th2": [["r1", "r2", "r3"]]}
+ # Remove a certification. Should remove the related symmetries.
+ dao.save_thermal_reserve_certifications({"fr": {"r3": {"th1": certification}, "r4": {"th1": certification}}})
+ assert dao.get_all_thermal_reserve_symmetries() == {}
+
def test_deleting_the_last_reserves_removes_their_symmetries(dao_10_2: StudyDao) -> None:
# Deleting a reserve cascades on the certifications and on the symmetries referencing it.
@@ -73,3 +83,47 @@ def test_deleting_the_last_reserves_removes_their_symmetries(dao_10_2: StudyDao)
symmetries = dao.get_thermal_reserve_symmetries("fr")
assert symmetries == {}
+
+
+def test_symmetries_removal_when_deleting_thermal_cluster_or_certification(dao_10_2: StudyDao) -> None:
+ dao = dao_10_2
+ _set_up(dao)
+
+ # Both clusters are certified on both reserves and symmetric on the r1/r2 pair.
+ dao.save_thermal_reserve_certifications(
+ {
+ "fr": {
+ "r1": {"th1": ThermalReserveCertification(), "th2": ThermalReserveCertification()},
+ "r2": {"th1": ThermalReserveCertification(), "th2": ThermalReserveCertification()},
+ }
+ }
+ )
+ dao.save_thermal_reserve_symmetries({"fr": {"th1": [["r1", "r2"]], "th2": [["r1", "r2"]]}})
+
+ # Removes the thermal cluster `th1`.
+ dao.delete_thermal("fr", "th1")
+
+ # The certifications of the deleted cluster are gone, th2 is untouched.
+ assert dao.get_thermal_reserve_certifications("fr") == {
+ "r1": {"th2": ThermalReserveCertification()},
+ "r2": {"th2": ThermalReserveCertification()},
+ }
+
+ # Same for the symmetries.
+ assert dao.get_thermal_reserve_symmetries("fr") == {"th2": [["r1", "r2"]]}
+
+ # Removing a certification should also clean the symmetries.
+ dao.save_thermal_reserve_certifications({"fr": {}})
+
+ assert dao.get_thermal_reserve_symmetries("fr") == {}
+
+
+def test_save_symmetry_without_certification_or_without_thermal(dao_10_2: StudyDao) -> None:
+ dao = dao_10_2
+ _set_up(dao)
+
+ with pytest.raises(ReserveCertificationsNotFound):
+ dao.save_thermal_reserve_symmetries({"fr": {"th1": [["r1", "r2"]]}})
+
+ with pytest.raises(ReserveCertificationsNotFound):
+ dao.save_thermal_reserve_symmetries({"fr": {"fake_thermal": [["r1", "r2"]]}})