from agentlib_mpc.modules import mpc_full, minlp_mpc
from agentlib_flexquant.utils.data_handling import strip_multi_index, fill_nans, MEAN, INTERPOLATE
from agentlib_flexquant.data_structures.globals import (
full_trajectory_prefix,
full_trajectory_suffix,
)
from typing import Dict, Union
from agentlib.core.datamodels import AgentVariable
[docs]
class FlexibilityShadowMPC(mpc_full.MPC):
config: mpc_full.MPCConfig
def __init__(self, *args, **kwargs):
# create instance variable
self._full_controls: Dict[str, Union[AgentVariable, None]] = {}
super().__init__(*args, **kwargs)
[docs]
def register_callbacks(self):
for control_var in self.config.controls:
self.agent.data_broker.register_callback(
name=f"{full_trajectory_prefix}{control_var.name}{full_trajectory_suffix}",
alias=f"{full_trajectory_prefix}{control_var.name}{full_trajectory_suffix}",
callback=self.calc_flex_callback,
)
for input_var in self.config.inputs:
if input_var.name.replace(full_trajectory_prefix, "", 1).replace(
full_trajectory_suffix, ""
) in [control_var.name for control_var in self.config.controls]:
self._full_controls[input_var.name] = input_var
super().register_callbacks()
[docs]
def calc_flex_callback(self, inp, name):
"""set the control trajectories before calculating the flexibility offer.
self.model should account for flexibility in its cost function
"""
# during provision dont calculate flex
if self.get("in_provision").value:
return
# do not trigger callback on self set variables
if self.agent.config.id == inp.source.agent_id:
return
vals = strip_multi_index(inp.value)
if vals.isna().any():
vals = fill_nans(series=vals, method=MEAN)
# the MPC Predictions starts at t=env.now not t=0
vals.index += self.env.time
self._full_controls[name].value = vals
self.set(name, vals)
# make sure all controls are set
if all(x.value is not None for x in self._full_controls.values()):
self.do_step()
for name in self._full_controls.keys():
self._full_controls[name].value = None
[docs]
def process(self):
# the shadow mpc should only be run after the results of the baseline are sent
yield self.env.event()
[docs]
class FlexibilityShadowMINLPMPC(minlp_mpc.MINLPMPC):
config: minlp_mpc.MINLPMPCConfig
def __init__(self, *args, **kwargs):
# create instance variable
self._full_controls: Dict[str, Union[AgentVariable, None]] = {}
super().__init__(*args, **kwargs)
[docs]
def register_callbacks(self):
for control_var in self.config.controls + self.config.binary_controls:
self.agent.data_broker.register_callback(
name=f"{full_trajectory_prefix}{control_var.name}{full_trajectory_suffix}",
alias=f"{full_trajectory_prefix}{control_var.name}{full_trajectory_suffix}",
callback=self.calc_flex_callback,
)
for input_var in self.config.inputs:
if input_var.name.replace(full_trajectory_prefix, "", 1).replace(
full_trajectory_suffix, ""
) in [control_var.name for control_var in self.config.controls + self.config.binary_controls]:
self._full_controls[input_var.name] = input_var
super().register_callbacks()
[docs]
def calc_flex_callback(self, inp, name):
"""set the control trajectories before calculating the flexibility offer.
self.model should account for flexibility in its cost function
"""
# during provision dont calculate flex
if self.get("in_provision").value:
return
# do not trigger callback on self set variables
if self.agent.config.id == inp.source.agent_id:
return
vals = strip_multi_index(inp.value)
if vals.isna().any():
vals = fill_nans(vals, method=MEAN)
# the MPC Predictions starts at t=env.now not t=0
vals.index += self.env.time
self._full_controls[name].value = vals
self.set(name, vals)
# make sure all controls are set
if all(x.value is not None for x in self._full_controls.values()):
self.do_step()
for name in self._full_controls.keys():
self._full_controls[name].value = None
[docs]
def process(self):
# the shadow mpc should only be run after the results of the baseline are sent
yield self.env.event()