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

84 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2026-05-13 02:47 +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 

7from .property_info_adapter import ValueAdapter 

8from core.auxiliary.enums.unitOpGraphics import ConType 

9from common.config_types import PropertyPackageKey 

10from ahuora_builder_types.custom_package_schema import ( 

11 CustomCompoundPropertiesSchema, 

12 CustomPropertyPackagePropertiesSchema, 

13) 

14from ..queryset_lookup import ( 

15 get_all_ports, 

16 get_property, 

17 get_index, 

18) 

19 

20if TYPE_CHECKING: 

21 from core.auxiliary.models.CustomPropertyPackage import CustomPropertyPackage 

22 from ..idaes_factory_context import IdaesFactoryContext 

23 

24 

25class PropertyPackageAdapter(ValueAdapter): 

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

27 self.package_name = package_name 

28 

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

30 return PropertyPackageArgSchema( 

31 type="property_package", 

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

33 ) 

34 

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

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

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

38 # and then return the id of the property package 

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

40 ports = get_all_ports(unit_model, keys=port_keys) 

41 

42 # set of compounds in the stream 

43 compounds = set() 

44 for port in ports: 

45 stream = ctx.get_simulation_object(port.stream_id) 

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

47 property_set = stream.properties 

48 mole_frac_comp = get_property(property_set, "mole_frac_comp") 

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

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

51 package_type = stream.propertyPackageType 

52 

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

54 if ctx.require_variables_fixed: 

55 raise ValueError( 

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

57 ) 

58 else: 

59 return ( 

60 -1 

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

62 

63 if len(compounds) == 0: 

64 if ctx.require_variables_fixed: 

65 raise ValueError( 

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

67 ) 

68 else: 

69 return -1 

70 

71 # check if property package is already in the flowsheet 

72 for existing_package in ctx.property_packages: 

73 if ( 

74 existing_package.type == package_type 

75 and set(existing_package.compounds) == compounds 

76 ): 

77 return existing_package.id 

78 

79 # create a new property package 

80 id = len(ctx.property_packages) 

81 if package_type == "custom": 

82 package_def = serialise_custom_package(stream.customPackage, id) 

83 else: 

84 package_def = PropertyPackageType( 

85 id=id, 

86 type=package_type, 

87 compounds=list(compounds), 

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

89 ) 

90 

91 ctx.property_packages.append(package_def) 

92 

93 return id 

94 

95 

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

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

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

99 (compounds, custom_compound_properties) = serialise_custom_compound_properties( 

100 package 

101 ) 

102 custom_package_properties = serialise_custom_package_properties(package) 

103 custom_kappa_values = serialise_custom_kappa_values(package) 

104 return PropertyPackageType( 

105 id=id, 

106 type="custom", 

107 compounds=compounds, 

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

109 custom_compound_properties=custom_compound_properties, 

110 custom_package_properties=custom_package_properties, 

111 custom_kappa_values=custom_kappa_values, 

112 ) 

113 

114 

115def serialise_custom_package_properties(package: CustomPropertyPackage): 

116 properties = package.properties.all() 

117 property_dict = {} 

118 for prop in properties: 

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

120 raise ValueError( 

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

122 ) 

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

124 

125 return CustomPropertyPackagePropertiesSchema.model_validate(property_dict) 

126 

127 

128def serialise_custom_kappa_values( 

129 package: CustomPropertyPackage, 

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

131 kappas = package.kappas.all() 

132 compounds = package.compounds.all() 

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

134 kappa_dict = {} 

135 for kappa in kappas: 

136 compound1_name = compound_names[kappa.compound1_id] 

137 compound2_name = compound_names[kappa.compound2_id] 

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

139 kappa_dict[compound1_name] = {} 

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

141 return kappa_dict 

142 

143 

144def serialise_custom_compound_properties( 

145 package: CustomPropertyPackage, 

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

147 compounds = package.compounds.all() 

148 compound_names = [] 

149 compound_properties = [] 

150 for compound in compounds: 

151 properties = compound.properties.all() 

152 property_dict = {} 

153 for prop in properties: 

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

155 raise ValueError( 

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

157 ) 

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

159 compound_properties.append( 

160 CustomCompoundPropertiesSchema.model_validate(property_dict) 

161 ) 

162 compound_names.append(compound.name) 

163 return (compound_names, compound_properties)