Skip to content

Commit

Permalink
terms
Browse files Browse the repository at this point in the history
  • Loading branch information
MartinBelthle committed Feb 13, 2025
1 parent e644caa commit 131b2b3
Show file tree
Hide file tree
Showing 3 changed files with 62 additions and 60 deletions.
70 changes: 36 additions & 34 deletions src/antares/craft/model/binding_constraint.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,12 @@
# This file is part of the Antares project.
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional, Union
from typing import Optional, Union

import pandas as pd

from antares.craft.service.base_services import BaseBindingConstraintService
from antares.craft.tools.contents_tool import EnumIgnoreCase, transform_name_to_id
from pydantic import BaseModel, Field, model_validator


class BindingConstraintFrequency(EnumIgnoreCase):
Expand All @@ -39,20 +38,8 @@ class ConstraintMatrixName(Enum):
GREATER_TERM = "gt"


class TermOperators(BaseModel):
weight: Optional[float] = None
offset: Optional[int] = None

def weight_offset(self) -> str:
if self.offset is not None:
# Rounded the weight to 6 decimals to be in line with other floats in the ini files
weight_offset = f"{(self.weight if self.weight is not None else 0):.6f}%{self.offset}"
else:
weight_offset = f"{self.weight if self.weight is not None else 0}"
return weight_offset


class LinkData(BaseModel):
@dataclass
class LinkData:
"""
DTO for a constraint term on a link between two areas.
"""
Expand All @@ -61,7 +48,8 @@ class LinkData(BaseModel):
area2: str


class ClusterData(BaseModel):
@dataclass
class ClusterData:
"""
DTO for a constraint term on a cluster in an area.
"""
Expand All @@ -70,24 +58,38 @@ class ClusterData(BaseModel):
cluster: str


class ConstraintTerm(TermOperators):
@dataclass
class ConstraintTermData:
data: Union[LinkData, ClusterData]
id: str = Field(init=False)

@model_validator(mode="before")
def fill_id(cls, v: dict[str, Any]) -> dict[str, Any]:
v["id"] = cls.generate_id(v["data"])
return v

@classmethod
def generate_id(cls, data: Union[dict[str, str], LinkData, ClusterData]) -> str:
if isinstance(data, dict):
if "area1" in data:
return "%".join(sorted((data["area1"].lower(), data["area2"].lower())))
return ".".join((data["area"].lower(), data["cluster"].lower()))
elif isinstance(data, LinkData):
return "%".join(sorted((data.area1.lower(), data.area2.lower())))
return ".".join((data.area.lower(), data.cluster.lower()))

@property
def id(self) -> str:
if isinstance(self.data, LinkData):
return "%".join(sorted((self.data.area1.lower(), self.data.area2.lower())))
return ".".join((self.data.area.lower(), self.data.cluster.lower()))

@staticmethod
def from_dict(input: dict[str, str]) -> Union[LinkData, ClusterData]:
if "area1" in input:
return LinkData(area1=input["area1"], area2=input["area2"])
elif "cluster" in input:
return ClusterData(area=input["area"], cluster=input["cluster"])
raise ValueError(f"Dict {input} couldn't be serialized as a ConstraintTermData object")


@dataclass
class ConstraintTerm(ConstraintTermData):
weight: float = 1
offset: int = 0

def weight_offset(self) -> str:
return f"{self.weight}%{self.offset}" if self.offset != 0 else f"{self.weight}"


@dataclass
class ConstraintTermUpdate(ConstraintTermData):
weight: Optional[float] = None
offset: Optional[int] = None


@dataclass
Expand Down
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 dataclasses import asdict
from pathlib import PurePosixPath
from typing import Optional

Expand All @@ -33,6 +33,7 @@
BindingConstraintPropertiesUpdate,
ConstraintMatrixName,
ConstraintTerm,
ConstraintTermData,
)
from antares.craft.service.api_services.models.binding_constraint import BindingConstraintPropertiesAPI
from antares.craft.service.api_services.utils import get_matrix
Expand Down Expand Up @@ -95,22 +96,18 @@ def create_binding_constraint(
del created_properties[key]
api_properties = BindingConstraintPropertiesAPI.model_validate(created_properties)
bc_properties = api_properties.to_user_model()
bc_terms: list[ConstraintTerm] = []

if terms:
json_terms = [term.model_dump() for term in terms]
json_terms = [
{"weight": term.weight, "offset": term.offset, "data": asdict(term.data)} for term in terms
]
url = f"{base_url}/{bc_id}/terms"
self._wrapper.post(url, json=json_terms)

url = f"{base_url}/{bc_id}"
response = self._wrapper.get(url)
created_terms = response.json()["terms"]
bc_terms = [ConstraintTerm.model_validate(term) for term in created_terms]

except APIError as e:
raise BindingConstraintCreationError(name, e.message) from e

constraint = BindingConstraint(name, self, bc_properties, bc_terms)
constraint = BindingConstraint(name, self, bc_properties, terms)

return constraint

Expand Down Expand Up @@ -173,7 +170,7 @@ def update_constraint_matrix(
def add_constraint_terms(self, constraint: BindingConstraint, terms: list[ConstraintTerm]) -> None:
url = f"{self._base_url}/studies/{self.study_id}/bindingconstraints/{constraint.id}"
try:
json_terms = [term.model_dump() for term in terms]
json_terms = [{"weight": term.weight, "offset": term.offset, "data": asdict(term.data)} for term in terms]
self._wrapper.post(f"{url}/terms", json=json_terms)

except APIError as e:
Expand All @@ -185,18 +182,20 @@ def read_binding_constraints(self) -> list[BindingConstraint]:
try:
response = self._wrapper.get(url)
constraints_json = response.json()
constraints = []

for constraint in constraints_json:
constraint_name = constraint.pop("name")
del constraint["id"]
api_terms = constraint.pop("terms")
api_properties = BindingConstraintPropertiesAPI.model_validate(constraint)
bc_properties = api_properties.to_user_model()
terms: list[ConstraintTerm] = []
for api_term in api_terms:
term_data = ConstraintTermData.from_dict(api_term["data"])
terms.append(ConstraintTerm(weight=api_term["weight"], offset=api_term["offset"], data=term_data))
constraints.append(BindingConstraint(constraint_name, self, bc_properties, terms))

constraints = [
BindingConstraint(
constraint["name"],
self,
BindingConstraintPropertiesAPI.model_validate(
{k: v for k, v in constraint.items() if k not in ["terms", "id", "name"]}
).to_user_model(),
[ConstraintTerm.model_validate(term) for term in constraint["terms"]],
)
for constraint in constraints_json
]
constraints.sort(key=lambda constraint: constraint.id)
return constraints
except APIError as e:
Expand Down
11 changes: 6 additions & 5 deletions tests/antares/services/local_services/test_study.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
BindingConstraintOperator,
BindingConstraintProperties,
ConstraintTerm,
LinkData,
)
from antares.craft.model.commons import FilterOption
from antares.craft.model.hydro import Hydro
Expand Down Expand Up @@ -1306,7 +1307,7 @@ def test_constraints_and_ini_have_custom_properties(self, local_study_with_const
assert actual_ini_content == expected_ini_content

def test_constraint_can_add_term(self, test_constraint):
new_term = [ConstraintTerm(data={"area1": "fr", "area2": "at"})]
new_term = [ConstraintTerm(data=LinkData(area1="fr", area2="at"))]
test_constraint.add_terms(new_term)
assert test_constraint.get_terms()

Expand All @@ -1322,11 +1323,11 @@ def test_constraint_term_and_ini_have_correct_defaults(self, local_study_with_co
filter-year-by-year = hourly
filter-synthesis = hourly
group = default
at%fr = 0
at%fr = 1
"""
# When
new_term = [ConstraintTerm(data={"area1": "fr", "area2": "at"})]
new_term = [ConstraintTerm(data=LinkData(area1="fr", area2="at"))]
test_constraint.add_terms(new_term)
with local_study_with_constraint._binding_constraints_service.ini_file.ini_path.open("r") as file:
actual_ini_content = file.read()
Expand All @@ -1347,11 +1348,11 @@ def test_constraint_term_with_offset_and_ini_have_correct_values(
filter-year-by-year = hourly
filter-synthesis = hourly
group = default
at%fr = 0.000000%1
at%fr = 1%1
"""
# When
new_term = [ConstraintTerm(offset=1, data={"area1": "fr", "area2": "at"})]
new_term = [ConstraintTerm(offset=1, data=LinkData(area1="fr", area2="at"))]
test_constraint.add_terms(new_term)
with local_study_with_constraint._binding_constraints_service.ini_file.ini_path.open("r") as file:
actual_ini_content = file.read()
Expand Down

0 comments on commit 131b2b3

Please sign in to comment.