Coverage for backend/django/idaes_factory/adapters/property_package_adapter.py: 87%

87 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2026-06-23 21:51 +0000

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING 

4 

5from ahuora_builder_types.flowsheet_schema import PropertyPackageType 

6from ahuora_builder_types.unit_model_schema import PropertyPackageArgSchema, PropertyPackageId 

7 

8from core.auxiliary.enums.unitOpData import SimulationObjectClass 

9from .property_info_adapter import ValueAdapter 

10from core.auxiliary.enums.unitOpGraphics import ConType 

11from common.config_types import PropertyPackageKey 

12from ahuora_builder_types.custom_package_schema import ( 

13 CustomCompoundPropertiesSchema, 

14 CustomPropertyPackagePropertiesSchema, 

15) 

16from ..queryset_lookup import ( 

17 get_all_ports, 

18 get_property, 

19 get_index, 

20) 

21 

22if TYPE_CHECKING: 

23 from core.auxiliary.models.CustomPropertyPackage import CustomPropertyPackage 

24 from ..idaes_factory_context import IdaesFactoryContext 

25 

26 

27class PropertyPackageAdapter(ValueAdapter): 

28 def __init__(self, package_name: PropertyPackageKey = ""): 

29 self.package_name = package_name 

30 

31 def serialise(self, ctx: "IdaesFactoryContext", model): 

32 return PropertyPackageArgSchema( 

33 type="property_package", 

34 id= PropertyPackageId(self.get_property_package_id(ctx, model)) 

35 ) 

36 

37 def get_property_package_id(self, ctx: "IdaesFactoryContext", unit_model): 

38 # TODO: This needs to figure out which compounds is in the stream is using, and then use that property 

39 # package (add it to the flowsheet if it's not already there) 

40 # and then return the id of the property package 

41 

42 # group is a container, not a unit op, therefore doesn't have property packages 

43 if unit_model.objectType == SimulationObjectClass.Group: 

44 return -1 

45 

46 port_keys = unit_model.schema.propertyPackagePorts[self.package_name] 

47 ports = get_all_ports(unit_model, keys=port_keys) 

48 

49 # set of compounds in the stream 

50 compounds = set() 

51 for port in ports: 

52 stream = ctx.get_simulation_object(port.stream_id) 

53 if stream is not None: 53 ↛ 51line 53 didn't jump to line 51 because the condition on line 53 was always true

54 property_set = stream.properties 

55 mole_frac_comp = get_property(property_set, "mole_frac_comp") 

56 for value in mole_frac_comp.values.all(): 

57 compounds.add(get_index(value, "compound").key) 

58 package_type = stream.propertyPackageType 

59 

60 if package_type is None: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true

61 if ctx.require_variables_fixed: 

62 raise ValueError( 

63 f"Property package {self.package_name} is not defined for unit model {unit_model.componentName}" 

64 ) 

65 else: 

66 return ( 

67 -1 

68 ) # if we are just trying a simple test scenario, we can return -1 to indicate that the property package is not defined 

69 

70 if len(compounds) == 0: 

71 if ctx.require_variables_fixed: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 raise ValueError( 

73 f"No compounds found in stream {port_keys} for unit model {unit_model.componentName}" 

74 ) 

75 else: 

76 return -1 

77 

78 # check if property package is already in the flowsheet 

79 for existing_package in ctx.property_packages: 

80 if ( 

81 existing_package.type == package_type 

82 and set(existing_package.compounds) == compounds 

83 ): 

84 return existing_package.id 

85 

86 # create a new property package 

87 id = len(ctx.property_packages) 

88 if package_type == "custom": 

89 package_def = serialise_custom_package(stream.customPackage, id) 

90 else: 

91 package_def = PropertyPackageType( 

92 id=id, 

93 type=package_type, 

94 compounds=list(compounds), 

95 phases=["Liq", "Vap"], 

96 ) 

97 

98 ctx.property_packages.append(package_def) 

99 

100 return id 

101 

102 

103def serialise_custom_package(package: "CustomPropertyPackage", id): 

104 if package is None: 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true

105 raise ValueError("Custom property package is not set") 

106 (compounds, custom_compound_properties) = serialise_custom_compound_properties( 

107 package 

108 ) 

109 custom_package_properties = serialise_custom_package_properties(package) 

110 custom_kappa_values = serialise_custom_kappa_values(package) 

111 return PropertyPackageType( 

112 id=id, 

113 type="custom", 

114 compounds=compounds, 

115 phases=["Liq", "Vap"], 

116 custom_compound_properties=custom_compound_properties, 

117 custom_package_properties=custom_package_properties, 

118 custom_kappa_values=custom_kappa_values, 

119 ) 

120 

121 

122def serialise_custom_package_properties(package: CustomPropertyPackage): 

123 properties = package.properties.all() 

124 property_dict = {} 

125 for prop in properties: 

126 if prop.value is None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 raise ValueError( 

128 f"Custom property package property {prop.package_property_key} is not set" 

129 ) 

130 property_dict[prop.package_property_key] = float(prop.value) 

131 

132 return CustomPropertyPackagePropertiesSchema.model_validate(property_dict) 

133 

134 

135def serialise_custom_kappa_values( 

136 package: CustomPropertyPackage, 

137) -> dict[str, dict[str, float]]: 

138 kappas = package.kappas.all() 

139 compounds = package.compounds.all() 

140 compound_names = {compound.id: compound.name for compound in compounds} 

141 kappa_dict = {} 

142 for kappa in kappas: 

143 compound1_name = compound_names[kappa.compound1_id] 

144 compound2_name = compound_names[kappa.compound2_id] 

145 if compound1_name not in kappa_dict: 145 ↛ 147line 145 didn't jump to line 147 because the condition on line 145 was always true

146 kappa_dict[compound1_name] = {} 

147 kappa_dict[compound1_name][compound2_name] = kappa.value 

148 return kappa_dict 

149 

150 

151def serialise_custom_compound_properties( 

152 package: CustomPropertyPackage, 

153) -> tuple[list[str], list[CustomCompoundPropertiesSchema]]: 

154 compounds = package.compounds.all() 

155 compound_names = [] 

156 compound_properties = [] 

157 for compound in compounds: 

158 properties = compound.properties.all() 

159 property_dict = {} 

160 for prop in properties: 

161 if prop.value is None: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true

162 raise ValueError( 

163 f"Custom compound property {prop.compound_property_key} for compound {compound.name} is not set" 

164 ) 

165 property_dict[prop.compound_property_key] = float(prop.value) 

166 compound_properties.append( 

167 CustomCompoundPropertiesSchema.model_validate(property_dict) 

168 ) 

169 compound_names.append(compound.name) 

170 return (compound_names, compound_properties)