Coverage for backend/ahuora-unit-ops/src/ahuora_unit_ops/json_config.py: 84%
167 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, Literal
7from pydantic import BaseModel, ConfigDict, Field, model_validator
10class UnitOpJsonConfigError(ValueError):
11 """Raised when a JSON unit-op config cannot be loaded safely."""
14class _StrictModel(BaseModel):
15 model_config = ConfigDict(extra="forbid", populate_by_name=True)
18class JsonPortConfig(_StrictModel):
19 displayName: str
20 type: Literal["inlet", "outlet", "branch"]
21 streamType: str = "stream"
22 many: bool = False
23 default: int = 1
24 minimum: int = 1
25 makeStream: bool = True
26 streamOffset: int | float = 0.75
27 streamName: str = "Stream"
30class JsonPropertyConfig(_StrictModel):
31 propertySetGroup: str = "default"
32 displayName: str
33 indexSets: list[str] | None = None
34 sumToOne: bool = False
35 value: int | float | bool | str | list[str] | None = None
36 unit: str | None = None
37 unitType: str
38 description: str | None = None
39 type: Literal["numeric", "dropdown", "checkbox", "segmented", "text", "numeric_arg"]
40 many: bool = False
41 default: int = 1
42 options: dict[str, str] = Field(default_factory=dict)
43 hasTimeIndex: bool = True
46class JsonPropertySetGroupConfig(_StrictModel):
47 type: Literal["All", "None", "composition", "stateVars", "exceptLast"]
48 displayName: str
49 stateVars: tuple[str, ...] = ()
50 toggle: str | None = None
53class JsonGraphicObjectConfig(_StrictModel):
54 kind: Literal["unitop_graphic", "stream_graphic", "explicit"]
55 width: int | float | None = None
56 height: int | float | None = None
57 autoHeight: bool = False
59 @model_validator(mode="after")
60 def _validate_kind_fields(self) -> "JsonGraphicObjectConfig":
61 if self.kind == "explicit":
62 if self.width is None or self.height is None: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true
63 raise ValueError("explicit graphicObject requires width and height")
64 return self
66 supplied_dimension_fields = {"width", "height", "autoHeight"} & self.model_fields_set
67 if supplied_dimension_fields: 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 fields = ", ".join(sorted(supplied_dimension_fields))
69 raise ValueError(f"{self.kind} graphicObject does not allow: {fields}")
70 return self
73AdapterArgKind = Literal[
74 "property_package",
75 "constant",
76 "constant_ref",
77 "constant_schema",
78 "dynamic",
79 "toggle",
80 "enum",
81 "port_count",
82 "num_inlets",
83 "num_outlets",
84 "dict",
85 "ml_model",
86 "ml_ids",
87 "ml_unitop_names",
88]
89AllowedSchemaName = Literal[
90 "ValueArgSchema",
91 "PropertyPackageArgSchema",
92 "PowerPPArgSchema",
93 "ACPPArgSchema",
94 "ReactionPPArgSchema",
95 "ReactorPPArgSchema",
96]
99class JsonAdapterArgConfig(_StrictModel):
100 kind: AdapterArgKind
101 label: str | None = None
102 value: Any = None
103 ref: str | None = None
104 schema_name: AllowedSchemaName | None = Field(default=None, alias="schema")
105 property_key: str | None = Field(default=None, alias="property")
106 port: str | None = None
107 args: dict[str, "JsonAdapterArgConfig"] | None = None
109 @model_validator(mode="after")
110 def _validate_kind_fields(self) -> "JsonAdapterArgConfig":
111 supplied_fields = _json_field_names(self.model_fields_set)
112 disallowed = supplied_fields - _ADAPTER_ARG_ALLOWED_FIELDS[self.kind]
113 if disallowed: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true
114 fields = ", ".join(sorted(disallowed))
115 raise ValueError(f"{self.kind} adapter arg does not allow: {fields}")
117 missing = _ADAPTER_ARG_REQUIRED_FIELDS.get(self.kind, set()) - supplied_fields
118 if missing: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 fields = ", ".join(sorted(missing))
120 raise ValueError(f"{self.kind} adapter arg requires: {fields}")
121 return self
124class JsonPortMapEntryConfig(_StrictModel):
125 port: str
126 inlet: bool = False
129class JsonSchemaPortMappingConfig(_StrictModel):
130 type: Literal["fixed", "group"]
131 port: str
132 outputKey: str | None = None
133 outputKeyTemplate: str | None = None
134 inlet: bool = False
135 optional: bool = False
136 connectedOnly: bool = False
137 sortByIndex: bool = False
138 activeDirection: Literal["inlet", "outlet", "branch"] | None = None
140 @model_validator(mode="after")
141 def _validate_mapping_type(self) -> "JsonSchemaPortMappingConfig":
142 if self.type == "fixed" and self.outputKey is None: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 raise ValueError("fixed schema port mapping requires outputKey")
144 if self.type == "group" and self.outputKeyTemplate is None: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 raise ValueError("group schema port mapping requires outputKeyTemplate")
146 return self
149class JsonPortAdapterConfig(_StrictModel):
150 kind: Literal[
151 "default",
152 "serialise",
153 "mixer",
154 "splitter",
155 "bus",
156 "port_map",
157 "schema_ports",
158 "machine_learning",
159 ]
160 mapping: dict[str, JsonPortMapEntryConfig] | None = None
161 mappings: list[JsonSchemaPortMappingConfig] | None = None
163 @model_validator(mode="after")
164 def _validate_kind_fields(self) -> "JsonPortAdapterConfig":
165 supplied_fields = self.model_fields_set
166 disallowed = supplied_fields - _PORT_ADAPTER_ALLOWED_FIELDS[self.kind]
167 if disallowed: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 fields = ", ".join(sorted(disallowed))
169 raise ValueError(f"{self.kind} port adapter does not allow: {fields}")
170 missing = _PORT_ADAPTER_REQUIRED_FIELDS.get(self.kind, set()) - supplied_fields
171 if missing: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 fields = ", ".join(sorted(missing))
173 raise ValueError(f"{self.kind} port adapter requires {fields}")
174 return self
177class JsonPropertyAdapterConfig(_StrictModel):
178 kind: Literal["serialise", "machine_learning"]
181class JsonIdaesAdapterConfig(_StrictModel):
182 constructor: str
183 args: dict[str, JsonAdapterArgConfig] = Field(default_factory=dict)
184 ports: JsonPortAdapterConfig | None = None
185 properties: JsonPropertyAdapterConfig | None = None
188class JsonVariantConfig(_StrictModel):
189 familyKey: str
190 label: str
191 selectorLabel: str = "Model Type"
192 default: bool = False
193 order: int | None = None
194 preservePorts: bool = True
195 preserveGraphic: bool = True
198class JsonFrontendConfig(_StrictModel):
199 showInPanel: bool = True
200 variant: JsonVariantConfig | None = None
203class JsonUnitOpConfig(_StrictModel):
204 key: str
205 objectType: str
206 enumMember: str
207 displayType: str
208 displayName: str | None = None
209 categoryPath: list[str]
210 ports: dict[str, JsonPortConfig] | None
211 propertyPackagePorts: dict[str, list[str]] | None = None
212 graphicObject: JsonGraphicObjectConfig
213 indexSets: list[str] = Field(default_factory=list)
214 properties: dict[str, JsonPropertyConfig] = Field(default_factory=dict)
215 propertySetGroups: dict[str, JsonPropertySetGroupConfig] = Field(default_factory=dict)
216 keyProperties: list[str] | dict[str, list[str]] | None = None
217 splitter_fraction_name: str | None = Field(default=None, alias="splitterFractionName")
218 idaes_adapter: JsonIdaesAdapterConfig | None = Field(default=None, alias="idaesAdapter")
219 frontend: JsonFrontendConfig | None = None
220 is_stream: bool = Field(default=False, alias="isStream")
221 info: str | None = None
224_LEGACY_CONFIG_FILENAMES_BY_KEY = {
225 "pl_turbine": "pl_willans_config.json",
226 "dts_turbine": "d_tsat_willans_config.json",
227 "bs_turbine": "bs_willans_config.json",
228 "cs_turbine": "cs_willans_config.json",
229 "heat_exchanger_ntu": "ntu_heat_exchanger_config.json",
230 "heat_exchanger_lc": "lumped_capacitance_hx_config.json",
231}
234def config_filename_from_unit_key(key: str) -> str:
235 """Return the generated JSON config filename for a unit key."""
237 legacy_name = _LEGACY_CONFIG_FILENAMES_BY_KEY.get(key)
238 if legacy_name is not None:
239 return legacy_name
240 return f"{key}_config.json"
243def render_unit_op_config_json(config: JsonUnitOpConfig) -> str:
244 """Render one unit-operation config as deterministic JSON."""
246 data = config.model_dump(mode="json", by_alias=True, exclude_unset=True)
247 return json.dumps(data, indent=2) + "\n"
250def load_unit_op_config_model(path: Path) -> JsonUnitOpConfig:
251 """Load one JSON file into the pure Pydantic unit-operation config model."""
253 return JsonUnitOpConfig.model_validate_json(path.read_text(encoding="utf-8"))
256_ADAPTER_ARG_ALLOWED_FIELDS = {
257 "property_package": {"kind", "label"},
258 "constant": {"kind", "value"},
259 "constant_ref": {"kind", "ref"},
260 "constant_schema": {"kind", "schema"},
261 "dynamic": {"kind"},
262 "toggle": {"kind", "property"},
263 "enum": {"kind", "property"},
264 "port_count": {"kind", "port"},
265 "num_inlets": {"kind"},
266 "num_outlets": {"kind"},
267 "dict": {"kind", "args"},
268 "ml_model": {"kind"},
269 "ml_ids": {"kind"},
270 "ml_unitop_names": {"kind"},
271}
273_ADAPTER_ARG_REQUIRED_FIELDS = {
274 "constant": {"value"},
275 "constant_ref": {"ref"},
276 "constant_schema": {"schema"},
277 "toggle": {"property"},
278 "enum": {"property"},
279 "port_count": {"port"},
280 "dict": {"args"},
281}
283_PORT_ADAPTER_ALLOWED_FIELDS = {
284 "default": {"kind"},
285 "serialise": {"kind"},
286 "mixer": {"kind"},
287 "splitter": {"kind"},
288 "bus": {"kind"},
289 "port_map": {"kind", "mapping"},
290 "schema_ports": {"kind", "mappings"},
291 "machine_learning": {"kind"},
292}
294_PORT_ADAPTER_REQUIRED_FIELDS = {
295 "port_map": {"mapping"},
296 "schema_ports": {"mappings"},
297}
300def _json_field_names(field_names: set[str]) -> set[str]:
301 aliases = {
302 "property_key": "property",
303 "schema_name": "schema",
304 }
305 return {aliases.get(field_name, field_name) for field_name in field_names}