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

1from core.auxiliary.enums.unitsLibrary import units_library 

2from compounds import CompoundDB 

3from core.auxiliary.enums.unitOpData 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): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

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) 

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) 

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(objectType=SimulationObjectClass.Stream) 

91 

92 for stream in streams: 

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

94 compounds = get_compounds(stream) 

95 current_compound_mode = stream.properties.compoundMode 

96 

97 if measure_type == MeasureType.FRACTION: 

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

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

100 mass_fractions = get_compound_mass_fractions(compounds) 

101 

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

103 object[key] = mass_fractions[i] 

104 columns.add(key) 

105 

106 else: # Mass fraction to Molar fraction 

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

108 columns.add(key) 

109 compound_value = prop.value # already mass fraction 

110 object[key] = compound_value 

111 

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

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

114 columns.add(key) 

115 compound_value = prop.displayValue 

116 object[key] = compound_value 

117 

118 else: 

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

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

121 mass_fractions = get_compound_mass_fractions(compounds) 

122 

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

124 if total_mass_flow: 

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

126 else: 

127 object[key] = None 

128 columns.add(key) 

129 

130 else: # Get Molar Flow 

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

132 

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

134 if total_molar_flow: 

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

136 else: 

137 object[key] = None 

138 columns.add(key) 

139 

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

141 

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

143 

144 return results 

145 

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

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

148 property_values = mole_frac_comp.values.all() 

149 result = [ 

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

151 for prop in property_values 

152 ] 

153 return result 

154 

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

156 results = [] 

157 for key, prop in compounds: 

158 try: 

159 value = float(prop.value) 

160 except (ValueError, TypeError): 

161 return [None for _ in compounds] 

162 

163 molar_fraction = value 

164 

165 compound = CompoundDB.get_compound(key) 

166 molecular_weight = compound["MolecularWeight"].value 

167 

168 results.append(molar_fraction * molecular_weight) 

169 

170 total_mass = sum(results) 

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

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

173 else: 

174 results = [None for _ in results] 

175 return results