Coverage for backend/django/idaes_factory/adapters/port_adapter.py: 93%
115 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from __future__ import annotations
3from dataclasses import dataclass
5from ahuora_builder_types import PortsSchema, PortSchema
6from .stream_properties import serialise_stream
7from core.auxiliary.enums import ConType
8from .. import queryset_lookup
9from ..queryset_lookup import get_port, get_all_ports
12class PortAdapter:
13 """
14 Handles running the relevant PropertyAdapter for each key in the schema.
15 """
17 def __init__(self, name: str, inlet: bool = False):
18 """
19 name: the name of the port that this adapter is for.
20 properties_schema: a dictionary of idaes propertyNames, where the value is a PropertyAdapter that can get the relevant property from the unit model.
21 """
22 self.inlet = inlet
23 self.name = name
25 def serialise(self, ctx, unit_model) -> PortSchema:
26 # properties of the port as part of the unit operation
27 properties = {}
28 port = get_port(unit_model, self.name)
29 stream_id = port.stream_id
30 if stream_id is None:
31 port_schema = unit_model.schema.ports[port.key]
32 if port_schema.makeStream is False: 32 ↛ 34line 32 didn't jump to line 34 because the condition on line 32 was always true
33 return None
34 if ctx.require_variables_fixed:
35 raise ValueError(
36 f"{unit_model.componentName}: No stream attached to port {self.name}"
37 )
38 else:
39 stream = ctx.get_simulation_object(stream_id)
40 properties = serialise_stream(ctx, stream, self.inlet)
42 return PortSchema(id=port.id, properties=properties)
45class PortListAdapter:
46 """
47 Handles the serialisation and reloading of a group of ports, where the port names are known ahead of time.
48 Used for unit models such as pumps, compressors, heaters, etc.
49 Compare with DynamicPortListAdapter
50 """
52 def __init__(self, schema: dict[str, PortAdapter]):
53 self.schema = schema
55 def serialise(self, ctx, unit_model) -> PortsSchema:
56 port_list = {}
57 for key, adapter in self.schema.items():
58 port = adapter.serialise(ctx, unit_model)
59 if port is not None:
60 port_list[key] = port
61 return port_list
64def serialise_port_with_stream(ctx, port, *, is_inlet: bool) -> dict:
65 """Serialise a connected port using the same shape as the existing dynamic adapters."""
67 stream_id = getattr(port, "stream_id", None)
68 if stream_id is not None:
69 stream = ctx.get_simulation_object(stream_id)
70 elif getattr(port, "stream", None) is not None: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 stream = port.stream
72 else:
73 raise ValueError(
74 f"Port {getattr(port, 'key', '<unknown>')} has no connected stream"
75 )
76 return {
77 "id": port.pk,
78 "properties": serialise_stream(ctx, stream, is_inlet=is_inlet),
79 }
82def is_connected(port) -> bool:
83 return (
84 getattr(port, "stream_id", None) is not None
85 or getattr(port, "stream", None) is not None
86 )
89@dataclass(frozen=True)
90class FixedPortMapping:
91 """Describe a single known port and the output key it should use."""
93 port_key: str
94 output_key: str
95 is_inlet: bool = False
96 optional: bool = False
97 connected_only: bool = False
99 def serialise(self, ctx, unit_model) -> PortsSchema:
100 try:
101 port = get_port(unit_model, self.port_key)
102 except ValueError:
103 if self.optional: 103 ↛ 105line 103 didn't jump to line 105 because the condition on line 103 was always true
104 return {}
105 raise
107 if self.connected_only:
108 if not is_connected(port):
109 return {}
110 return {
111 self.output_key: serialise_port_with_stream(
112 ctx, port, is_inlet=self.is_inlet
113 )
114 }
116 serialised_port = PortAdapter(self.port_key, inlet=self.is_inlet).serialise(
117 ctx, unit_model
118 )
119 if serialised_port is None: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 return {}
121 return {self.output_key: serialised_port}
124@dataclass(frozen=True)
125class PortGroupMapping:
126 """Describe a repeated port group and how its serialised keys are generated."""
128 port_key: str
129 output_key_template: str
130 is_inlet: bool
131 active_direction: ConType | None = None
132 connected_only: bool = False
133 sort_by_index: bool = False
135 def get_ports(self, unit_model):
136 if self.active_direction is not None:
137 ports = queryset_lookup.get_active_ports_for_direction(
138 unit_model, self.active_direction
139 )
140 else:
141 ports = get_all_ports(unit_model, self.port_key)
143 if self.connected_only:
144 ports = [port for port in ports if is_connected(port)]
145 if self.sort_by_index:
146 ports = sorted(ports, key=lambda port: getattr(port, "index", 0))
147 return ports
149 def serialise(self, ctx, unit_model) -> PortsSchema:
150 port_list = {}
151 for index, port in enumerate(self.get_ports(unit_model), start=1):
152 port_list[self.output_key_template.format(index=index)] = (
153 serialise_port_with_stream(ctx, port, is_inlet=self.is_inlet)
154 )
155 return port_list
158class SchemaPortListAdapter:
159 """Serialise ports from an ordered set of fixed and grouped schema mappings."""
161 def __init__(self, mappings: list[FixedPortMapping | PortGroupMapping]) -> None:
162 self.mappings = mappings
164 def serialise(self, ctx, unit_model) -> PortsSchema:
165 port_list = {}
166 for mapping in self.mappings:
167 port_list.update(mapping.serialise(ctx, unit_model))
168 return port_list
171class SerialisePortAdapter:
172 def serialise(self, ctx, model) -> PortsSchema:
173 """
174 Serialise all ports in the model.
175 """
177 def isinlet(key) -> bool:
178 if "inlet" in key:
179 inlet: bool = True
180 else:
181 inlet = False
182 return inlet
184 # get the port from the unit model
185 schema = {
186 port_key: PortAdapter(port_key, isinlet(port_key))
187 for port_key in model.schema.ports
188 }
189 # serialise the port
190 result = PortListAdapter(schema).serialise(ctx=ctx, unit_model=model)
191 return result
194class MixerPortListAdapter(SchemaPortListAdapter):
195 """
196 Handles the serialisation and reloading of the ports of a mixer.
197 """
199 def __init__(self):
200 super().__init__(
201 [
202 PortGroupMapping(
203 port_key="inlet",
204 output_key_template="inlet_{index}",
205 is_inlet=True,
206 ),
207 FixedPortMapping(
208 port_key="outlet",
209 output_key="outlet",
210 is_inlet=False,
211 ),
212 ]
213 )
216class SplitterPortListAdapter(SchemaPortListAdapter):
217 """
218 Handles the serialisation and reloading of the ports of a splitter.
219 """
221 def __init__(self):
222 super().__init__(
223 [
224 FixedPortMapping(
225 port_key="inlet",
226 output_key="inlet",
227 is_inlet=True,
228 ),
229 PortGroupMapping(
230 port_key="outlet",
231 output_key_template="outlet_{index}",
232 is_inlet=False,
233 active_direction=ConType.Outlet,
234 ),
235 ]
236 )
239class BusPortListAdapter(SchemaPortListAdapter):
240 """
241 Handles the serialisation and reloading of the ports of a Bus.
242 """
244 def __init__(self):
245 super().__init__(
246 [
247 PortGroupMapping(
248 port_key="inlet",
249 output_key_template="inlet_{index}",
250 is_inlet=True,
251 ),
252 PortGroupMapping(
253 port_key="outlet",
254 output_key_template="outlet_{index}",
255 is_inlet=False,
256 active_direction=ConType.Outlet,
257 ),
258 ]
259 )