Coverage for backend/django/flowsheetInternals/unitops/models/summary_table_factory.py: 95%

1from core.auxiliary.enums.unitsLibrary import units_library 

2from ahuora_compounds import CompoundDB 

3from core.auxiliary.enums import SimulationObjectClass 

4from flowsheetInternals.unitops.config.config_base import stream_classes, unitop_classes 

5from idaes_factory.unit_conversion.unit_conversion import convert_value 

6from core.auxiliary.models.PropertyValue import PropertyValue 

7from django.db.models import QuerySet 

8from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

9from pydantic import BaseModel 

10from django.db import models 

11from idaes_factory.queryset_lookup import get_property 

12 

13class MeasureType(models.TextChoices): 

14 FLOW = "Flow" # absolute value 

15 FRACTION = "Fraction" # relative value 

16 

17class CompoundMode(models.TextChoices): 

18 MASS = "Mass" 

19 MOLAR = "Molar" 

20class TableData(BaseModel): 

21 units: dict[str, list[str]] 

22 data: list[dict[str, str | float | int]] 

23 

24PropertyKey = str 

25Unit = str 

26UnitMap = dict[PropertyKey, Unit] 

27ObjectType = str 

28 

29 

30def serialize_objects_for_table(objects_by_type: list[list[SimulationObject]], unit_map: UnitMap) -> dict[ObjectType, TableData]: 

31 serialized = {} 

32 

33 for queryset in objects_by_type: 

34 object_type = queryset[0].objectType 

35 

36 objs_to_add = [] 

37 units = {} 

38 for obj in queryset: 

39 properties = {"name": obj.componentName} 

40 for prop_info in obj.properties.ContainedProperties.all(): 

41 if prop_info.key == "mole_frac_comp": 

42 continue 

43 for prop_value in prop_info.values.all(): 

44 indexedItem = prop_value.indexedItems.first() 

45 key = f"{prop_value.property.displayName} {indexedItem.displayName}" if indexedItem else prop_value.property.displayName 

46 

47 target_unit = unit_map.get(key) 

48 value = prop_value.value 

49 if value and target_unit and (target_unit["value"] != prop_info.unit): 

50 value = convert_value(float(value), prop_info.unit, target_unit["value"]) 

51 

52 units[key] = units_library.get(prop_info.unitType, [{"value": "dimensionless", "label": "—"}]) 

53 properties[key] = value 

54 

55 objs_to_add.append(properties) 

56 

57 serialized[object_type] = { 

58 "units": units, 

59 "data": objs_to_add 

60 } 

61 

62 return serialized 

63 

64def get_stream_summary_table_data(queryset: QuerySet[SimulationObject], unit_map: UnitMap): 

65 all_streams = queryset.filter(objectType__in=stream_classes).order_by("pk") 

66 

67 streams_by_type = {} 

68 for stream in all_streams: 

69 streams_by_type.setdefault(stream.objectType, []).append(stream) 

70 

71 streams_by_type = list(streams_by_type.values()) 

72 

73 return serialize_objects_for_table(streams_by_type, unit_map) 

74 

75def get_unitops_summary_table_data(queryset: QuerySet[SimulationObject], unit_map: UnitMap): 

76 all_unitops = queryset.filter(objectType__in=unitop_classes).order_by("pk") 

77 

78 unitops_by_type = {} 

79 for unitop in all_unitops: 

80 unitops_by_type.setdefault(unitop.objectType, []).append(unitop) 

81 

82 unitops_by_type = list(unitops_by_type.values()) 

83 

84 return serialize_objects_for_table(unitops_by_type, unit_map) 

85 

86def get_composition_summary_table_data(queryset: QuerySet[SimulationObject], target_compound_mode: CompoundMode, measure_type: MeasureType): 

87 results = {"data": [], "columns": None} 

88 columns = set() # This is for collecting all possible columns to display in the table 

89 

90 streams = queryset.filter( 

91 objectType=SimulationObjectClass.Stream 

92 ).order_by("pk") 

93 

94 for stream in streams: 

95 object = {"name": stream.componentName} 

96 compounds = get_compounds(stream) 

97 current_compound_mode = stream.properties.compoundMode 

98 

99 if measure_type == MeasureType.FRACTION: 

100 if current_compound_mode != target_compound_mode: 100 ↛ 115line 100 didn't jump to line 115 because the condition on line 100 was always true

101 if target_compound_mode == CompoundMode.MASS: # Molar fraction to Mass fraction 

102 mass_fractions = get_compound_mass_fractions(compounds) 

103 

104 for i, (key, prop) in enumerate(compounds): 

105 object[key] = mass_fractions[i] 

106 columns.add(key) 

107 

108 else: # Mass fraction to Molar fraction 

109 for i, (key, prop) in enumerate(compounds): 

110 columns.add(key) 

111 compound_value = prop.value # already mass fraction 

112 object[key] = compound_value 

113 

114 else: # things are already calculated in the right mode -> use displayValue directly 

115 for i, (key, prop) in enumerate(compounds): 

116 columns.add(key) 

117 compound_value = prop.displayValue 

118 object[key] = compound_value 

119 

120 else: 

121 if target_compound_mode == CompoundMode.MASS: # Get Mass Flow 

122 total_mass_flow = get_property(stream.properties, "flow_mass").get_value() 

123 mass_fractions = get_compound_mass_fractions(compounds) 

124 

125 for i, (key, prop) in enumerate(compounds): 

126 if total_mass_flow: 

127 object[key] = total_mass_flow * mass_fractions[i] 

128 else: 

129 object[key] = None 

130 columns.add(key) 

131 

132 else: # Get Molar Flow 

133 total_molar_flow = get_property(stream.properties, "flow_mol").get_value() 

134 

135 for i, (key, prop) in enumerate(compounds): 

136 if total_molar_flow: 

137 object[key] = float(total_molar_flow) * float(prop.value) 

138 else: 

139 object[key] = None 

140 columns.add(key) 

141 

142 results["data"].append(object) 

143 

144 results["columns"] = ["name"] + list(columns) 

145 

146 return results 

147 

148def get_compounds(stream) -> set[tuple[str, PropertyValue]]: 

149 mole_frac_comp = get_property(stream.properties, "mole_frac_comp") 

150 property_values = mole_frac_comp.values.all() 

151 result = [ 

152 (prop.get_index("compound").key, prop) 

153 for prop in property_values 

154 ] 

155 return result 

156 

157def get_compound_mass_fractions(compounds: set[tuple[str, PropertyValue]]) -> list[float]: 

158 results = [] 

159 for key, prop in compounds: 

160 try: 

161 value = float(prop.value) 

162 except (ValueError, TypeError): 

163 return [None for _ in compounds] 

164 

165 molar_fraction = value 

166 

167 compound = CompoundDB.get_compound(key) 

168 molecular_weight = compound.MolecularWeight.value 

169 

170 results.append(molar_fraction * molecular_weight) 

171 

172 total_mass = sum(results) 

173 if total_mass > 0: 173 ↛ 176line 173 didn't jump to line 176 because the condition on line 173 was always true

174 results = [mass / total_mass for mass in results] 

175 else: 

176 results = [None for _ in results] 

177 return results