Skip to content

Commit

Permalink
refactor(bc): create user class (#84)
Browse files Browse the repository at this point in the history
  • Loading branch information
MartinBelthle authored Feb 14, 2025
1 parent 3864f14 commit 4ae8e2a
Show file tree
Hide file tree
Showing 14 changed files with 421 additions and 267 deletions.
12 changes: 12 additions & 0 deletions src/antares/craft/exceptions/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,12 @@ def __init__(self, constraint_name: str, message: str) -> None:
super().__init__(self.message)


class ConstraintDoesNotExistError(Exception):
def __init__(self, constraint_name: str) -> None:
self.message = f"The binding constraint {constraint_name} doesn't exist: "
super().__init__(self.message)


class ConstraintMatrixUpdateError(Exception):
def __init__(self, constraint_name: str, matrix_name: str, message: str) -> None:
self.message = f"Could not update matrix {matrix_name} for binding constraint {constraint_name}: " + message
Expand Down Expand Up @@ -238,6 +244,12 @@ def __init__(self, constraint_id: str, term_id: str, message: str) -> None:
super().__init__(self.message)


class ConstraintTermEditionError(Exception):
def __init__(self, constraint_id: str, term_id: str, message: str) -> None:
self.message = f"Could not update the term {term_id} of the binding constraint {constraint_id}: " + message
super().__init__(self.message)


class StudyCreationError(Exception):
def __init__(self, study_name: str, message: str) -> None:
self.message = f"Could not create the study {study_name}: " + message
Expand Down
121 changes: 58 additions & 63 deletions src/antares/craft/model/binding_constraint.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,17 +9,14 @@
# SPDX-License-Identifier: MPL-2.0
#
# 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.all_optional_meta import all_optional_model
from antares.craft.tools.contents_tool import EnumIgnoreCase, transform_name_to_id
from pydantic import BaseModel, Field, model_validator
from pydantic.alias_generators import to_camel


class BindingConstraintFrequency(EnumIgnoreCase):
Expand All @@ -41,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 @@ -63,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 @@ -72,40 +58,53 @@ 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()))


class DefaultBindingConstraintProperties(BaseModel, extra="forbid", populate_by_name=True, alias_generator=to_camel):
"""Default properties for binding constraints
Attributes:
enabled (bool): True
time_step (BindingConstraintFrequency): BindingConstraintFrequency.HOURLY
operator (BindingConstraintOperator): BindingConstraintOperator.LESS
comments (str): None
filter_year_by_year (str): "hourly"
filter_synthesis (str): "hourly"
group (str): "default"

"""
@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 ConstraintTermUpdate(ConstraintTermData):
weight: Optional[float] = None
offset: Optional[int] = None


@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 BindingConstraintPropertiesUpdate:
enabled: Optional[bool] = None
time_step: Optional[BindingConstraintFrequency] = None
operator: Optional[BindingConstraintOperator] = None
comments: Optional[str] = None
filter_year_by_year: Optional[str] = None
filter_synthesis: Optional[str] = None
group: Optional[str] = None


@dataclass
class BindingConstraintProperties:
enabled: bool = True
time_step: BindingConstraintFrequency = BindingConstraintFrequency.HOURLY
operator: BindingConstraintOperator = BindingConstraintOperator.LESS
Expand All @@ -115,11 +114,6 @@ class DefaultBindingConstraintProperties(BaseModel, extra="forbid", populate_by_
group: str = "default"


@all_optional_model
class BindingConstraintProperties(DefaultBindingConstraintProperties):
pass


class BindingConstraint:
def __init__(
self,
Expand All @@ -146,23 +140,24 @@ def id(self) -> str:
def properties(self) -> BindingConstraintProperties:
return self._properties

@properties.setter
def properties(self, new_properties: BindingConstraintProperties) -> None:
self._properties = new_properties

def get_terms(self) -> dict[str, ConstraintTerm]:
return self._terms

def add_terms(self, terms: list[ConstraintTerm]) -> None:
added_terms = self._binding_constraint_service.add_constraint_terms(self, terms)
for term in added_terms:
self._binding_constraint_service.add_constraint_terms(self, terms)
for term in terms:
self._terms[term.id] = term

def delete_term(self, term: ConstraintTerm) -> None:
self._binding_constraint_service.delete_binding_constraint_term(self.id, term.id)
self._terms.pop(term.id)

def update_properties(self, properties: BindingConstraintProperties) -> None:
def update_term(self, term: ConstraintTermUpdate) -> None:
existing_term = self._terms[term.id]
new_term = self._binding_constraint_service.update_binding_constraint_term(self.id, term, existing_term)
self._terms[term.id] = new_term

def update_properties(self, properties: BindingConstraintPropertiesUpdate) -> None:
new_properties = self._binding_constraint_service.update_binding_constraint_properties(self, properties)
self._properties = new_properties

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# Copyright (c) 2024, RTE (https://www.rte-france.com)
#
# See AUTHORS.txt
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
#
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.

from dataclasses import asdict
from typing import Union

from antares.craft.model.binding_constraint import (
BindingConstraintFrequency,
BindingConstraintOperator,
BindingConstraintProperties,
BindingConstraintPropertiesUpdate,
)
from antares.craft.service.api_services.models.base_model import APIBaseModel
from antares.craft.tools.all_optional_meta import all_optional_model

BindingConstraintPropertiesType = Union[BindingConstraintProperties, BindingConstraintPropertiesUpdate]


@all_optional_model
class BindingConstraintPropertiesAPI(APIBaseModel):
enabled: bool
time_step: BindingConstraintFrequency
operator: BindingConstraintOperator
comments: str
filter_year_by_year: str
filter_synthesis: str
group: str

@staticmethod
def from_user_model(user_class: BindingConstraintPropertiesType) -> "BindingConstraintPropertiesAPI":
user_dict = asdict(user_class)
return BindingConstraintPropertiesAPI.model_validate(user_dict)

def to_user_model(self) -> BindingConstraintProperties:
return BindingConstraintProperties(
enabled=self.enabled,
time_step=self.time_step,
operator=self.operator,
comments=self.comments,
filter_year_by_year=self.filter_year_by_year,
filter_synthesis=self.filter_synthesis,
group=self.group,
)
Loading

0 comments on commit 4ae8e2a

Please sign in to comment.