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