Coverage for agentlib_flexquant/data_structures/flexquant.py: 92%
88 statements
« prev ^ index » next coverage.py v7.4.4, created at 2026-03-26 09:43 +0000
1"""
2Pydantic data models for FlexQuant configuration and validation.
3"""
4from pathlib import Path
5from typing import Optional, Union
7from pydantic import (field_validator, ConfigDict, model_validator, Field, BaseModel,
8 field_serializer)
9from agentlib.core.agent import AgentConfig
10from agentlib.core.errors import ConfigurationError
11from agentlib_mpc.data_structures.mpc_datamodels import AgentVariable, MPCVariable
13from agentlib_flexquant.data_structures.mpcs import (
14 BaselineMPCData,
15 NFMPCData,
16 PFMPCData,
17)
19excluded_fields = [
20 "rdf_class",
21 "source",
22 "type",
23 "timestamp",
24 "description",
25 "unit",
26 "clip",
27 "interpolation_method",
28 "allowed_values",
29 ]
32class ShadowMPCConfigGeneratorConfig(BaseModel):
33 """Class defining the options to initialize the shadow mpc config generation."""
35 model_config = ConfigDict(
36 json_encoders={MPCVariable: lambda v: v.dict(), AgentVariable: lambda v: v.dict()}, extra="forbid"
37 )
38 weights: list[MPCVariable] = Field(
39 default=[], description="Name and value of weights",
40 )
41 custom_inputs: list[AgentVariable] = Field(
42 default=[], description="Additional Inputs for the Shadow-MPCs. E.g. the baseline power prediction P_el_base"
43 )
45 pos_flex: PFMPCData = Field(default=None, description="Data for PF-MPC")
46 neg_flex: NFMPCData = Field(default=None, description="Data for NF-MPC")
48 @model_validator(mode="after")
49 def assign_weights_to_flex(self):
50 """Validate flexibility cost function fields and assign weights to them."""
51 if self.pos_flex is None:
52 raise ValueError(
53 "Missing required field: 'pos_flex' specifying the pos flex "
54 "cost function."
55 )
56 if self.neg_flex is None:
57 raise ValueError(
58 "Missing required field: 'neg_flex' specifying the neg flex "
59 "cost function."
60 )
61 if self.weights:
62 self.pos_flex.config_parameters_appendix.extend(self.weights)
63 self.neg_flex.config_parameters_appendix.extend(self.weights)
65 if self.custom_inputs:
66 self.pos_flex.config_inputs_appendix.extend(self.custom_inputs)
67 self.neg_flex.config_inputs_appendix.extend(self.custom_inputs)
69 return self
71 @field_serializer('weights', 'custom_inputs')
72 def serialize_mpc_variables(self, variables: list[MPCVariable], _info):
73 return [v.dict(exclude=excluded_fields) for v in variables]
76class FlexibilityMarketConfig(BaseModel):
77 """Class defining the options to initialize the market."""
79 model_config = ConfigDict(extra="forbid")
80 agent_config: Union[AgentConfig, Path, str]
81 name_of_created_file: str = Field(
82 default="flexibility_market.json",
83 description="Name of the config that is created by the generator",
84 )
86 @model_validator(mode="after")
87 def check_file_extension(self):
88 """Validate that name_of_created_file has a .json extension."""
89 if self.name_of_created_file:
90 file_path = Path(self.name_of_created_file)
91 if file_path.suffix != ".json":
92 raise ConfigurationError(
93 f"Invalid file extension in market_config for "
94 f"name_of_created_file: '{self.name_of_created_file}'. "
95 f"Expected a '.json' file."
96 )
97 return self
100class FlexibilityIndicatorConfig(BaseModel):
101 """Class defining the options for the flexibility indicators."""
103 model_config = ConfigDict(
104 json_encoders={Path: str, AgentConfig: lambda v: v.model_dump()}, extra="forbid"
105 )
106 agent_config: Union[AgentConfig, Path, str]
107 name_of_created_file: str = Field(
108 default="indicator.json",
109 description="Name of the config that is created by the generator",
110 )
112 @model_validator(mode="after")
113 def check_file_extension(self):
114 """Validate that name_of_created_file has a .json extension."""
115 if self.name_of_created_file:
116 file_path = Path(self.name_of_created_file)
117 if file_path.suffix != ".json":
118 raise ConfigurationError(
119 f"Invalid file extension for indicator config "
120 f"name_of_created_file: '{self.name_of_created_file}'. "
121 f"Expected a '.json' file."
122 )
123 return self
126class FlexQuantConfig(BaseModel):
127 """Class defining the options to initialize the FlexQuant generation."""
129 model_config = ConfigDict(json_encoders={Path: str}, extra="forbid")
130 prep_time: int = Field(
131 default=1800,
132 ge=0,
133 unit="s",
134 description="Preparation time before the flexibility event",
135 )
136 flex_event_duration: int = Field(
137 default=7200, ge=0, unit="s", description="Flexibility event duration",
138 )
139 market_time: int = Field(
140 default=900, ge=0, unit="s", description="Time for market interaction",
141 )
142 indicator_config: Union[FlexibilityIndicatorConfig, Path] = Field(
143 description="Path to the file or dict of flexibility indicator config",
144 )
145 market_config: Optional[Union[FlexibilityMarketConfig, Path]] = Field(
146 default=None, description="Path to the file or dict of market config",
147 )
148 baseline_config_generator_data: BaselineMPCData = Field(
149 description="Baseline generator data config file or dict",
150 )
151 shadow_mpc_config_generator_data: ShadowMPCConfigGeneratorConfig = Field(
152 description="Shadow mpc generator data config file or dict",
153 )
154 casadi_sim_time_step: int = Field(
155 default=0,
156 description="Simulate over the prediction horizon with a defined resolution "
157 "using Casadi simulator. "
158 "Only use it when the power depends on the states. "
159 "Don't use it when power itself is the control variable."
160 "Set to 0 to skip simulation",
161 )
162 flex_base_directory_path: Optional[Path] = Field(
163 default_factory=lambda: Path.cwd() / "flex_output_data",
164 description="Base path where generated flex data is stored",
165 )
166 flex_files_directory: Path = Field(
167 default=Path("created_flex_files"),
168 description="Directory where generated files (jsons) should be stored",
169 )
170 results_directory: Path = Field(
171 default=Path("results"),
172 description="Directory where generated result files (CSVs) should be stored",
173 )
174 delete_files: bool = Field(
175 default=True, description="If generated files should be deleted afterwards",
176 )
177 overwrite_files: bool = Field(
178 default=False,
179 description="If generated files should be overwritten by new files",
180 )
182 @model_validator(mode="after")
183 def check_config_file_extension(self):
184 """Validate that the indicator and market config file paths have a '.json'
185 extension.
187 Raises:
188 ValueError: If either file does not have the expected '.json' extension.
190 """
191 if (
192 isinstance(self.indicator_config, Path)
193 and self.indicator_config.suffix != ".json"
194 ):
195 raise ValueError(
196 f"Invalid file extension for indicator "
197 f"config: '{self.indicator_config}'. "
198 f"Expected a '.json' file."
199 )
200 if (
201 isinstance(self.market_config, Path)
202 and self.market_config.suffix != ".json"
203 ):
204 raise ValueError(
205 f"Invalid file extension for market "
206 f"config: '{self.market_config}'. "
207 f"Expected a '.json' file."
208 )
209 return self
211 @field_validator('casadi_sim_time_step', mode='after')
212 @classmethod
213 def is_none_negative_integer(cls, value: int) -> int:
214 if value < 0:
215 raise ValueError(f'{value} is not a non-negative integer')
216 return value
218 @model_validator(mode="after")
219 def adapt_paths_and_create_directory(self):
220 """Adjust and ensure the directory structure for flex file generation and
221 results storage.
223 This method:
224 - Updates `flex_files_directory` and `results_directory` paths, so they are
225 relative to
226 the base flex directory, using only the directory names (ignoring any
227 user-supplied paths).
228 - Creates the base, flex files, and results directories if they do not
229 already exist.
231 """
232 # adapt paths and use only names for user supplied data
233 self.flex_files_directory = (
234 self.flex_base_directory_path / self.flex_files_directory.name
235 )
236 self.results_directory = (
237 self.flex_base_directory_path / self.results_directory.name
238 )
239 # create directories if not already existing
240 self.flex_base_directory_path.mkdir(parents=True, exist_ok=True)
241 self.flex_files_directory.mkdir(parents=True, exist_ok=True)
242 self.results_directory.mkdir(parents=True, exist_ok=True)
243 return self