agentlib_flexquant.modules package

Submodules

agentlib_flexquant.modules.baseline_mpc module

Defines MPC and MINLP-MPC for baseline flexibility quantification.

class agentlib_flexquant.modules.baseline_mpc.FlexibilityBaselineMINLPMPC(config, agent)

Bases: MINLPMPC

MINLP-MPC for baseline flexibility quantification with mixed-integer optimization.

pre_computation_hook()

Calculate relative start and end times for flexibility provision.

When in provision mode, computes the relative timing for flexibility events based on the external power profile timestamps and current environment time.

set_actuation(solution: Results)

Takes the solution from optimization backend and sends the first step to AgentVariables.

set_output(solution)

Takes the solution from optimization backend and sends it to AgentVariables.

sim_flex_model(solution)

simulate the flex model over the preditcion horizon and save results

pydantic model agentlib_flexquant.modules.baseline_mpc.FlexibilityBaselineMINLPMPCConfig

Bases: MINLPMPCConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• casadi_sim_time_step (int)

• full_controls (list[agentlib.core.datamodels.AgentVariable])

• power_variable_name (str)

• storage_variable_name (str | None)

Validators:

field casadi_sim_time_step: int = 0

Time step for simulation with Casadi simulator. Value is read from FlexQuantConfig

field full_controls: list[agentlib.core.datamodels.AgentVariable] = []

field power_variable_name: str = None

Name of the power variable in the baseline mpc model.

field storage_variable_name: str | None = None

Name of the storage variable in the baseline mpc model.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

class agentlib_flexquant.modules.baseline_mpc.FlexibilityBaselineMPC(config, agent)

Bases: MPC

MPC for baseline flexibility quantification.

pre_computation_hook()

Calculate relative start and end times for flexibility provision.

When in provision mode, computes the relative timing for flexibility events based on the external power profile timestamps and current environment time.

set_actuation(solution: Results)

Takes the solution from optimization backend and sends the first step to AgentVariables.

set_output(solution)

Takes the solution from optimization backend and sends it to AgentVariables.

sim_flex_model(solution)

simulate the flex model over the preditcion horizon and save results

pydantic model agentlib_flexquant.modules.baseline_mpc.FlexibilityBaselineMPCConfig

Bases: MPCConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• casadi_sim_time_step (int)

• full_controls (list[agentlib.core.datamodels.AgentVariable])

• power_variable_name (str)

• storage_variable_name (str | None)

Validators:

field casadi_sim_time_step: int = 0

Time step for simulation with Casadisimulator. Value is read from FlexQuantConfig

field full_controls: list[agentlib.core.datamodels.AgentVariable] = []

field power_variable_name: str = None

Name of the power variable in the baseline mpc model.

field storage_variable_name: str | None = None

Name of the storage variable in the baseline mpc model.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib_flexquant.modules.flexibility_indicator module

Flexibility indicator module for calculating and distributing energy flexibility offers.

This module processes power and energy profiles from baseline and shadow MPCs to calculate flexibility KPIs, validate profile consistency, and generate flexibility offers for energy markets. It handles both positive and negative flexibility with optional cost calculations and energy storage corrections.

class agentlib_flexquant.modules.flexibility_indicator.FlexibilityIndicatorModule(*args, **kwargs)

Bases: BaseModule

Module for calculating flexibility KPIs and generating flexibility offers from MPC power/energy profiles.

calc_and_send_offer()

Calculate the flexibility KPIs for current predictions, send the flex offer and set the outputs, write and save the results.

callback(inp, name)

Handle incoming data by storing power/energy profiles and triggering flexibility calculations when all required inputs are available.

check_power_end_deviation(tol: float)

Calculate the deviation of the final value of the power profiles and warn the user if it exceeds the tolerance.

cleanup_results()

Remove the existing result files.

data: FlexibilityData

get_results() → DataFrame | None

Open results file of flexibility_indicator.py.

process()

Yield control to the simulation environment and wait for events.

register_callbacks()

send_flex_offer(name: str, base_power_profile: Series, pos_diff_profile: Series, pos_price: float, neg_diff_profile: Series, neg_price: float, timestamp: float = None)

Send a flex offer as an agent Variable.

The first offer is dismissed, since the different MPCs need one time step to fully initialize.

Parameters:

• name – name of the agent variable

• base_power_profile – time series of power from baseline mpc

• pos_diff_profile – power profile for the positive difference (base-pos)

• grid (in flexibility event time) –

• pos_price – price for positive flexibility

• neg_diff_profile – power profile for the negative difference (neg-base)

• grid –

• neg_price – price for negative flexibility

• timestamp – the time offer was generated

write_results(df: DataFrame, ts: float, n: int) → DataFrame

Write every data of variables in self.var_list in an DataFrame.

DataFrame will be updated every time step

Parameters:

• df – DataFrame which is initialised as an empty DataFrame with columns

• self.var_list (according to) –

• ts – time step

• n – number of time steps during prediction horizon

Returns:

DataFrame with results of every variable in self.var_list

pydantic model agentlib_flexquant.modules.flexibility_indicator.FlexibilityIndicatorModuleConfig

Bases: BaseModuleConfig

Configuration for flexibility indicator module with power/energy inputs, KPI outputs, and cost calculation settings.

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• calculate_costs (agentlib_flexquant.modules.flexibility_indicator.InputsForCalculateFlexCosts)

• correct_costs (agentlib_flexquant.modules.flexibility_indicator.InputsForCorrectFlexCosts)

• inputs (list[agentlib.core.datamodels.AgentVariable])

• integration_method (Literal['linear', 'constant'])

• outputs (list[agentlib.core.datamodels.AgentVariable])

• parameters (list[agentlib.core.datamodels.AgentVariable])

• power_unit (str)

• price_variable (str)

• results_file (pathlib.Path | None)

• save_results (bool | None)

• shared_variable_fields (list[str])

Validators:

• check_results_file_extension » all fields

• check_valid_fields » shared_variable_fields

field calculate_costs: InputsForCalculateFlexCosts = InputsForCalculateFlexCosts(use_constant_electricity_price=False, calculate_flex_costs=True, const_electricity_price=nan)

Validated by:

• check_results_file_extension

field correct_costs: InputsForCorrectFlexCosts = InputsForCorrectFlexCosts(enable_energy_costs_correction=False, absolute_power_deviation_tolerance=0.1, stored_energy_variable=None)

Validated by:

• check_results_file_extension

field inputs: list[agentlib.core.datamodels.AgentVariable] = [AgentVariable(name='_P_el_base', type='pd.Series', timestamp=None, unit='W', description='The power input to the system', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_P_el_base', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='_P_el_neg', type='pd.Series', timestamp=None, unit='W', description='The power input to the system', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_P_el_neg', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='_P_el_pos', type='pd.Series', timestamp=None, unit='W', description='The power input to the system', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_P_el_pos', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='_E_stored_base', type='pd.Series', timestamp=None, unit='kWh', description='Energy stored in the system w.r.t. 0K', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_E_stored_base', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='_E_stored_neg', type='pd.Series', timestamp=None, unit='kWh', description='Energy stored in the system w.r.t. 0K', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_E_stored_neg', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='_E_stored_pos', type='pd.Series', timestamp=None, unit='kWh', description='Energy stored in the system w.r.t. 0K', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_E_stored_pos', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

Validated by:

• check_results_file_extension

field integration_method: Literal['linear', 'constant'] = 'linear'

Method set to integrate series variable

Validated by:

• check_results_file_extension

field outputs: list[agentlib.core.datamodels.AgentVariable] = [AgentVariable(name='FlexibilityOffer', type='FlexOffer', timestamp=None, unit='Not defined', description='Not defined', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='FlexibilityOffer', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_power_flex_full', type='pd.Series', timestamp=None, unit='W', description='Negative power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_power_flex_full', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_power_flex_full', type='pd.Series', timestamp=None, unit='W', description='Positive power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_power_flex_full', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_power_flex_offer', type='pd.Series', timestamp=None, unit='W', description='Negative power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_power_flex_offer', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_power_flex_offer', type='pd.Series', timestamp=None, unit='W', description='Positive power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_power_flex_offer', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_power_flex_offer_min', type='float', timestamp=None, unit='W', description='Minimum of negative power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_power_flex_offer_min', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_power_flex_offer_min', type='float', timestamp=None, unit='W', description='Minimum of positive power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_power_flex_offer_min', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_power_flex_offer_max', type='float', timestamp=None, unit='W', description='Maximum of negative power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_power_flex_offer_max', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_power_flex_offer_max', type='float', timestamp=None, unit='W', description='Maximum of positive power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_power_flex_offer_max', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_power_flex_offer_avg', type='float', timestamp=None, unit='W', description='Average of negative power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_power_flex_offer_avg', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_power_flex_offer_avg', type='float', timestamp=None, unit='W', description='Average of positive power flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_power_flex_offer_avg', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_power_flex_within_boundary', type='bool', timestamp=None, unit='-', description='Variable indicating whether the baseline power and flex power align at the horizon end', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_power_flex_within_boundary', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_power_flex_within_boundary', type='bool', timestamp=None, unit='-', description='Variable indicating whether the baseline power and flex power align at the horizon end', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_power_flex_within_boundary', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_energy_flex', type='float', timestamp=None, unit='kWh', description='Negative energy flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_energy_flex', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_energy_flex', type='float', timestamp=None, unit='kWh', description='Positive energy flexibility', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_energy_flex', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_costs', type='float', timestamp=None, unit='ct', description='Saved costs due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_costs', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_costs', type='float', timestamp=None, unit='ct', description='Saved costs due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_costs', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_corrected_costs', type='float', timestamp=None, unit='ct', description='Corrected saved costs due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_corrected_costs', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_corrected_costs', type='float', timestamp=None, unit='ct', description='Corrected saved costs due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_corrected_costs', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_costs_rel', type='float', timestamp=None, unit='ct/kWh', description='Saved costs due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_costs_rel', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_costs_rel', type='float', timestamp=None, unit='ct/kWh', description='Saved costs due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_costs_rel', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='negative_corrected_costs_rel', type='float', timestamp=None, unit='ct/kWh', description='Corrected saved costs per energy due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='negative_corrected_costs_rel', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='positive_corrected_costs_rel', type='float', timestamp=None, unit='ct/kWh', description='Corrected saved costs per energy due to baseline', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='positive_corrected_costs_rel', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

Validated by:

• check_results_file_extension

field parameters: list[agentlib.core.datamodels.AgentVariable] = [AgentVariable(name='prep_time', type=None, timestamp=None, unit='s', description='Preparation time', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='prep_time', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='market_time', type=None, timestamp=None, unit='s', description='Market time', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='market_time', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='flex_event_duration', type=None, timestamp=None, unit='s', description='time to switch objective', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='flex_event_duration', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='time_step', type=None, timestamp=None, unit='s', description='timestep of the mpc solution', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='time_step', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='prediction_horizon', type=None, timestamp=None, unit='-', description='prediction horizon of the mpc solution', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='prediction_horizon', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='collocation_time_grid', type=None, timestamp=None, unit='Not defined', description='Time grid of the mpc model output', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='collocation_time_grid', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

Validated by:

• check_results_file_extension

field power_unit: str = 'kW'

Unit of the power variable

Validated by:

• check_results_file_extension

field price_variable: str = 'c_pel'

Name of the price variable sent by a predictor

Validated by:

• check_results_file_extension

field results_file: Path | None = PosixPath('flexibility_indicator.csv')

User specified results file name

Validated by:

• check_results_file_extension

field save_results: bool | None = True

Validated by:

• check_results_file_extension

field shared_variable_fields: list[str] = ['outputs']

Validated by:

• check_results_file_extension

• check_valid_fields

validator check_results_file_extension  »  all fields

Validate that results_file has a .csv extension.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

pydantic model agentlib_flexquant.modules.flexibility_indicator.InputsForCalculateFlexCosts

Bases: BaseModel

Configuration for flexibility cost calculation with optional constant pricing.

Fields:

• calculate_flex_costs (bool)

• const_electricity_price (float)

• use_constant_electricity_price (bool)

Validators:

• validate_constant_price » all fields

field calculate_flex_costs: bool = True

Calculate the flexibility cost

Validated by:

• validate_constant_price

field const_electricity_price: float = nan

constant electricity price in ct/kWh

Validated by:

• validate_constant_price

field use_constant_electricity_price: bool = False

Use constant electricity price

Validated by:

• validate_constant_price

validator validate_constant_price  »  all fields

Validate that a valid constant electricity price is provided when constant pricing is enabled.

pydantic model agentlib_flexquant.modules.flexibility_indicator.InputsForCorrectFlexCosts

Bases: BaseModel

Configuration for flexibility cost correction.

Fields:

• absolute_power_deviation_tolerance (float)

• enable_energy_costs_correction (bool)

• stored_energy_variable (str | None)

field absolute_power_deviation_tolerance: float = 0.1

Absolute tolerance in kW within which no warning is thrown

field enable_energy_costs_correction: bool = False

Variable determining whether to correct the costs of the flexible energy Define the variable for stored electrical energy in the base MPC model and config as output if the correction of costs is enabled

field stored_energy_variable: str | None = None

Name of the variable representing the stored electrical energy in the baseline config

agentlib_flexquant.modules.flexibility_market module

class agentlib_flexquant.modules.flexibility_market.FlexibilityMarketModule(*, config: dict, agent: Agent)

Bases: BaseModule

Class to emulate flexibility market. Receives flex offers and accepts these.

abs_flex_event_end: int | float = 0

cleanup_results()

Remove the results if they already exist.

custom_flexibility_callback(inp: AgentVariable, name: str)

Placeholder for a custom flexibility callback.

dummy_callback(inp: AgentVariable, name: str)

Dummy function that is included, when market type is not specified.

flex_offer_df: DataFrame = None

get_results() → DataFrame | None

Open results file of flexibility_indicators.py.

process()

This abstract method must be implemented in order to sync the module with the other processes of the agent and the whole MAS.

random_flexibility_callback(inp: AgentVariable, name: str)

When a flexibility offer is sent, this function is called.

The offer is accepted randomly. The factor self.offer_acceptance_rate determines the random factor for offer acceptance. self.pos_neg_rate is the random factor for the direction of the flexibility. A higher rate means that more positive offers will be accepted.

Constraints:

cooldown: during $cooldown steps after a flexibility event no offer is accepted minimum_average_flex: min amount of flexibility to be accepted, to account for the model error

register_callbacks()

set_random_seed(random_seed: int)

Set the random seed for reproducibility.

single_flexibility_callback(inp: AgentVariable, name: str)

Callback to activate a single, predefined flexibility offer.

write_results(offer: FlexOffer)

Save the flex offer results depending on the config.

pydantic model agentlib_flexquant.modules.flexibility_market.FlexibilityMarketModuleConfig

Bases: BaseModuleConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• inputs (list[agentlib.core.datamodels.AgentVariable])

• market_specs (agentlib_flexquant.data_structures.market.MarketSpecifications)

• outputs (list[agentlib.core.datamodels.AgentVariable])

• parameters (list[agentlib.core.datamodels.AgentVariable])

• results_file (pathlib.Path | None)

• save_results (bool | None)

• shared_variable_fields (list[str])

Validators:

• check_results_file_extension » all fields

• check_valid_fields » shared_variable_fields

field inputs: list[agentlib.core.datamodels.AgentVariable] = [AgentVariable(name='FlexibilityOffer', type=None, timestamp=None, unit='Not defined', description='Not defined', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='FlexibilityOffer', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

Validated by:

• check_results_file_extension

field market_specs: MarketSpecifications [Required]

Validated by:

• check_results_file_extension

field outputs: list[agentlib.core.datamodels.AgentVariable] = [AgentVariable(name='_P_external', type=None, timestamp=None, unit='Not defined', description='External Power IO', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='_P_external', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='rel_start', type=None, timestamp=None, unit='Not defined', description='relative start time of the flexibility event', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='rel_start', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='rel_end', type=None, timestamp=None, unit='Not defined', description='relative end time of the flexibility event', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='rel_end', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='in_provision', type=None, timestamp=None, unit='Not defined', description='Set if the system is in provision', ub=inf, lb=-inf, clip=False, allowed_values=[], value=False, alias='in_provision', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

Validated by:

• check_results_file_extension

field parameters: list[agentlib.core.datamodels.AgentVariable] = [AgentVariable(name='collocation_time_grid', type=None, timestamp=None, unit='Not defined', description='Time grid of the mpc model output', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='collocation_time_grid', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None), AgentVariable(name='time_step', type=None, timestamp=None, unit='s', description='Time step of the mpc', ub=inf, lb=-inf, clip=False, allowed_values=[], value=None, alias='time_step', source=Source(agent_id=None, module_id=None), shared=None, rdf_class=None)]

Validated by:

• check_results_file_extension

field results_file: Path | None = PosixPath('flexibility_market.csv')

User specified results file name

Validated by:

• check_results_file_extension

field save_results: bool | None = True

Validated by:

• check_results_file_extension

field shared_variable_fields: list[str] = ['outputs']

Validated by:

• check_results_file_extension

• check_valid_fields

validator check_results_file_extension  »  all fields

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

agentlib_flexquant.modules.shadow_mpc module

Defines shadow MPC and MINLP-MPC for positive/negative flexibility quantification.

class agentlib_flexquant.modules.shadow_mpc.FlexibilityShadowMINLPMPC(*args, **kwargs)

Bases: MINLPMPC

Shadow MINLP-MPC for calculating positive/negatives flexibility offers.

calc_flex_callback(inp: AgentVariable, name: str)

Set the control trajectories before calculating the flexibility offer.

self.model should account for flexibility in its cost function

process()

This abstract method must be implemented in order to sync the module with the other processes of the agent and the whole MAS.

register_callbacks()

Registers the init_optimization callback to all parameters which cannot be changed without recreating the optimization problem.

set_output(solution)

Takes the solution from optimization backend and sends it to AgentVariables.

sim_flex_model(solution)

simulate the flex model over the preditcion horizon and save results

pydantic model agentlib_flexquant.modules.shadow_mpc.FlexibilityShadowMINLPMPCConfig

Bases: MINLPMPCConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• casadi_sim_time_step (int)

• power_variable_name (str)

• storage_variable_name (str | None)

Validators:

field casadi_sim_time_step: int = 0

Time step for simulation with Casadi simulator. Value is read from FlexQuantConfig

field power_variable_name: str = None

Name of the power variable in the shadow mpc model.

field storage_variable_name: str | None = None

Name of the storage variable in the shadow mpc model.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

class agentlib_flexquant.modules.shadow_mpc.FlexibilityShadowMPC(*args, **kwargs)

Bases: MPC

Shadow MPC for calculating positive/negative flexibility offers.

calc_flex_callback(inp: AgentVariable, name: str)

Set the control trajectories before calculating the flexibility offer.

self.model should account for flexibility in its cost function.

process()

This abstract method must be implemented in order to sync the module with the other processes of the agent and the whole MAS.

register_callbacks()

Registers the init_optimization callback to all parameters which cannot be changed without recreating the optimization problem.

set_output(solution)

Takes the solution from optimization backend and sends it to AgentVariables.

sim_flex_model(solution)

simulate the flex model over the preditcion horizon and save results

pydantic model agentlib_flexquant.modules.shadow_mpc.FlexibilityShadowMPCConfig

Bases: MPCConfig

Config:

• arbitrary_types_allowed: bool = True

• validate_assignment: bool = True

• extra: str = forbid

• frozen: bool = True

Fields:

• casadi_sim_time_step (int)

• power_variable_name (str)

• storage_variable_name (str | None)

Validators:

field casadi_sim_time_step: int = 0

Time step for simulation with Casadi simulator. Value is read from FlexQuantConfig

field power_variable_name: str = None

Name of the power variable in the shadow mpc model.

field storage_variable_name: str | None = None

Name of the storage variable in the shadow mpc model.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.