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

1from common.models.idaes import PropertiesSchema 

2from flowsheetInternals.unitops.models import SimulationObject 

3from .property_info_adapter import PropertyKeyAdapter, PropertyDictAdapter 

4from ..queryset_lookup import get_property 

5 

6""" 

7A list of properties that can be serialised and reloaded for a material stream. 

8Used in reloading the properties of a material stream, since we will recieve properties 

9outside of the ones specified in the state definition. 

10""" 

11 

12# Todo: deprecate the reliance on propertyDictAdapter here so we can remove it. 

13stream_properties = PropertyDictAdapter({ 

14 "flow_mol": PropertyKeyAdapter("flow_mol"), 

15 "pressure": PropertyKeyAdapter("pressure"), 

16 "enth_mass": PropertyKeyAdapter("enth_mass"), 

17 "enth_mol": PropertyKeyAdapter("enth_mol"), 

18 "mole_frac_comp": PropertyKeyAdapter("mole_frac_comp"), 

19 "mole_frac_phase_comp": PropertyKeyAdapter("mole_frac_phase_comp"), 

20 "temperature": PropertyKeyAdapter("temperature"), 

21 "vapor_frac": PropertyKeyAdapter("vapor_frac"), 

22 "flow_mass": PropertyKeyAdapter("flow_mass"), 

23 "flow_vol": PropertyKeyAdapter("flow_vol"), 

24 "entr_mass": PropertyKeyAdapter("entr_mass"), 

25 "entr_mol": PropertyKeyAdapter("entr_mol"), 

26 "total_energy_flow": PropertyKeyAdapter("total_energy_flow"), 

27}) 

28energy_stream_properties = PropertyDictAdapter({ 

29 "power": PropertyKeyAdapter("power"), 

30}) 

31humid_air_properties = PropertyDictAdapter({ 

32 "flow_mol": PropertyKeyAdapter("flow_mol"), 

33 "pressure": PropertyKeyAdapter("pressure"), 

34 "enth_mass": PropertyKeyAdapter("enth_mass"), 

35 "enth_mol": PropertyKeyAdapter("enth_mol"), 

36 "mole_frac_comp": PropertyKeyAdapter("mole_frac_comp"), 

37 # "mole_frac_phase_comp": PropertyKeyAdapter("mole_frac_phase_comp"), 

38 "temperature": PropertyKeyAdapter("temperature"), 

39 "temperature_wet_bulb": PropertyKeyAdapter("temperature_wet_bulb"), 

40 # "vapor_frac": PropertyKeyAdapter("vaporFrac"), 

41 "flow_mass": PropertyKeyAdapter("flow_mass"), 

42 "flow_vol": PropertyKeyAdapter("flow_vol"), 

43 "entr_mass": PropertyKeyAdapter("entr_mass"), 

44 "entr_mol": PropertyKeyAdapter("entr_mol"), 

45 "total_energy_flow": PropertyKeyAdapter("total_energy_flow"), 

46 "relative_humidity": PropertyKeyAdapter("relative_humidity") 

47}) 

48ac_stream_properties = PropertyDictAdapter({ 

49 "active_power": PropertyKeyAdapter("active_power"), 

50 "reactive_power": PropertyKeyAdapter("reactive_power"), 

51 "voltage": PropertyKeyAdapter("voltage"), 

52}) 

53transformer_stream_properties = PropertyDictAdapter({ 

54 "current": PropertyKeyAdapter("current"), 

55 "voltage": PropertyKeyAdapter("voltage"), 

56}) 

57 

58 

59 

60stream_types = { 

61 "stream": stream_properties, 

62 "energy_stream": energy_stream_properties, 

63 "humid_air_stream": humid_air_properties, 

64 "ac_stream": ac_stream_properties, 

65 "transformer_stream": transformer_stream_properties, 

66} 

67 

68 

69def get_stream_adapter(stream: SimulationObject) -> PropertyDictAdapter: 

70 """ 

71 Get the property adapter for a material stream. 

72 """ 

73 stream_type = stream.objectType 

74 adapter = stream_types.get(stream_type) 

75 if adapter: 75 ↛ 78line 75 didn't jump to line 78 because the condition on line 75 was always true

76 return adapter 

77 else: 

78 raise ValueError(f"Stream type {stream_type} not supported (stream_properties.py)") 

79 

80 

81def should_serialise_stream(stream: SimulationObject, is_inlet: bool) -> bool: 

82 """ 

83 Check if a stream should be serialised. 

84 """ 

85 if stream.connectedPorts.count() == 1: 

86 # ie. inlet or outlet port 

87 return True 

88 if stream.has_recycle_connection: 

89 # serialise both the inlet (for guesses) and outlet (for set points) 

90 return True 

91 else: 

92 # serialise the outlet for intermediate ports 

93 return not is_inlet 

94 

95 

96def serialise_stream(ctx, stream: SimulationObject, is_inlet: bool) -> PropertiesSchema: 

97 """ 

98 Serialise the properties of a material stream. 

99 """ 

100 if not should_serialise_stream(stream, is_inlet): 

101 return {} 

102 

103 dict_adapter: PropertyDictAdapter = get_stream_adapter(stream) 

104 result = {} 

105 property_set = stream.properties 

106 

107 tear_inlet = stream.has_recycle_connection and is_inlet 

108 tear_outlet = stream.has_recycle_connection and not is_inlet 

109 

110 for key, adapter in dict_adapter.schema.items(): 

111 # get value from adapter 

112 prop = get_property(property_set, adapter._prop_key) 

113 for values in prop.values.all(): 

114 if ( 

115 tear_outlet 

116 and ( 

117 not values.is_control_set_point() 

118 or not values.is_externally_controlled() 

119 ) 

120 ): 

121 # for the other side of a tear, skip if not a set point 

122 continue 

123 # Note: This will not work if you are controlling composition. This logic 

124 # needs to be updated for controlling an indexed variable. 

125 # It expects that there is only one value in prop.values.all() 

126 value = adapter.property_adapter.serialise(ctx, prop, is_tear=tear_inlet) 

127 if value is not None: 127 ↛ 113line 127 didn't jump to line 113 because the condition on line 127 was always true

128 result[key] = value 

129 return result