Skip to content

Commit

Permalink
binding_constraints terms properly written
Browse files Browse the repository at this point in the history
  • Loading branch information
vargastat authored and MartinBelthle committed Dec 4, 2024
1 parent 4973fe0 commit f9b161e
Show file tree
Hide file tree
Showing 4 changed files with 94 additions and 39 deletions.
2 changes: 1 addition & 1 deletion src/antares/model/study.py
Original file line number Diff line number Diff line change
Expand Up @@ -287,7 +287,7 @@ def create_binding_constraint(
binding_constraint = self._binding_constraints_service.create_binding_constraint(
name, properties, terms, less_term_matrix, equal_term_matrix, greater_term_matrix
)
self._binding_constraints[binding_constraint.name] = binding_constraint
self._binding_constraints[binding_constraint.id] = binding_constraint
return binding_constraint

def update_settings(self, settings: StudySettings) -> None:
Expand Down
101 changes: 68 additions & 33 deletions src/antares/service/local_services/binding_constraint_local.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.
from typing import Any, Optional
from typing import Any, Optional, Union

import numpy as np
import pandas as pd
Expand All @@ -23,14 +23,29 @@
BindingConstraintProperties,
BindingConstraintPropertiesLocal,
ConstraintMatrixName,
ConstraintTerm,
ConstraintTerm, LinkData, ClusterData,
)
from antares.service.base_services import BaseBindingConstraintService
from antares.tools.ini_tool import IniFile, IniFileTypes
from antares.tools.matrix_tool import df_save
from antares.tools.time_series_tool import TimeSeriesFileType


def serialize_term_data(data: Union[LinkData,ClusterData], offset: Optional[int], weight: Optional[float]) -> Union[str, None]:
"""
Serializes the term data to be correctly written in INI.
"""
if isinstance(data, LinkData):

if offset is not None:
return f"0.000000%{offset}"
if weight is not None:
return f"{weight}"
if weight is None:
return "0"
else:
return None

class BindingConstraintLocalService(BaseBindingConstraintService):
def __init__(self, config: LocalConfiguration, study_name: str, **kwargs: Any) -> None:
super().__init__(**kwargs)
Expand All @@ -55,6 +70,13 @@ def create_binding_constraint(
)
constraint.properties = constraint.local_properties.yield_binding_constraint_properties()

current_ini_content = self.ini_file.ini_dict_binding_constraints or {}
if any(
values.get("name") == name
for values in current_ini_content.values()
):
raise BindingConstraintCreationError(constraint_name=name, message= f"A binding constraint with the name {name} already exists.")

self._write_binding_constraint_ini(constraint.properties, name, name, terms)

self._store_time_series(constraint, less_term_matrix, equal_term_matrix, greater_term_matrix)
Expand Down Expand Up @@ -96,45 +118,58 @@ def _check_if_empty_ts(time_step: BindingConstraintFrequency, time_series: Optio
return time_series if time_series is not None else pd.DataFrame(np.zeros([time_series_length, 1]))

def _write_binding_constraint_ini(
self,
properties: BindingConstraintProperties,
constraint_name: str,
constraint_id: str,
terms: Optional[list[ConstraintTerm]] = None,
self,
properties: BindingConstraintProperties,
constraint_name: str,
constraint_id: str,
terms: Optional[list[ConstraintTerm]] = None,
) -> None:
"""
Write a single binding constraint to the INI file, reconstructing a full BindingConstraintPropertiesLocal instance.
Write or update a binding constraint in the INI file.
Args:
properties (BindingConstraintProperties): Basic properties of the binding constraint.
constraint_name (str): The name of the constraint.
constraint_id (str): The ID of the constraint.
terms (dict[str, ConstraintTerm], optional): Terms applying to the binding constraint. Defaults to None.
Raises:
BindingConstraintCreationError: If a binding constraint with the same name already exists in the INI file.
"""

current_ini_content = self.ini_file.ini_dict or {}

if constraint_name in current_ini_content:
if terms in [None, {}]:
raise BindingConstraintCreationError(
constraint_name=constraint_name,
message=f"A binding constraint with the name {constraint_name} already exists with terms.",
)
current_ini_content = self.ini_file.ini_dict_binding_constraints or {}

terms_dict = {term.id: term for term in terms} if terms else {}

full_properties = BindingConstraintPropertiesLocal(
constraint_name=constraint_name, constraint_id=constraint_id, terms=terms_dict, **properties.model_dump()
existing_section = next(
(section for section, values in current_ini_content.items() if values.get("name") == constraint_name),
None,
)

current_ini_content = self.ini_file.ini_dict or {}
if existing_section:
# If constraint exists, update the terms
existing_terms = current_ini_content[existing_section]

# Serialize the terms data (this assumes you want to serialize LinkData or ClusterData in `terms`)
serialized_terms = {term.id: serialize_term_data(term.data, term.offset, term.weight) for term in terms} if terms else {}


existing_terms.update(serialized_terms) # type: ignore
current_ini_content[existing_section] = existing_terms

# Persist the updated INI content
self.ini_file.ini_dict_binding_constraints = current_ini_content
self.ini_file.write_ini_file()
else:


terms_dict = {
term.id: ConstraintTerm(data=term.data, offset=term.offset, weight=term.weight)
if isinstance(term.data, (LinkData, ClusterData))
else term
for term in terms
} if terms else {}

current_ini_content[full_properties.constraint_name] = full_properties.list_ini_fields
full_properties = BindingConstraintPropertiesLocal(
constraint_name=constraint_name, constraint_id=constraint_id, terms=terms_dict,
**properties.model_dump()
)

self.ini_file.ini_dict = current_ini_content
section_index = len(current_ini_content)
current_ini_content[str(section_index)] = full_properties.list_ini_fields

self.ini_file.ini_dict_binding_constraints = current_ini_content
self.ini_file.write_ini_file()

def add_constraint_terms(self, constraint: BindingConstraint, terms: list[ConstraintTerm]) -> list[ConstraintTerm]:
Expand All @@ -150,19 +185,19 @@ def add_constraint_terms(self, constraint: BindingConstraint, terms: list[Constr
"""

new_terms = {
**constraint.local_properties.terms, # Existing terms
**{term.id: term for term in terms if term.id not in constraint.get_terms()}, # New terms
**constraint.local_properties.terms,
**{term.id: term for term in terms if term.id not in constraint.get_terms()},
}

constraint.local_properties.terms = new_terms

list(new_terms.values())
terms_values = list(new_terms.values())

self._write_binding_constraint_ini(
properties=constraint.properties,
constraint_name=constraint.name,
constraint_id=constraint.id,
terms=list(new_terms.values()),
terms=terms_values,
)

return list(new_terms.values())
Expand Down
20 changes: 20 additions & 0 deletions src/antares/tools/ini_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from pydantic import BaseModel

from antares.model.binding_constraint import ConstraintTerm
from antares.tools.custom_raw_config_parser import CustomRawConfigParser
from antares.tools.model_tools import filter_out_empty_model_fields

Expand Down Expand Up @@ -90,6 +91,22 @@ def ini_dict(self, new_ini_dict: dict[str, dict[str, str]]) -> None:
self._ini_contents = CustomRawConfigParser()
self._ini_contents.read_dict(new_ini_dict)

@property
def ini_dict_binding_constraints(self) -> dict[str, dict[str, str]]:
return {
section: dict(self._ini_contents[section])
for section in self._ini_contents.sections()
}

@ini_dict_binding_constraints.setter
def ini_dict_binding_constraints(self, new_ini_dict: dict[str, dict[str, str]]) -> None:
"""Set INI file contents for binding constraints."""
self._ini_contents = CustomRawConfigParser()
for index, (section, values) in enumerate(new_ini_dict.items()):
self._ini_contents.add_section(str(index))
for key, value in values.items():
self._ini_contents.set(str(index), key, value)

@property
def parsed_ini(self) -> CustomRawConfigParser:
"""Ini contents as a CustomRawConfigParser"""
Expand Down Expand Up @@ -132,6 +149,9 @@ def update_from_ini_file(self) -> None:

self._ini_contents = parsed_ini

def update_binding_constraints_from_ini_file(self) -> None:
self.update_from_ini_file()

def write_ini_file(
self,
sort_sections: bool = False,
Expand Down
10 changes: 5 additions & 5 deletions tests/antares/services/local_services/test_study.py
Original file line number Diff line number Diff line change
Expand Up @@ -2099,7 +2099,7 @@ def test_constraints_ini_have_correct_default_content(
self, local_study_with_constraint, test_constraint, default_constraint_properties
):
# Given
expected_ini_contents = """[test constraint]
expected_ini_contents = """[0]
name = test constraint
id = test constraint
enabled = true
Expand Down Expand Up @@ -2133,7 +2133,7 @@ def test_constraints_and_ini_have_custom_properties(self, local_study_with_const
filter_synthesis="monthly",
group="test group",
)
expected_ini_content = """[test constraint]
expected_ini_content = """[0]
name = test constraint
id = test constraint
enabled = true
Expand All @@ -2143,7 +2143,7 @@ def test_constraints_and_ini_have_custom_properties(self, local_study_with_const
filter-synthesis = hourly
group = default
[test constraint two]
[1]
name = test constraint two
id = test constraint two
enabled = false
Expand Down Expand Up @@ -2176,7 +2176,7 @@ def test_constraint_can_add_term(self, test_constraint):

def test_constraint_term_and_ini_have_correct_defaults(self, local_study_with_constraint, test_constraint):
# Given
expected_ini_contents = """[test constraint]
expected_ini_contents = """[0]
name = test constraint
id = test constraint
enabled = true
Expand All @@ -2200,7 +2200,7 @@ def test_constraint_term_with_offset_and_ini_have_correct_values(
self, local_study_with_constraint, test_constraint
):
# Given
expected_ini_contents = """[test constraint]
expected_ini_contents = """[0]
name = test constraint
id = test constraint
enabled = true
Expand Down

0 comments on commit f9b161e

Please sign in to comment.