17 skip mpc

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.6.7
+- Add option to skip MPC calculation in given time intervals, e.g. during summer period
+
 ## 0.6.6
 - self.time available in mpc and ml mpc (not yet available for admm, minlp, etc)
 

agentlib_mpc/__init__.py

@@ -3,4 +3,5 @@
 
 from .modules import MODULE_TYPES
 from .models import MODEL_TYPES
-__version__ = "0.6.6"
+
+__version__ = "0.6.7"

agentlib_mpc/modules/__init__.py

@@ -66,4 +66,7 @@ def import_class(self):
     "mhe": ModuleImport(
         module_path="agentlib_mpc.modules.estimation.mhe", class_name="MHE"
     ),
+    "skip_mpc_intervals": ModuleImport(
+        module_path="agentlib_mpc.modules.deactivate_mpc", class_name="SkipMPCInIntervals"
+    )
 }

agentlib_mpc/modules/deactivate_mpc.py

@@ -0,0 +1,52 @@
+from agentlib.core import BaseModule, BaseModuleConfig, AgentVariable
+from pydantic import Field
+
+from agentlib_mpc import utils
+
+
+class SkipMPCInIntervalsConfig(BaseModuleConfig):
+    """
+    Config for a module which deactivates any MPC by sending the variable
+    `active` in the specified intervals.
+    """
+    intervals: list[tuple[float, float]] = Field(
+        default=[],
+        description="If environment time is within these intervals"
+    )
+    time_unit: utils.TimeConversionTypes = Field(
+        default="seconds",
+        description="Specifies the unit of the given "
+                    "`skip_mpc_in_intervals`, e.g. seconds or days."
+    )
+    active: AgentVariable = Field(
+        default=AgentVariable(name="active", description="MPC is active", type="bool", value=True, shared=False),
+        description="Variable used to activate or deactivate the MPC operation"
+    )
+    t_sample: float = Field(
+        default=60,
+        description="Sends the active variable every other t_sample"
+    )
+
+
+class SkipMPCInIntervals(BaseModule):
+    """
+    Module which deactivates any MPC by sending the variable
+    `active` in the specified intervals.
+    """
+    config: SkipMPCInIntervalsConfig
+
+    def process(self):
+        """Write the current data values into data_broker every t_sample"""
+        while True:
+            if utils.is_time_in_intervals(
+                    time=self.env.now / utils.TIME_CONVERSION[self.config.time_unit],
+                    intervals=self.config.intervals
+            ):
+                self.logger.debug("Current time is in skip_mpc_in_intervals, sending active=False to MPC")
+                self.set("active", False)
+            else:
+                self.set("active", True)
+            yield self.env.timeout(self.config.t_sample)
+
+    def register_callbacks(self):
+        """Don't do anything as this module is not event-triggered"""

agentlib_mpc/modules/mpc_full.py

@@ -1,4 +1,5 @@
 """Holds the class for full featured MPCs."""
+from typing import Dict, Union
 
 import numpy as np
 import pandas as pd
@@ -20,6 +21,30 @@ class MPCConfig(BaseMPCConfig):
         default={},
         description="Weights that are applied to the change in control variables.",
     )
+    enable_deactivate_mpc: bool = Field(
+        default=False,
+        description="If true, the MPC module uses an AgentVariable `active` which"
+        "other modules may change to disable the MPC operation "
+        "temporarily",
+    )
+    active: AgentVariable = Field(
+        default=AgentVariable(
+            name="active",
+            description="MPC is active",
+            type="bool",
+            value=True,
+            shared=False,
+        ),
+        description="Variable used to activate or deactivate the MPC operation",
+    )
+    control_values_when_deactivated: Dict[str, Union[float, bool, int]] = Field(
+        default={},
+        description="When the MPC is deactivated, send the controls with these values"
+        "specified as `{control_name: control_value}`."
+        "In case of variables to send which are not listed as model variables,"
+        "a plain AgentVariable is send. This may be used to deactivate "
+        "supervisory control in a simulation / real PLC.",
+    )
 
     @field_validator("r_del_u")
     def check_r_del_u_in_controls(
@@ -80,7 +105,41 @@ def _init_optimization(self):
         self.history: dict[str, dict[float, float]] = history
         self.register_callbacks_for_lagged_variables()
 
+    def check_if_mpc_step_should_be_skipped(self):
+        """Checks if mpc steps should be skipped based on external activation flag."""
+        if not self.config.enable_deactivate_mpc:
+            return False
+        active = self.get("active")
+        if active.value:
+            return False
+        source = str(active.source)
+        if source == "None_None":
+            source = "unknown (not specified in config)"
+        if not self.config.control_values_when_deactivated:
+            self.logger.info("MPC was deactivated by source %s", source)
+            return True
+        self.logger.info(
+            "MPC was deactivated by source %s, sending control_values_when_deactivated %s",
+            source,
+            self.config.control_values_when_deactivated,
+        )
+        for control_name, value in self.config.control_values_when_deactivated.items():
+            if control_name in self.config.controls:
+                self.set(control_name, value)
+            else:
+                self.agent.data_broker.send_variable(
+                    AgentVariable(
+                        name=control_name,
+                        value=value,
+                        source=self.source,
+                        shared=True
+                    )
+                )
+        return True
+
     def do_step(self):
+        if self.check_if_mpc_step_should_be_skipped():
+            return
         super().do_step()
         self._remove_old_values_from_history()
 

agentlib_mpc/optimization_backends/backend.py

@@ -185,13 +185,15 @@ def get_lags_per_variable(self) -> dict[str, float]:
         variables"""
         return {}
 
-    def results_file_exists(self) -> bool:
-        """Checks if the results file already exists, and if not, creates it with
-        headers."""
+    def results_folder_exists(self) -> bool:
+        """
+        Checks if the results folder already exists, and if not, creates it with
+        headers.
+        """
         if self._created_file:
             return True
 
-        if self.config.results_file.is_file():
+        if self.results_file_exists():
             # todo, this case is weird, as it is the mistake-append
             self._created_file = True
             return True
@@ -201,6 +203,12 @@ def results_file_exists(self) -> bool:
         self._created_file = True
         return False
 
+    def results_file_exists(self) -> bool:
+        """
+        Checks if the results file already exists.
+        """
+        return self.config.results_file.is_file()
+
     def update_model_variables(self, current_vars: Dict[str, AgentVariable]):
         """
         Internal method to write current data_broker to model variables.

agentlib_mpc/optimization_backends/casadi_/admm.py

@@ -379,7 +379,7 @@ def save_result_df(
 
         res_file = self.config.results_file
 
-        if self.results_file_exists():
+        if self.results_folder_exists():
             self.it += 1
             if now != self.now:  # means we advanced to next step
                 self.it = 0

agentlib_mpc/optimization_backends/casadi_/core/casadi_backend.py

@@ -50,6 +50,13 @@ class CasadiBackendConfig(BackendConfig):
         description="Boolean to turn JIT of the optimization problems on or off.",
         validate_default=True,
     )
+    save_only_stats: bool = pydantic.Field(
+        default=False,
+        description="If results should be saved, setting this to True will only save"
+                    "the optimization statistics. May be useful for longer timespans,"
+                    "if the results with all predictions gets too large."
+    )
+
 
     @pydantic.field_validator("do_jit")
     @classmethod
@@ -281,13 +288,24 @@ def save_result_df(
             return
 
         res_file = self.config.results_file
-        if not self.results_file_exists():
+        stats_file = stats_path(res_file)
+
+        first_entry = False
+        # Handle stats
+        if not self.results_folder_exists():
+            results.write_stats_columns(stats_file)
+            first_entry = True
+
+        with open(stats_file, "a") as f:
+            f.writelines(results.stats_line(str(now)))
+
+        if self.config.save_only_stats:
+            return
+
+        # Handle all results, including predictions
+        if first_entry:
             results.write_columns(res_file)
-            results.write_stats_columns(stats_path(res_file))
 
         df = results.df
         df.index = list(map(lambda x: str((now, x)), df.index))
         df.to_csv(res_file, mode="a", header=False)
-
-        with open(stats_path(res_file), "a") as f:
-            f.writelines(results.stats_line(str(now)))

agentlib_mpc/utils/__init__.py

@@ -11,3 +11,17 @@
     "hours": 3600,
     "days": 86400,
 }
+
+
+def is_time_in_intervals(time: float, intervals: list[tuple[float, float]]) -> bool:
+    """
+    Check if given time is within any of the provided intervals.
+
+    Args:
+        time: The time value to check
+        intervals: List of tuples, each containing (start_time, end_time)
+
+    Returns:
+        True if time falls within any interval, False otherwise
+    """
+    return any(start <= time <= end for start, end in intervals)