Coverage for agentlib_flexquant/generate_flex_agents.py: 86%
255 statements
« prev ^ index » next coverage.py v7.4.4, created at 2025-08-01 15:10 +0000
1import ast
2import atexit
3import inspect
4import logging
5import os
6from copy import deepcopy
7from pathlib import Path
8from typing import List, Union
10import astor
11import black
12import json
13from agentlib.core.agent import AgentConfig
14from agentlib.core.datamodels import AgentVariable
15from agentlib.core.errors import ConfigurationError
16from agentlib.core.module import BaseModuleConfig
17from agentlib.utils import custom_injection, load_config
18from agentlib_mpc.data_structures.mpc_datamodels import MPCVariable
19from agentlib_mpc.models.casadi_model import CasadiModelConfig
20from agentlib_mpc.modules.mpc_full import BaseMPCConfig
21from pydantic import FilePath
23import agentlib_flexquant.data_structures.globals as glbs
24import agentlib_flexquant.utils.config_management as cmng
25from agentlib_flexquant.utils.parsing import (
26 SetupSystemModifier,
27 add_import_to_tree
28)
29from agentlib_flexquant.data_structures.flexquant import (
30 FlexibilityIndicatorConfig,
31 FlexibilityMarketConfig,
32 FlexQuantConfig
33)
34from agentlib_flexquant.data_structures.mpcs import (
35 BaselineMPCData,
36 BaseMPCData
37)
38from agentlib_flexquant.modules.flexibility_indicator import (
39 FlexibilityIndicatorModuleConfig
40)
41from agentlib_flexquant.modules.flexibility_market import (
42 FlexibilityMarketModuleConfig
43)
46class FlexAgentGenerator:
47 orig_mpc_module_config: BaseMPCConfig
48 baseline_mpc_module_config: BaseMPCConfig
49 pos_flex_mpc_module_config: BaseMPCConfig
50 neg_flex_mpc_module_config: BaseMPCConfig
51 indicator_module_config: FlexibilityIndicatorModuleConfig
52 market_module_config: FlexibilityMarketModuleConfig
54 def __init__(
55 self,
56 flex_config: Union[str, FilePath, FlexQuantConfig],
57 mpc_agent_config: Union[str, FilePath, AgentConfig],
58 ):
59 self.logger = logging.getLogger(__name__)
61 if isinstance(flex_config, str or FilePath):
62 self.flex_config_file_name = os.path.basename(flex_config)
63 else:
64 # provide default name for json
65 self.flex_config_file_name = "flex_config.json"
66 # load configs
67 self.flex_config = load_config.load_config(
68 flex_config, config_type=FlexQuantConfig
69 )
71 # original mpc agent
72 self.orig_mpc_agent_config = load_config.load_config(
73 mpc_agent_config, config_type=AgentConfig
74 )
75 # baseline agent
76 self.baseline_mpc_agent_config = self.orig_mpc_agent_config.__deepcopy__()
77 # pos agent
78 self.pos_flex_mpc_agent_config = self.orig_mpc_agent_config.__deepcopy__()
79 # neg agent
80 self.neg_flex_mpc_agent_config = self.orig_mpc_agent_config.__deepcopy__()
82 # original mpc module
83 self.orig_mpc_module_config = cmng.get_module(
84 config=self.orig_mpc_agent_config,
85 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
86 )
87 # baseline module
88 self.baseline_mpc_module_config = cmng.get_module(
89 config=self.baseline_mpc_agent_config,
90 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
91 )
92 # pos module
93 self.pos_flex_mpc_module_config = cmng.get_module(
94 config=self.pos_flex_mpc_agent_config,
95 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
96 )
97 # neg module
98 self.neg_flex_mpc_module_config = cmng.get_module(
99 config=self.neg_flex_mpc_agent_config,
100 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
101 )
102 # load indicator config
103 self.indicator_config = load_config.load_config(
104 self.flex_config.indicator_config, config_type=FlexibilityIndicatorConfig
105 )
106 # load indicator module config
107 self.indicator_agent_config = load_config.load_config(
108 self.indicator_config.agent_config, config_type=AgentConfig
109 )
110 self.indicator_module_config = cmng.get_module(
111 config=self.indicator_agent_config, module_type=cmng.INDICATOR_CONFIG_TYPE
112 )
113 # load market config
114 if self.flex_config.market_config:
115 self.market_config = load_config.load_config(
116 self.flex_config.market_config, config_type=FlexibilityMarketConfig
117 )
118 # load market module config
119 self.market_agent_config = load_config.load_config(
120 self.market_config.agent_config, config_type=AgentConfig
121 )
122 self.market_module_config = cmng.get_module(
123 config=self.market_agent_config, module_type=cmng.MARKET_CONFIG_TYPE
124 )
125 else:
126 self.flex_config.market_time = 0
128 self.run_config_validations()
130 def generate_flex_agents(
131 self,
132 ) -> [
133 BaseMPCConfig,
134 BaseMPCConfig,
135 BaseMPCConfig,
136 FlexibilityIndicatorModuleConfig,
137 FlexibilityMarketModuleConfig,
138 ]:
139 """Generates the configs and the python module for the flexibility agents.
140 Power variable must be defined in the mpc config.
142 """
143 # adapt modules to include necessary communication variables
144 baseline_mpc_config = self.adapt_mpc_module_config(
145 module_config=self.baseline_mpc_module_config,
146 mpc_dataclass=self.flex_config.baseline_config_generator_data,
147 )
148 pf_mpc_config = self.adapt_mpc_module_config(
149 module_config=self.pos_flex_mpc_module_config,
150 mpc_dataclass=self.flex_config.shadow_mpc_config_generator_data.pos_flex,
151 )
152 nf_mpc_config = self.adapt_mpc_module_config(
153 module_config=self.neg_flex_mpc_module_config,
154 mpc_dataclass=self.flex_config.shadow_mpc_config_generator_data.neg_flex,
155 )
156 indicator_module_config = self.adapt_indicator_config(
157 module_config=self.indicator_module_config
158 )
159 if self.flex_config.market_config:
160 market_module_config = self.adapt_market_config(
161 module_config=self.market_module_config
162 )
164 # dump jsons of the agents including the adapted module configs
165 self.append_module_and_dump_agent(
166 module=baseline_mpc_config,
167 agent=self.baseline_mpc_agent_config,
168 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
169 config_name=self.flex_config.baseline_config_generator_data.name_of_created_file,
170 )
171 self.append_module_and_dump_agent(
172 module=pf_mpc_config,
173 agent=self.pos_flex_mpc_agent_config,
174 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
175 config_name=self.flex_config.shadow_mpc_config_generator_data.pos_flex.name_of_created_file,
176 )
177 self.append_module_and_dump_agent(
178 module=nf_mpc_config,
179 agent=self.neg_flex_mpc_agent_config,
180 module_type=cmng.get_orig_module_type(self.orig_mpc_agent_config),
181 config_name=self.flex_config.shadow_mpc_config_generator_data.neg_flex.name_of_created_file,
182 )
183 self.append_module_and_dump_agent(
184 module=indicator_module_config,
185 agent=self.indicator_agent_config,
186 module_type=cmng.INDICATOR_CONFIG_TYPE,
187 config_name=self.indicator_config.name_of_created_file,
188 )
189 if self.flex_config.market_config:
190 self.append_module_and_dump_agent(
191 module=market_module_config,
192 agent=self.market_agent_config,
193 module_type=cmng.MARKET_CONFIG_TYPE,
194 config_name=self.market_config.name_of_created_file,
195 )
197 # generate python files for the shadow mpcs
198 self._generate_flex_model_definition()
200 # save flex config to created flex files
201 with open(os.path.join(self.flex_config.flex_files_directory, self.flex_config_file_name), "w") as f:
202 config_json = self.flex_config.model_dump_json(exclude_defaults=True)
203 f.write(config_json)
205 # register the exit function if the corresponding flag is set
206 if self.flex_config.delete_files:
207 atexit.register(lambda: self._delete_created_files())
208 return self.get_config_file_paths()
210 def append_module_and_dump_agent(
211 self,
212 module: BaseModuleConfig,
213 agent: AgentConfig,
214 module_type: str,
215 config_name: str,
216 ):
217 """Appends the given module config to the given agent config and dumps the agent config to a
218 json file. The json file is named based on the config_name."""
220 # if module is not from the baseline, set a new agent id, based on module id
221 if module.type is not self.baseline_mpc_module_config.type:
222 agent.id = module.module_id
223 # get the module as a dict without default values
224 module_dict = cmng.to_dict_and_remove_unnecessary_fields(module=module)
225 # write given module to agent config
226 for i, agent_module in enumerate(agent.modules):
227 if (
228 cmng.MODULE_TYPE_DICT[module_type]
229 is cmng.MODULE_TYPE_DICT[agent_module["type"]]
230 ):
231 agent.modules[i] = module_dict
233 # dump agent config
234 if agent.modules:
235 if self.flex_config.overwrite_files:
236 try:
237 Path(
238 os.path.join(self.flex_config.flex_files_directory, config_name)
239 ).unlink()
240 except OSError:
241 pass
242 with open(
243 os.path.join(self.flex_config.flex_files_directory, config_name), "w+"
244 ) as f:
245 module_json = agent.model_dump_json(exclude_defaults=True)
246 f.write(module_json)
247 else:
248 logging.error("Provided agent config does not contain any modules.")
250 def get_config_file_paths(self) -> List[str]:
251 """Returns a list of paths with the created config files
253 """
254 paths = [
255 os.path.join(
256 self.flex_config.flex_files_directory,
257 self.flex_config.baseline_config_generator_data.name_of_created_file,
258 ),
259 os.path.join(
260 self.flex_config.flex_files_directory,
261 self.flex_config.shadow_mpc_config_generator_data.pos_flex.name_of_created_file,
262 ),
263 os.path.join(
264 self.flex_config.flex_files_directory,
265 self.flex_config.shadow_mpc_config_generator_data.neg_flex.name_of_created_file,
266 ),
267 os.path.join(
268 self.flex_config.flex_files_directory,
269 self.indicator_config.name_of_created_file,
270 ),
271 ]
272 if self.flex_config.market_config:
273 paths.append(
274 os.path.join(
275 self.flex_config.flex_files_directory,
276 self.market_config.name_of_created_file,
277 )
278 )
279 return paths
281 def _delete_created_files(self):
282 """Function to run at exit if the files are to be deleted
284 """
285 to_be_deleted = self.get_config_file_paths()
286 to_be_deleted.append(
287 os.path.join(
288 self.flex_config.flex_files_directory,
289 self.flex_config_file_name,
290 ))
291 # delete files
292 for file in to_be_deleted:
293 Path(file).unlink()
294 # also delete folder
295 Path(self.flex_config.flex_files_directory).rmdir()
297 def adapt_mpc_module_config(
298 self, module_config: BaseMPCConfig, mpc_dataclass: BaseMPCData
299 ) -> BaseMPCConfig:
300 """Adapts the mpc module config for automated flexibility quantification.
301 Things adapted among others are:
302 - the file name/path of the mpc config file
303 - names of the control variables for the shadow mpcs
304 - reduce communicated variables of shadow mpcs to outputs
305 - add the power variable to the outputs
306 - add the Time variable to the inputs
307 - add parameters for the activation and quantification of flexibility
309 """
310 # allow the module config to be changed
311 module_config.model_config["frozen"] = False
313 module_config.module_id = mpc_dataclass.module_id
315 # append the new weights as parameter to the MPC or update its value
316 parameter_dict = {
317 parameter.name: parameter for parameter in module_config.parameters
318 }
319 for weight in mpc_dataclass.weights:
320 if weight.name in parameter_dict:
321 parameter_dict[weight.name].value = weight.value
322 else:
323 module_config.parameters.append(weight)
325 # set new MPC type
326 module_config.type = mpc_dataclass.module_types[
327 cmng.get_orig_module_type(self.orig_mpc_agent_config)
328 ]
329 # set new id (needed for plotting)
330 module_config.module_id = mpc_dataclass.module_id
331 # update optimization backend to use the created mpc files and classes
332 module_config.optimization_backend["model"]["type"] = {
333 "file": os.path.join(
334 self.flex_config.flex_files_directory,
335 mpc_dataclass.created_flex_mpcs_file,
336 ),
337 "class_name": mpc_dataclass.class_name,
338 }
339 # extract filename from results file and update it with suffix and parent directory
340 result_filename = Path(
341 module_config.optimization_backend["results_file"]
342 ).name.replace(".csv", mpc_dataclass.results_suffix)
343 full_path = (
344 self.flex_config.results_directory
345 / result_filename
346 )
347 module_config.optimization_backend["results_file"] = str(full_path)
348 # change cia backend to custom backend of flexquant
349 if module_config.optimization_backend["type"] == "casadi_cia":
350 module_config.optimization_backend["type"] = "casadi_cia_cons"
351 module_config.optimization_backend["market_time"] = (
352 self.flex_config.market_time
353 )
355 # add the control signal of the baseline to outputs (used during market time)
356 # and as inputs for the shadow mpcs
357 if type(mpc_dataclass) is not BaselineMPCData:
358 for control in module_config.controls:
359 module_config.inputs.append(
360 MPCVariable(
361 name=glbs.full_trajectory_prefix
362 + control.name
363 + glbs.full_trajectory_suffix,
364 value=control.value,
365 )
366 )
367 # also include binary controls
368 if hasattr(module_config, "binary_controls"):
369 for control in module_config.binary_controls:
370 module_config.inputs.append(
371 MPCVariable(
372 name=glbs.full_trajectory_prefix
373 + control.name
374 + glbs.full_trajectory_suffix,
375 value=control.value,
376 )
377 )
379 # only communicate outputs for the shadow mpcs
380 module_config.shared_variable_fields = ["outputs"]
381 else:
382 for control in module_config.controls:
383 module_config.outputs.append(
384 MPCVariable(
385 name=glbs.full_trajectory_prefix
386 + control.name
387 + glbs.full_trajectory_suffix,
388 value=control.value,
389 )
390 )
391 # also include binary controls
392 if hasattr(module_config, "binary_controls"):
393 for control in module_config.binary_controls:
394 module_config.outputs.append(
395 MPCVariable(
396 name=glbs.full_trajectory_prefix
397 + control.name
398 + glbs.full_trajectory_suffix,
399 value=control.value,
400 )
401 )
402 module_config.set_outputs = True
403 # add outputs for the power variables, for easier handling create a lookup dict
404 output_dict = {output.name: output for output in module_config.outputs}
405 if (
406 self.flex_config.baseline_config_generator_data.power_variable
407 in output_dict
408 ):
409 output_dict[
410 self.flex_config.baseline_config_generator_data.power_variable
411 ].alias = mpc_dataclass.power_alias
412 else:
413 module_config.outputs.append(
414 MPCVariable(
415 name=self.flex_config.baseline_config_generator_data.power_variable,
416 alias=mpc_dataclass.power_alias,
417 )
418 )
419 # add or change alias for stored energy variable
420 if self.indicator_module_config.correct_costs.enable_energy_costs_correction:
421 output_dict[
422 self.indicator_module_config.correct_costs.stored_energy_variable
423 ].alias = mpc_dataclass.stored_energy_alias
425 # add extra inputs needed for activation of flex
426 module_config.inputs.extend(mpc_dataclass.config_inputs_appendix)
427 # CONFIG_PARAMETERS_APPENDIX only includes dummy values
428 # overwrite dummy values with values from flex config and append it to module config
429 for var in mpc_dataclass.config_parameters_appendix:
430 if var.name in self.flex_config.model_fields:
431 var.value = getattr(self.flex_config, var.name)
432 if var.name in self.flex_config.baseline_config_generator_data.model_fields:
433 var.value = getattr(self.flex_config.baseline_config_generator_data, var.name)
434 module_config.parameters.extend(mpc_dataclass.config_parameters_appendix)
436 # freeze the config again
437 module_config.model_config["frozen"] = True
439 return module_config
441 def adapt_indicator_config(
442 self, module_config: FlexibilityIndicatorModuleConfig
443 ) -> FlexibilityIndicatorModuleConfig:
444 """Adapts the indicator module config for automated flexibility quantification.
446 """
447 # append user-defined price var to indicator module config
448 module_config.inputs.append(
449 AgentVariable(
450 name=module_config.price_variable,
451 unit="ct/kWh",
452 type="pd.Series",
453 description="electricity price"
454 )
455 )
456 # allow the module config to be changed
457 module_config.model_config["frozen"] = False
458 for parameter in module_config.parameters:
459 if parameter.name == glbs.PREP_TIME:
460 parameter.value = self.flex_config.prep_time
461 if parameter.name == glbs.MARKET_TIME:
462 parameter.value = self.flex_config.market_time
463 if parameter.name == glbs.FLEX_EVENT_DURATION:
464 parameter.value = self.flex_config.flex_event_duration
465 if parameter.name == "time_step":
466 parameter.value = self.baseline_mpc_module_config.time_step
467 if parameter.name == "prediction_horizon":
468 parameter.value = self.baseline_mpc_module_config.prediction_horizon
470 # set power unit
471 module_config.power_unit = (
472 self.flex_config.baseline_config_generator_data.power_unit
473 )
474 module_config.results_file = (
475 self.flex_config.results_directory
476 / module_config.results_file.name
477 )
478 module_config.model_config["frozen"] = True
479 return module_config
481 def adapt_market_config(
482 self, module_config: FlexibilityMarketModuleConfig
483 ) -> FlexibilityMarketModuleConfig:
484 """Adapts the market module config for automated flexibility quantification.
486 """
487 # allow the module config to be changed
488 module_config.model_config["frozen"] = False
489 for field in module_config.__fields__:
490 if field in self.market_module_config.__fields__.keys():
491 module_config.__setattr__(
492 field, getattr(self.market_module_config, field)
493 )
494 module_config.results_file = (
495 self.flex_config.results_directory
496 / module_config.results_file.name
497 )
498 module_config.model_config["frozen"] = True
499 return module_config
501 def _generate_flex_model_definition(self):
502 """Generates a python module for negative and positive flexibility agents from
503 the Baseline MPC model
505 """
506 output_file = os.path.join(
507 self.flex_config.flex_files_directory,
508 self.flex_config.baseline_config_generator_data.created_flex_mpcs_file,
509 )
510 opt_backend = self.orig_mpc_module_config.optimization_backend["model"]["type"]
512 # Extract the config class of the casadi model to check cost functions
513 config_class = inspect.get_annotations(custom_injection(opt_backend))["config"]
514 config_instance = config_class()
515 self.check_variables_in_casadi_config(
516 config_instance,
517 self.flex_config.shadow_mpc_config_generator_data.neg_flex.flex_cost_function,
518 )
519 self.check_variables_in_casadi_config(
520 config_instance,
521 self.flex_config.shadow_mpc_config_generator_data.pos_flex.flex_cost_function,
522 )
524 # parse mpc python file
525 with open(opt_backend["file"], "r") as f:
526 source = f.read()
527 tree = ast.parse(source)
529 # create modifiers for python file
530 modifier_base = SetupSystemModifier(
531 mpc_data=self.flex_config.baseline_config_generator_data,
532 controls=self.baseline_mpc_module_config.controls,
533 binary_controls=self.baseline_mpc_module_config.binary_controls if hasattr(self.baseline_mpc_module_config, "binary_controls") else None,
534 )
535 modifier_pos = SetupSystemModifier(
536 mpc_data=self.flex_config.shadow_mpc_config_generator_data.pos_flex,
537 controls=self.pos_flex_mpc_module_config.controls,
538 binary_controls=self.pos_flex_mpc_module_config.binary_controls if hasattr(self.pos_flex_mpc_module_config, "binary_controls") else None,
539 )
540 modifier_neg = SetupSystemModifier(
541 mpc_data=self.flex_config.shadow_mpc_config_generator_data.neg_flex,
542 controls=self.neg_flex_mpc_module_config.controls,
543 binary_controls=self.neg_flex_mpc_module_config.binary_controls if hasattr(self.neg_flex_mpc_module_config, "binary_controls") else None,
544 )
545 # run the modification
546 modified_tree_base = modifier_base.visit(deepcopy(tree))
547 modified_tree_pos = modifier_pos.visit(deepcopy(tree))
548 modified_tree_neg = modifier_neg.visit(deepcopy(tree))
549 # combine modifications to one file
550 modified_tree = ast.Module(body=[], type_ignores=[])
551 modified_tree.body.extend(
552 modified_tree_base.body + modified_tree_pos.body + modified_tree_neg.body
553 )
554 modified_source = astor.to_source(modified_tree)
555 # Use black to format the generated code
556 formatted_code = black.format_str(modified_source, mode=black.FileMode())
558 if self.flex_config.overwrite_files:
559 try:
560 Path(
561 os.path.join(
562 self.flex_config.flex_files_directory,
563 self.flex_config.baseline_config_generator_data.created_flex_mpcs_file,
564 )
565 ).unlink()
566 except OSError:
567 pass
569 with open(output_file, "w") as f:
570 f.write(formatted_code)
572 def check_variables_in_casadi_config(self, config: CasadiModelConfig, expr: str):
573 """Check if all variables in the expression are defined in the config.
575 Args:
576 config (CasadiModelConfig): casadi model config.
577 expr (str): The expression to check.
579 Raises:
580 ValueError: If any variable in the expression is not defined in the config.
582 """
583 variables_in_config = set(config.get_variable_names())
584 variables_in_cost_function = set(ast.walk(ast.parse(expr)))
585 variables_in_cost_function = {
586 node.attr
587 for node in variables_in_cost_function
588 if isinstance(node, ast.Attribute)
589 }
590 variables_newly_created = set(
591 weight.name
592 for weight in self.flex_config.shadow_mpc_config_generator_data.weights
593 )
594 unknown_vars = (
595 variables_in_cost_function - variables_in_config - variables_newly_created
596 )
597 if unknown_vars:
598 raise ValueError(f"Unknown variables in new cost function: {unknown_vars}")
600 def run_config_validations(self):
601 """
602 Function to validate integrity of user-supplied flex config.
604 The following checks are performed:
605 1. Ensures the specified power variable exists in the MPC model outputs.
606 2. Ensures the specified comfort variable exists in the MPC model states.
607 3. Validates that the stored energy variable exists in MPC outputs if energy cost correction is enabled.
608 4. Verifies the supported collocation method is used; otherwise, switches to 'legendre' and raises a warning.
609 5. Ensures that the sum of prep time, market time, and flex event duration does not exceed the prediction horizon.
610 6. Ensures market time equals the MPC model time step if market config is present.
611 7. Ensures that all flex time values are multiples of the MPC model time step.
612 8. Checks for mismatches between time-related parameters in the flex/MPC and indicator configs and issues warnings
613 when discrepancies exist, using the flex/MPC config values as the source of truth.
615 """
616 # check if the power variable exists in the mpc config
617 if self.flex_config.baseline_config_generator_data.power_variable not in [
618 output.name for output in self.baseline_mpc_module_config.outputs
619 ]:
620 raise ConfigurationError(
621 f"Given power variable {self.flex_config.baseline_config_generator_data.power_variable} is not defined as output in baseline mpc config."
622 )
624 # check if the comfort variable exists in the mpc slack variables
625 if self.flex_config.baseline_config_generator_data.comfort_variable:
626 file_path = self.baseline_mpc_module_config.optimization_backend["model"]["type"]["file"]
627 class_name = self.baseline_mpc_module_config.optimization_backend["model"]["type"]["class_name"]
628 # Get the class
629 dynamic_class = cmng.get_class_from_file(file_path, class_name)
630 if self.flex_config.baseline_config_generator_data.comfort_variable not in [
631 state.name for state in dynamic_class().states
632 ]:
633 raise ConfigurationError(
634 f"Given comfort variable {self.flex_config.baseline_config_generator_data.comfort_variable} is not defined as state in baseline mpc config."
635 )
637 # check if the energy storage variable exists in the mpc config
638 if self.indicator_module_config.correct_costs.enable_energy_costs_correction:
639 if self.indicator_module_config.correct_costs.stored_energy_variable not in [
640 output.name for output in self.baseline_mpc_module_config.outputs
641 ]:
642 raise ConfigurationError(
643 f"The stored energy variable {self.indicator_module_config.correct_costs.stored_energy_variable} is not defined in baseline mpc config. "
644 f"It must be defined in the base MPC model and config as output if the correction of costs is enabled."
645 )
647 # raise warning if unsupported collocation method is used and change to supported method
648 if self.baseline_mpc_module_config.optimization_backend["discretization_options"]["collocation_method"] != "legendre":
649 self.logger.warning(f'Collocation method {self.baseline_mpc_module_config.optimization_backend["discretization_options"]["collocation_method"]} is not supported. '
650 f'Switching to method legendre.')
651 self.baseline_mpc_module_config.optimization_backend["discretization_options"]["collocation_method"] = "legendre"
652 self.pos_flex_mpc_module_config.optimization_backend["discretization_options"]["collocation_method"] = "legendre"
653 self.neg_flex_mpc_module_config.optimization_backend["discretization_options"]["collocation_method"] = "legendre"
655 #time data validations
656 flex_times = {
657 glbs.PREP_TIME: self.flex_config.prep_time,
658 glbs.MARKET_TIME: self.flex_config.market_time,
659 glbs.FLEX_EVENT_DURATION: self.flex_config.flex_event_duration
660 }
661 mpc_times = {
662 glbs.TIME_STEP: self.baseline_mpc_module_config.time_step,
663 glbs.PREDICTION_HORIZON: self.baseline_mpc_module_config.prediction_horizon
664 }
665 # total time length check (prep+market+flex_event)
666 if sum(flex_times.values()) > mpc_times["time_step"] * mpc_times["prediction_horizon"]:
667 raise ConfigurationError(f'Market time + prep time + flex event duration can not exceed the prediction horizon.')
668 # market time val check
669 if self.flex_config.market_config:
670 if flex_times["market_time"] != mpc_times["time_step"]:
671 raise ConfigurationError(f'Market time must be equal to the time step.')
672 # check for divisibility of flex_times by time_step
673 for name, value in flex_times.items():
674 if value % mpc_times["time_step"] != 0:
675 raise ConfigurationError(f'{name} is not a multiple of the time step. Please redefine.')
676 # raise warning if parameter value in flex indicator module config differs from value in flex config/ baseline mpc module config
677 for parameter in self.indicator_module_config.parameters:
678 if parameter.value is not None:
679 if parameter.name in flex_times:
680 flex_value = flex_times[parameter.name]
681 if parameter.value != flex_value:
682 self.logger.warning(f'Value mismatch for {parameter.name} in flex config (field) and indicator module config (parameter). '
683 f'Flex config value will be used.')
684 elif parameter.name in mpc_times:
685 mpc_value = mpc_times[parameter.name]
686 if parameter.value != mpc_value:
687 self.logger.warning(f'Value mismatch for {parameter.name} in baseline MPC module config (field) and indicator module config (parameter). '
688 f'Baseline MPC module config value will be used.')
690 def adapt_sim_results_path(self, simulator_agent_config: Union[str, Path]) -> dict:
691 """
692 Optional helper function to adapt file path for simulator results in sim config
693 so that sim results land in the same results directory as flex results.
694 Args:
695 simulator_agent_config (Union[str, Path]): Path to the simulator agent config JSON file.
697 Returns:
698 dict: The updated simulator config with the modified result file path.
700 Raises:
701 FileNotFoundError: If the specified config file does not exist.
703 """
704 # open config and extract sim module
705 with open(simulator_agent_config, "r") as f:
706 sim_config = json.load(f)
707 sim_module_config = next(
708 (module for module in sim_config["modules"] if module["type"] == "simulator"),
709 None
710 )
711 # convert filename string to path and extract the name
712 sim_file_name = Path(sim_module_config["result_filename"]).name
713 # set results path so that sim results lands in same directory as flex result CSVs
714 sim_module_config["result_filename"] = str(self.flex_config.results_directory / sim_file_name)
715 return sim_config