Coverage for agentlib_flexquant/data_structures/flexquant.py: 88%
121 statements
« prev ^ index » next coverage.py v7.4.4, created at 2026-06-17 09:09 +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
12import agentlib_flexquant.utils.config_management as cmng
14from agentlib_flexquant.data_structures.mpcs import (
15 BaselineMPCData,
16 NFMPCData,
17 PFMPCData,
18)
20excluded_fields = [
21 "rdf_class",
22 "source",
23 "type",
24 "timestamp",
25 "description",
26 "unit",
27 "clip",
28 "interpolation_method",
29 "allowed_values",
30 ]
33class ShadowMPCConfigGeneratorConfig(BaseModel):
34 """Class defining the options to initialize the shadow mpc config generation."""
36 model_config = ConfigDict(
37 json_encoders={MPCVariable: lambda v: v.dict(), AgentVariable: lambda v: v.dict()}, extra="forbid"
38 )
39 weights: list[MPCVariable] = Field(
40 default=[], description="Name and value of weights",
41 )
42 custom_inputs: list[AgentVariable] = Field(
43 default=[], description="Additional Inputs for the Shadow-MPCs. E.g. the baseline power prediction P_el_base"
44 )
46 pos_flex: PFMPCData = Field(default=None, description="Data for PF-MPC")
47 neg_flex: NFMPCData = Field(default=None, description="Data for NF-MPC")
49 @model_validator(mode="after")
50 def assign_weights_to_flex(self):
51 """Validate flexibility cost function fields and assign weights to them."""
52 if self.pos_flex is None:
53 raise ValueError(
54 "Missing required field: 'pos_flex' specifying the pos flex "
55 "cost function."
56 )
57 if self.neg_flex is None:
58 raise ValueError(
59 "Missing required field: 'neg_flex' specifying the neg flex "
60 "cost function."
61 )
62 if self.weights:
63 self.pos_flex.config_parameters_appendix.extend(self.weights)
64 self.neg_flex.config_parameters_appendix.extend(self.weights)
66 if self.custom_inputs:
67 self.pos_flex.config_inputs_appendix.extend(self.custom_inputs)
68 self.neg_flex.config_inputs_appendix.extend(self.custom_inputs)
70 return self
72 @field_serializer('weights', 'custom_inputs')
73 def serialize_mpc_variables(self, variables: list[MPCVariable], _info):
74 return [v.dict(exclude=excluded_fields) for v in variables]
77class FlexibilityMarketConfig(BaseModel):
78 """Class defining the options to initialize the market."""
80 model_config = ConfigDict(extra="forbid")
81 agent_config: Union[AgentConfig, Path, str]
82 name_of_created_file: str = Field(
83 default="flexibility_market.json",
84 description="Name of the config that is created by the generator",
85 )
86 module_type: Union[dict, str] = Field(
87 default=None,
88 description="Module type or dict with type and path for local files",
89 )
91 @model_validator(mode="after")
92 def check_file_extension(self):
93 """Validate that name_of_created_file has a .json extension."""
94 if self.name_of_created_file:
95 file_path = Path(self.name_of_created_file)
96 if file_path.suffix != ".json":
97 raise ConfigurationError(
98 f"Invalid file extension in market_config for "
99 f"name_of_created_file: '{self.name_of_created_file}'. "
100 f"Expected a '.json' file."
101 )
102 return self
104 @model_validator(mode="after")
105 def validate_module_type(self):
106 """Ensure module_type is str or dict and set default if None."""
108 if self.module_type is None:
109 self.module_type = cmng.MARKET_CONFIG_TYPE
110 return self
112 if isinstance(self.module_type, dict):
113 if 'file' not in self.module_type or 'class_name' not in self.module_type:
114 raise ConfigurationError("module_type dict must contain 'file' and 'class_name' keys")
116 elif not isinstance(self.module_type, str):
117 raise TypeError("module_type must be either a string or a dictionary")
119 return self
122class FlexibilityIndicatorConfig(BaseModel):
123 """Class defining the options for the flexibility indicators."""
125 model_config = ConfigDict(
126 json_encoders={Path: str, AgentConfig: lambda v: v.model_dump()}, extra="forbid"
127 )
128 agent_config: Union[AgentConfig, Path, str]
129 name_of_created_file: str = Field(
130 default="indicator.json",
131 description="Name of the config that is created by the generator",
132 )
133 module_type: Union[dict, str] = Field(
134 default=None,
135 description="Module type or dict with type and path for local files",
136 )
138 @model_validator(mode="after")
139 def check_file_extension(self):
140 """Validate that name_of_created_file has a .json extension."""
141 if self.name_of_created_file:
142 file_path = Path(self.name_of_created_file)
143 if file_path.suffix != ".json":
144 raise ConfigurationError(
145 f"Invalid file extension for indicator config "
146 f"name_of_created_file: '{self.name_of_created_file}'. "
147 f"Expected a '.json' file."
148 )
149 return self
151 @model_validator(mode="after")
152 def validate_module_type(self):
153 """Ensure module_type is str or dict and set default if None."""
155 if self.module_type is None:
156 self.module_type = cmng.INDICATOR_CONFIG_TYPE
157 return self
159 if isinstance(self.module_type, dict):
160 if 'file' not in self.module_type or 'class_name' not in self.module_type:
161 raise ConfigurationError("module_type dict must contain 'file' and 'class_name' keys")
163 elif not isinstance(self.module_type, str):
164 raise TypeError("module_type must be either a string or a dictionary")
166 return self
169class FlexQuantConfig(BaseModel):
170 """Class defining the options to initialize the FlexQuant generation."""
172 model_config = ConfigDict(json_encoders={Path: str}, extra="forbid")
173 prep_time: int = Field(
174 default=1800,
175 ge=0,
176 unit="s",
177 description="Preparation time before the flexibility event",
178 )
179 flex_event_duration: int = Field(
180 default=7200, ge=0, unit="s", description="Flexibility event duration",
181 )
182 market_time: int = Field(
183 default=900, ge=0, unit="s", description="Time for market interaction",
184 )
185 indicator_config: Union[FlexibilityIndicatorConfig, Path] = Field(
186 description="Path to the file or dict of flexibility indicator config",
187 )
188 market_config: Optional[Union[FlexibilityMarketConfig, Path]] = Field(
189 default=None, description="Path to the file or dict of market config",
190 )
191 baseline_config_generator_data: BaselineMPCData = Field(
192 description="Baseline generator data config file or dict",
193 )
194 shadow_mpc_config_generator_data: ShadowMPCConfigGeneratorConfig = Field(
195 description="Shadow mpc generator data config file or dict",
196 )
197 casadi_sim_time_step: int = Field(
198 default=0,
199 description="Simulate over the prediction horizon with a defined resolution "
200 "using Casadi simulator. "
201 "Only use it when the power depends on the states. "
202 "Don't use it when power itself is the control variable."
203 "Set to 0 to skip simulation",
204 )
205 flex_base_directory_path: Optional[Path] = Field(
206 default_factory=lambda: Path.cwd() / "flex_output_data",
207 description="Base path where generated flex data is stored",
208 )
209 flex_files_directory: Path = Field(
210 default=Path("created_flex_files"),
211 description="Directory where generated files (jsons) should be stored",
212 )
213 results_directory: Path = Field(
214 default=Path("results"),
215 description="Directory where generated result files (CSVs) should be stored",
216 )
217 delete_files: bool = Field(
218 default=True, description="If generated files should be deleted afterwards",
219 )
220 overwrite_files: bool = Field(
221 default=False,
222 description="If generated files should be overwritten by new files",
223 )
224 custom_plugins: list[str] = Field(
225 default=None,
226 description="Custom AgentLib plugins to be loaded",
227 )
229 @model_validator(mode="after")
230 def check_config_file_extension(self):
231 """Validate that the indicator and market config file paths have a '.json'
232 extension.
234 Raises:
235 ValueError: If either file does not have the expected '.json' extension.
237 """
238 if (
239 isinstance(self.indicator_config, Path)
240 and self.indicator_config.suffix != ".json"
241 ):
242 raise ValueError(
243 f"Invalid file extension for indicator "
244 f"config: '{self.indicator_config}'. "
245 f"Expected a '.json' file."
246 )
247 if (
248 isinstance(self.market_config, Path)
249 and self.market_config.suffix != ".json"
250 ):
251 raise ValueError(
252 f"Invalid file extension for market "
253 f"config: '{self.market_config}'. "
254 f"Expected a '.json' file."
255 )
256 return self
258 @field_validator('casadi_sim_time_step', mode='after')
259 @classmethod
260 def is_none_negative_integer(cls, value: int) -> int:
261 if value < 0:
262 raise ValueError(f'{value} is not a non-negative integer')
263 return value
265 @model_validator(mode="after")
266 def adapt_paths_and_create_directory(self):
267 """Adjust and ensure the directory structure for flex file generation and
268 results storage.
270 This method:
271 - Updates `flex_files_directory` and `results_directory` paths, so they are
272 relative to
273 the base flex directory, using only the directory names (ignoring any
274 user-supplied paths).
275 - Creates the base, flex files, and results directories if they do not
276 already exist.
278 """
279 # adapt paths and use only names for user supplied data
280 self.flex_files_directory = (
281 self.flex_base_directory_path / self.flex_files_directory.name
282 )
283 self.results_directory = (
284 self.flex_base_directory_path / self.results_directory.name
285 )
286 # create directories if not already existing
287 self.flex_base_directory_path.mkdir(parents=True, exist_ok=True)
288 self.flex_files_directory.mkdir(parents=True, exist_ok=True)
289 self.results_directory.mkdir(parents=True, exist_ok=True)
290 return self
292 @model_validator(mode="after")
293 def validate_custom_plugins(self):
294 """Ensure custom_plugins is a list of strings (or None)."""
295 if self.custom_plugins is None:
296 return self
298 if not isinstance(self.custom_plugins, list) or not all(
299 isinstance(p, str) for p in self.custom_plugins
300 ):
301 raise ConfigurationError(
302 f"Invalid custom_plugins: {self.custom_plugins!r} "
303 f"(type: {type(self.custom_plugins).__name__}). "
304 "Expected a list of strings."
305 )
307 return self