Coverage for agentlib_flexquant/data_structures/mpcs.py: 99%
101 statements
« prev ^ index » next coverage.py v7.4.4, created at 2026-03-26 09:43 +0000
1"""
2Data models for MPC configurations in flexibility quantification.
4This module defines Pydantic data models that encapsulate configuration parameters
5for baseline, positive flexibility, and negative flexibility MPC controllers used
6in flexquant. The models handle file paths, module configurations, variable
7mappings, and optimization weights for MPC implementations.
8"""
9import pydantic
10from agentlib_mpc.data_structures.mpc_datamodels import MPCVariable
11from pydantic import model_validator, field_serializer, Field
13import agentlib_flexquant.data_structures.globals as glbs
14import agentlib_flexquant.utils.config_management as cmng
16excluded_fields = [
17 "rdf_class",
18 "source",
19 "type",
20 "timestamp",
21 "description",
22 "unit",
23 "clip",
24 "interpolation_method",
25 "allowed_values",
26 ]
28default_parameters = [
29 MPCVariable(
30 name=glbs.PREP_TIME,
31 value=0,
32 unit="s",
33 type="int",
34 description="Preparation time before switching objective",
35 ),
36 MPCVariable(
37 name=glbs.FLEX_EVENT_DURATION,
38 value=0,
39 unit="s",
40 type="int",
41 description="Duration of the flexibility event",
42 ),
43 MPCVariable(
44 name=glbs.MARKET_TIME,
45 value=0,
46 unit="s",
47 type="int",
48 description="Market time associated with the objective switch",
49 ),
50 ]
52default_inputs = [
53 MPCVariable(
54 name=glbs.PROVISION_VAR_NAME,
55 value=False,
56 type="bool",
57 description="Flag indicating whether flexibility should be provisioned",
58 ),
59 ]
61def _ensure_defaults_in_appendix(
62 data: dict,
63 field_name: str,
64 default_variables: list[MPCVariable],
65) -> dict:
66 """
67 Helper to ensure that all default_variables are present in
68 data[field_name], based on their ``name`` attribute.
69 """
70 provided_variables = data.get(field_name, [])
71 provided_variables = [MPCVariable(**var) if isinstance(var, dict) else var for var in provided_variables]
72 provided_names = {param.name for param in provided_variables}
73 for default_var in default_variables:
74 if default_var.name not in provided_names:
75 provided_variables.append(default_var)
77 data[field_name] = provided_variables
78 return data
81class BaseMPCData(pydantic.BaseModel):
82 """Base class containing necessary data for the code creation
83 of the different mpcs
85 """
87 # files and paths
88 created_flex_mpcs_file: str = "flex_agents.py"
89 name_of_created_file: str
90 results_suffix: str
91 # modules
92 module_types: dict
93 class_name: str
94 module_id: str
95 agent_id: str
96 # variables
97 power_alias: str
98 stored_energy_alias: str
99 config_inputs_appendix: list[MPCVariable] = Field(
100 default=[],
101 description="Inputs, which are appended to the MPCs' config "
102 "(.json file and ConfigClass).")
103 config_parameters_appendix: list[MPCVariable] = Field(
104 default=[],
105 description="Parameters, which are appended to the MPCs' config "
106 "(.json file and ConfigClass).")
108 @field_serializer('config_inputs_appendix', 'config_parameters_appendix')
109 def serialize_mpc_variables(self, variables: list[MPCVariable], _info):
110 return [v.dict(exclude=excluded_fields) for v in variables]
113class BaselineMPCData(BaseMPCData):
114 """Data class for Baseline MPC"""
116 # files and paths
117 results_suffix: str = "_base.csv"
118 name_of_created_file: str = "baseline.json"
119 # modules
120 module_types: dict = cmng.BASELINE_MODULE_TYPE_DICT
121 class_name: str = "BaselineMPCModel"
122 module_id: str = "Baseline"
123 agent_id: str = "Baseline"
124 # variables
125 power_alias: str = glbs.POWER_ALIAS_BASE
126 stored_energy_alias: str = glbs.STORED_ENERGY_ALIAS_BASE
127 power_variable: str = pydantic.Field(
128 default="P_el",
129 description=(
130 "Name of the variable representing the electrical "
131 "power in the baseline config"
132 ),
133 )
134 profile_deviation_weight: float = pydantic.Field(
135 default=0,
136 description="Weight of soft constraint for deviation from "
137 "accepted flexible profile",
138 )
139 power_unit: str = pydantic.Field(
140 default="kW", description="Unit of the power variable"
141 )
142 comfort_variable: str = pydantic.Field(
143 default=None,
144 description=(
145 "Name of the slack variable representing the thermal "
146 "comfort in the baseline config"
147 ),
148 )
149 profile_comfort_weight: float = pydantic.Field(
150 default=1, description="Weight of soft constraint for discomfort",
151 )
152 config_inputs_appendix: list[MPCVariable] = [
153 MPCVariable(name=glbs.ACCEPTED_POWER_VAR_NAME, value=0, unit="W"),
154 MPCVariable(name=glbs.PROVISION_VAR_NAME, value=False),
155 MPCVariable(name=glbs.RELATIVE_EVENT_START_TIME_VAR_NAME, value=0, unit="s"),
156 MPCVariable(name=glbs.RELATIVE_EVENT_END_TIME_VAR_NAME, value=0, unit="s"),
157 ]
159 config_parameters_appendix: list[MPCVariable] = []
162 @field_serializer('config_inputs_appendix', 'config_parameters_appendix')
163 def serialize_mpc_variables(self, variables: list[MPCVariable], _info):
164 return [v.dict(exclude=excluded_fields) for v in variables]
166 @model_validator(mode="after")
167 def update_config_parameters_appendix(self) -> "BaselineMPCData":
168 """Update the config parameters appendix with profile deviation and comfort
169 weights.
171 Adds the profile deviation weight parameter and optionally the profile
172 comfort weight parameter (if comfort_variable is enabled) to the
173 config_parameters_appendix list as MPCVariable instances.
174 """
175 self.config_parameters_appendix = [
176 MPCVariable(
177 name=glbs.PROFILE_DEVIATION_WEIGHT,
178 value=self.profile_deviation_weight,
179 unit="-",
180 description=(
181 "Weight of soft constraint for deviation from accepted "
182 "flexible profile"
183 ),
184 )
185 ]
186 if self.comfort_variable:
187 self.config_parameters_appendix.append(
188 MPCVariable(
189 name=glbs.PROFILE_COMFORT_WEIGHT,
190 value=self.profile_comfort_weight,
191 unit="-",
192 description="Weight of soft constraint for discomfort",
193 )
194 )
195 return self
198class PFMPCData(BaseMPCData):
199 """Data class for PF-MPC"""
201 # files and paths
202 results_suffix: str = "_pos_flex.csv"
203 name_of_created_file: str = "pos_flex.json"
204 # modules
205 module_types: dict = cmng.SHADOW_MODULE_TYPE_DICT
206 class_name: str = "PosFlexModel"
207 module_id: str = "PosFlexMPC"
208 agent_id: str = "PosFlexMPC"
209 # variables
210 power_alias: str = glbs.POWER_ALIAS_POS
211 stored_energy_alias: str = glbs.STORED_ENERGY_ALIAS_POS
212 flex_cost_function: str = pydantic.Field(
213 default=None, description="Cost function of the PF-MPC during the event",
214 )
215 flex_cost_function_appendix: str = pydantic.Field(
216 default=None, description="Cost function appendix of the PF-MPC added "
217 "to the Baseline cost function",
218 )
219 # initialize market parameters with dummy values (0)
220 config_parameters_appendix: list[MPCVariable] = pydantic.Field(
221 default=[], description="Parameters, which need to be appended to the shadow MPCs"
222 )
223 config_inputs_appendix: list[MPCVariable] = pydantic.Field(
224 default=[], description="Inputs, which need to be appended to the shadow MPCs"
225 )
228 @field_serializer('config_inputs_appendix', 'config_parameters_appendix')
229 def serialize_mpc_variables(self, variables: list[MPCVariable], _info):
230 return [v.dict(exclude=excluded_fields) for v in variables]
232 @model_validator(mode="before")
233 @classmethod
234 def add_defaults_to_appendix(cls, data):
235 """
236 Ensure that all required framework variables are included in both
237 ``config_inputs_appendix`` and ``config_parameters_appendix``.
238 If any default framework input or parameter (e.g., PROVISION_VAR_NAME)
239 is missing, it is appended to the corresponding list.
240 """
242 data = _ensure_defaults_in_appendix(
243 data=data,
244 field_name="config_inputs_appendix",
245 default_variables=default_inputs,
246 )
248 data = _ensure_defaults_in_appendix(
249 data=data,
250 field_name="config_parameters_appendix",
251 default_variables=default_parameters,
252 )
253 return data
256class NFMPCData(BaseMPCData):
257 """Data class for NF-MPC"""
259 # files and paths
260 results_suffix: str = "_neg_flex.csv"
261 name_of_created_file: str = "neg_flex.json"
262 # modules
263 module_types: dict = cmng.SHADOW_MODULE_TYPE_DICT
264 class_name: str = "NegFlexModel"
265 module_id: str = "NegFlexMPC"
266 agent_id: str = "NegFlexMPC"
267 # variables
268 power_alias: str = glbs.POWER_ALIAS_NEG
269 stored_energy_alias: str = glbs.STORED_ENERGY_ALIAS_NEG
270 flex_cost_function: str = pydantic.Field(
271 default=None, description="Cost function of the NF-MPC during the event",
272 )
273 flex_cost_function_appendix: str = pydantic.Field(
274 default=None, description="Cost function appendix of the NF-MPC added "
275 "to the Baseline cost function",
276 )
277 # initialize market parameters with dummy values (0)
278 config_parameters_appendix: list[MPCVariable] = pydantic.Field(
279 default=[], description="Parameters, which need to be appended to the shadow MPCs"
280 )
281 config_inputs_appendix: list[MPCVariable] = pydantic.Field(
282 default=[], description = "Inputs, which need to be appended to the shadow MPCs"
283 )
286 @field_serializer('config_inputs_appendix', 'config_parameters_appendix')
287 def serialize_mpc_variables(self, variables: list[MPCVariable], _info):
288 return [v.dict(exclude=excluded_fields) for v in variables]
291 @model_validator(mode="before")
292 @classmethod
293 def add_defaults_to_appendix(cls, data):
294 """
295 Ensures that all required framework input and parameter variables are included in
296 ``config_inputs_appendix`` and ``config_parameters_appendix``. If any default
297 framework input or parameter (e.g., PROVISION_VAR_NAME) is missing, it is appended
298 to the corresponding list.
299 """
301 data = _ensure_defaults_in_appendix(
302 data=data,
303 field_name="config_inputs_appendix",
304 default_variables=default_inputs,
305 )
307 data = _ensure_defaults_in_appendix(
308 data=data,
309 field_name="config_parameters_appendix",
310 default_variables=default_parameters,
311 )
312 return data