Skip to content

Change callback structure MPCs #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lwaer merged 1 commit into from
Jan 19, 2026
Merged

Change callback structure MPCs #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 61 additions & 39 deletions agentlib_flexquant/data_structures/flex_results.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,13 +66,14 @@ def load_market(file_path: Union[str, FilePath]) -> pd.DataFrame:
class Results:
"""
Loads the results for the baseline, positive and negative flexibility,
the indicator, market and simulator results/data. Additionally the MPC stats are loaded.
the indicator, market and simulator results/data. Additionally the MPC stats
are loaded.

Results can be loaded either from a user-specified custom base path or from the
(default) base path specified in the flex config.

Loaded results are stored in pandas DataFrames which can be used for further processing,
e.g. plotting and analysis.
Loaded results are stored in pandas DataFrames which can be used for further
processing, e.g. plotting and analysis.
"""

# Configs:
Expand Down Expand Up @@ -186,14 +187,16 @@ def _get_config_filenames(self):
if self.flex_config.market_config:
if isinstance(self.flex_config.market_config, Union[str, Path]):
self.config_filename_market = load_config.load_config(
config=self.flex_config.market_config, config_type=FlexibilityMarketConfig
config=self.flex_config.market_config,
config_type=FlexibilityMarketConfig
).name_of_created_file
else: # is dict
self.config_filename_market = FlexibilityMarketConfig.model_validate(
self.flex_config.market_config).name_of_created_file

def _load_agent_module_configs(self):
"""Load agent and module configs."""
files_found = []
for file_path in Path(self.flex_config.flex_files_directory).rglob(
"*.json"
):
Expand All @@ -203,26 +206,32 @@ def _load_agent_module_configs(self):
)
self.baseline_module_config = cmng.get_module(
config=self.baseline_agent_config,
module_type=self._get_flexquant_mpc_module_type(self.baseline_agent_config),
module_type=
self._get_flexquant_mpc_module_type(self.baseline_agent_config),
)
files_found.append(self.config_filename_baseline)

elif file_path.name in self.config_filename_pos_flex:
self.pos_flex_agent_config = load_config.load_config(
config=file_path, config_type=AgentConfig
)
self.pos_flex_module_config = cmng.get_module(
config=self.pos_flex_agent_config,
module_type=self._get_flexquant_mpc_module_type(self.pos_flex_agent_config),
module_type=
self._get_flexquant_mpc_module_type(self.pos_flex_agent_config),
)
files_found.append(self.config_filename_pos_flex)

elif file_path.name in self.config_filename_neg_flex:
self.neg_flex_agent_config = load_config.load_config(
config=file_path, config_type=AgentConfig
)
self.neg_flex_module_config = cmng.get_module(
config=self.neg_flex_agent_config,
module_type=self._get_flexquant_mpc_module_type(self.neg_flex_agent_config),
module_type=
self._get_flexquant_mpc_module_type(self.neg_flex_agent_config),
)
files_found.append(self.config_filename_neg_flex)

elif file_path.name in self.config_filename_indicator:
self.indicator_agent_config = load_config.load_config(
Expand All @@ -232,6 +241,7 @@ def _load_agent_module_configs(self):
config=self.indicator_agent_config,
module_type=cmng.INDICATOR_CONFIG_TYPE,
)
files_found.append(self.config_filename_indicator)

elif (
self.flex_config.market_config
Expand All @@ -243,16 +253,18 @@ def _load_agent_module_configs(self):
self.market_module_config = cmng.get_module(
config=self.market_agent_config, module_type=cmng.MARKET_CONFIG_TYPE
)
else:
import warnings
warnings.warn(f"The file {file_path.name} is not listed in any of the known "
f"filenames of the created files: {self.config_filename_baseline}, "
f"{self.config_filename_pos_flex}, {self.config_filename_neg_flex}, "
f"{self.config_filename_indicator}" +
(f", {self.config_filename_market}" if
self.flex_config.market_config else "") +
". This can cause trouble in loading the files. "
"Please check the filenames.")
files_found.append(self.config_filename_market)
files_needed = [self.config_filename_baseline,
self.config_filename_pos_flex, self.config_filename_neg_flex,
self.config_filename_indicator]
if self.flex_config.market_config:
files_needed.append(self.config_filename_market)
difference = list(set(files_needed) - set(files_found))
if difference:
import warnings
warnings.warn(f"The files {difference} have not been found in the "
f"given Path. This will most likely cause problems "
f"later on. Please check the filenames.")

def _load_simulator_config(self, simulator_agent_config):
"""Load simulator agent and module config separately.
Expand Down Expand Up @@ -281,11 +293,12 @@ def _load_simulator_config(self, simulator_agent_config):
self.simulator_agent_config = AgentConfig.model_validate(sim_config)
# instantiate sim module config by skipping validation for result_filename
# to prevent file deletion, if overwrite_result_file in sim config is true
self.simulator_module_config = self.create_simulator_config_with_skipped_validation(
sim_config_class=SimulatorConfig,
sim_config=sim_module_config,
skip_fields=["result_filename"],
)
self.simulator_module_config = (
self.create_simulator_config_with_skipped_validation(
sim_config_class=SimulatorConfig,
sim_config=sim_module_config,
skip_fields=["result_filename"],
))

def _get_flexquant_mpc_module_type(self, agent_config: AgentConfig) -> str:
"""Get the mpc module type from agent_config.
Expand All @@ -300,12 +313,15 @@ def _get_flexquant_mpc_module_type(self, agent_config: AgentConfig) -> str:

"""
for module in agent_config.modules:
if module['type'] in [cmng.BASELINEMPC_CONFIG_TYPE, cmng.BASELINEMINLPMPC_CONFIG_TYPE,
cmng.SHADOWMPC_CONFIG_TYPE, cmng.SHADOWMINLPMPC_CONFIG_TYPE]:
if module['type'] in [cmng.BASELINEMPC_CONFIG_TYPE,
cmng.BASELINEMINLPMPC_CONFIG_TYPE,
cmng.SHADOWMPC_CONFIG_TYPE,
cmng.SHADOWMINLPMPC_CONFIG_TYPE]:
return module['type']

raise ModuleNotFoundError(f'There is no matching mpc module type in Agentlib_FlexQuant for '
f'modules in agent {agent_config.id}.')
raise ModuleNotFoundError(f'There is no matching mpc module type in '
f'Agentlib_FlexQuant for modules in agent '
f'{agent_config.id}.')

def _resolve_sim_results_path(
self, sim_result_filename: str, results_path: Union[str, Path]
Expand Down Expand Up @@ -340,8 +356,8 @@ def _resolve_sim_results_path(
if not sim_results_path.is_absolute() and sim_results_path.exists():
return sim_results_path

# Strategy 3: Try in results directory (handles both relative paths and just filenames)
# (fallback for helper function usage)
# Strategy 3: Try in results directory (handles both relative paths
# and just filenames) (fallback for helper function usage)
results_dir_path = results_path / sim_results_path.name
if results_dir_path.exists():
return results_dir_path
Expand Down Expand Up @@ -514,8 +530,9 @@ def get_intersection_mpcs_sim(self) -> dict[str, dict[str, str]]:
"""Get the intersection of the MPCs and the simulator variables.

Returns:
dictionary with the following structure: Key: variable alias (from baseline)
Value: {module id: variable name}
dictionary with the following structure:
Key: variable alias (from baseline)
Value: {module id: variable name}

"""
id_alias_name_dict = {}
Expand Down Expand Up @@ -552,11 +569,13 @@ def create_simulator_config_with_skipped_validation(
sim_config: Dict[str, Any],
skip_fields: Optional[list[str]] = None,
) -> SimulatorConfig:
"""Create a Pydantic model instance while skipping validation for specified fields.
"""Create a Pydantic model instance while skipping validation for
specified fields.

This function allows partial validation of a model's config dictionary by validating
all fields except those listed in `skip_fields`. Skipped fields are set on the instance
after construction without triggering their validators.
This function allows partial validation of a model's config dictionary
by validating all fields except those listed in `skip_fields`.
Skipped fields are set on the instance after construction without
triggering their validators.

Args:
sim_config_class: The Pydantic model class to instantiate.
Expand All @@ -565,17 +584,20 @@ def create_simulator_config_with_skipped_validation(
These fields will be manually set after instantiation.

Returns:
SimulatorConfig: An instance of the model_class with validated and skipped fields assigned.
SimulatorConfig: An instance of the model_class with validated and
skipped fields assigned.

"""
if skip_fields is None:
skip_fields = []
# Separate data into validated and skipped fields
validated_fields = {
field: value for field, value in sim_config.items() if field not in skip_fields
field: value for field, value in sim_config.items() if
field not in skip_fields
}
skipped_fields = {
field: value for field, value in sim_config.items() if field in skip_fields
field: value for field, value in sim_config.items() if
field in skip_fields
}
# Create instance with validation for non-skipped fields
if validated_fields:
Expand All @@ -596,8 +618,8 @@ def create_simulator_config_with_skipped_validation(
def __deepcopy__(self, memo: Dict[int, Any]) -> "Results":
"""Custom deepcopy implementation that handles Pydantic models with bypassed
validation.
Needed, if an Results object should be copied with copy.deepcopy, without
deleting the simulator results due to its üydantic validators.
Needed, if a Results object should be copied with copy.deepcopy, without
deleting the simulator results due to its pydantic validators.
"""
# Create a new instance of the same class
new_instance = self.__class__.__new__(self.__class__)
Expand Down
36 changes: 22 additions & 14 deletions agentlib_flexquant/data_structures/flexquant.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
"description",
"unit",
"clip",
"shared",
"interpolation_method",
"allowed_values",
]
Expand All @@ -47,11 +46,13 @@ def assign_weights_to_flex(self):
"""Validate flexibility cost function fields and assign weights to them."""
if self.pos_flex is None:
raise ValueError(
"Missing required field: 'pos_flex' specifying the pos flex cost function."
"Missing required field: 'pos_flex' specifying the pos flex "
"cost function."
)
if self.neg_flex is None:
raise ValueError(
"Missing required field: 'neg_flex' specifying the neg flex cost function."
"Missing required field: 'neg_flex' specifying the neg flex "
"cost function."
)
if self.weights:
self.pos_flex.weights = self.weights
Expand Down Expand Up @@ -143,10 +144,10 @@ class FlexQuantConfig(BaseModel):
)
casadi_sim_time_step: int = Field(
default=0,
description="Simulate over the prediction horizon with a defined resolution using Casadi "
"simulator. "
"Only use it when the power depends on the states. Don't use it when power "
"itself is the control variable."
description="Simulate over the prediction horizon with a defined resolution "
"using Casadi simulator. "
"Only use it when the power depends on the states. "
"Don't use it when power itself is the control variable."
"Set to 0 to skip simulation",
)
flex_base_directory_path: Optional[Path] = Field(
Expand All @@ -171,7 +172,8 @@ class FlexQuantConfig(BaseModel):

@model_validator(mode="after")
def check_config_file_extension(self):
"""Validate that the indicator and market config file paths have a '.json' extension.
"""Validate that the indicator and market config file paths have a '.json'
extension.

Raises:
ValueError: If either file does not have the expected '.json' extension.
Expand All @@ -182,15 +184,17 @@ def check_config_file_extension(self):
and self.indicator_config.suffix != ".json"
):
raise ValueError(
f"Invalid file extension for indicator config: '{self.indicator_config}'. "
f"Invalid file extension for indicator "
f"config: '{self.indicator_config}'. "
f"Expected a '.json' file."
)
if (
isinstance(self.market_config, Path)
and self.market_config.suffix != ".json"
):
raise ValueError(
f"Invalid file extension for market config: '{self.market_config}'. "
f"Invalid file extension for market "
f"config: '{self.market_config}'. "
f"Expected a '.json' file."
)
return self
Expand All @@ -204,12 +208,16 @@ def is_none_negative_integer(cls, value: int) -> int:

@model_validator(mode="after")
def adapt_paths_and_create_directory(self):
"""Adjust and ensure the directory structure for flex file generation and results storage.
"""Adjust and ensure the directory structure for flex file generation and
results storage.

This method:
- Updates `flex_files_directory` and `results_directory` paths, so they are relative to
the base flex directory, using only the directory names (ignoring any user-supplied paths).
- Creates the base, flex files, and results directories if they do not already exist.
- Updates `flex_files_directory` and `results_directory` paths, so they are
relative to
the base flex directory, using only the directory names (ignoring any
user-supplied paths).
- Creates the base, flex files, and results directories if they do not
already exist.

"""
# adapt paths and use only names for user supplied data
Expand Down
22 changes: 13 additions & 9 deletions agentlib_flexquant/data_structures/globals.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,13 @@
STORED_ENERGY_ALIAS_NEG = "_E_stored_neg"
STORED_ENERGY_ALIAS_POS = "_E_stored_pos"
full_trajectory_suffix: str = "_full"
full_trajectory_prefix: str = "_"
base_vars_to_communicate_suffix: str = "_base"
shadow_suffix: str = "_shadow"
COLLOCATION_TIME_GRID = 'collocation_time_grid'
PROVISION_VAR_NAME = "in_provision"
ACCEPTED_POWER_VAR_NAME = "_P_external"
RELATIVE_EVENT_START_TIME_VAR_NAME = "rel_start"
RELATIVE_EVENT_END_TIME_VAR_NAME = "rel_end"

# cost function in the shadow mpc. obj_std and obj_flex are to be evaluated according
# to user definition
Expand All @@ -51,19 +55,19 @@ def return_baseline_cost_function(power_variable: str, comfort_variable: str) ->
"""
if comfort_variable:
cost_func = (
"return ca.if_else(self.in_provision.sym, "
"ca.if_else(self.time < self.rel_start.sym, obj_std, "
"ca.if_else(self.time >= self.rel_end.sym, obj_std, "
f"return ca.if_else(self.{PROVISION_VAR_NAME}.sym, "
f"ca.if_else(self.time < self.{RELATIVE_EVENT_START_TIME_VAR_NAME}.sym, obj_std, "
f"ca.if_else(self.time >= self.{RELATIVE_EVENT_END_TIME_VAR_NAME}.sym, obj_std, "
f"sum([self.profile_deviation_weight*(self.{power_variable} - "
f"self._P_external)**2, "
f"self.{ACCEPTED_POWER_VAR_NAME})**2, "
f"self.{comfort_variable}**2 * self.profile_comfort_weight]))),obj_std)"
)
else:
cost_func = (
"return ca.if_else(self.in_provision.sym, "
"ca.if_else(self.time < self.rel_start.sym, obj_std, "
"ca.if_else(self.time >= self.rel_end.sym, obj_std, "
f"return ca.if_else(self.{PROVISION_VAR_NAME}.sym, "
f"ca.if_else(self.time < self.{RELATIVE_EVENT_START_TIME_VAR_NAME}.sym, obj_std, "
f"ca.if_else(self.time >= self.{RELATIVE_EVENT_END_TIME_VAR_NAME}.sym, obj_std, "
f"sum([self.profile_deviation_weight*(self.{power_variable} - "
f"self._P_external)**2]))),obj_std)"
f"self.{ACCEPTED_POWER_VAR_NAME})**2]))),obj_std)"
)
return cost_func
Loading