Coverage for backend/idaes_factory/adapters/port_adapter.py: 95%

1from common.models.idaes import PortsSchema, PortSchema 

2from flowsheetInternals.unitops.models import SimulationObject 

3from .stream_properties import serialise_stream 

4from core.auxiliary.enums import ConType 

5from .. import queryset_lookup 

6from ..queryset_lookup import get_port, get_all_ports 

7from ..idaes_factory_context import IdaesFactoryContext 

8 

9 

10class PortAdapter(): 

11 """ 

12 Handles running the relevant PropertyAdapter for each key in the schema. 

13 """ 

14 def __init__(self, name: str, inlet: bool = False): 

15 """ 

16 name: the name of the port that this adapter is for. 

17 properties_schema: a dictionary of idaes propertyNames, where the value is a PropertyAdapter that can get the relevant property from the unit model. 

18 """ 

19 self.inlet = inlet 

20 self.name = name 

21 

22 

23 def serialise(self, ctx: IdaesFactoryContext, unit_model: SimulationObject) -> PortSchema: 

24 # properties of the port as part of the unit operation 

25 properties = {} 

26 port = get_port(unit_model, self.name) 

27 stream_id = port.stream_id 

28 if stream_id is None: 

29 port_schema = unit_model.schema.ports[port.key] 

30 if port_schema.makeStream is False: 30 ↛ 32line 30 didn't jump to line 32 because the condition on line 30 was always true

31 return None 

32 if ctx.require_variables_fixed: 

33 raise ValueError(f"{unit_model.componentName}: No stream attached to port {self.name}") 

34 else : 

35 stream = ctx.get_simulation_object(stream_id) 

36 properties = serialise_stream(ctx, stream, self.inlet) 

37 

38 return PortSchema( 

39 id=port.id, 

40 properties=properties 

41 ) 

42 

43 

44class PortListAdapter: 

45 """ 

46 Handles the serialisation and reloading of a group of ports, where the port names are known ahead of time. 

47 Used for unit models such as pumps, compressors, heaters, etc. 

48 Compare with DynamicPortListAdapter 

49 """ 

50 def __init__(self, schema:dict[str,PortAdapter]): 

51 self.schema = schema 

52 

53 def serialise(self, ctx: IdaesFactoryContext, unit_model: SimulationObject) -> PortsSchema: 

54 port_list = {} 

55 for key, adapter in self.schema.items(): 

56 port = adapter.serialise(ctx, unit_model) 

57 if port is not None: 

58 port_list[key] = port 

59 return port_list 

60 

61class SerialisePortAdapter: 

62 def serialise(self, ctx: IdaesFactoryContext, model: SimulationObject) -> PortsSchema: 

63 """ 

64 Serialise all ports in the model. 

65 """ 

66 def isinlet(key) -> bool: 

67 if "inlet" in key: 

68 inlet:bool = True 

69 else: 

70 inlet = False 

71 return inlet 

72 

73 # get the port from the unit model 

74 schema = {port_key: PortAdapter(port_key, isinlet(port_key)) for port_key in model.schema.ports} 

75 # serialise the port 

76 result = PortListAdapter(schema).serialise(ctx = ctx,unit_model = model) 

77 return result 

78 

79class MixerPortListAdapter(PortListAdapter): 

80 """ 

81 Handles the serialisation and reloading of the ports of a mixer. 

82 """ 

83 def __init__(self): 

84 pass 

85 

86 def serialise(self, ctx, unit_model: SimulationObject) -> dict: 

87 inlets = get_all_ports(unit_model, "inlet") 

88 

89 portList = {} 

90 id = 1 

91 

92 for inlet in inlets: 

93 inlet_stream = ctx.get_simulation_object(inlet.stream_id) 

94 

95 portList[f"inlet_{id}"] = { 

96 "id": inlet.pk, 

97 "properties": serialise_stream(ctx, inlet_stream, is_inlet=True) 

98 } 

99 id += 1 

100 

101 portList["outlet"] = PortAdapter("outlet").serialise(ctx, unit_model) 

102 

103 return portList 

104 

105class SplitterPortListAdapter(PortListAdapter): 

106 """ 

107 Handles the serialisation and reloading of the ports of a splitter. 

108 """ 

109 def __init__(self): 

110 pass 

111 

112 def serialise(self, ctx, unit_model: SimulationObject): 

113 portList = {} 

114 portList["inlet"] = PortAdapter("inlet", inlet=True).serialise(ctx, unit_model) 

115 

116 outlets = queryset_lookup.get_active_ports_for_direction(unit_model, ConType.Outlet) 

117 

118 id = 1 

119 for outlet in outlets: 

120 outlet_stream = ctx.get_simulation_object(outlet.stream_id) 

121 

122 portList[f"outlet_{id}"] = { 

123 "id": outlet.pk, 

124 "properties": serialise_stream(ctx, outlet_stream, is_inlet=False) 

125 } 

126 id += 1 

127 

128 return portList 

129 

130class BusPortListAdapter(PortListAdapter): 

131 """ 

132 Handles the serialisation and reloading of the ports of a Bus. 

133 """ 

134 def __init__(self): 

135 pass 

136 

137 def serialise(self, ctx, unit_model: SimulationObject) -> dict: 

138 inlets = get_all_ports(unit_model, "inlet") 

139 

140 portList = {} 

141 id = 1 

142 

143 for inlet in inlets: 

144 inlet_stream = ctx.get_simulation_object(inlet.stream_id) 

145 

146 portList[f"inlet_{id}"] = { 

147 "id": inlet.pk, 

148 "properties": serialise_stream(ctx, inlet_stream, is_inlet=True) 

149 } 

150 id += 1 

151 

152 outlets = queryset_lookup.get_active_ports_for_direction(unit_model, ConType.Outlet) 

153 

154 id = 1 

155 for outlet in outlets: 

156 outlet_stream = ctx.get_simulation_object(outlet.stream_id) 

157 

158 portList[f"outlet_{id}"] = { 

159 "id": outlet.pk, 

160 "properties": serialise_stream(ctx, outlet_stream, is_inlet=False) 

161 } 

162 id += 1 

163 

164 return portList