-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(commands): add ST-Storage commands (#1630)
1 parent
c1145de
commit e88d4ee
Showing
24 changed files
with
2,327 additions
and
113 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
115 changes: 115 additions & 0 deletions
115
antarest/study/storage/rawstudy/model/filesystem/config/st_storage.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| from typing import Dict, Any | ||
|
|
||
| from pydantic import BaseModel, Extra, Field, root_validator | ||
|
|
||
| from antarest.study.business.enum_ignore_case import EnumIgnoreCase | ||
|
|
||
|
|
||
| class STStorageGroup(EnumIgnoreCase): | ||
| """ | ||
| This class defines the specific energy storage systems. | ||
| Enum values: | ||
| - PSP_OPEN: Represents an open pumped storage plant. | ||
| - PSP_CLOSED: Represents a closed pumped storage plant. | ||
| - PONDAGE: Represents a pondage storage system (reservoir storage system). | ||
| - BATTERY: Represents a battery storage system. | ||
| - OTHER: Represents other energy storage systems. | ||
| """ | ||
|
|
||
| PSP_OPEN = "PSP_open" | ||
| PSP_CLOSED = "PSP_closed" | ||
| PONDAGE = "Pondage" | ||
| BATTERY = "Battery" | ||
| OTHER = "Other" | ||
|
|
||
|
|
||
| # noinspection SpellCheckingInspection | ||
| class STStorageConfig(BaseModel): | ||
| """ | ||
| Manage the configuration files in the context of Short-Term Storage. | ||
| It provides a convenient way to read and write configuration data from/to an INI file format. | ||
| """ | ||
|
|
||
| class Config: | ||
| extra = Extra.forbid | ||
| allow_population_by_field_name = True | ||
|
|
||
| # The `id` field is a calculated field that is excluded when converting | ||
| # the model to a dictionary or JSON format (`model_dump`). | ||
| id: str = Field( | ||
| description="Short-term storage ID", | ||
| regex=r"[a-zA-Z0-9_(),& -]+", | ||
| exclude=True, | ||
| ) | ||
| name: str = Field( | ||
| description="Short-term storage name", | ||
| regex=r"[a-zA-Z0-9_(),& -]+", | ||
| ) | ||
| group: STStorageGroup = Field( | ||
| ..., | ||
| description="Energy storage system group (mandatory)", | ||
| ) | ||
| injection_nominal_capacity: float = Field( | ||
| 0, | ||
| description="Injection nominal capacity (MW)", | ||
| ge=0, | ||
| alias="injectionnominalcapacity", | ||
| ) | ||
| withdrawal_nominal_capacity: float = Field( | ||
| 0, | ||
| description="Withdrawal nominal capacity (MW)", | ||
| ge=0, | ||
| alias="withdrawalnominalcapacity", | ||
| ) | ||
| reservoir_capacity: float = Field( | ||
| 0, | ||
| description="Reservoir capacity (MWh)", | ||
| ge=0, | ||
| alias="reservoircapacity", | ||
| ) | ||
| efficiency: float = Field( | ||
| 1, | ||
| description="Efficiency of the storage system", | ||
| ge=0, | ||
| le=1, | ||
| ) | ||
| initial_level: float = Field( | ||
| 0, | ||
| description="Initial level of the storage system", | ||
| ge=0, | ||
| alias="initiallevel", | ||
| ) | ||
| initial_level_optim: bool = Field( | ||
| False, | ||
| description="Flag indicating if the initial level is optimized", | ||
| alias="initialleveloptim", | ||
| ) | ||
|
|
||
| @root_validator(pre=True) | ||
| def calculate_storage_id(cls, values: Dict[str, Any]) -> Dict[str, Any]: | ||
| """ | ||
| Calculate the short-term storage ID based on the storage name, if not provided. | ||
| Args: | ||
| values: values used to construct the object. | ||
| Returns: | ||
| The updated values. | ||
| """ | ||
| # Avoid circular imports | ||
| from antarest.study.storage.rawstudy.model.filesystem.config.model import ( | ||
| transform_name_to_id, | ||
| ) | ||
|
|
||
| if values.get("id") or not values.get("name"): | ||
| return values | ||
| storage_name = values["name"] | ||
| if storage_id := transform_name_to_id(storage_name): | ||
| values["id"] = storage_id | ||
| else: | ||
| raise ValueError( | ||
| f"Invalid short term storage name '{storage_name}'." | ||
| ) | ||
| return values |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
antarest/study/storage/variantstudy/business/matrix_constants/__init__.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1 +1 @@ | ||
| from . import hydro, prepro, thermals, link | ||
| from . import hydro, prepro, thermals, link, st_storage |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| from . import series |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
381 changes: 381 additions & 0 deletions
381
antarest/study/storage/variantstudy/model/command/create_st_storage.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,381 @@ | ||
| import json | ||
| from typing import Any, Dict, List, Optional, Tuple, Union, cast | ||
|
|
||
| import numpy as np | ||
| from antarest.core.model import JSON | ||
| from antarest.matrixstore.model import MatrixData | ||
| from antarest.study.storage.rawstudy.model.filesystem.config.model import ( | ||
| Area, | ||
| FileStudyTreeConfig, | ||
| ) | ||
| from antarest.study.storage.rawstudy.model.filesystem.config.st_storage import ( | ||
| STStorageConfig, | ||
| ) | ||
| from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy | ||
| from antarest.study.storage.variantstudy.business.matrix_constants_generator import ( | ||
| GeneratorMatrixConstants, | ||
| ) | ||
| from antarest.study.storage.variantstudy.business.utils import ( | ||
| strip_matrix_protocol, | ||
| validate_matrix, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.common import ( | ||
| CommandName, | ||
| CommandOutput, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.icommand import ( | ||
| MATCH_SIGNATURE_SEPARATOR, | ||
| ICommand, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.model import CommandDTO | ||
| from pydantic import Field, validator, Extra | ||
| from pydantic.fields import ModelField | ||
|
|
||
| # noinspection SpellCheckingInspection | ||
| _MATRIX_NAMES = ( | ||
| "pmax_injection", | ||
| "pmax_withdrawal", | ||
| "lower_rule_curve", | ||
| "upper_rule_curve", | ||
| "inflows", | ||
| ) | ||
|
|
||
| # Minimum required version. | ||
| REQUIRED_VERSION = 860 | ||
|
|
||
| MatrixType = List[List[MatrixData]] | ||
|
|
||
|
|
||
| # noinspection SpellCheckingInspection | ||
| class CreateSTStorage(ICommand): | ||
| """ | ||
| Command used to create a short-terme storage in an area. | ||
| """ | ||
|
|
||
| class Config: | ||
| extra = Extra.forbid | ||
|
|
||
| # Overloaded parameters | ||
| # ===================== | ||
|
|
||
| command_name = CommandName.CREATE_ST_STORAGE | ||
| version = 1 | ||
|
|
||
| # Command parameters | ||
| # ================== | ||
|
|
||
| area_id: str = Field(description="Area ID", regex=r"[a-z0-9_(),& -]+") | ||
| parameters: STStorageConfig | ||
| pmax_injection: Optional[Union[MatrixType, str]] = Field( | ||
| None, | ||
| description="Charge capacity (modulation)", | ||
| ) | ||
| pmax_withdrawal: Optional[Union[MatrixType, str]] = Field( | ||
| None, | ||
| description="Discharge capacity (modulation)", | ||
| ) | ||
| lower_rule_curve: Optional[Union[MatrixType, str]] = Field( | ||
| None, | ||
| description="Lower rule curve (coefficient)", | ||
| ) | ||
| upper_rule_curve: Optional[Union[MatrixType, str]] = Field( | ||
| None, | ||
| description="Upper rule curve (coefficient)", | ||
| ) | ||
| inflows: Optional[Union[MatrixType, str]] = Field( | ||
| None, | ||
| description="Inflows (MW)", | ||
| ) | ||
|
|
||
| @property | ||
| def storage_id(self) -> str: | ||
| """The normalized version of the storage's name used as the ID.""" | ||
| return self.parameters.id | ||
|
|
||
| @property | ||
| def storage_name(self) -> str: | ||
| """The label representing the name of the storage for the user.""" | ||
| return self.parameters.name | ||
|
|
||
| @validator(*_MATRIX_NAMES, always=True) | ||
| def register_matrix( | ||
| cls, | ||
| v: Optional[Union[MatrixType, str]], | ||
| values: Dict[str, Any], | ||
| field: ModelField, | ||
| ) -> Optional[Union[MatrixType, str]]: | ||
| """ | ||
| Validates a matrix array or link, and store the matrix array in the matrix repository. | ||
| This method is used to validate the matrix array or link provided as input. | ||
| - If the input is `None`, it retrieves a default matrix from the | ||
| generator matrix constants. | ||
| - If the input is a string, it validates the matrix link. | ||
| - If the input is a list of lists, it validates the matrix values | ||
| and creates the corresponding matrix link. | ||
| Args: | ||
| v: The matrix array or link to be validated and registered. | ||
| values: A dictionary containing additional values used for validation. | ||
| field: The field being validated. | ||
| Returns: | ||
| The ID of the validated and stored matrix prefixed by "matrix://". | ||
| Raises: | ||
| ValueError: If the matrix has an invalid shape, contains NaN values, | ||
| or violates specific constraints. | ||
| TypeError: If the input datatype is not supported. | ||
| """ | ||
| if v is None: | ||
| # use an already-registered default matrix | ||
| constants: GeneratorMatrixConstants | ||
| constants = values["command_context"].generator_matrix_constants | ||
| # Directly access the methods instead of using `getattr` for maintainability | ||
| methods = { | ||
| "pmax_injection": constants.get_st_storage_pmax_injection, | ||
| "pmax_withdrawal": constants.get_st_storage_pmax_withdrawal, | ||
| "lower_rule_curve": constants.get_st_storage_lower_rule_curve, | ||
| "upper_rule_curve": constants.get_st_storage_upper_rule_curve, | ||
| "inflows": constants.get_st_storage_inflows, | ||
| } | ||
| method = methods[field.name] | ||
| return method() | ||
| if isinstance(v, str): | ||
| # Check the matrix link | ||
| return validate_matrix(v, values) | ||
| if isinstance(v, list): | ||
| # Check the matrix values and create the corresponding matrix link | ||
| array = np.array(v, dtype=np.float64) | ||
| if array.shape != (8760, 1): | ||
| raise ValueError( | ||
| f"Invalid matrix shape {array.shape}, expected (8760, 1)" | ||
| ) | ||
| if np.isnan(array).any(): | ||
| raise ValueError("Matrix values cannot contain NaN") | ||
| # All matrices except "inflows" are constrained between 0 and 1 | ||
| constrained = set(_MATRIX_NAMES) - {"inflows"} | ||
| if field.name in constrained and ( | ||
| np.any(array < 0) or np.any(array > 1) | ||
| ): | ||
| raise ValueError("Matrix values should be between 0 and 1") | ||
| v = cast(MatrixType, array.tolist()) | ||
| return validate_matrix(v, values) | ||
| # Invalid datatype | ||
| # pragma: no cover | ||
| raise TypeError(repr(v)) | ||
|
|
||
| def _apply_config( | ||
| self, study_data: FileStudyTreeConfig | ||
| ) -> Tuple[CommandOutput, Dict[str, Any]]: | ||
| """ | ||
| Applies configuration changes to the study data: add the short-term storage in the storages list. | ||
| Args: | ||
| study_data: The study data configuration. | ||
| Returns: | ||
| A tuple containing the command output and a dictionary of extra data. | ||
| On success, the dictionary of extra data is `{"storage_id": storage_id}`. | ||
| """ | ||
|
|
||
| # Check if the study version is above the minimum required version. | ||
| version = study_data.version | ||
| if version < REQUIRED_VERSION: | ||
| return ( | ||
| CommandOutput( | ||
| status=False, | ||
| message=( | ||
| f"Invalid study version {version}," | ||
| f" at least version {REQUIRED_VERSION} is required." | ||
| ), | ||
| ), | ||
| {}, | ||
| ) | ||
|
|
||
| # Search the Area in the configuration | ||
| if self.area_id not in study_data.areas: | ||
| return ( | ||
| CommandOutput( | ||
| status=False, | ||
| message=f"Area '{self.area_id}' does not exist in the study configuration.", | ||
| ), | ||
| {}, | ||
| ) | ||
| area: Area = study_data.areas[self.area_id] | ||
|
|
||
| # Check if the short-term storage already exists in the area | ||
| if any(s.id == self.storage_id for s in area.st_storages): | ||
| return ( | ||
| CommandOutput( | ||
| status=False, | ||
| message=( | ||
| f"Short-term storage '{self.storage_name}' already exists" | ||
| f" in the area '{self.area_id}'." | ||
| ), | ||
| ), | ||
| {}, | ||
| ) | ||
|
|
||
| # Create a new short-term storage and add it to the area | ||
| area.st_storages.append(self.parameters) | ||
|
|
||
| return ( | ||
| CommandOutput( | ||
| status=True, | ||
| message=( | ||
| f"Short-term st_storage '{self.storage_name}' successfully added" | ||
| f" to area '{self.area_id}'." | ||
| ), | ||
| ), | ||
| {"storage_id": self.storage_id}, | ||
| ) | ||
|
|
||
| def _apply(self, study_data: FileStudy) -> CommandOutput: | ||
| """ | ||
| Applies the study data to update storage configurations and saves the changes. | ||
| Saves the changes made to the storage configurations. | ||
| Args: | ||
| study_data: The study data to be applied. | ||
| Returns: | ||
| The output of the command execution. | ||
| """ | ||
| output, data = self._apply_config(study_data.config) | ||
| if not output.status: | ||
| return output | ||
|
|
||
| # Fill-in the "list.ini" file with the parameters | ||
| config = study_data.tree.get( | ||
| ["input", "st-storage", "clusters", self.area_id, "list"] | ||
| ) | ||
| config[self.storage_id] = json.loads( | ||
| self.parameters.json(by_alias=True) | ||
| ) | ||
|
|
||
| new_data: JSON = { | ||
| "input": { | ||
| "st-storage": { | ||
| "clusters": {self.area_id: {"list": config}}, | ||
| "series": { | ||
| self.area_id: { | ||
| self.storage_id: { | ||
| attr: getattr(self, attr) | ||
| for attr in _MATRIX_NAMES | ||
| } | ||
| } | ||
| }, | ||
| } | ||
| } | ||
| } | ||
| study_data.tree.save(new_data) | ||
|
|
||
| return output | ||
|
|
||
| def to_dto(self) -> CommandDTO: | ||
| """ | ||
| Converts the current object to a Data Transfer Object (DTO) | ||
| which is stored in the `CommandBlock` in the database. | ||
| Returns: | ||
| The DTO object representing the current command. | ||
| """ | ||
| parameters = json.loads(self.parameters.json(by_alias=True)) | ||
| return CommandDTO( | ||
| action=self.command_name.value, | ||
| args={ | ||
| "area_id": self.area_id, | ||
| "parameters": parameters, | ||
| **{ | ||
| attr: strip_matrix_protocol(getattr(self, attr)) | ||
| for attr in _MATRIX_NAMES | ||
| }, | ||
| }, | ||
| ) | ||
|
|
||
| def match_signature(self) -> str: | ||
| """Returns the command signature.""" | ||
| return str( | ||
| self.command_name.value | ||
| + MATCH_SIGNATURE_SEPARATOR | ||
| + self.area_id | ||
| + MATCH_SIGNATURE_SEPARATOR | ||
| + self.storage_id | ||
| ) | ||
|
|
||
| def match(self, other: "ICommand", equal: bool = False) -> bool: | ||
| """ | ||
| Checks if the current instance matches another `ICommand` object. | ||
| Args: | ||
| other: Another `ICommand` object to compare against. | ||
| equal: Flag indicating whether to perform a deep comparison. | ||
| Returns: | ||
| bool: `True` if the current instance matches the other object, `False` otherwise. | ||
| """ | ||
| if not isinstance(other, CreateSTStorage): | ||
| return False | ||
| if equal: | ||
| # Deep comparison | ||
| return self.__eq__(other) | ||
| else: | ||
| return ( | ||
| self.area_id == other.area_id | ||
| and self.storage_id == other.storage_id | ||
| ) | ||
|
|
||
| def _create_diff(self, other: "ICommand") -> List["ICommand"]: | ||
| """ | ||
| Creates a list of commands representing the differences between | ||
| the current instance and another `ICommand` object. | ||
| Args: | ||
| other: Another ICommand object to compare against. | ||
| Returns: | ||
| A list of commands representing the differences between | ||
| the two `ICommand` objects. | ||
| """ | ||
| from antarest.study.storage.variantstudy.model.command.replace_matrix import ( | ||
| ReplaceMatrix, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.update_config import ( | ||
| UpdateConfig, | ||
| ) | ||
|
|
||
| other = cast(CreateSTStorage, other) | ||
| commands: List[ICommand] = [ | ||
| ReplaceMatrix( | ||
| target=f"input/st-storage/series/{self.area_id}/{self.storage_id}/{attr}", | ||
| matrix=strip_matrix_protocol(getattr(other, attr)), | ||
| command_context=self.command_context, | ||
| ) | ||
| for attr in _MATRIX_NAMES | ||
| if getattr(self, attr) != getattr(other, attr) | ||
| ] | ||
| if self.parameters != other.parameters: | ||
| data: Dict[str, Any] = json.loads( | ||
| other.parameters.json(by_alias=True) | ||
| ) | ||
| commands.append( | ||
| UpdateConfig( | ||
| target=f"input/st-storage/clusters/{self.area_id}/list/{self.storage_id}", | ||
| data=data, | ||
| command_context=self.command_context, | ||
| ) | ||
| ) | ||
| return commands | ||
|
|
||
| def get_inner_matrices(self) -> List[str]: | ||
| """ | ||
| Retrieves the list of matrix IDs. | ||
| """ | ||
| matrices: List[str] = [ | ||
| strip_matrix_protocol(getattr(self, attr)) | ||
| for attr in _MATRIX_NAMES | ||
| ] | ||
| return matrices |
169 changes: 169 additions & 0 deletions
169
antarest/study/storage/variantstudy/model/command/remove_st_storage.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,169 @@ | ||
| from typing import Any, Dict, Tuple, List | ||
|
|
||
| from antarest.study.storage.rawstudy.model.filesystem.config.model import ( | ||
| Area, | ||
| FileStudyTreeConfig, | ||
| ) | ||
| from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy | ||
| from antarest.study.storage.variantstudy.model.command.common import ( | ||
| CommandName, | ||
| CommandOutput, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.icommand import ( | ||
| ICommand, | ||
| MATCH_SIGNATURE_SEPARATOR, | ||
| ) | ||
| from pydantic import Field | ||
|
|
||
| from antarest.study.storage.variantstudy.model.model import CommandDTO | ||
|
|
||
| # minimum required version. | ||
| REQUIRED_VERSION = 860 | ||
|
|
||
|
|
||
| class RemoveSTStorage(ICommand): | ||
| """ | ||
| Command used to remove a short-terme storage from an area. | ||
| """ | ||
|
|
||
| area_id: str = Field(description="Area ID", regex=r"[a-z0-9_(),& -]+") | ||
| storage_id: str = Field( | ||
| description="Short term storage ID", | ||
| regex=r"[a-z0-9_(),& -]+", | ||
| ) | ||
|
|
||
| def __init__(self, **data: Any) -> None: | ||
| super().__init__( | ||
| command_name=CommandName.REMOVE_ST_STORAGE, version=1, **data | ||
| ) | ||
|
|
||
| def _apply_config( | ||
| self, study_data: FileStudyTreeConfig | ||
| ) -> Tuple[CommandOutput, Dict[str, Any]]: | ||
| """ | ||
| Applies configuration changes to the study data: remove the storage from the storages list. | ||
| Args: | ||
| study_data: The study data configuration. | ||
| Returns: | ||
| A tuple containing the command output and a dictionary of extra data. | ||
| On success, the dictionary is empty. | ||
| """ | ||
| # Check if the study version is above the minimum required version. | ||
| version = study_data.version | ||
| if version < REQUIRED_VERSION: | ||
| return ( | ||
| CommandOutput( | ||
| status=False, | ||
| message=( | ||
| f"Invalid study version {version}," | ||
| f" at least version {REQUIRED_VERSION} is required." | ||
| ), | ||
| ), | ||
| {}, | ||
| ) | ||
|
|
||
| # Search the Area in the configuration | ||
| if self.area_id not in study_data.areas: | ||
| return ( | ||
| CommandOutput( | ||
| status=False, | ||
| message=( | ||
| f"Area '{self.area_id}' does not exist" | ||
| f" in the study configuration." | ||
| ), | ||
| ), | ||
| {}, | ||
| ) | ||
| area: Area = study_data.areas[self.area_id] | ||
|
|
||
| # Search the Short term storage in the area | ||
| for st_storage in area.st_storages: | ||
| if st_storage.id == self.storage_id: | ||
| break | ||
| else: | ||
| return ( | ||
| CommandOutput( | ||
| status=False, | ||
| message=( | ||
| f"Short term storage '{self.storage_id}' does not exist" | ||
| f" in the area '{self.area_id}'." | ||
| ), | ||
| ), | ||
| {}, | ||
| ) | ||
|
|
||
| # Remove the Short term storage from the configuration | ||
| area.st_storages.remove(st_storage) | ||
|
|
||
| return ( | ||
| CommandOutput( | ||
| status=True, | ||
| message=( | ||
| f"Short term storage '{self.storage_id}' removed" | ||
| f" from the area '{self.area_id}'." | ||
| ), | ||
| ), | ||
| {}, | ||
| ) | ||
|
|
||
| def _apply(self, study_data: FileStudy) -> CommandOutput: | ||
| """ | ||
| Applies the study data to update storage configurations and saves the changes: | ||
| remove the storage from the configuration and remove the attached time series. | ||
| Args: | ||
| study_data: The study data to be applied. | ||
| Returns: | ||
| The output of the command execution. | ||
| """ | ||
| # It is required to delete the files and folders that correspond to the short-term storage | ||
| # BEFORE updating the configuration, as we need the configuration to do so. | ||
| # Specifically, deleting the time series uses the list of short-term storages from the configuration. | ||
| # fmt: off | ||
| paths = [ | ||
| ["input", "st-storage", "clusters", self.area_id, "list", self.storage_id], | ||
| ["input", "st-storage", "series", self.area_id, self.storage_id], | ||
| ] | ||
| # fmt: on | ||
| for path in paths: | ||
| study_data.tree.delete(path) | ||
| # Deleting the short-term storage in the configuration must be done AFTER | ||
| # deleting the files and folders. | ||
| return self._apply_config(study_data.config)[0] | ||
|
|
||
| def to_dto(self) -> CommandDTO: | ||
| """ | ||
| Converts the current object to a Data Transfer Object (DTO) | ||
| which is stored in the `CommandBlock` in the database. | ||
| Returns: | ||
| The DTO object representing the current command. | ||
| """ | ||
| return CommandDTO( | ||
| action=self.command_name.value, | ||
| args={"area_id": self.area_id, "storage_id": self.storage_id}, | ||
| ) | ||
|
|
||
| def match_signature(self) -> str: | ||
| """Returns the command signature.""" | ||
| return str( | ||
| self.command_name.value | ||
| + MATCH_SIGNATURE_SEPARATOR | ||
| + self.area_id | ||
| + MATCH_SIGNATURE_SEPARATOR | ||
| + self.storage_id | ||
| ) | ||
|
|
||
| def match(self, other: "ICommand", equal: bool = False) -> bool: | ||
| # always perform a deep comparison, as there are no parameters | ||
| # or matrices, so that shallow and deep comparisons are identical. | ||
| return self.__eq__(other) | ||
|
|
||
| def _create_diff(self, other: "ICommand") -> List["ICommand"]: | ||
| return [] | ||
|
|
||
| def get_inner_matrices(self) -> List[str]: | ||
| return [] |
Large diffs are not rendered by default.
Oops, something went wrong.
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,240 @@ | ||
| import http | ||
| from unittest.mock import ANY | ||
|
|
||
| import numpy as np | ||
| import pytest | ||
| from starlette.testclient import TestClient | ||
|
|
||
| from antarest.core.tasks.model import TaskStatus | ||
| from antarest.study.storage.rawstudy.model.filesystem.config.model import ( | ||
| transform_name_to_id, | ||
| ) | ||
| from tests.integration.utils import wait_task_completion | ||
|
|
||
|
|
||
| @pytest.mark.integration_test | ||
| class TestSTStorage: | ||
| """ | ||
| This unit test is designed to demonstrate the creation, modification of properties and | ||
| updating of matrices, and the deletion of one or more short-term storages. | ||
| """ | ||
|
|
||
| # noinspection SpellCheckingInspection | ||
| def test_lifecycle( | ||
| self, | ||
| client: TestClient, | ||
| user_access_token: str, | ||
| study_id: str, | ||
| ): | ||
| # ======================= | ||
| # Study version upgrade | ||
| # ======================= | ||
|
|
||
| # We have an "old" study that we need to upgrade to version 860 | ||
| min_study_version = 860 | ||
| res = client.put( | ||
| f"/v1/studies/{study_id}/upgrade", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| params={"target_version": min_study_version}, | ||
| ) | ||
| res.raise_for_status() | ||
| task_id = res.json() | ||
| task = wait_task_completion(client, user_access_token, task_id) | ||
| assert task.status == TaskStatus.COMPLETED, task | ||
|
|
||
| # We can check that the study is upgraded to the required version | ||
| res = client.get( | ||
| f"/v1/studies/{study_id}", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| ) | ||
| res.raise_for_status() | ||
| assert res.json() == { | ||
| "id": study_id, | ||
| "name": "STA-mini", | ||
| "version": min_study_version, | ||
| "created": ANY, # ISO8601 Date/time | ||
| "updated": ANY, # ISO8601 Date/time | ||
| "type": "rawstudy", | ||
| "owner": {"id": None, "name": ANY}, | ||
| "groups": [], | ||
| "public_mode": "FULL", | ||
| "workspace": "ext", | ||
| "managed": False, | ||
| "archived": False, | ||
| "horizon": "2030", | ||
| "scenario": None, | ||
| "status": None, | ||
| "doc": None, | ||
| "folder": "STA-mini", | ||
| "tags": [], | ||
| } | ||
|
|
||
| # Here is the list of available areas | ||
| res = client.get( | ||
| f"/v1/studies/{study_id}/areas", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| ) | ||
| res.raise_for_status() | ||
| areas = res.json() | ||
| area_ids = {a["id"] for a in areas if a["type"] == "AREA"} | ||
| assert area_ids == {"es", "it", "de", "fr"} | ||
|
|
||
| # ============================================= | ||
| # Short-Term Storage Creation w/o Time Series | ||
| # ============================================= | ||
|
|
||
| # First, we will define a short-term storage in the geographical | ||
| # area "FR" called "Siemens Battery" with the bellow arguments. | ||
| # We will use the default values for the time series: | ||
| # - `pmax_injection`: Charge capacity, | ||
| # - `pmax_withdrawal`: Discharge capacity, | ||
| # - `lower_rule_curve`: Lower rule curve, | ||
| # - `upper_rule_curve`: Upper rule curve, | ||
| # - `inflows`: Inflows | ||
| area_id = transform_name_to_id("FR") | ||
| siemens_battery = "Siemens Battery" | ||
| args = { | ||
| "area_id": area_id, | ||
| "parameters": { | ||
| "name": siemens_battery, | ||
| "group": "Battery", | ||
| "injection_nominal_capacity": 150, | ||
| "withdrawal_nominal_capacity": 150, | ||
| "reservoir_capacity": 600, | ||
| "efficiency": 0.94, | ||
| "initial_level_optim": True, | ||
| }, | ||
| } | ||
| res = client.post( | ||
| f"/v1/studies/{study_id}/commands", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| json=[{"action": "create_st_storage", "args": args}], | ||
| ) | ||
| res.raise_for_status() | ||
|
|
||
| # ======================================= | ||
| # Short-Term Storage Time Series Update | ||
| # ======================================= | ||
|
|
||
| # Then, it is possible to update a time series. | ||
| # For instance, we want to initialize the `inflows` time series | ||
| # with random values (for this demo). | ||
| # To do that, we can use the `replace_matrix` command like bellow: | ||
| siemens_battery_id = transform_name_to_id(siemens_battery) | ||
| inflows = np.random.randint(0, 1001, size=(8760, 1)) | ||
| args1 = { | ||
| "target": f"input/st-storage/series/{area_id}/{siemens_battery_id}/inflows", | ||
| "matrix": inflows.tolist(), | ||
| } | ||
| pmax_injection = np.random.rand(8760, 1) | ||
| args2 = { | ||
| "target": f"input/st-storage/series/{area_id}/{siemens_battery_id}/pmax_injection", | ||
| "matrix": pmax_injection.tolist(), | ||
| } | ||
| res = client.post( | ||
| f"/v1/studies/{study_id}/commands", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| json=[ | ||
| {"action": "replace_matrix", "args": args1}, | ||
| {"action": "replace_matrix", "args": args2}, | ||
| ], | ||
| ) | ||
| res.raise_for_status() | ||
|
|
||
| # ============================================== | ||
| # Short-Term Storage Creation with Time Series | ||
| # ============================================== | ||
|
|
||
| # Another way to create a Short-Term Storage is by providing | ||
| # both the parameters and the time series arrays. | ||
| # Here is an example where we populate some arrays with random values. | ||
| pmax_injection = np.random.rand(8760, 1) | ||
| pmax_withdrawal = np.random.rand(8760, 1) | ||
| inflows = np.random.randint(0, 1001, size=(8760, 1)) | ||
| grand_maison = "Grand'Maison" | ||
| args = { | ||
| "area_id": area_id, | ||
| "parameters": { | ||
| "name": grand_maison, | ||
| "group": "PSP_closed", | ||
| "injectionnominalcapacity": 1500, | ||
| "withdrawalnominalcapacity": 1800, | ||
| "reservoircapacity": 20000, | ||
| "efficiency": 0.78, | ||
| "initiallevel": 10000, | ||
| }, | ||
| "pmax_injection": pmax_injection.tolist(), | ||
| "pmax_withdrawal": pmax_withdrawal.tolist(), | ||
| "inflows": inflows.tolist(), | ||
| } | ||
| res = client.post( | ||
| f"/v1/studies/{study_id}/commands", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| json=[{"action": "create_st_storage", "args": args}], | ||
| ) | ||
| res.raise_for_status() | ||
|
|
||
| # ============================ | ||
| # Short-Term Storage Removal | ||
| # ============================ | ||
|
|
||
| # The `remove_st_storage` command allows you to delete a Short-Term Storage. | ||
| args = {"area_id": area_id, "storage_id": siemens_battery_id} | ||
| res = client.post( | ||
| f"/v1/studies/{study_id}/commands", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| json=[{"action": "remove_st_storage", "args": args}], | ||
| ) | ||
| res.raise_for_status() | ||
|
|
||
| # ======================================= | ||
| # Parameters and Time Series Validation | ||
| # ======================================= | ||
|
|
||
| # When creating a Short-Term Storage, both the validity of the parameters | ||
| # (value type and valid range) and the validity of the time series | ||
| # (value range) are checked. | ||
| # In the example below, multiple parameters are invalid, and one matrix contains | ||
| # values outside the valid range. Upon executing the request, an HTTP 422 | ||
| # error occurs, and a response specifies the invalid values. | ||
| pmax_injection = np.random.rand(8760, 1) | ||
| pmax_withdrawal = np.random.rand(8760, 1) * 10 # Oops! | ||
| inflows = np.random.randint(0, 1001, size=(8760, 1)) | ||
| args = { | ||
| "area_id": area_id, | ||
| "parameters": { | ||
| "name": "Bad Storage", | ||
| "group": "Wonderland", # Oops! | ||
| "injection_nominal_capacity": -2000, # Oops! | ||
| "withdrawal_nominal_capacity": 1500, | ||
| "reservoir_capacity": 20000, | ||
| "efficiency": 0.78, | ||
| "initial_level": 10000, | ||
| "initial_level_optim": "BlurBool", # Oops! | ||
| }, | ||
| "pmax_injection": pmax_injection.tolist(), | ||
| "pmax_withdrawal": pmax_withdrawal.tolist(), | ||
| "inflows": inflows.tolist(), | ||
| } | ||
| res = client.post( | ||
| f"/v1/studies/{study_id}/commands", | ||
| headers={"Authorization": f"Bearer {user_access_token}"}, | ||
| json=[{"action": "create_st_storage", "args": args}], | ||
| ) | ||
| assert res.status_code == http.HTTPStatus.UNPROCESSABLE_ENTITY | ||
| description = res.json()["description"] | ||
| """ | ||
| 4 validation errors for CreateSTStorage | ||
| parameters -> group | ||
| value is not a valid enumeration member […] | ||
| parameters -> injectionnominalcapacity | ||
| ensure this value is greater than or equal to 0 (type=value_error.number.not_ge; limit_value=0) | ||
| parameters -> initialleveloptim | ||
| value could not be parsed to a boolean (type=type_error.bool) | ||
| pmax_withdrawal | ||
| Matrix values should be between 0 and 1 (type=value_error) | ||
| """ | ||
| assert "parameters -> group" in description | ||
| assert "parameters -> injectionnominalcapacity" in description | ||
| assert "parameters -> initialleveloptim" in description | ||
| assert "pmax_withdrawal" in description |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
Empty file.
54 changes: 54 additions & 0 deletions
54
tests/study/storage/variantstudy/business/test_matrix_constants_generator.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| import numpy as np | ||
| from antarest.matrixstore.service import SimpleMatrixService | ||
| from antarest.study.storage.variantstudy.business import matrix_constants | ||
| from antarest.study.storage.variantstudy.business.matrix_constants_generator import ( | ||
| MATRIX_PROTOCOL_PREFIX, | ||
| GeneratorMatrixConstants, | ||
| ) | ||
|
|
||
|
|
||
| class TestGeneratorMatrixConstants: | ||
| def test_get_st_storage(self, tmp_path): | ||
| generator = GeneratorMatrixConstants( | ||
| matrix_service=SimpleMatrixService(bucket_dir=tmp_path) | ||
| ) | ||
|
|
||
| ref1 = generator.get_st_storage_pmax_injection() | ||
| matrix_id1 = ref1.split(MATRIX_PROTOCOL_PREFIX)[1] | ||
| matrix_dto1 = generator.matrix_service.get(matrix_id1) | ||
| assert ( | ||
| np.array(matrix_dto1.data).all() | ||
| == matrix_constants.st_storage.series.pmax_injection.all() | ||
| ) | ||
|
|
||
| ref2 = generator.get_st_storage_pmax_withdrawal() | ||
| matrix_id2 = ref2.split(MATRIX_PROTOCOL_PREFIX)[1] | ||
| matrix_dto2 = generator.matrix_service.get(matrix_id2) | ||
| assert ( | ||
| np.array(matrix_dto2.data).all() | ||
| == matrix_constants.st_storage.series.pmax_withdrawal.all() | ||
| ) | ||
|
|
||
| ref3 = generator.get_st_storage_lower_rule_curve() | ||
| matrix_id3 = ref3.split(MATRIX_PROTOCOL_PREFIX)[1] | ||
| matrix_dto3 = generator.matrix_service.get(matrix_id3) | ||
| assert ( | ||
| np.array(matrix_dto3.data).all() | ||
| == matrix_constants.st_storage.series.lower_rule_curve.all() | ||
| ) | ||
|
|
||
| ref4 = generator.get_st_storage_upper_rule_curve() | ||
| matrix_id4 = ref4.split(MATRIX_PROTOCOL_PREFIX)[1] | ||
| matrix_dto4 = generator.matrix_service.get(matrix_id4) | ||
| assert ( | ||
| np.array(matrix_dto4.data).all() | ||
| == matrix_constants.st_storage.series.upper_rule_curve.all() | ||
| ) | ||
|
|
||
| ref5 = generator.get_st_storage_inflows() | ||
| matrix_id5 = ref5.split(MATRIX_PROTOCOL_PREFIX)[1] | ||
| matrix_dto5 = generator.matrix_service.get(matrix_id5) | ||
| assert ( | ||
| np.array(matrix_dto5.data).all() | ||
| == matrix_constants.st_storage.series.inflows.all() | ||
| ) |
534 changes: 534 additions & 0 deletions
534
tests/variantstudy/model/command/test_create_st_storage.py
Large diffs are not rendered by default.
Oops, something went wrong.
259 changes: 259 additions & 0 deletions
259
tests/variantstudy/model/command/test_remove_st_storage.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,259 @@ | ||
| import re | ||
|
|
||
| import pytest | ||
| from antarest.study.storage.rawstudy.model.filesystem.config.model import ( | ||
| transform_name_to_id, | ||
| ) | ||
| from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy | ||
| from antarest.study.storage.study_upgrader import upgrade_study | ||
| from antarest.study.storage.variantstudy.model.command.common import ( | ||
| CommandName, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.create_area import ( | ||
| CreateArea, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.create_st_storage import ( | ||
| CreateSTStorage, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command.remove_st_storage import ( | ||
| REQUIRED_VERSION, | ||
| RemoveSTStorage, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.command_context import ( | ||
| CommandContext, | ||
| ) | ||
| from antarest.study.storage.variantstudy.model.model import CommandDTO | ||
| from pydantic import ValidationError | ||
|
|
||
|
|
||
| @pytest.fixture(name="recent_study") | ||
| def recent_study_fixture(empty_study: FileStudy) -> FileStudy: | ||
| """ | ||
| Fixture for creating a recent version of the FileStudy object. | ||
| Args: | ||
| empty_study: The empty FileStudy object used as model. | ||
| Returns: | ||
| FileStudy: The FileStudy object upgraded to the required version. | ||
| """ | ||
| upgrade_study(empty_study.config.study_path, str(REQUIRED_VERSION)) | ||
| empty_study.config.version = REQUIRED_VERSION | ||
| return empty_study | ||
|
|
||
|
|
||
| # The parameter names to be used are those in the INI file. | ||
| # Non-string values are automatically converted into strings. | ||
| # noinspection SpellCheckingInspection | ||
| PARAMETERS = { | ||
| "name": "Storage1", | ||
| "group": "Battery", | ||
| "injectionnominalcapacity": 1500, | ||
| "withdrawalnominalcapacity": 1500, | ||
| "reservoircapacity": 20000, | ||
| "efficiency": 0.94, | ||
| "initialleveloptim": True, | ||
| } | ||
|
|
||
|
|
||
| class TestRemoveSTStorage: | ||
| # noinspection SpellCheckingInspection | ||
| def test_init(self, command_context: CommandContext): | ||
| cmd = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="area_fr", | ||
| storage_id="storage_1", | ||
| ) | ||
|
|
||
| # Check the attribues | ||
| assert cmd.command_name == CommandName.REMOVE_ST_STORAGE | ||
| assert cmd.version == 1 | ||
| assert cmd.command_context == command_context | ||
| assert cmd.area_id == "area_fr" | ||
| assert cmd.storage_id == "storage_1" | ||
|
|
||
| def test_init__invalid_storage_id( | ||
| self, recent_study: FileStudy, command_context: CommandContext | ||
| ): | ||
| # When we apply the config for a new ST Storage with a bad name | ||
| with pytest.raises(ValidationError) as ctx: | ||
| RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="dummy", | ||
| storage_id="?%$$", # bad name | ||
| ) | ||
| assert ctx.value.errors() == [ | ||
| { | ||
| "ctx": {"pattern": "[a-z0-9_(),& -]+"}, | ||
| "loc": ("storage_id",), | ||
| "msg": 'string does not match regex "[a-z0-9_(),& -]+"', | ||
| "type": "value_error.str.regex", | ||
| } | ||
| ] | ||
|
|
||
| def test_apply_config__invalid_version( | ||
| self, empty_study: FileStudy, command_context: CommandContext | ||
| ): | ||
| # Given an old study in version 720 | ||
| # When we apply the config to add a new ST Storage | ||
| remove_st_storage = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="foo", | ||
| storage_id="bar", | ||
| ) | ||
| command_output = remove_st_storage.apply_config(empty_study.config) | ||
|
|
||
| # Then, the output should be an error | ||
| assert command_output.status is False | ||
| assert re.search( | ||
| rf"Invalid.*version {empty_study.config.version}", | ||
| command_output.message, | ||
| flags=re.IGNORECASE, | ||
| ) | ||
|
|
||
| def test_apply_config__missing_area( | ||
| self, recent_study: FileStudy, command_context: CommandContext | ||
| ): | ||
| # Given a study without "unknown area" area | ||
| # When we apply the config to add a new ST Storage | ||
| remove_st_storage = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="unknown area", # bad ID | ||
| storage_id="storage_1", | ||
| ) | ||
| command_output = remove_st_storage.apply_config(recent_study.config) | ||
|
|
||
| # Then, the output should be an error | ||
| assert command_output.status is False | ||
| assert re.search( | ||
| rf"'{re.escape(remove_st_storage.area_id)}'.*does not exist", | ||
| command_output.message, | ||
| flags=re.IGNORECASE, | ||
| ) | ||
|
|
||
| def test_apply_config__missing_storage( | ||
| self, recent_study: FileStudy, command_context: CommandContext | ||
| ): | ||
| # First, prepare a new Area | ||
| create_area = CreateArea( | ||
| command_context=command_context, | ||
| area_name="Area FR", | ||
| ) | ||
| create_area.apply(recent_study) | ||
|
|
||
| # Then, apply the config for a new ST Storage | ||
| remove_st_storage = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id=transform_name_to_id(create_area.area_name), | ||
| storage_id="storage 1", | ||
| ) | ||
| command_output = remove_st_storage.apply_config(recent_study.config) | ||
|
|
||
| # Then, the output should be an error | ||
| assert command_output.status is False | ||
| assert re.search( | ||
| rf"'{re.escape(remove_st_storage.storage_id)}'.*does not exist", | ||
| command_output.message, | ||
| flags=re.IGNORECASE, | ||
| ) | ||
|
|
||
| def test_apply_config__nominal_case( | ||
| self, recent_study: FileStudy, command_context: CommandContext | ||
| ): | ||
| # First, prepare a new Area | ||
| create_area = CreateArea( | ||
| area_name="Area FR", | ||
| command_context=command_context, | ||
| ) | ||
| create_area.apply(recent_study) | ||
|
|
||
| # Then, prepare a new Storage | ||
| create_st_storage = CreateSTStorage( | ||
| command_context=command_context, | ||
| area_id=transform_name_to_id(create_area.area_name), | ||
| parameters=PARAMETERS, # type: ignore | ||
| ) | ||
| create_st_storage.apply(recent_study) | ||
|
|
||
| # Then, apply the config for a new ST Storage | ||
| remove_st_storage = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id=transform_name_to_id(create_area.area_name), | ||
| storage_id=create_st_storage.storage_id, | ||
| ) | ||
| command_output = remove_st_storage.apply_config(recent_study.config) | ||
|
|
||
| # Check the command output and extra dict | ||
| assert command_output.status is True | ||
| assert re.search( | ||
| rf"'{re.escape(remove_st_storage.storage_id)}'.*removed", | ||
| command_output.message, | ||
| flags=re.IGNORECASE, | ||
| ) | ||
|
|
||
| def test_to_dto(self, command_context: CommandContext): | ||
| cmd = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="area_fr", | ||
| storage_id="storage_1", | ||
| ) | ||
| actual = cmd.to_dto() | ||
|
|
||
| # noinspection SpellCheckingInspection | ||
| assert actual == CommandDTO( | ||
| action=CommandName.REMOVE_ST_STORAGE.value, | ||
| args={"area_id": "area_fr", "storage_id": "storage_1"}, | ||
| ) | ||
|
|
||
| def test_match_signature(self, command_context: CommandContext): | ||
| cmd = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="area_fr", | ||
| storage_id="storage_1", | ||
| ) | ||
| assert cmd.match_signature() == "remove_st_storage%area_fr%storage_1" | ||
|
|
||
| @pytest.mark.parametrize("area_id", ["area_fr", "area_en"]) | ||
| @pytest.mark.parametrize("storage_id", ["storage_1", "storage_2"]) | ||
| def test_match( | ||
| self, | ||
| command_context: CommandContext, | ||
| area_id, | ||
| storage_id, | ||
| ): | ||
| cmd1 = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="area_fr", | ||
| storage_id="storage_1", | ||
| ) | ||
| cmd2 = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id=area_id, | ||
| storage_id=storage_id, | ||
| ) | ||
| is_equal = area_id == cmd1.area_id and storage_id == cmd1.storage_id | ||
| assert cmd1.match(cmd2, equal=False) == is_equal | ||
| assert cmd1.match(cmd2, equal=True) == is_equal | ||
|
|
||
| def test_create_diff(self, command_context: CommandContext): | ||
| cmd = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="area_fr", | ||
| storage_id="storage_1", | ||
| ) | ||
| other = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id=cmd.area_id, | ||
| storage_id=cmd.storage_id, | ||
| ) | ||
| actual = cmd.create_diff(other) | ||
| assert not actual | ||
|
|
||
| def test_get_inner_matrices(self, command_context: CommandContext): | ||
| cmd = RemoveSTStorage( | ||
| command_context=command_context, | ||
| area_id="area_fr", | ||
| storage_id="storage_1", | ||
| ) | ||
| actual = cmd.get_inner_matrices() | ||
| assert actual == [] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters