Coverage for backend/django/flowsheetInternals/unitops/config/json_loader.py: 91%
149 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3import json
4from pathlib import Path
5from typing import Any
7from pydantic import ValidationError
9from ahuora_unit_ops.json_config import (
10 JsonAdapterArgConfig,
11 JsonGraphicObjectConfig,
12 JsonIdaesAdapterConfig,
13 JsonPortAdapterConfig,
14 JsonPropertyAdapterConfig,
15 JsonSchemaPortMappingConfig,
16 JsonUnitOpConfig,
17 UnitOpJsonConfigError,
18)
20from ahuora_builder_types.unit_model_schema import (
21 ACPPArgSchema,
22 PowerPPArgSchema,
23 PropertyPackageArgSchema,
24 ReactionPPArgSchema,
25 ReactorPPArgSchema,
26 ValueArgSchema,
27)
28from common.config_types import (
29 GraphicObjectType,
30 ObjectType,
31 PortType,
32 PropertiesType,
33 PropertySetType,
34 PropertyType,
35)
36from common.config_utils import stream_graphic, unitop_graphic
37from core.auxiliary.enums.unitOpGraphics import ConType as RuntimeConType
38from idaes.models.control.controller import ControllerMVBoundType, ControllerType
39from idaes.models.unit_models.separator import EnergySplittingType, SplittingType
40from idaes_factory.adapters.arg_adapter import (
41 ArgAdapter,
42 ConstantArg,
43 ConstantSchemaAdapter,
44 DictArgAdapter,
45)
46from idaes_factory.adapters.dynamic_adapter import DynamicAdapter
47from idaes_factory.adapters.enum_adapter import EnumAdapter
48from idaes_factory.adapters.generic_adapters import (
49 NumInletsAdapter,
50 NumOutletsAdapter,
51 PortCountAdapter,
52)
53from idaes_factory.adapters.ml_adapters import (
54 IDAdapter,
55 JSONModelAdapter,
56 MLPortListAdapter,
57 MLPropertiesAdapter,
58 UnitopNamesAdapter,
59)
60from idaes_factory.adapters.port_adapter import (
61 BusPortListAdapter,
62 FixedPortMapping,
63 MixerPortListAdapter,
64 PortAdapter,
65 PortGroupMapping,
66 PortListAdapter,
67 SchemaPortListAdapter,
68 SerialisePortAdapter,
69 SplitterPortListAdapter,
70)
71from idaes_factory.adapters.property_info_adapter import SerialisePropertiesAdapter
72from idaes_factory.adapters.property_package_adapter import PropertyPackageAdapter
73from idaes_factory.adapters.toggle_adapter import ToggleAdapter
74from idaes_factory.adapters.unit_model_adapter import UnitModelAdapter
77def _runtime_idaes_adapter(config: JsonIdaesAdapterConfig) -> UnitModelAdapter:
78 _validate_trusted_constructor_path(config.constructor)
79 kwargs: dict[str, Any] = {
80 "args": ArgAdapter(
81 {key: _runtime_adapter_arg(arg) for key, arg in config.args.items()}
82 )
83 }
84 if config.properties is not None:
85 kwargs["properties"] = _runtime_property_adapter(config.properties)
86 if config.ports is not None:
87 ports = _runtime_port_adapter(config.ports)
88 if ports is not None: 88 ↛ 90line 88 didn't jump to line 90 because the condition on line 88 was always true
89 kwargs["ports"] = ports
90 return UnitModelAdapter(**kwargs)
93def _runtime_unit_op_config(config: JsonUnitOpConfig) -> ObjectType:
94 kwargs: dict[str, Any] = {
95 "displayType": config.displayType,
96 "displayName": config.displayName,
97 "ports": (
98 {
99 key: PortType(**port.model_dump(mode="python"))
100 for key, port in config.ports.items()
101 }
102 if config.ports is not None
103 else None
104 ),
105 "propertyPackagePorts": config.propertyPackagePorts,
106 "graphicObject": _runtime_graphic_object(config.graphicObject),
107 "indexSets": config.indexSets,
108 "properties": PropertiesType(
109 {
110 key: PropertyType(**prop.model_dump(mode="python"))
111 for key, prop in config.properties.items()
112 }
113 ),
114 "propertySetGroups": {
115 key: PropertySetType(**group.model_dump(mode="python"))
116 for key, group in config.propertySetGroups.items()
117 },
118 "splitter_fraction_name": config.splitter_fraction_name,
119 "is_stream": config.is_stream,
120 "info": config.info,
121 "idaes_adapter": (
122 _runtime_idaes_adapter(config.idaes_adapter)
123 if config.idaes_adapter is not None
124 else None
125 ),
126 }
127 if config.keyProperties is not None:
128 kwargs["keyProperties"] = config.keyProperties
129 return ObjectType(**kwargs)
132def load_unit_op_config_json(path: Path) -> tuple[str, ObjectType]:
133 """Load and validate one JSON unit-op config file."""
135 try:
136 data = json.loads(path.read_text(encoding="utf-8"))
137 config = JsonUnitOpConfig.model_validate(data)
138 except (OSError, json.JSONDecodeError, ValidationError, ValueError) as exc:
139 raise UnitOpJsonConfigError(
140 f"Invalid unit-op JSON config: {path}\n{exc}"
141 ) from exc
142 return config.objectType, _runtime_unit_op_config(config)
145def load_unit_op_configuration_from_json_dir(path: Path) -> dict[str, ObjectType]:
146 """Load all JSON unit-op configs from a directory."""
148 loaded: dict[str, ObjectType] = {}
149 seen_paths: dict[str, Path] = {}
150 for config_path in sorted(path.glob("*.json")):
151 object_type, config = load_unit_op_config_json(config_path)
152 if object_type in loaded: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 raise UnitOpJsonConfigError(
154 "Duplicate unit-op JSON objectType "
155 f"{object_type!r}: {seen_paths[object_type]} and {config_path}"
156 )
157 loaded[object_type] = config
158 seen_paths[object_type] = config_path
160 configuration: dict[str, ObjectType] = {}
161 for object_type in _LEGACY_CONFIGURATION_ORDER:
162 if object_type in loaded: 162 ↛ 161line 162 didn't jump to line 161 because the condition on line 162 was always true
163 configuration[object_type] = loaded.pop(object_type)
164 configuration.update(loaded)
165 return configuration
168def load_unit_op_configuration() -> dict[str, ObjectType]:
169 """Load runtime unit-op configuration from the conventional JSON directory."""
171 return load_unit_op_configuration_from_json_dir(Path(__file__).parent / "objects")
174_TRUSTED_CONSTRUCTOR_PREFIXES = (
175 "ahuora_builder.custom.",
176 "idaes.models.",
177 "idaes.models_extra.",
178 "watertap.",
179)
181_LEGACY_CONFIGURATION_ORDER = (
182 "decisionNode",
183 "stream",
184 "humid_air_stream",
185 "pump",
186 "compressor",
187 "cooler",
188 "valve",
189 "turbine",
190 "willans_turbine",
191 "pl_turbine",
192 "dts_turbine",
193 "bs_turbine",
194 "cs_turbine",
195 "heater",
196 "Tank",
197 "recycle",
198 "mixer",
199 "splitter",
200 "header",
201 "phaseSeparator",
202 "energy_stream",
203 "ac_stream",
204 "transformer_stream",
205 "compoundSeparator",
206 "heatExchanger",
207 "heatPump",
208 "group",
209 "machineLearningBlock",
210 "link",
211 "grid",
212 "boiler",
213 "solar",
214 "hydro",
215 "wind",
216 "energy_mixer",
217 "bus",
218 "storage",
219 "acBus",
220 "transmissionLine",
221 "load",
222 "transformer",
223 "convertor",
224 "direct_steam_injection",
225 "translator",
226 "RCT_CSTR",
227 "stoich_hda",
228 "pid_controller",
229 "custom_variable",
230 "integration",
231 "energy_splitter",
232 "mdb",
233 "simple_header",
234 "heat_user",
235 "steam_user",
236 "desuperheater",
237 "heat_exchanger_ntu",
238 "duty_heat_exchanger",
239 "waterpipe",
240 "heat_exchanger_lc",
241 "heat_exchanger_1d",
242 "plate_heat_exchanger",
243 "reverse_osmosis_0d",
244 "pressure_exchanger",
245 "crystallizer",
246)
249_SCHEMA_CLASSES = {
250 "ValueArgSchema": ValueArgSchema,
251 "PropertyPackageArgSchema": PropertyPackageArgSchema,
252 "PowerPPArgSchema": PowerPPArgSchema,
253 "ACPPArgSchema": ACPPArgSchema,
254 "ReactionPPArgSchema": ReactionPPArgSchema,
255 "ReactorPPArgSchema": ReactorPPArgSchema,
256}
259_CONSTANT_REFS = {
260 "idaes.models.control.controller.ControllerType.PI": ControllerType.PI,
261 "idaes.models.control.controller.ControllerMVBoundType.SMOOTH_BOUND": (
262 ControllerMVBoundType.SMOOTH_BOUND
263 ),
264 "idaes.models.unit_models.separator.SplittingType.phaseFlow": (
265 SplittingType.phaseFlow
266 ),
267 "idaes.models.unit_models.separator.SplittingType.componentFlow": (
268 SplittingType.componentFlow
269 ),
270 "idaes.models.unit_models.separator.EnergySplittingType.enthalpy_split": (
271 EnergySplittingType.enthalpy_split
272 ),
273}
276def _runtime_graphic_object(config: JsonGraphicObjectConfig) -> GraphicObjectType:
277 match config.kind:
278 case "unitop_graphic":
279 return unitop_graphic()
280 case "stream_graphic":
281 return stream_graphic()
282 case "explicit": 282 ↛ exitline 282 didn't return from function '_runtime_graphic_object' because the pattern on line 282 always matched
283 return GraphicObjectType(
284 width=config.width,
285 height=config.height,
286 autoHeight=config.autoHeight,
287 )
290def _runtime_adapter_arg(config: JsonAdapterArgConfig):
291 match config.kind:
292 case "property_package":
293 return PropertyPackageAdapter(config.label or "")
294 case "constant":
295 return ConstantArg(config.value)
296 case "constant_ref":
297 return ConstantArg(_constant_ref(config.ref))
298 case "constant_schema":
299 return ConstantSchemaAdapter(_schema_instance(config.schema_name))
300 case "dynamic":
301 return DynamicAdapter()
302 case "toggle":
303 return ToggleAdapter(config.property_key)
304 case "enum":
305 return EnumAdapter(config.property_key)
306 case "port_count":
307 return PortCountAdapter(config.port)
308 case "num_inlets":
309 return NumInletsAdapter()
310 case "num_outlets":
311 return NumOutletsAdapter()
312 case "dict":
313 return DictArgAdapter(
314 {
315 key: _runtime_adapter_arg(arg)
316 for key, arg in (config.args or {}).items()
317 }
318 )
319 case "ml_model":
320 return JSONModelAdapter()
321 case "ml_ids":
322 return IDAdapter()
323 case "ml_unitop_names": 323 ↛ exitline 323 didn't return from function '_runtime_adapter_arg' because the pattern on line 323 always matched
324 return UnitopNamesAdapter()
327def _runtime_schema_port_mapping(config: JsonSchemaPortMappingConfig):
328 match config.type:
329 case "fixed":
330 return FixedPortMapping(
331 port_key=config.port,
332 output_key=config.outputKey,
333 is_inlet=config.inlet,
334 optional=config.optional,
335 connected_only=config.connectedOnly,
336 )
337 case "group": 337 ↛ exitline 337 didn't return from function '_runtime_schema_port_mapping' because the pattern on line 337 always matched
338 return PortGroupMapping(
339 port_key=config.port,
340 output_key_template=config.outputKeyTemplate,
341 is_inlet=config.inlet,
342 active_direction=(
343 RuntimeConType(config.activeDirection)
344 if config.activeDirection is not None
345 else None
346 ),
347 connected_only=config.connectedOnly,
348 sort_by_index=config.sortByIndex,
349 )
352def _runtime_port_adapter(config: JsonPortAdapterConfig):
353 match config.kind:
354 case "default": 354 ↛ 355line 354 didn't jump to line 355 because the pattern on line 354 never matched
355 return None
356 case "serialise":
357 return SerialisePortAdapter()
358 case "mixer":
359 return MixerPortListAdapter()
360 case "splitter":
361 return SplitterPortListAdapter()
362 case "bus":
363 return BusPortListAdapter()
364 case "port_map":
365 return PortListAdapter(
366 {
367 key: PortAdapter(entry.port, inlet=entry.inlet)
368 for key, entry in (config.mapping or {}).items()
369 }
370 )
371 case "schema_ports":
372 return SchemaPortListAdapter(
373 [
374 _runtime_schema_port_mapping(mapping)
375 for mapping in (config.mappings or [])
376 ]
377 )
378 case "machine_learning": 378 ↛ exitline 378 didn't return from function '_runtime_port_adapter' because the pattern on line 378 always matched
379 return MLPortListAdapter()
382def _runtime_property_adapter(config: JsonPropertyAdapterConfig):
383 match config.kind:
384 case "serialise":
385 return SerialisePropertiesAdapter()
386 case "machine_learning": 386 ↛ exitline 386 didn't return from function '_runtime_property_adapter' because the pattern on line 386 always matched
387 return MLPropertiesAdapter()
390def _validate_trusted_constructor_path(dotted_path: str) -> None:
391 if not dotted_path.startswith(_TRUSTED_CONSTRUCTOR_PREFIXES): 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 raise UnitOpJsonConfigError(
393 f"Disallowed idaesAdapter constructor path: {dotted_path!r}"
394 )
395 module_name, _, object_name = dotted_path.rpartition(".")
396 if not module_name or not object_name: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 raise UnitOpJsonConfigError(
398 f"Invalid idaesAdapter constructor path: {dotted_path!r}"
399 )
402def _schema_instance(schema_name: str | None):
403 schema_class = _SCHEMA_CLASSES.get(schema_name or "")
404 if schema_class is None: 404 ↛ 405line 404 didn't jump to line 405 because the condition on line 404 was never true
405 raise UnitOpJsonConfigError(
406 f"Disallowed constant_schema adapter schema: {schema_name!r}"
407 )
408 return schema_class()
411def _constant_ref(ref: str | None):
412 try:
413 return _CONSTANT_REFS[ref or ""]
414 except KeyError as exc:
415 raise UnitOpJsonConfigError(
416 f"Disallowed constant_ref adapter value: {ref!r}"
417 ) from exc